romfast/atm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(canary): auto-rebase pe layout change (2↔1 ferestre)

Browse Source 
Când canary drift coincide cu schimbare strip-count pe același frame
(ex: TS comută 2→1 chart-uri și mută menu-ul peste care e ancorat ROI),
sistemul rescrie automat baseline_phash în TOML, commit layout change
și trimite o singură alertă combinată — fără pauză, fără /rebase manual.

Drift fără strip-count change rămâne pauză ca azi (drift real).
Gate pe două semnale independente previne fals-pozitive.

Canary.check() despărțit în măsurare pură + commit_pause/rebase
explicit; tick-loop-ul orchestrează decizia.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
Marius Mutu
2026-05-05 18:30:57 +03:00
parent 839caacc95
commit bb241bf050
4 changed files with 576 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

376
tests/test_main.py
View File

@@ -2089,3 +2089,379 @@ def test_commit_layout_change_alert_is_silent(monkeypatch):
layout_alerts = [a for a in notifier_alerts if "Layout TS schimbat" in a.title]
assert len(layout_alerts) == 1
assert layout_alerts[0].silent is True
# ---------------------------------------------------------------------------
# Auto-rebase on layout change (two-signal gate: drift + strip-count change)
# ---------------------------------------------------------------------------
class _FakeCanary:
"""Stand-in for atm.canary.Canary that lets tests drive drift state.
Mirrors the new split API (check pure, commit_pause + rebase explicit).
"""
def __init__(self, drift_distance: int = 0):
self.drift_distance = drift_distance
self.is_paused = False
self.commit_pause_calls: list[int] = []
self.rebase_calls: list[str] = []
self.resume_calls = 0
def check(self, _frame):
return types.SimpleNamespace(
distance=self.drift_distance,
drifted=self.drift_distance > 0,
paused=self.is_paused,
)
def commit_pause(self, distance: int) -> None:
if self.is_paused:
return
self.is_paused = True
self.commit_pause_calls.append(distance)
def rebase(self, new_phash: str) -> None:
self.rebase_calls.append(new_phash)
def resume(self) -> None:
self.is_paused = False
self.resume_calls += 1
def _build_multi_tick_ctx(tmp_path, cfg, canary, initial_strips):
"""Construct a RunContext suitable for driving _run_multi_tick.
Returns (ctx, notifier_alerts, audit_events, detector_stubs).
Charts are seeded with stub Detectors that record dot_roi updates.
"""
import atm.main as _main
from atm.detector import DetectionResult
notifier_alerts: list = []
audit_events: list = []
class _Det:
def __init__(self):
self.last_roi = None
def step(self, ts, frame=None):
return DetectionResult(
ts=ts, window_found=True, dot_found=False,
rgb=None, match=None, accepted=False, color=None,
)
def update_dot_roi(self, roi):
self.last_roi = roi
class _FSM:
state = types.SimpleNamespace(value="IDLE")
_last_fire: dict = {}
class _N:
def send(self, a): notifier_alerts.append(a)
class _A:
def log(self, e): audit_events.append(e)
class _S:
is_running = False
def start(self, s): pass
def stop(self): pass
state_obj = _main._LoopState()
detectors = [_Det() for _ in initial_strips]
charts = [
_main.ChartState(
chart_id=("left" if i == 0 else "right") if len(initial_strips) == 2 else "only",
sub_roi=strip, detector=detectors[i], fsm=_FSM(),
)
for i, strip in enumerate(initial_strips)
]
ctx = _main.RunContext(
cfg=cfg, capture=lambda: np.zeros((200, 1800, 3), dtype=np.uint8),
canary=canary, detector=MagicMock(), fsm=_FSM(),
notifier=_N(), audit=_A(), detection_log=_A(),
scheduler=_S(), samples_dir=tmp_path, fires_dir=tmp_path,
cmd_queue=MagicMock(), state=state_obj,
levels_extractor_factory=lambda *a, **kw: None,
lifecycle=None,
)
ctx.charts = charts
return ctx, notifier_alerts, audit_events, detectors
def _make_minimal_cfg_for_multi_tick():
"""Minimal cfg that satisfies _run_multi_tick + _commit_layout_change paths."""
from atm.config import ROI
cfg = MagicMock()
cfg.lockout_s = 60
cfg.colors = {}
cfg.debounce_depth = 3
cfg.confidence_min = 0.5
cfg.dot_roi = ROI(x=0, y=0, w=600, h=35)
cfg.window_title = None
cfg.attach_screenshots = types.SimpleNamespace(
arm=False, prime=False, trigger=False, late_start=False,
catchup=False, opposite_rearm=False, rearm=False, phase_skip=False,
)
cfg.config_version = "test-cfg"
cfg.canary = types.SimpleNamespace(
roi=ROI(x=0, y=0, w=50, h=50),
baseline_phash="0" * 64,
drift_threshold=8,
)
return cfg
@pytest.mark.asyncio
async def test_multi_tick_drift_with_strip_count_change_auto_rebases(monkeypatch, tmp_path):
"""Drift + strip count 2→1 → silent auto-rebase, no pause, single combined alert."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=156)
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
new_strips = [ROI(x=0, y=0, w=900, h=35)] # 2 → 1
ctx, notifier_alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: new_strips)
rewrite_calls: list = []
def _stub_rewrite(path, new_phash):
rewrite_calls.append((path, new_phash))
return "OLD_PHASH"
monkeypatch.setattr(_main, "_rewrite_baseline_phash", _stub_rewrite)
await _main._run_multi_tick(ctx)
# Layout committed: charts shrank to 1.
assert len(ctx.charts) == 1
assert ctx.charts[0].sub_roi == new_strips[0]
# Canary NOT paused (auto-rebase path).
assert canary.is_paused is False
assert canary.commit_pause_calls == []
# rebase() called with the new phash that was also passed to _rewrite_baseline_phash.
assert len(canary.rebase_calls) == 1
assert len(rewrite_calls) == 1
new_phash = canary.rebase_calls[0]
assert rewrite_calls[0][1] == new_phash
# Audit: layout_change_with_rebase + standard layout_change (from _commit_layout_change).
rebase_events = [e for e in audit_events if e.get("event") == "layout_change_with_rebase"]
assert len(rebase_events) == 1
assert rebase_events[0]["old_n"] == 2
assert rebase_events[0]["new_n"] == 1
assert rebase_events[0]["distance"] == 156
assert rebase_events[0]["old_phash"] == "OLD_PHASH"
assert rebase_events[0]["new_phash"] == new_phash
# No "paused" audit event.
assert not any(e.get("event") == "paused" for e in audit_events)
# Exactly one combined alert. Generic layout_change alert was suppressed.
combined = [a for a in notifier_alerts if "auto-rebased" in a.title]
assert len(combined) == 1
assert "2→1" in combined[0].title
plain_layout = [a for a in notifier_alerts
if "Layout TS schimbat" in a.title and "auto-rebased" not in a.title]
assert plain_layout == []
@pytest.mark.asyncio
async def test_multi_tick_drift_with_same_strip_count_pauses(monkeypatch, tmp_path):
"""Drift without strip count change → real drift → pause as before."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=156)
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
same_count = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
ctx, notifier_alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: same_count)
monkeypatch.setattr(_main, "_rewrite_baseline_phash",
lambda *a, **kw: pytest.fail("should not be called"))
results = await _main._run_multi_tick(ctx)
assert results == []
assert canary.is_paused is True
assert canary.commit_pause_calls == [156]
assert len(ctx.charts) == 2 # unchanged
paused_events = [e for e in audit_events if e.get("event") == "paused"]
assert len(paused_events) == 1
assert paused_events[0]["drift"] == 156
assert canary.rebase_calls == []
@pytest.mark.asyncio
async def test_multi_tick_drift_with_zero_strips_pauses(monkeypatch, tmp_path):
"""Strip detection returns [] (chart blackout) → pause, do NOT auto-rebase."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=156)
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
ctx, _alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: [])
monkeypatch.setattr(_main, "_rewrite_baseline_phash",
lambda *a, **kw: pytest.fail("should not be called"))
results = await _main._run_multi_tick(ctx)
assert results == []
assert canary.is_paused is True
assert canary.commit_pause_calls == [156]
assert canary.rebase_calls == []
assert any(e.get("event") == "paused" for e in audit_events)
@pytest.mark.asyncio
async def test_multi_tick_no_drift_strip_count_change_unchanged_path(monkeypatch, tmp_path):
"""Strip count change without drift → existing _commit_layout_change path,
canary baseline untouched, silent layout-change alert."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=0) # no drift
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
new_strips = [ROI(x=0, y=0, w=900, h=35)]
ctx, notifier_alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: new_strips)
monkeypatch.setattr(_main, "_rewrite_baseline_phash",
lambda *a, **kw: pytest.fail("should not be called"))
await _main._run_multi_tick(ctx)
assert len(ctx.charts) == 1
assert canary.rebase_calls == []
# Standard layout_change alert (not suppressed).
layout_alerts = [a for a in notifier_alerts if "Layout TS schimbat" in a.title]
assert len(layout_alerts) == 1
assert "auto-rebased" not in layout_alerts[0].title
assert any(e.get("event") == "layout_change" for e in audit_events)
assert not any(e.get("event") == "layout_change_with_rebase" for e in audit_events)
@pytest.mark.asyncio
async def test_multi_tick_auto_rebase_toml_failure_falls_back_to_pause(monkeypatch, tmp_path):
"""When _rewrite_baseline_phash raises, fall back to commit_pause + audit."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=156)
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
new_strips = [ROI(x=0, y=0, w=900, h=35)]
ctx, notifier_alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: new_strips)
def _boom(_path, _new):
raise ValueError("expected exactly 1 baseline_phash line, found 0")
monkeypatch.setattr(_main, "_rewrite_baseline_phash", _boom)
results = await _main._run_multi_tick(ctx)
assert results == []
assert canary.is_paused is True
assert canary.commit_pause_calls == [156]
# Charts NOT mutated.
assert len(ctx.charts) == 2
fail_events = [e for e in audit_events if e.get("event") == "auto_rebase_failed"]
assert len(fail_events) == 1
paused_events = [e for e in audit_events
if e.get("event") == "paused" and e.get("reason") == "auto_rebase_failed"]
assert len(paused_events) == 1
@pytest.mark.asyncio
async def test_multi_tick_symmetric_1_to_2(monkeypatch, tmp_path):
"""1→2 transition is symmetric: same auto-rebase + layout commit path."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=120)
initial = [ROI(x=0, y=0, w=900, h=35)]
new_strips = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
ctx, notifier_alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
monkeypatch.setattr(_main, "_detect_strips_for_ctx", lambda c, f: new_strips)
monkeypatch.setattr(_main, "_rewrite_baseline_phash",
lambda path, new_phash: "OLD")
await _main._run_multi_tick(ctx)
assert len(ctx.charts) == 2
assert canary.is_paused is False
assert len(canary.rebase_calls) == 1
rebase_events = [e for e in audit_events if e.get("event") == "layout_change_with_rebase"]
assert len(rebase_events) == 1
assert rebase_events[0]["old_n"] == 1
assert rebase_events[0]["new_n"] == 2
combined = [a for a in notifier_alerts if "auto-rebased" in a.title]
assert len(combined) == 1
assert "1→2" in combined[0].title
@pytest.mark.asyncio
async def test_multi_tick_already_paused_short_circuits(monkeypatch, tmp_path):
"""is_paused=True at tick start → return [], no auto-rebase attempted."""
import atm.main as _main
from atm.config import ROI
monkeypatch.chdir(tmp_path)
cfg = _make_minimal_cfg_for_multi_tick()
canary = _FakeCanary(drift_distance=156)
canary.is_paused = True # already paused from a prior tick
initial = [ROI(x=0, y=0, w=400, h=35), ROI(x=500, y=0, w=400, h=35)]
new_strips = [ROI(x=0, y=0, w=900, h=35)]
ctx, _alerts, audit_events, _ = _build_multi_tick_ctx(
tmp_path, cfg, canary, initial,
)
# Strip detection should NOT run on the already-paused short-circuit path.
detect_calls: list = []
def _detect(c, f):
detect_calls.append((c, f))
return new_strips
monkeypatch.setattr(_main, "_detect_strips_for_ctx", _detect)
monkeypatch.setattr(_main, "_rewrite_baseline_phash",
lambda *a, **kw: pytest.fail("should not be called"))
results = await _main._run_multi_tick(ctx)
assert results == []
assert detect_calls == [] # short-circuited before strip detection
assert canary.rebase_calls == []
assert canary.commit_pause_calls == [] # already paused; no new commit
assert len(ctx.charts) == 2
assert any(e.get("event") == "paused" for e in audit_events)