diff --git a/scripts/launchd/throughput-watchdog.py b/scripts/launchd/throughput-watchdog.py index a8acb97a..8e4445d4 100755 --- a/scripts/launchd/throughput-watchdog.py +++ b/scripts/launchd/throughput-watchdog.py @@ -21,6 +21,11 @@ DEFAULT_WATCH_LABEL = "com.brainlayer.watch" DEFAULT_NOTIFY_ENDPOINT = "http://localhost:3847/notify" +# Alert-framing (orc law, stalker postmortem): after a wedge episode is paged once, +# the operator is re-notified for the NEXT episode only — i.e. after this many +# consecutive healthy (progressing) ticks clear the latch. Prevents a chronic +# wedge sawtooth from paging on every kickstart. +_EPISODE_RESET_TICKS = 3 DEFAULT_COMMAND_TIMEOUT_SECONDS = 15 KICKSTART_TIMEOUT_SECONDS = 45 @@ -491,11 +496,21 @@ def run_once( elif config.dry_run: result.action = "would_kickstart" else: - try: - alert_fn(config, result) - except Exception as exc: - result.alert_error = str(exc) - print(f"throughput-watchdog alert failed: {exc}", file=sys.stderr) + # Alert-framing: page on the FIRST wedge of an episode only — never + # per-kick of a chronic sawtooth. The latch clears after a sustained + # healthy run (episode reset below). Recovery still runs every tick. + # The latch is set ONLY when the page actually sent — a transient + # notify failure on the first wedge must NOT silence the whole + # episode (page-once must never become page-zero); the next tick + # retries the alert instead. + episode_alerted = bool(state.get("episode_alerted")) + if not episode_alerted: + try: + alert_fn(config, result) + episode_alerted = True + except Exception as exc: + result.alert_error = str(exc) + print(f"throughput-watchdog alert failed: {exc}", file=sys.stderr) attempt_state = dict(state) attempt_state.update( { @@ -512,6 +527,7 @@ def run_once( "last_action": "recovery_attempt", "last_restart_epoch": checked_at, "restart_attempt_count": int(state.get("restart_attempt_count", 0)) + 1, + "episode_alerted": episode_alerted, } ) _atomic_write_json(config.state_path, attempt_state) @@ -549,6 +565,25 @@ def run_once( next_state["last_alert_error"] = result.alert_error else: next_state.pop("last_alert_error", None) + + # Episode reset (alert-framing): clear the first-wedge page latch after a + # SUSTAINED healthy run, so the next genuine wedge pages again without + # re-paging every cycle of a chronic sawtooth. "Healthy" = real watcher + # progress by EITHER measure — chunk highwater OR liveness rowid — because + # a DB without watcher_liveness_events reports liveness_rowid==0 forever + # while chunks still advance (else the latch would never reset there). A + # kickstart tick is never healthy: its own state pre-advances both rowids to + # the current values, so neither comparison is strictly-greater on that tick. + prev_chunk = int(state.get("watcher_highwater_rowid", 0) or 0) + prev_liveness = int(state.get("watcher_liveness_highwater_rowid", 0) or 0) + progressed = current_highwater > prev_chunk or current_liveness_highwater > prev_liveness + if result.stalled_ticks == 0 and progressed: + healthy_ticks = int(state.get("healthy_ticks", 0)) + 1 + next_state["healthy_ticks"] = healthy_ticks + if healthy_ticks >= _EPISODE_RESET_TICKS: + next_state["episode_alerted"] = False + else: + next_state["healthy_ticks"] = 0 _atomic_write_json(config.state_path, next_state) return result diff --git a/tests/test_throughput_watchdog.py b/tests/test_throughput_watchdog.py index 401f76b0..d3c9905c 100644 --- a/tests/test_throughput_watchdog.py +++ b/tests/test_throughput_watchdog.py @@ -606,3 +606,93 @@ def test_launchagent_runs_every_minute_and_invokes_installed_script() -> None: assert plist["EnvironmentVariables"]["BRAINLAYER_ENV_FILE"] == "__BRAINLAYER_ENV_FILE__" assert plist["EnvironmentVariables"]["BRAINLAYER_LAUNCHD_SERVICE"] == "watch" assert plist["SoftResourceLimits"]["NumberOfFiles"] >= 4096 + + +def test_alert_framing_pages_once_per_episode_not_per_kickstart(tmp_path: Path) -> None: + """Alert-framing: a chronic wedge sawtooth pages on the FIRST wedge of an + episode only, re-paging for a NEW episode after a sustained healthy run.""" + module = _load_module() + config = _config(module, tmp_path, stall_threshold=3, cooldown_seconds=0) + alerts: list[int] = [] + + def command_runner(args: list[str]): + stdout = "state = running\npid = 4321\n" if args[:2] == ["launchctl", "print"] else "" + return SimpleNamespace(returncode=0, stdout=stdout, stderr="") + + clock = {"t": 1_000} + + def tick(rowid: int, *, pending: bool) -> object: + clock["t"] += 60 + evidence = module.SourceEvidence(2, 120, 3, 999.0) if pending else module.SourceEvidence(0, 0, 0, 999.0) + res = module.run_once( + config, + now_epoch=clock["t"], + progress_reader=lambda _p, r=rowid: _progress(module, r, r), + source_probe=lambda _c, _n, e=evidence: e, + command_runner=command_runner, + alert_fn=lambda _c, _r: alerts.append(clock["t"]), + ) + return res + + tick(40, pending=True) # baseline + tick(40, pending=True) # stall 1 + tick(40, pending=True) # stall 2 + k1 = tick(40, pending=True) # stall 3 -> kickstart + FIRST alert + assert k1.action.startswith("kickstart:") + assert len(alerts) == 1 # episode 1 paged once + + tick(50, pending=True) # brief recovery (1 healthy tick, < reset) + tick(50, pending=True) # stall 1 + tick(50, pending=True) # stall 2 + k2 = tick(50, pending=True) # stall 3 -> kickstart, SUPPRESSED (same episode) + assert k2.action.startswith("kickstart:") + assert len(alerts) == 1 # sawtooth did NOT re-page + + tick(60, pending=True) # sustained recovery: healthy 1 + tick(70, pending=True) # healthy 2 + tick(80, pending=True) # healthy 3 -> episode latch clears + tick(80, pending=True) # stall 1 + tick(80, pending=True) # stall 2 + k3 = tick(80, pending=True) # stall 3 -> kickstart + NEW-episode alert + assert k3.action.startswith("kickstart:") + assert len(alerts) == 2 # new episode paged again + + +def test_failed_alert_does_not_latch_episode_and_retries(tmp_path: Path) -> None: + """A transient notify failure on the first wedge must NOT silence the + episode — page-once must never become page-zero; the next wedge retries.""" + module = _load_module() + config = _config(module, tmp_path, stall_threshold=3, cooldown_seconds=0) + calls = {"n": 0} + + def command_runner(args): + stdout = "state = running\npid = 4321\n" if args[:2] == ["launchctl", "print"] else "" + return SimpleNamespace(returncode=0, stdout=stdout, stderr="") + + def flaky_alert(_config, _result): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("notify endpoint down") + + clock = {"t": 1_000} + + def tick(): + clock["t"] += 60 + return module.run_once( + config, + now_epoch=clock["t"], + progress_reader=lambda _p: _progress(module, 40, 40), + source_probe=lambda _c, _n: module.SourceEvidence(2, 120, 3, 999.0), + command_runner=command_runner, + alert_fn=flaky_alert, + ) + + tick() # baseline (establishes highwater; not a stall) + tick() # stall 1 + tick() # stall 2 + tick() # stall 3 -> kickstart, alert RAISES (call 1), NOT latched + assert calls["n"] == 1 + tick() # stall 1 + tick() # stall 2 + tick() # stall 3 -> kickstart, alert RETRIES in same episode (call 2) + assert calls["n"] == 2 # retried, not silenced by the earlier failure