fix(status): self-heal a stuck-PROCESSING terminal via a fresh capture-pane read

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,29 @@
     }
 )
 
+# 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 this to at most once per terminal per interval -- 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
+
 
 class StatusMonitor:
     """Accumulates terminal output into rolling buffers and detects status changes."""
@@ -68,6 +92,15 @@ 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]] = {}
         # --- 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
@@ -510,6 +543,7 @@ 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)
             handle = self._quiesce_handle.pop(terminal_id, None)
         self._cancel_quiesce_handle(handle)
 
@@ -530,6 +564,7 @@ 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)
             handle = self._quiesce_handle.pop(terminal_id, None)
         self._cancel_quiesce_handle(handle)
 
@@ -587,8 +622,75 @@ 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.
+            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
+                ):
+                    self._apply_detection(terminal_id, fresh_capture)
+                    return fresh_capture
         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.
+
+        Returns ``None`` when skipped (rate-limited, 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 here just means "try again next poll", not a regression from today's behavior.
+        """
+        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:
+            return provider.get_status(fresh_output)
+        except Exception as e:
+            logger.debug(f"_fresh_capture_pane_status [{terminal_id}]: detection failed: {e}")
+            return None
+
     def get_buffer(self, terminal_id: str) -> str:
         """Get accumulated output buffer for a terminal."""
         with self._lock:

test/services/test_status_monitor.py

@@ -85,6 +85,189 @@ 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."""
+
+    @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(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"] = ""
+
+        assert sm.get_status("t1") == TerminalStatus.IDLE
+        backend.get_history.assert_called_once_with("s1", "w1")
+        provider.get_status.assert_called_once_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_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"] = ""
+
+        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"] = ""
+
+        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"] = ""
+
+        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"] = ""
+
+        assert sm.get_status("t1") == TerminalStatus.PROCESSING
+        backend.get_history.assert_not_called()
+
+    @patch("cli_agent_orchestrator.services.status_monitor.time")
+    @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, mock_time):
+        # 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 within the rate-limit window must only shell out once.
+        mock_time.monotonic.side_effect = [0.0, 0.1]  # one time.monotonic() call per get_status()
+        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.get_status("t1")
+        sm.get_status("t1")
+
+        backend.get_history.assert_called_once()
+
+    @patch("cli_agent_orchestrator.services.status_monitor.time")
+    @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, mock_time
+    ):
+        mock_time.monotonic.side_effect = [0.0, 10.0]  # one time.monotonic() call per get_status()
+        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.get_status("t1")
+        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()
+
+
 class TestScreenDetection:
     """Rendered-screen detection should fail soft and keep monitoring alive."""