From d7305fbbfcc7ec7df235b484f71b43332cc8b68f Mon Sep 17 00:00:00 2001 From: Claude Agent Date: Thu, 16 Apr 2026 18:54:03 +0000 Subject: [PATCH] fix(run): drop first_accepted gate from catchup synth-arm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/atm/main.py | 2 +- tests/test_handle_tick.py | 38 ++++++++++++ tests/test_main.py | 120 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 1 deletion(-) diff --git a/src/atm/main.py b/src/atm/main.py index e93ae97..3eb72ec 100644 --- a/src/atm/main.py +++ b/src/atm/main.py @@ -371,7 +371,7 @@ def _handle_tick( # Drive FSM through a synthetic arm so the real PRIME transition fires a # normal prime alert below. Audit entry is tagged catchup:true. catchup = False - if first_accepted and color in ("dark_green", "dark_red") and fsm.state == State.IDLE: + if color in ("dark_green", "dark_red") and fsm.state == State.IDLE: assert fsm.state == State.IDLE, "synth-arm invariant: FSM must be IDLE" arm_color = "turquoise" if color == "dark_green" else "yellow" direction = "BUY" if color == "dark_green" else "SELL" diff --git a/tests/test_handle_tick.py b/tests/test_handle_tick.py index e7cc1ff..20c40e2 100644 --- a/tests/test_handle_tick.py +++ b/tests/test_handle_tick.py @@ -218,3 +218,41 @@ def test_handle_tick_first_turquoise_no_catchup_label(): # Decision 3A: cannot distinguish fresh arm from catchup-at-arm-phase assert "catchup" not in notif.alerts[0].title.lower() assert "catchup" not in notif.alerts[0].body.lower() + + +# --------------------------------------------------------------------------- +# Regression: user bug 2026-04-16. Catchup must fire whenever IDLE + dark_* is +# observed, regardless of first_accepted. Prior gray noise tick consumed +# first_accepted and the subsequent dark_red was gated out of catchup. +# --------------------------------------------------------------------------- + +def test_handle_tick_catchup_dark_red_when_not_first_accepted(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "dark_red", 1.0, notif, audit, first_accepted=False) + + assert tr is not None + assert tr.next == State.PRIMED_SELL + assert len(notif.alerts) == 2 + assert notif.alerts[0].kind == "arm" + assert notif.alerts[0].direction == "SELL" + assert "catchup" in (notif.alerts[0].title + notif.alerts[0].body).lower() + assert notif.alerts[1].kind == "prime" + assert notif.alerts[1].direction == "SELL" + + +def test_handle_tick_catchup_dark_green_when_not_first_accepted(): + """Mirror — BUY side.""" + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "dark_green", 1.0, notif, audit, first_accepted=False) + + assert tr is not None + assert tr.next == State.PRIMED_BUY + assert len(notif.alerts) == 2 + assert notif.alerts[0].direction == "BUY" + assert notif.alerts[1].direction == "BUY" diff --git a/tests/test_main.py b/tests/test_main.py index 99492b5..96d855d 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -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"