fix: opposite_rearm/rearm alerts + /resume unified + canary-pause UX guards
Trei găuri observate în sesiunea 2026-04-21: A. _handle_tick nu avea branch pentru reason=opposite_rearm (PRIMED_* ↔ ARMED_opus) sau reason=rearm (PRIMED_* → ARMED_* aceeași direcție). La 17:45 yellow a trecut FSM-ul PRIMED_BUY→ARMED_SELL corect, dar zero alert pe Telegram. Adaugă helper _emit_arm_alert (DRY cu branch-ul arm existent) și două branch-uri noi cu kind=opposite_rearm / kind=rearm. B. Canary drift se curăța doar cu /resume force — user ușor confundă /set_interval cu „relansare" și rămâne în drift-pause (cazul 18:09 azi). /resume acum curăță user_paused + canary.resume() într-o singură comandă. /resume force rămâne alias acceptat (muscle memory legacy). C. Heartbeat-ul afișa „activ ARMED_SELL" deși detecția era oprită de 3 ore (state FSM înghețat). Extract _build_heartbeat_alert care arată „⚠️ pauzat (drift)" + „[drift-pause]" când canary.is_paused. Guard-uri pentru comenzi când canary e paused: - /set_interval: refuzat cu warn „Trimite /resume" - /ss: screenshot trimis + body-ul include „⚠️ DETECȚIE OPRITĂ" 11 teste noi (1 critical regression pentru bug-ul A observat azi), plus actualizarea test-ului /resume existent care aserta vechiul comportament. Total: 235 passed + 8 scenarii regresie. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -567,3 +567,169 @@ def test_state_machine_is_locked_and_record_fire_public_api():
|
||||
assert fsm.is_locked("BUY", 50.0) is True # within 100s
|
||||
assert fsm.is_locked("BUY", 150.0) is False # past lockout
|
||||
assert fsm.is_locked("SELL", 50.0) is False # other direction unaffected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# opposite_rearm — bug observat 2026-04-21 17:45
|
||||
# PRIMED_BUY + yellow → ARMED_SELL; reason=opposite_rearm; must emit alert
|
||||
# with screenshot attached. CRITICAL regression.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_opposite_rearm_primed_buy_to_armed_sell_emits_alert():
|
||||
"""REGRESSION (2026-04-21): PRIMED_BUY + yellow → ARMED_SELL silently."""
|
||||
from pathlib import Path
|
||||
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
def snap(kind, label):
|
||||
return Path(f"/tmp/{label}.png")
|
||||
|
||||
# Drive to PRIMED_BUY
|
||||
_handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
_handle_tick(fsm, "dark_green", 2.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
assert fsm.state == State.PRIMED_BUY
|
||||
notif.alerts.clear()
|
||||
|
||||
# Yellow → ARMED_SELL via opposite_rearm
|
||||
tr = _handle_tick(fsm, "yellow", 3.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_SELL
|
||||
assert tr.reason == "opposite_rearm"
|
||||
assert len(notif.alerts) == 1
|
||||
a = notif.alerts[0]
|
||||
assert a.kind == "opposite_rearm"
|
||||
assert a.direction == "SELL"
|
||||
assert "yellow" in a.title
|
||||
assert "opus" in a.title.lower()
|
||||
assert a.image_path == Path("/tmp/opposite_rearm_sell.png")
|
||||
|
||||
|
||||
def test_opposite_rearm_primed_sell_to_armed_buy_emits_alert():
|
||||
"""Mirror: PRIMED_SELL + turquoise → ARMED_BUY."""
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
_handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False)
|
||||
_handle_tick(fsm, "dark_red", 2.0, notif, audit, first_accepted=False)
|
||||
assert fsm.state == State.PRIMED_SELL
|
||||
notif.alerts.clear()
|
||||
|
||||
tr = _handle_tick(fsm, "turquoise", 3.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_BUY
|
||||
assert tr.reason == "opposite_rearm"
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "opposite_rearm"
|
||||
assert notif.alerts[0].direction == "BUY"
|
||||
|
||||
|
||||
def test_opposite_rearm_armed_buy_to_armed_sell_emits_alert():
|
||||
"""Flip direct ARMED_BUY → ARMED_SELL (fără prime între)."""
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
_handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False)
|
||||
assert fsm.state == State.ARMED_BUY
|
||||
notif.alerts.clear()
|
||||
|
||||
tr = _handle_tick(fsm, "yellow", 2.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_SELL
|
||||
assert tr.reason == "opposite_rearm"
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "opposite_rearm"
|
||||
assert notif.alerts[0].direction == "SELL"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# rearm — PRIMED_* + arm-color aceeași direcție → ARMED_* (reset ciclu)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_rearm_primed_buy_to_armed_buy_emits_alert():
|
||||
from pathlib import Path
|
||||
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
def snap(kind, label):
|
||||
return Path(f"/tmp/{label}.png")
|
||||
|
||||
_handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
_handle_tick(fsm, "dark_green", 2.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
assert fsm.state == State.PRIMED_BUY
|
||||
notif.alerts.clear()
|
||||
|
||||
tr = _handle_tick(fsm, "turquoise", 3.0, notif, audit, first_accepted=False, snapshot=snap)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_BUY
|
||||
assert tr.reason == "rearm"
|
||||
assert len(notif.alerts) == 1
|
||||
a = notif.alerts[0]
|
||||
assert a.kind == "rearm"
|
||||
assert a.direction == "BUY"
|
||||
assert "reluat" in a.title.lower()
|
||||
assert a.image_path == Path("/tmp/rearm_buy.png")
|
||||
|
||||
|
||||
def test_rearm_primed_sell_to_armed_sell_emits_alert():
|
||||
fsm = StateMachine(lockout_s=60)
|
||||
notif = FakeNotifier()
|
||||
audit = FakeAudit()
|
||||
|
||||
_handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False)
|
||||
_handle_tick(fsm, "dark_red", 2.0, notif, audit, first_accepted=False)
|
||||
assert fsm.state == State.PRIMED_SELL
|
||||
notif.alerts.clear()
|
||||
|
||||
tr = _handle_tick(fsm, "yellow", 3.0, notif, audit, first_accepted=False)
|
||||
|
||||
assert tr is not None
|
||||
assert tr.next == State.ARMED_SELL
|
||||
assert tr.reason == "rearm"
|
||||
assert len(notif.alerts) == 1
|
||||
assert notif.alerts[0].kind == "rearm"
|
||||
assert notif.alerts[0].direction == "SELL"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _emit_arm_alert helper — unit test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_emit_arm_alert_helper_builds_expected_alert():
|
||||
from pathlib import Path
|
||||
from atm.main import _emit_arm_alert
|
||||
|
||||
notif = FakeNotifier()
|
||||
calls: list[tuple[str, str]] = []
|
||||
|
||||
def snap(kind, label):
|
||||
calls.append((kind, label))
|
||||
return Path(f"/tmp/{label}.png")
|
||||
|
||||
_emit_arm_alert(
|
||||
notif,
|
||||
kind="opposite_rearm",
|
||||
direction="SELL",
|
||||
now=1700000000.0,
|
||||
title="SELL re-armat (yellow) — ciclu opus",
|
||||
snap=snap,
|
||||
snap_kind="opposite_rearm",
|
||||
snap_label="opposite_rearm_sell",
|
||||
)
|
||||
|
||||
assert len(notif.alerts) == 1
|
||||
a = notif.alerts[0]
|
||||
assert a.kind == "opposite_rearm"
|
||||
assert a.direction == "SELL"
|
||||
assert a.title == "SELL re-armat (yellow) — ciclu opus"
|
||||
assert a.image_path == Path("/tmp/opposite_rearm_sell.png")
|
||||
assert calls == [("opposite_rearm", "opposite_rearm_sell")]
|
||||
|
||||
Reference in New Issue
Block a user