diff --git a/src/cli_agent_orchestrator/constants.py b/src/cli_agent_orchestrator/constants.py index 85ddcd3e0..1c001aa64 100644 --- a/src/cli_agent_orchestrator/constants.py +++ b/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 diff --git a/src/cli_agent_orchestrator/services/fifo_reader.py b/src/cli_agent_orchestrator/services/fifo_reader.py index 7c3cfed2f..8c367b707 100644 --- a/src/cli_agent_orchestrator/services/fifo_reader.py +++ b/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 diff --git a/test/services/test_fifo_reader.py b/test/services/test_fifo_reader.py index 5cf675566..fad7d6575 100644 --- a/test/services/test_fifo_reader.py +++ b/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."""