fix: recover tmux terminal streams after server restart

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

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)

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,

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()