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

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:

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",

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

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."""

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