fix(fifo): bound the probe-throws-because-gone liveness storm

src/cli_agent_orchestrator/constants.py

@@ -209,6 +209,20 @@ def _env_positive_float(name: str, default: float) -> float:
 # After this many attempts, give up loudly and drop the terminal from the
 # watchdog, exactly like the rearm()-exception path already does.
 PIPE_LIVENESS_MAX_COLD_START_ATTEMPTS = _env_int("CAO_PIPE_LIVENESS_MAX_COLD_START_ATTEMPTS", 5)
+# Cap on consecutive liveness-PROBE failures per terminal (harness-control#845). The
+# probe (a tmux ``capture-pane``/``get_history``) raises — e.g. libtmux
+# ``ObjectDoesNotExist`` — when the session, window, or the whole tmux server is gone.
+# That exception path reaches NEITHER the rearm-failure NOR the cold-start counter above
+# (both sit downstream of a probe that RETURNED), so before this bound a terminal whose
+# session/server had died was re-probed every PIPE_LIVENESS_CHECK_INTERVAL_S forever, each
+# tick emitting a full-traceback ERROR — an unbounded, self-amplifying log/CPU storm across
+# every ghost terminal exactly when the box is already unhealthy (live incident: ~578k
+# error lines, a strong contributor to a near-simultaneous mass session teardown). After
+# this many consecutive probe failures, give up loudly ONCE and drop the terminal from the
+# watchdog, exactly like the rearm-exception and cold-start paths already do. The counter
+# resets on any successful probe, so a brief transient (a session momentarily unavailable
+# but not gone) never accumulates to a false drop.
+PIPE_LIVENESS_MAX_PROBE_FAILURES = _env_int("CAO_PIPE_LIVENESS_MAX_PROBE_FAILURES", 5)
 
 # pyte-rendered status detection. When enabled, the StatusMonitor feeds each
 # terminal's output through a pyte terminal emulator and runs detection against

src/cli_agent_orchestrator/services/fifo_reader.py

@@ -15,6 +15,7 @@
     PIPE_LIVENESS_CHECK_INTERVAL_S,
     PIPE_LIVENESS_COLD_START_GRACE_S,
     PIPE_LIVENESS_MAX_COLD_START_ATTEMPTS,
+    PIPE_LIVENESS_MAX_PROBE_FAILURES,
     PIPE_LIVENESS_MAX_REARM_FAILURES,
     PIPE_LIVENESS_STALL_CHECKS,
 )
@@ -136,6 +137,12 @@ def __init__(self):
         # any successful re-arm; once it hits PIPE_LIVENESS_MAX_REARM_FAILURES
         # the terminal is dropped from the watchdog instead of retrying forever.
         self._rearm_failures: Dict[str, int] = {}
+        # Consecutive liveness-*probe* failures per terminal (probe() raised — the
+        # session/window/whole tmux server is gone). Reset on any successful probe;
+        # once it hits PIPE_LIVENESS_MAX_PROBE_FAILURES the terminal is dropped from
+        # the watchdog instead of re-probing (and logging a traceback) every tick
+        # forever (harness-control#845).
+        self._probe_failures: Dict[str, int] = {}
         self._watchdog_stop = threading.Event()
         self._watchdog_thread: Optional[threading.Thread] = None
 
@@ -212,6 +219,7 @@ def stop_reader(self, terminal_id: str) -> None:
             self._registered_at.pop(terminal_id, None)
             self._ever_delivered.pop(terminal_id, None)
             self._cold_start_attempts.pop(terminal_id, None)
+            self._probe_failures.pop(terminal_id, None)
 
         # Deliberately NOT stopping the watchdog thread here even when this was
         # the last enrolled terminal: doing it under a "now idle" check raced
@@ -463,7 +471,52 @@ def _check_pipe_liveness(self, terminal_id: str) -> None:
         # probe() is a slow tmux `capture-pane` call — deliberately made
         # without holding self._lock so it never blocks stop_reader() (or
         # other terminals' housekeeping) for its duration.
-        content = probe()
+        try:
+            content = probe()
+        except Exception:
+            # The session/window/whole tmux server is gone, so probe() raises
+            # (e.g. libtmux ObjectDoesNotExist) and will keep raising every tick.
+            # Nothing downstream (the re-arm / cold-start counters) is ever reached
+            # on this path, so without a dedicated bound a dead terminal produces an
+            # unbounded per-tick traceback storm across every ghost terminal
+            # (harness-control#845). Bound it exactly like the re-arm and cold-start
+            # give-up paths: count consecutive probe failures and, after
+            # PIPE_LIVENESS_MAX_PROBE_FAILURES, drop the terminal from the watchdog,
+            # emitting ONE summary WARNING instead of a traceback per tick.
+            with self._lock:
+                # stop_reader() may have unenrolled this terminal while probe()
+                # was in flight; don't resurrect state for a terminal that's gone.
+                if terminal_id not in self._pane_probe:
+                    return
+                failures = self._probe_failures.get(terminal_id, 0) + 1
+                if failures >= PIPE_LIVENESS_MAX_PROBE_FAILURES:
+                    self._pane_probe.pop(terminal_id, None)
+                    self._rearm.pop(terminal_id, None)
+                    self._liveness.pop(terminal_id, None)
+                    self._rearm_failures.pop(terminal_id, None)
+                    self._registered_at.pop(terminal_id, None)
+                    self._ever_delivered.pop(terminal_id, None)
+                    self._cold_start_attempts.pop(terminal_id, None)
+                    self._probe_failures.pop(terminal_id, None)
+                    give_up = True
+                else:
+                    self._probe_failures[terminal_id] = failures
+                    give_up = False
+            if give_up:
+                logger.warning(
+                    "pipe-pane liveness probe for terminal %s failed %d consecutive "
+                    "times (session/window/server gone); dropping it from the watchdog",
+                    terminal_id,
+                    PIPE_LIVENESS_MAX_PROBE_FAILURES,
+                )
+            else:
+                logger.debug(
+                    "pipe-pane liveness probe for terminal %s failed (%d/%d); will retry",
+                    terminal_id,
+                    failures,
+                    PIPE_LIVENESS_MAX_PROBE_FAILURES,
+                )
+            return
         now = time.monotonic()
 
         do_rearm = False
@@ -479,6 +532,10 @@ def _check_pipe_liveness(self, terminal_id: str) -> None:
             # and never cleaned up, leaking slowly across create/stop churn.
             if terminal_id not in self._pane_probe:
                 return
+            # probe() succeeded — the session is reachable again, so clear any
+            # accumulated probe-failure strikes (harness-control#845): a brief
+            # transient must never accumulate across recoveries into a false drop.
+            self._probe_failures.pop(terminal_id, None)
             last_data_at = self._last_data_at.get(terminal_id, 0.0)
 
             # ---- cold-start check (harness-control#93) ----
@@ -521,6 +578,7 @@ def _check_pipe_liveness(self, terminal_id: str) -> None:
                     self._registered_at.pop(terminal_id, None)
                     self._ever_delivered.pop(terminal_id, None)
                     self._cold_start_attempts.pop(terminal_id, None)
+                    self._probe_failures.pop(terminal_id, None)
                 else:
                     self._cold_start_attempts[terminal_id] = attempts
                     # Reset the grace-period clock so the NEXT evaluation is
@@ -664,6 +722,7 @@ def _rearm_stalled_pipe(
                     self._registered_at.pop(terminal_id, None)
                     self._ever_delivered.pop(terminal_id, None)
                     self._cold_start_attempts.pop(terminal_id, None)
+                    self._probe_failures.pop(terminal_id, None)
             if give_up:
                 # Not a silent retry-forever: a re-arm that keeps failing
                 # (e.g. the tmux pane is gone) previously re-struck and

test/services/test_fifo_reader.py

@@ -557,6 +557,76 @@ def failing_rearm():
         assert "term" not in manager._rearm
         assert "term" not in manager._rearm_failures
 
+    def test_probe_failure_is_bounded_and_terminal_dropped(self, tmp_path, monkeypatch):
+        """harness-control#845: when probe() itself raises (session/window/whole
+        tmux server gone, e.g. libtmux ObjectDoesNotExist) the exception must NOT
+        propagate to _watchdog_loop (which would log a full traceback per terminal
+        per tick forever — the storm). Instead it is caught and bounded: after
+        PIPE_LIVENESS_MAX_PROBE_FAILURES consecutive failures the terminal is
+        dropped from the watchdog, once, with a single summary log."""
+        monkeypatch.setattr(fr, "PIPE_LIVENESS_MAX_PROBE_FAILURES", 3)
+
+        manager = self._manager(tmp_path, monkeypatch)
+        probe_calls: list = []
+
+        def gone_probe():
+            probe_calls.append(True)
+            raise RuntimeError("No objects found: session gone")  # mimics libtmux ObjectDoesNotExist
+
+        manager._pane_probe["term"] = gone_probe
+        manager._rearm["term"] = lambda: None
+        manager._last_data_at["term"] = time.monotonic()
+
+        # Each call must return normally (NOT raise) — this is the storm fix: the
+        # exception is swallowed here so _watchdog_loop never logs a per-tick traceback.
+        for _ in range(3):
+            manager._check_pipe_liveness("term")  # must not raise
+
+        assert len(probe_calls) == 3
+        assert "term" not in manager._pane_probe, "gone terminal must be dropped after the probe-failure cap"
+        assert "term" not in manager._rearm
+        assert "term" not in manager._probe_failures
+
+        # A further watchdog pass no longer probes it at all (storm over): the
+        # terminal is unenrolled, so _watchdog_loop wouldn't even iterate it.
+        manager._check_pipe_liveness("term")
+        assert len(probe_calls) == 3, "a dropped terminal must never be probed again"
+
+    def test_probe_failure_counter_resets_on_success(self, tmp_path, monkeypatch):
+        """A brief transient probe failure (session momentarily unavailable but not
+        gone) must not accumulate toward the cap across recoveries: a successful
+        probe resets the counter, so the terminal is never falsely dropped."""
+        monkeypatch.setattr(fr, "PIPE_LIVENESS_MAX_PROBE_FAILURES", 3)
+        monkeypatch.setattr(fr, "PIPE_LIVENESS_STALL_CHECKS", 1)
+
+        manager = self._manager(tmp_path, monkeypatch)
+        state = {"fail": True, "content": "l0"}
+
+        def flaky_probe():
+            if state["fail"]:
+                raise RuntimeError("transient: No objects found")
+            return state["content"]
+
+        manager._pane_probe["term"] = flaky_probe
+        manager._rearm["term"] = lambda: None
+        manager._last_data_at["term"] = time.monotonic()
+
+        # Two failures (below the cap of 3), then a success.
+        manager._check_pipe_liveness("term")
+        manager._check_pipe_liveness("term")
+        assert manager._probe_failures.get("term") == 2
+        state["fail"] = False
+        manager._check_pipe_liveness("term")  # success -> resets the counter
+        assert "term" in manager._pane_probe, "must not be dropped after recovering"
+        assert "term" not in manager._probe_failures, "counter must reset on a successful probe"
+
+        # Two more failures must again NOT drop it (proves it didn't secretly carry 2+2).
+        state["fail"] = True
+        manager._check_pipe_liveness("term")
+        manager._check_pipe_liveness("term")
+        assert "term" in manager._pane_probe
+        assert manager._probe_failures.get("term") == 2
+
     def test_create_reader_enrolls_and_starts_watchdog(self, tmp_path, monkeypatch):
         """A tmux caller passing probe+rearm enrolls the terminal and starts the
         watchdog; stop_reader unenrolls it and clears its liveness state."""