fix(run): drop first_accepted gate from catchup synth-arm
Catchup branch gated on first_accepted, but an earlier accepted gray tick consumes the flag before a dark_* arrives, so the real prime-phase color falls through to noise classification and no alert fires. Gate on IDLE + dark_* alone — self-sufficient and correct. Regression: 2 unit tests for _handle_tick + 1 integration test feeding run_live a scripted gray→gray→dark_red→dark_red→light_red sequence. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -135,3 +135,123 @@ def test_run_live_dry(monkeypatch):
|
||||
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["duration_s"] == pytest.approx(0.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regression integration test — user bug 2026-04-16.
|
||||
# Session starts with an accepted gray tick followed by dark_red. Catchup
|
||||
# synth-arm must fire on dark_red (the previous gray consumed first_accepted),
|
||||
# then light_red triggers SELL. Proves run_live wiring dispatches alerts for
|
||||
# the user's exact scenario.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
|
||||
import numpy as np
|
||||
import atm.main as _main
|
||||
from atm.detector import DetectionResult
|
||||
|
||||
captured = []
|
||||
|
||||
class FakeFanout:
|
||||
def __init__(self, *a, **kw):
|
||||
pass
|
||||
def send(self, alert):
|
||||
captured.append(alert)
|
||||
def stop(self):
|
||||
pass
|
||||
|
||||
class FakeCanaryResult:
|
||||
distance = 0
|
||||
drifted = False
|
||||
paused = False
|
||||
|
||||
class FakeCanary:
|
||||
def __init__(self, *a, **kw):
|
||||
self.is_paused = False
|
||||
def check(self, frame):
|
||||
return FakeCanaryResult()
|
||||
def resume(self):
|
||||
pass
|
||||
|
||||
class _StopLoop(Exception):
|
||||
pass
|
||||
|
||||
class ScriptedDetector:
|
||||
_script = [
|
||||
("gray", True),
|
||||
("gray", True),
|
||||
("dark_red", True),
|
||||
("dark_red", True),
|
||||
("light_red", True),
|
||||
]
|
||||
def __init__(self, *a, **kw):
|
||||
self._i = 0
|
||||
def step(self, ts):
|
||||
if self._i >= len(self._script):
|
||||
raise _StopLoop
|
||||
color, accepted = self._script[self._i]
|
||||
self._i += 1
|
||||
return DetectionResult(
|
||||
ts=ts,
|
||||
window_found=True,
|
||||
dot_found=True,
|
||||
rgb=(1, 1, 1),
|
||||
match=None,
|
||||
accepted=accepted,
|
||||
color=color,
|
||||
)
|
||||
|
||||
def fake_build_capture(cfg, capture_stub=False):
|
||||
return lambda: np.zeros((50, 50, 3), dtype=np.uint8)
|
||||
|
||||
cfg = MagicMock()
|
||||
cfg.lockout_s = 60
|
||||
cfg.heartbeat_min = 999
|
||||
cfg.loop_interval_s = 0
|
||||
cfg.config_version = "test"
|
||||
cfg.dead_letter_path = str(tmp_path / "dl.jsonl")
|
||||
cfg.canary.drift_threshold = 10
|
||||
cfg.dot_roi.x = 0
|
||||
cfg.dot_roi.y = 0
|
||||
cfg.dot_roi.w = 10
|
||||
cfg.dot_roi.h = 10
|
||||
cfg.chart_window_region = None
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
class _Stub:
|
||||
def __init__(self, *a, **kw):
|
||||
pass
|
||||
def log(self, *a, **kw):
|
||||
pass
|
||||
def close(self, *a, **kw):
|
||||
pass
|
||||
def step(self, *a, **kw):
|
||||
return types.SimpleNamespace(status="pending", levels=None)
|
||||
|
||||
monkeypatch.setattr("atm.detector.Detector", ScriptedDetector)
|
||||
monkeypatch.setattr("atm.canary.Canary", FakeCanary)
|
||||
monkeypatch.setattr("atm.notifier.fanout.FanoutNotifier", FakeFanout)
|
||||
monkeypatch.setattr("atm.notifier.discord.DiscordNotifier", _Stub)
|
||||
monkeypatch.setattr("atm.notifier.telegram.TelegramNotifier", _Stub)
|
||||
monkeypatch.setattr("atm.audit.AuditLog", _Stub)
|
||||
monkeypatch.setattr("atm.levels.LevelsExtractor", _Stub)
|
||||
monkeypatch.setattr("atm.main._build_capture", fake_build_capture)
|
||||
monkeypatch.setattr("atm.main.time.sleep", lambda s: None)
|
||||
|
||||
with pytest.raises(_StopLoop):
|
||||
_main.run_live(cfg, duration_s=None)
|
||||
|
||||
arm = [a for a in captured if a.kind == "arm"]
|
||||
prime = [a for a in captured if a.kind == "prime"]
|
||||
trigger = [a for a in captured if a.kind == "trigger"]
|
||||
|
||||
assert len(arm) == 1, f"expected 1 arm alert, got {len(arm)} ({[a.title for a in captured]})"
|
||||
assert arm[0].direction == "SELL"
|
||||
assert "catchup" in (arm[0].title + arm[0].body).lower()
|
||||
|
||||
assert len(prime) == 1
|
||||
assert prime[0].direction == "SELL"
|
||||
assert "catchup" in (prime[0].title + prime[0].body).lower()
|
||||
|
||||
assert len(trigger) == 1
|
||||
assert trigger[0].direction == "SELL"
|
||||
|
||||
Reference in New Issue
Block a user