Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions src/cli_agent_orchestrator/api/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
21 changes: 21 additions & 0 deletions src/cli_agent_orchestrator/services/status_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:])
Comment on lines +506 to +510
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)
Expand Down
80 changes: 80 additions & 0 deletions src/cli_agent_orchestrator/services/terminal_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
)
Comment on lines +213 to +215
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)
Comment on lines +239 to +240
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,
Expand Down
48 changes: 48 additions & 0 deletions test/services/test_terminal_recovery.py
Original file line number Diff line number Diff line change
@@ -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()
Loading