fix(watchdog): alert-framing — page once per wedge episode, not per kickstart

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
 

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