diff --git a/src/cli_agent_orchestrator/api/main.py b/src/cli_agent_orchestrator/api/main.py index 20ae04a50..82a00af8f 100644 --- a/src/cli_agent_orchestrator/api/main.py +++ b/src/cli_agent_orchestrator/api/main.py @@ -2029,7 +2029,13 @@ async def get_session(session_name: str) -> Dict: except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) try: - return session_service.get_session(session_name) + # session_service.get_session() calls status_monitor.get_status() per terminal in the + # session, which for a PROCESSING terminal can shell out to a real tmux capture-pane + # subprocess (the stale-PROCESSING fallback). A session with N processing terminals would + # otherwise fork N times inline on the event loop per request -- this endpoint is polled + # by the web UI, so run it off the loop, matching GET /terminals/{id}'s own established + # pattern just below for the identical hazard. + return await asyncio.to_thread(session_service.get_session, session_name) except ValueError as e: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) except Exception as e: diff --git a/src/cli_agent_orchestrator/services/agent_step.py b/src/cli_agent_orchestrator/services/agent_step.py index 9f00c2708..cd3eaf520 100644 --- a/src/cli_agent_orchestrator/services/agent_step.py +++ b/src/cli_agent_orchestrator/services/agent_step.py @@ -181,7 +181,9 @@ async def _wait_for_completion( if cancel_event is not None and cancel_event.is_set(): raise StepCancelledError(terminal_id=terminal_id) - current = status_monitor.get_status(terminal_id) + # Off the event loop -- get_status() can shell out to a real tmux capture-pane + # subprocess (status_monitor.py's stale-PROCESSING fallback) or a herdr CLI call. + current = await asyncio.to_thread(status_monitor.get_status, terminal_id) if current == TerminalStatus.ERROR: raise StepExecutionError( f"terminal {terminal_id} reached ERROR status", @@ -215,7 +217,9 @@ async def _wait_for_completion( if time.monotonic() >= deadline: # Defensive: a terminal that flipped to ERROR right at the deadline is # a crash, not a slow run (preserve the kind="error" vs "timeout" split). - if status_monitor.get_status(terminal_id) == TerminalStatus.ERROR: + if ( + await asyncio.to_thread(status_monitor.get_status, terminal_id) + ) == TerminalStatus.ERROR: raise StepExecutionError( f"terminal {terminal_id} reached ERROR status", kind="error", diff --git a/src/cli_agent_orchestrator/services/status_monitor.py b/src/cli_agent_orchestrator/services/status_monitor.py index 134d3184e..8191809af 100644 --- a/src/cli_agent_orchestrator/services/status_monitor.py +++ b/src/cli_agent_orchestrator/services/status_monitor.py @@ -7,6 +7,7 @@ import asyncio import logging import threading +import time from typing import Dict, List, Optional, Tuple from cli_agent_orchestrator.constants import ( @@ -45,6 +46,41 @@ } ) +# Live production incident (2026-08-02, app.workain.ai, harness-control#617/#618 investigation): +# get_status()'s own stale-PROCESSING re-check (below) re-derives from the SAME rolling +# self._buffers[terminal_id] the FIFO push pipeline feeds -- which stops changing the moment the +# underlying process goes genuinely idle and stops emitting output. If the buffer's last content +# never happened to parse as a ready state (a truncated escape sequence, or the true idle marker +# rotated out of the bounded window before it was ever sampled as ready), re-running detection on +# that SAME unchanging buffer produces the SAME PROCESSING/UNKNOWN result forever -- a session can +# be genuinely idle, with the model's real response already fully rendered in the pane, while +# get_status() reports PROCESSING indefinitely. Confirmed live TWICE in one operator session +# (`cao-support`, workspace 227): a real chat message queued behind PROCESSING sat undelivered for +# ~10 minutes until a manual tmux resize (forcing a fresh redraw) unstuck it -- no automatic +# self-healing existed for this case at all. `_handle_trust_prompt` (codex.py) already solved the +# identical staleness problem for init-time dialog detection by reading `get_backend(). +# get_history()` directly (a real `tmux capture-pane`, NOT the FIFO-fed buffer) -- tmux itself +# always holds the correct, current rendered pane state regardless of output volume, so a fresh +# capture-pane read can see what the stale FIFO buffer cannot. `STALE_PROCESSING_CAPTURE_INTERVAL_S` +# rate-limits RE-CHECKS to at most once per terminal per interval once the fallback is already +# eligible to run -- get_status() is a hot path (every wait_until_status poll, every UI status +# refresh, across the whole fleet), and unlike the existing cheap buffer re-check, a capture-pane +# read is a real subprocess call; unbounded, it would repeat the exact "fork storm freezes the +# server" class of problem `run()`'s own docstring already documents for status detection in +# general. +STALE_PROCESSING_CAPTURE_INTERVAL_S = 3.0 + +# Round-2 review fix (call-me-ram): the interval above alone rate-limits how OFTEN the fallback +# can re-run, but does not gate WHETHER it should run at all -- without this second gate, a +# terminal that is genuinely PROCESSING (actively streaming chunks) would still get a real +# capture-pane subprocess call every ~3s for the ENTIRE duration of every busy turn, not just when +# it's actually stuck. The incident this whole fallback exists for has a specific signature -- +# "the rolling buffer stopped changing" -- so require that signature directly: only attempt a +# fresh capture-pane read once the buffer has gone quiet (no new chunk appended) for at least this +# long. A terminal mid-burst never reaches this gate at all; only one that has genuinely stopped +# producing output does. +STALE_PROCESSING_BUFFER_QUIET_S = 3.0 + class StatusMonitor: """Accumulates terminal output into rolling buffers and detects status changes.""" @@ -68,6 +104,24 @@ def __init__(self): # IDLE/COMPLETED would freeze the terminal forever even when the # agent is genuinely processing new work. self._allow_processing_revert: Dict[str, bool] = {} + # Per-terminal timestamp of the last stale-PROCESSING fresh capture-pane read (see + # STALE_PROCESSING_CAPTURE_INTERVAL_S / get_status()) -- rate-limits that fallback so a + # terminal genuinely stuck reprocessing doesn't get a real tmux subprocess call on every + # single get_status() poll. Absence (never checked) is `None`, deliberately NOT `0.0` -- + # `time.monotonic()`'s reference point is arbitrary and a `0.0` sentinel would collide + # with a genuinely-elapsed `0.0` reading (as it did in this fix's own tests, mocked with + # `time.monotonic() == 0.0` on the first call), incorrectly rate-limiting the very first + # check before it ever runs. + self._last_stale_capture_check: Dict[str, Optional[float]] = {} + # Per-terminal monotonic timestamp of the last time _process_chunk actually appended a + # chunk (i.e. the buffer changed) -- see STALE_PROCESSING_BUFFER_QUIET_S. Same None-vs-0.0 + # sentinel concern as _last_stale_capture_check above. + self._buffer_changed_at: Dict[str, Optional[float]] = {} + # Per-terminal pending capture-pane candidate awaiting a second, confirming read before + # being honored (round-2 review fix, gutosantos82: a single mid-repaint capture-pane + # sample can catch Ink between clear/rewrite -- see _fresh_capture_pane_status's own + # comment). Cleared once confirmed (and applied) or once it fails to reproduce. + self._pending_stale_capture: Dict[str, TerminalStatus] = {} # --- pyte rendered-screen detection state (only used when CAO_PYTE_STATUS # is on AND the provider opts in via supports_screen_detection) --- # Per-terminal pyte Screen+Stream that composites the raw byte stream @@ -148,6 +202,8 @@ def _process_chunk(self, terminal_id: str, chunk: str) -> None: if len(buffer) > state_buffer_max: buffer = buffer[-state_buffer_max:] self._buffers[terminal_id] = buffer + # Real new output just arrived -- see STALE_PROCESSING_BUFFER_QUIET_S. + self._buffer_changed_at[terminal_id] = time.monotonic() if use_screen: self._feed_screen_locked(terminal_id, chunk) @@ -510,6 +566,9 @@ def clear_terminal(self, terminal_id: str) -> None: self._allow_processing_revert.pop(terminal_id, None) self._screens.pop(terminal_id, None) self._bursting.pop(terminal_id, None) + self._last_stale_capture_check.pop(terminal_id, None) + self._buffer_changed_at.pop(terminal_id, None) + self._pending_stale_capture.pop(terminal_id, None) handle = self._quiesce_handle.pop(terminal_id, None) self._cancel_quiesce_handle(handle) @@ -530,6 +589,9 @@ def reset_buffer(self, terminal_id: str) -> None: # detected against a fresh viewport, not the failed attempt's. self._screens.pop(terminal_id, None) self._bursting.pop(terminal_id, None) + self._last_stale_capture_check.pop(terminal_id, None) + self._buffer_changed_at.pop(terminal_id, None) + self._pending_stale_capture.pop(terminal_id, None) handle = self._quiesce_handle.pop(terminal_id, None) self._cancel_quiesce_handle(handle) @@ -587,8 +649,161 @@ def get_status(self, terminal_id: str) -> TerminalStatus: if fresh != TerminalStatus.PROCESSING and fresh != TerminalStatus.UNKNOWN: self._apply_detection(terminal_id, fresh) return fresh + + if cached == TerminalStatus.PROCESSING: + # The cheap re-check above re-derives from the SAME rolling buffer the FIFO pipeline + # feeds -- if the terminal has genuinely gone idle and stopped emitting output, that + # buffer stops changing too, so the re-check above can return PROCESSING/UNKNOWN + # forever even though the real pane already shows a ready state. See + # STALE_PROCESSING_CAPTURE_INTERVAL_S's own comment for the live incident this closes. + # + # Only even attempt this once the buffer has actually gone quiet for + # STALE_PROCESSING_BUFFER_QUIET_S -- a terminal mid-burst (new chunks still arriving) + # is not the stuck case this exists for, and gating on quiescence rather than just a + # re-check interval keeps the real subprocess call from firing every ~3s for the + # entire duration of every ordinary busy turn. + with self._lock: + changed_at = self._buffer_changed_at.get(terminal_id) + buffer_is_quiet = ( + changed_at is not None + and time.monotonic() - changed_at >= STALE_PROCESSING_BUFFER_QUIET_S + ) + if buffer_is_quiet: + fresh_capture = self._fresh_capture_pane_status(terminal_id) + if fresh_capture is not None: + logger.debug( + f"get_status [{terminal_id}]: cached=PROCESSING stale-buffer re-check " + f"still PROCESSING/UNKNOWN, fresh capture-pane={fresh_capture.value}" + ) + if ( + fresh_capture != TerminalStatus.PROCESSING + and fresh_capture != TerminalStatus.UNKNOWN + ): + # TOCTOU re-validation (round-2 review fix, gutosantos82): the capture-pane + # read above ran OUTSIDE the lock (a real subprocess call, seconds not + # microseconds), so real new output may have arrived and genuinely resumed + # PROCESSING in the meantime via a fresh _process_chunk call. Re-check + # under the lock that this terminal is STILL the same stale-PROCESSING + # terminal before applying a capture taken against what may now be a + # stale snapshot of a since-superseded state -- applying it unconditionally + # could downgrade a terminal that is, right now, genuinely processing again. + with self._lock: + current_last_status = self._last_status.get(terminal_id) + if current_last_status == TerminalStatus.PROCESSING: + self._apply_detection(terminal_id, fresh_capture) + return fresh_capture + logger.debug( + f"get_status [{terminal_id}]: fresh capture-pane result discarded -- " + "terminal status changed while the capture-pane read was in flight" + ) + # Something else (the real pipeline) already resolved this terminal to a + # fresher status while the capture-pane subprocess call was in flight -- + # return THAT rather than the `cached` value snapshotted at function entry, + # so this call doesn't hand back a status that's already one step stale. + if current_last_status is not None: + return current_last_status return cached + def _fresh_capture_pane_status(self, terminal_id: str) -> Optional[TerminalStatus]: + """Rate-limited fallback for a terminal stuck showing PROCESSING against a buffer that's + stopped changing: reads the pane directly via ``get_backend().get_history()`` (a real + ``tmux capture-pane``, not the FIFO-fed rolling buffer) and re-runs provider detection + against that. tmux always holds the correct, current rendered pane state regardless of + output volume, so this can see a genuine idle/ready state the stale buffer cannot. + + A single capture is not enough to trust on its own (round-2 review fix, gutosantos82): + this fallback's own detection design elsewhere (_schedule_screen_detection) deliberately + never samples mid-burst, because Ink-style TUIs repaint by clear-then-rewrite and a sample + caught between those two steps can miss the spinner and read the PREVIOUS turn's response + box as the current state -- a false ready that, applied here, sticky-latches (see + _apply_detection) and can leave a still-processing agent blocked from its own genuine + PROCESSING transition. Since this capture-pane read runs at an arbitrary moment (not + edge-debounced like the screen path), it requires the SAME ready status on two consecutive + calls before it's honored -- the same "confirm, don't trust a single sample" pattern + claude_code.py's own wait_until_input_ready already uses for an analogous settle-race. + + Returns ``None`` when skipped (rate-limited, not yet confirmed, no provider, or the + read/detection itself failed) -- the caller treats that identically to "still PROCESSING", + never as a signal to change status. Only ever called when cached status is already + PROCESSING, so a transient failure/non-confirmation here just means "try again next poll", + not a regression from today's behavior. + + Known limitations, disclosed rather than fixed in this round (review, gutosantos82): + - ``provider.get_status()`` is not pure for every provider -- kimi latches + ``_has_received_input`` from whatever text it's fed, and codex's detection forks a + second subprocess internally, so feeding this fallback's capture-pane text (rather than + the normal FIFO-fed buffer) can perturb pipeline-shared state on those two providers. + - Under the default-ON pyte path this always consults the RAW detector + (``provider.get_status(fresh_output)``), never ``get_status_from_screen`` -- the + pipeline's own normal detection for a pyte-opted-in provider uses the screen path + instead, and antigravity's own get_status() docstring calls its raw detector + unreliable. This fallback only ever runs when a terminal is ALREADY stuck PROCESSING + (both paths already failed to resolve it), so the raw-vs-screen choice here is a + best-effort second opinion rather than the primary detection signal either way. + """ + now = time.monotonic() + with self._lock: + last_check = self._last_stale_capture_check.get(terminal_id) + if last_check is not None and now - last_check < STALE_PROCESSING_CAPTURE_INTERVAL_S: + return None + self._last_stale_capture_check[terminal_id] = now + + try: + provider = provider_manager.get_provider(terminal_id) + except Exception as e: + # get_provider() raises (not returns None) for a terminal it doesn't recognize + # (e.g. not yet/no longer in the DB) -- matches the defensive pattern get_status()'s + # own event-inbox branch above already uses for the identical call. + logger.debug(f"_fresh_capture_pane_status [{terminal_id}]: get_provider failed: {e}") + return None + if provider is None: + return None + + try: + from cli_agent_orchestrator.backends.registry import get_backend + + fresh_output = get_backend().get_history(provider.session_name, provider.window_name) + except Exception as e: + logger.debug( + f"_fresh_capture_pane_status [{terminal_id}]: capture-pane read failed: {e}" + ) + return None + if not fresh_output: + return None + + try: + detected = provider.get_status(fresh_output) + except Exception as e: + logger.debug(f"_fresh_capture_pane_status [{terminal_id}]: detection failed: {e}") + return None + + if detected == TerminalStatus.PROCESSING or detected == TerminalStatus.UNKNOWN: + # Not a ready candidate at all -- nothing to confirm. Clear any prior pending + # candidate: a PROCESSING/UNKNOWN read in between two ready reads means the terminal + # is genuinely still busy, not settled, so the earlier candidate no longer counts. + with self._lock: + self._pending_stale_capture.pop(terminal_id, None) + return detected + + with self._lock: + pending = self._pending_stale_capture.get(terminal_id) + if pending == detected: + # Second consecutive confirming read -- honor it. + self._pending_stale_capture.pop(terminal_id, None) + confirmed = True + else: + # First read of this candidate (or it differs from a still-pending one from + # before) -- record it and wait for the next poll to confirm. + self._pending_stale_capture[terminal_id] = detected + confirmed = False + if not confirmed: + logger.debug( + f"_fresh_capture_pane_status [{terminal_id}]: candidate {detected.value} seen " + "once, awaiting a second confirming read before honoring it" + ) + return None + return detected + def get_buffer(self, terminal_id: str) -> str: """Get accumulated output buffer for a terminal.""" with self._lock: diff --git a/src/cli_agent_orchestrator/utils/terminal.py b/src/cli_agent_orchestrator/utils/terminal.py index 8dab084da..557644c9d 100644 --- a/src/cli_agent_orchestrator/utils/terminal.py +++ b/src/cli_agent_orchestrator/utils/terminal.py @@ -169,6 +169,12 @@ async def wait_until_status( it returns the pushed pipeline status, and for event-inbox backends (herdr) it derives status on demand from the provider's native status. So this poll works for both backends without special-casing here. + + get_status() can occasionally shell out to a real tmux capture-pane subprocess (the + stale-PROCESSING fallback in status_monitor.py) or to a herdr CLI call -- offload each poll + via asyncio.to_thread so that blocking I/O can't fork/exec on the shared event loop. Matches + the existing pattern at GET /terminals/{id} (api/main.py), which wraps the same call for the + identical reason. """ from cli_agent_orchestrator.services.status_monitor import status_monitor @@ -179,7 +185,7 @@ async def wait_until_status( ) start = time.time() while time.time() - start < timeout: - current = status_monitor.get_status(terminal_id) + current = await asyncio.to_thread(status_monitor.get_status, terminal_id) if current in targets: logger.info(f"wait_until_status [{terminal_id}]: reached {current.value}") return True diff --git a/test/api/test_api_endpoints.py b/test/api/test_api_endpoints.py index 9c7415f84..440edb9b7 100644 --- a/test/api/test_api_endpoints.py +++ b/test/api/test_api_endpoints.py @@ -695,6 +695,31 @@ def test_get_session_server_error(self, client): assert response.status_code == 500 assert "Failed to get session" in response.json()["detail"] + def test_get_session_dispatches_via_to_thread(self, client): + """#558 review (gutosantos82): session_service.get_session() calls + status_monitor.get_status() once per terminal in the session, which for a PROCESSING + terminal can shell out to a real tmux capture-pane subprocess -- a session with N + processing terminals would fork N times inline on the event loop per request otherwise. + Pin the asyncio.to_thread wrapping directly (the tests above mock session_service + entirely and can't see HOW it was called -- a regression back to a bare synchronous call + would stay green).""" + mock_session = {"id": "test-session", "windows": []} + with ( + patch("cli_agent_orchestrator.api.main.session_service") as mock_svc, + patch( + "cli_agent_orchestrator.api.main.asyncio.to_thread", wraps=asyncio.to_thread + ) as mock_to_thread, + ): + mock_svc.get_session.return_value = mock_session + response = client.get("/sessions/test-session") + get_session_calls = [ + c for c in mock_to_thread.call_args_list if c.args[0] == mock_svc.get_session + ] + + assert response.status_code == 200 + assert get_session_calls, "session_service.get_session was never dispatched via to_thread" + assert get_session_calls[0].args[1] == "test-session" + class TestDeleteSession: """Tests for DELETE /sessions/{session_name} endpoint.""" diff --git a/test/services/test_agent_step.py b/test/services/test_agent_step.py index a044c346f..49313a4b5 100644 --- a/test/services/test_agent_step.py +++ b/test/services/test_agent_step.py @@ -589,6 +589,38 @@ def test_error_still_raises_even_after_working(self): asyncio.run(run_agent_step("kiro_cli", "dev", "x")) assert exc_info.value.kind == "error" + def test_completion_poll_dispatches_get_status_via_to_thread(self): + """#558 review (gutosantos82): status_monitor.get_status() can shell out to a real tmux + capture-pane subprocess; calling it inline on the event loop in _wait_for_completion's + poll loop would fork tmux ON the loop every poll. Pin the asyncio.to_thread wrapping + directly (every other test here mocks status_monitor.get_status itself, which can't see + HOW it was called -- a regression back to a bare synchronous call would stay green).""" + create, send, delete, get_output, exit_cli, get_wd, wait, status = _patch_terminal_layer( + final_status=TerminalStatus.COMPLETED, + ) + with ( + create, + send, + delete, + get_output, + exit_cli, + wait, + status, + patch(f"{_MODULE}.asyncio.to_thread", wraps=asyncio.to_thread) as mock_to_thread, + ): + from cli_agent_orchestrator.services.agent_step import status_monitor + + asyncio.run(run_agent_step("kiro_cli", "dev", "x")) + + # Captured while still inside the patch context -- status_monitor.get_status is the + # active mock here (patched by `status` above); comparing against it after the + # patches unwind would compare against the restored, unpatched method instead. + get_status_calls = [ + c for c in mock_to_thread.call_args_list if c.args[0] == status_monitor.get_status + ] + assert get_status_calls, "status_monitor.get_status was never dispatched via to_thread" + assert all(c.args[1] == "abc12345" for c in get_status_calls) + class TestInterruptibleCancel: """#409b: an in-flight completion wait is interruptible via cancel_event, so a diff --git a/test/services/test_status_monitor.py b/test/services/test_status_monitor.py index 8052fbbf2..296e3af7e 100644 --- a/test/services/test_status_monitor.py +++ b/test/services/test_status_monitor.py @@ -85,6 +85,348 @@ def test_unknown_when_provider_get_status_raises(self, mock_get_backend, mock_pm assert sm.get_status("t1") == TerminalStatus.UNKNOWN +class TestStaleProcessingCapturePane: + """Live incident regression (2026-08-02, app.workain.ai, harness-control#617/#618 + investigation): a terminal that goes genuinely idle can leave get_status() reporting + PROCESSING forever, because the cheap re-check re-derives from the SAME rolling buffer that + stopped changing the moment the process stopped emitting output. These pin the fresh + capture-pane fallback that self-heals this without waiting for a manual nudge. + + Round-2 review fixes (call-me-ram, gutosantos82) added two more gates on top of the original + rate limit: + 1. The fallback only even attempts once the buffer has gone quiet for + STALE_PROCESSING_BUFFER_QUIET_S -- most tests below set _buffer_changed_at to a value far + in the past directly, rather than mocking the `time` module wholesale, since real + time.monotonic() is always far more than a few seconds past its arbitrary reference point. + 2. A single capture-pane read is not trusted on its own -- it must be confirmed by a second, + matching read (STALE_PROCESSING_CAPTURE_INTERVAL_S apart) before being honored. + """ + + @staticmethod + def _quiet_since(): + """A _buffer_changed_at value old enough to satisfy the quiet gate unconditionally.""" + return -1000.0 + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_stale_processing_self_heals_via_capture_pane_after_two_confirming_reads( + self, mock_pm, mock_get_backend + ): + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.IDLE + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "the real pane -- idle composer, fully rendered" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + # Empty buffer -- as if the process stopped emitting output entirely, exactly the shape + # that leaves the cheap re-check (which requires a truthy buffer) unable to help at all. + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + # First read: a genuine ready candidate, but a single sample is never trusted -- must NOT + # self-heal yet (see the class docstring on why a lone capture can catch an Ink repaint + # mid-clear/rewrite and read the wrong turn's response box). + assert sm.get_status("t1") == TerminalStatus.PROCESSING + assert sm._last_status["t1"] == TerminalStatus.PROCESSING + assert backend.get_history.call_count == 1 + + # Second, matching read confirms it. Reset the internal rate-limit gate directly instead + # of waiting out STALE_PROCESSING_CAPTURE_INTERVAL_S for real. + sm._last_stale_capture_check["t1"] = None + assert sm.get_status("t1") == TerminalStatus.IDLE + assert backend.get_history.call_count == 2 + provider.get_status.assert_called_with("the real pane -- idle composer, fully rendered") + # Self-healing must actually update the latched status, not just this one return value -- + # otherwise the very next poll would go right back through the same stale path. + assert sm._last_status["t1"] == TerminalStatus.IDLE + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_differing_second_read_never_confirms(self, mock_pm, mock_get_backend): + """Two DIFFERENT ready candidates in a row must never be honored -- only two + IDENTICAL consecutive reads count as confirmed.""" + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "some pane content" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + provider.get_status.return_value = TerminalStatus.IDLE + assert sm.get_status("t1") == TerminalStatus.PROCESSING # 1st read: pending=IDLE + + sm._last_stale_capture_check["t1"] = None + provider.get_status.return_value = TerminalStatus.COMPLETED + assert sm.get_status("t1") == TerminalStatus.PROCESSING # 2nd read differs -> not + # confirmed; pending is now COMPLETED, not IDLE + + assert sm._last_status["t1"] == TerminalStatus.PROCESSING + assert backend.get_history.call_count == 2 + + # A THIRD read matching the second (COMPLETED) now confirms it. + sm._last_stale_capture_check["t1"] = None + assert sm.get_status("t1") == TerminalStatus.COMPLETED + assert sm._last_status["t1"] == TerminalStatus.COMPLETED + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_capture_pane_still_processing_stays_processing_no_crash( + self, mock_pm, mock_get_backend + ): + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.PROCESSING + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "• Working (12s • esc to interrupt)" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_capture_pane_read_failure_stays_processing_no_crash(self, mock_pm, mock_get_backend): + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.side_effect = RuntimeError("tmux not reachable") + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + provider.get_status.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_no_provider_stays_processing_and_skips_capture_pane_entirely( + self, mock_pm, mock_get_backend + ): + mock_pm.get_provider.return_value = None + backend = MagicMock() + backend.supports_event_inbox.return_value = False + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + backend.get_history.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_get_provider_raising_stays_processing_no_crash(self, mock_pm, mock_get_backend): + # get_provider() raises (not returns None) for a terminal it no longer recognizes -- + # matches get_status()'s own event-inbox branch, which already defends against this. + mock_pm.get_provider.side_effect = ValueError("terminal not in db") + backend = MagicMock() + backend.supports_event_inbox.return_value = False + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + backend.get_history.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_capture_pane_fallback_is_rate_limited(self, mock_pm, mock_get_backend): + # get_status() is a hot path (every poll, across the whole fleet) -- the capture-pane + # fallback is a real tmux subprocess call and must not fire on every single poll while a + # terminal is stuck. Two calls back-to-back (real time.monotonic(), so well within the + # rate-limit window) must only shell out once. + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.PROCESSING + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "still working" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + sm.get_status("t1") + sm.get_status("t1") + + backend.get_history.assert_called_once() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_capture_pane_fallback_retried_after_rate_limit_window(self, mock_pm, mock_get_backend): + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.PROCESSING + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "still working" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + + sm.get_status("t1") + # Simulate the rate-limit window having elapsed for real, without a real sleep. + sm._last_stale_capture_check["t1"] = None + sm.get_status("t1") + + assert backend.get_history.call_count == 2 + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_buffer_recheck_resolving_skips_capture_pane_entirely(self, mock_pm, mock_get_backend): + # When the existing cheap buffer re-check already resolves the status, the (more + # expensive) capture-pane fallback must not run at all -- no regression in the common + # case where the original mechanism already worked. + provider = MagicMock() + provider.get_status.return_value = TerminalStatus.COMPLETED + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "a real, non-empty buffer that resolves cleanly" + + assert sm.get_status("t1") == TerminalStatus.COMPLETED + backend.get_history.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_recently_changed_buffer_skips_capture_pane_entirely(self, mock_pm, mock_get_backend): + """Round-2 review fix (call-me-ram): a terminal mid-burst -- new chunks still actively + arriving -- is not the stuck case this fallback exists for. Without the buffer-quiet + gate, this would shell out to a real tmux subprocess on every ~3s poll for the ENTIRE + duration of every ordinary busy turn, not just when genuinely stuck.""" + import time as time_module + + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.IDLE + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "some content" + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + # A chunk "just arrived" -- well within STALE_PROCESSING_BUFFER_QUIET_S. + sm._buffer_changed_at["t1"] = time_module.monotonic() + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + backend.get_history.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_no_buffer_changed_at_recorded_skips_capture_pane_entirely( + self, mock_pm, mock_get_backend + ): + """A terminal that has never had _process_chunk record a change (e.g. buffer set + directly, or a very old code path) must not be treated as "quiet since forever" -- + the gate requires a real recorded quiet duration, not the absence of one.""" + provider = MagicMock() + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + mock_get_backend.return_value = backend + + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + # _buffer_changed_at deliberately left unset. + + assert sm.get_status("t1") == TerminalStatus.PROCESSING + backend.get_history.assert_not_called() + + @patch("cli_agent_orchestrator.backends.registry.get_backend") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_toctou_stale_capture_discarded_if_status_changed_meanwhile( + self, mock_pm, mock_get_backend + ): + """Round-2 review fix (call-me-ram): the capture-pane read runs OUTSIDE the lock (a real + subprocess call, seconds not microseconds). If the real pipeline independently resolves + the terminal to something else WHILE that read is in flight, the stale capture result + must be discarded rather than applied over the fresher, real status.""" + sm = StatusMonitor() + sm._last_status["t1"] = TerminalStatus.PROCESSING + sm._buffers["t1"] = "" + sm._buffer_changed_at["t1"] = self._quiet_since() + # Pre-seed the pending candidate as already-confirmed-eligible: same value on both of + # the two reads _fresh_capture_pane_status will see, via a stubbed provider. + provider = MagicMock() + provider.session_name = "s1" + provider.window_name = "w1" + provider.get_status.return_value = TerminalStatus.IDLE + mock_pm.get_provider.return_value = provider + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.get_history.return_value = "idle pane" + mock_get_backend.return_value = backend + + # First read establishes the pending candidate. + assert sm.get_status("t1") == TerminalStatus.PROCESSING + sm._last_stale_capture_check["t1"] = None + + # Simulate the real pipeline resolving this terminal to ERROR WHILE the second, + # confirming capture-pane read is "in flight" -- mutate _last_status from inside the + # mocked backend call, which is where the real (slow, unlocked) subprocess call happens. + def mutate_then_return_history(*args, **kwargs): + sm._last_status["t1"] = TerminalStatus.ERROR + return "idle pane" + + backend.get_history.side_effect = mutate_then_return_history + + result = sm.get_status("t1") + + # The stale IDLE confirmation must be discarded, not applied over the real ERROR that + # arrived while the capture-pane read was in flight. + assert result == TerminalStatus.ERROR + assert sm._last_status["t1"] == TerminalStatus.ERROR + + class TestScreenDetection: """Rendered-screen detection should fail soft and keep monitoring alive.""" diff --git a/test/utils/test_terminal.py b/test/utils/test_terminal.py index 7322b4b82..77cf645a8 100644 --- a/test/utils/test_terminal.py +++ b/test/utils/test_terminal.py @@ -318,6 +318,25 @@ async def test_wait_until_status_eventually_succeeds(self, mock_monitor): assert result is True + @pytest.mark.asyncio + @patch("cli_agent_orchestrator.utils.terminal.asyncio.to_thread") + async def test_wait_until_status_dispatches_get_status_via_to_thread(self, mock_to_thread): + """#558 review (gutosantos82): status_monitor.get_status() can shell out to a real tmux + capture-pane subprocess (the stale-PROCESSING fallback); calling it inline on the event + loop here would fork tmux ON the loop every poll. Pin the asyncio.to_thread wrapping so a + regression back to a direct synchronous call stays red instead of silently passing every + other test (which mock status_monitor entirely and can't see how it was called).""" + mock_to_thread.return_value = TerminalStatus.IDLE + + result = await wait_until_status( + "test-terminal", TerminalStatus.IDLE, timeout=1.0, polling_interval=0.1 + ) + + assert result is True + from cli_agent_orchestrator.services.status_monitor import status_monitor + + mock_to_thread.assert_called_once_with(status_monitor.get_status, "test-terminal") + class TestWaitUntilTerminalStatus: """Tests for wait_until_terminal_status function."""