diff --git a/src/cli_agent_orchestrator/api/main.py b/src/cli_agent_orchestrator/api/main.py index 07aee2cee..bd3e651b0 100644 --- a/src/cli_agent_orchestrator/api/main.py +++ b/src/cli_agent_orchestrator/api/main.py @@ -1088,6 +1088,19 @@ async def lifespan(app: FastAPI): inbox_service_task = asyncio.create_task(inbox_service.run(registry)) logger.info("Event bus consumers started (StatusMonitor, LogWriter, InboxService)") + # Tmux panes intentionally survive API-process restarts. Recreate their + # FIFO readers and seed status before inbox reconciliation begins, otherwise + # pending worker callbacks can remain stranded behind UNKNOWN forever. + from cli_agent_orchestrator.services.terminal_service import ( + recover_persisted_terminal_output_streams, + ) + + await asyncio.to_thread(recover_persisted_terminal_output_streams) + # Recovery publishes initial status before the consumer tasks get their first + # scheduling turn. Reconcile once here so a callback already pending at + # restart is not left waiting for the periodic sweep. + await asyncio.to_thread(inbox_service.reconcile_orphaned_messages, registry) + # Start ApprovalBridge when AG-UI surface is enabled approval_bridge_task: Optional[asyncio.Task] = None from cli_agent_orchestrator.services.agui_enablement import agui_surface_enabled diff --git a/src/cli_agent_orchestrator/services/status_monitor.py b/src/cli_agent_orchestrator/services/status_monitor.py index 134d3184e..cdb59a9be 100644 --- a/src/cli_agent_orchestrator/services/status_monitor.py +++ b/src/cli_agent_orchestrator/services/status_monitor.py @@ -490,6 +490,27 @@ def clear_rolling_buffer(self, terminal_id: str) -> None: with self._lock: self._buffers[terminal_id] = "" + def seed_from_snapshot(self, terminal_id: str, output: str) -> TerminalStatus: + """Prime status from a rendered pane snapshot after server recovery. + + A tmux pane survives a ``cao-server`` restart, but its old pipe-pane + target points at the previous process's FIFO reader. Reattaching the + pipe only observes *future* output, so a terminal already waiting at an + input prompt would otherwise remain UNKNOWN indefinitely. Seed the + rolling buffer from ``capture-pane`` once, then run the ordinary + provider detector and publish the resulting state. This is deliberately + not an input action: it never writes to the terminal. + """ + state_buffer_max = get_server_settings()["state_buffer_max"] + with self._lock: + self._buffers[terminal_id] = output[-state_buffer_max:] + # A persisted terminal is quiescent at recovery time. Do not let a + # prior process's debounce state suppress its first real status. + self._bursting[terminal_id] = False + detected = self._detect_status(terminal_id, output[-state_buffer_max:]) + self._apply_detection(terminal_id, detected) + return detected + def _detect_status(self, terminal_id: str, buffer: str) -> TerminalStatus: """Detect status: provider-specific patterns or UNKNOWN if no provider.""" provider = provider_manager.get_provider(terminal_id) diff --git a/src/cli_agent_orchestrator/services/terminal_service.py b/src/cli_agent_orchestrator/services/terminal_service.py index e42a940e6..68f901967 100644 --- a/src/cli_agent_orchestrator/services/terminal_service.py +++ b/src/cli_agent_orchestrator/services/terminal_service.py @@ -35,6 +35,7 @@ from cli_agent_orchestrator.clients.database import delete_terminal as db_delete_terminal from cli_agent_orchestrator.clients.database import ( get_terminal_metadata, + list_terminals_by_session, list_siblings_by_group_prefix, update_last_active, update_terminal_group, @@ -169,6 +170,85 @@ class OutputMode(str, Enum): } +def recover_persisted_terminal_output_streams() -> int: + """Reattach FIFO/status monitoring to tmux terminals surviving a restart. + + Terminal rows and tmux panes intentionally outlive the CAO API process. On + restart, however, the old process's FIFO readers disappear and tmux keeps + forwarding to their now-unread FIFO paths. Re-arm each live, persisted + pane, then seed status from a read-only rendered snapshot so InboxService + can deliver messages already waiting for an idle supervisor. + + Returns the number of panes successfully recovered. Missing/stale database + rows are expected and skipped; this function must never prevent server + startup or send input to an existing agent. + """ + backend = get_backend() + if backend.supports_event_inbox(): + return 0 + + recovered = 0 + try: + sessions = backend.list_sessions() + except Exception as exc: + logger.warning("Cannot enumerate persisted terminal streams at startup: %s", exc) + return 0 + + for session in sessions: + session_name = session.get("id") + if not session_name: + continue + try: + terminals = list_terminals_by_session(session_name) + except Exception as exc: + logger.warning("Cannot list persisted terminals for %s: %s", session_name, exc) + continue + + for terminal in terminals: + terminal_id = terminal["id"] + window_name = terminal["tmux_window"] + try: + # This doubles as a non-mutating liveness check for a stale DB + # row whose tmux window was already removed. + snapshot = backend.get_history( + session_name, window_name, tail_lines=PIPE_LIVENESS_TAIL_LINES + ) + except Exception as exc: + logger.info( + "Skipping persisted terminal %s: pane %s:%s is unavailable (%s)", + terminal_id, + session_name, + window_name, + exc, + ) + continue + + fifo_path = FIFO_DIR / f"{terminal_id}.fifo" + + def _probe_pane(s=session_name, w=window_name) -> str: + return get_backend().get_history(s, w, tail_lines=PIPE_LIVENESS_TAIL_LINES) + + def _rearm_pipe(s=session_name, w=window_name, p=str(fifo_path)) -> None: + get_backend().stop_pipe_pane(s, w) + get_backend().pipe_pane(s, w, p) + + try: + fifo_manager.create_reader(terminal_id, pane_probe=_probe_pane, rearm=_rearm_pipe) + # pipe-pane has one destination. Stop first: ``pipe-pane -o`` + # would otherwise toggle an inherited stale target off. + _rearm_pipe() + status_monitor.seed_from_snapshot(terminal_id, snapshot) + recovered += 1 + except Exception as exc: + # The terminal itself is untouched; a later restart or manual + # send can retry recovery. Leave the process up for healthy panes. + logger.warning("Failed to recover output stream for %s: %s", terminal_id, exc) + + if recovered: + logger.info("Recovered output/status streams for %d persisted terminal(s)", recovered) + return recovered + + async def create_terminal( provider: str, agent_profile: str, diff --git a/test/services/test_terminal_recovery.py b/test/services/test_terminal_recovery.py new file mode 100644 index 000000000..ab270505e --- /dev/null +++ b/test/services/test_terminal_recovery.py @@ -0,0 +1,48 @@ +"""Recovery of tmux output streams after a CAO API-process restart.""" + +from unittest.mock import MagicMock, patch + +from cli_agent_orchestrator.models.terminal import TerminalStatus + + +@patch("cli_agent_orchestrator.services.terminal_service.status_monitor") +@patch("cli_agent_orchestrator.services.terminal_service.fifo_manager") +@patch("cli_agent_orchestrator.services.terminal_service.list_terminals_by_session") +@patch("cli_agent_orchestrator.services.terminal_service.get_backend") +def test_recover_persisted_tmux_terminal_rearms_pipe_and_seeds_status( + backend_getter, list_terminals, fifo_manager, status_monitor +): + from cli_agent_orchestrator.services.terminal_service import ( + recover_persisted_terminal_output_streams, + ) + + backend = MagicMock() + backend.supports_event_inbox.return_value = False + backend.list_sessions.return_value = [{"id": "cao-live"}] + backend.get_history.return_value = "❯ ready" + backend_getter.return_value = backend + list_terminals.return_value = [ + {"id": "abc12345", "tmux_window": "tech_lead-a1b2"} + ] + status_monitor.seed_from_snapshot.return_value = TerminalStatus.COMPLETED + + assert recover_persisted_terminal_output_streams() == 1 + + fifo_manager.create_reader.assert_called_once() + backend.stop_pipe_pane.assert_called_once_with("cao-live", "tech_lead-a1b2") + backend.pipe_pane.assert_called_once() + status_monitor.seed_from_snapshot.assert_called_once_with("abc12345", "❯ ready") + + +@patch("cli_agent_orchestrator.services.terminal_service.get_backend") +def test_recover_persisted_terminal_streams_skips_event_inbox_backend(backend_getter): + from cli_agent_orchestrator.services.terminal_service import ( + recover_persisted_terminal_output_streams, + ) + + backend = MagicMock() + backend.supports_event_inbox.return_value = True + backend_getter.return_value = backend + + assert recover_persisted_terminal_output_streams() == 0 + backend.list_sessions.assert_not_called()