Realtime and WebSockets

Real-time & WebSockets

Task progress, streamed synthesis, and human-in-the-loop questions all flow over WebSocket. The design survives reconnects (full snapshot replay), scales across workers (Redis pub/sub), and degrades to HTTP polling when WebSockets are blocked.

Event bus

utils/events.py provides an in-process event_bus (subscribe/publish). When REDIS_URL is set, a Redis pub/sub EventBus fans events out across workers:

• Events channel — maestro:events:{task_id} carries agent events.
• Control channel — maestro:ctrl:{task_id} carries cross-worker control messages: cancel and HITL answers. This is how a cancel issued on worker B reaches a task running on worker A.

The single-worker in-process bus is the automatic fallback when REDIS_URL is empty (dev/test).

Event envelope

Every event is a versioned, sequence-stamped envelope: {v, seq, type, ts, ...}. The MongoDB agent_logs collection is the seq-ordered source of truth; task_sessions.events is a capped mirror for quick reads (TASK_EVENTS_MONGO_KEEP = 200).

task_service._make_emit produces these envelopes and dual-writes them (source-of-truth agent_logs + capped task_sessions mirror). events_since provides the resume cursor.

Event types

Type	Meaning
node_update	An agent node changed state (used to drive the architect graph).
agent_delta	A streamed text chunk (synthesis output; flushed every STREAM_DELTA_FLUSH_CHARS = 80).
agent_question	The main agent is asking a HITL question.
user_answer	The user's answer was accepted.
review_result	Reviewer verdict (approved / issues).
task_completed / task_completed_with_warnings	Terminal success (with optional known gaps).
task_cancelled	Terminal cancel.
task_failed	Terminal failure (wire status may be failed / timeout / cancelled).

WebSocket endpoints

Both authenticate via ?token=<access> query param, are owner-only, and rate-limit the handshake by calling check_websocket before accept() (closing 1013 on excess — a Depends can't reject a handshake early enough). See Security.

Path	Purpose
WS /api/v1/tasks/{task_id}/stream	Task stream: an initial snapshot then live events; ?after_seq=N resumes from a cursor; accepts inbound answer messages for HITL.
WS /api/v1/architect/live?task_id=	Architect communication stream for the live agent graph.

websocket.py carries a logger: _run_stream wraps unexpected errors with logger.exception (extra: task_id/user_id) + close(1011) instead of dying silently inside uvicorn; _receive_answers warns and continues on malformed JSON rather than killing the socket.

Frontend consumption

The client helper is frontend/src/lib/ws.ts → openTaskStream(taskId, handlers). See Frontend-Reference for how the Zustand tasks store wires it up. Key behaviors:

1. Message shapes — either {type:'snapshot', status, events[]} (full replay on every reconnect, so missed events are recovered) or a bare event.
2. Token freshness — ensureFreshAccessToken() runs before every (re)connect so a token can't expire mid-handshake.
3. Reconnect — exponential backoff (1s → 8s cap); reconnects immediately on visibilitychange when the tab returns to foreground (beats browser timer throttling); stops once terminal.
4. HTTP-poll fallback — after WS_FALLBACK_AFTER_FAILURES = 3 connects that die before any message, it degrades to polling GET /tasks/{id} every TASK_POLL_INTERVAL_MS = 4s through the same snapshot path until terminal. A 404 means the task was deleted.

Human-in-the-loop (HITL)

When the main agent asks a question (agent_question), the user answers:

• Primary path — handle.answer(text) sends {type:'answer', answer} over the open socket.
• Fallback — if the socket isn't open, it calls POST /tasks/{id}/answer (HTTP), which delivers via the Redis control channel if the task runs on another worker.

The backend persists a task_questions row, flips status to awaiting_answer, awaits the answer (bounded by HITL_TIMEOUT_SECONDS = 180), and re-plans once. See Agent-Orchestration.