Skip to main content

Real-Time Streaming Architecture (SSE & WebSocket)

This document details the real-time event streaming and telemetry architecture in the GitHub Backup Automation System, covering Server-Sent Events (SSE) in the Python Agentic Observatory and the live log streams (SSE and WebSocket) of the Go Backend API.

1. High-Level Streaming Architecture​

┌─────────────────────────────────────────────────────────────┐
│                      Next.js Frontend                       │
│   (AI Playground & Live Logs, via same-origin /api/proxy)   │
└──────────────┬───────────────────────────────┬──────────────┘
               │                               │
       POST /chat/stream (SSE)      GET /api/logs/stream (SSE)
               │                               │
               ▼                               ▼
┌─────────────────────────────┐ ┌─────────────────────────────┐
│  Python Agent Observatory   │ │        Go Backend API       │
│  (FastAPI StreamingResponse)│ │  (shared livefeed poller)   │
└──────────────┬──────────────┘ └──────────────┬──────────────┘
               │                               │
       Async LLM Generator            PostgreSQL execution_logs
               │                               │
               ▼                               ▼
      OpenRouter AI API               Live Log Event Stream

2. Python Agent Observatory (Server-Sent Events / SSE)​

Lifecycle of an SSE Stream​

  1. Client Request: The dashboard sends POST /chat/stream with a bearer token and {"question": "...", "session_id": "<uuid, optional>", "model": "<optional>"}.
  2. Checks before streaming: The model must be the default OPENROUTER_MODEL or a free model from the OpenRouter registry (otherwise 400 MODEL_NOT_ALLOWED), and the caller's OpenRouter credentials are resolved (their own key if connected, else the server pool). These fail as plain HTTP errors, not as stream events.
  3. User message saved first: chat_service.start_turn() stores the question in ai_chat_messages before the response starts, so it is kept even if the stream fails or the client disconnects.
  4. Long-Lived Stream: FastAPI returns a StreamingResponse (media_type="text/event-stream") over chat_service.stream_chat_turn(), which wraps agent.openrouter.stream_agent().
  5. Real token streaming: The model is called with astream() and each text chunk is sent as a token event as it arrives. Text the model writes before a tool call (for example "Let me check the logs") is streamed too and is part of the answer; tool-call chunks only build up the tool call and are never sent as text. Each event is one SSE frame: 
    data: {"type": "token", "text": "Analyzing recent backup logs..."}\n\n
  6. Deadline: The whole turn has AGENT_REQUEST_DEADLINE_SECONDS (default 280). Each model request has OPENROUTER_REQUEST_TIMEOUT_SECONDS (default 30) and no SDK retries; a 401, 402 or 429 while opening the stream moves to the next server key.
  7. Turn saved with its real status: When the stream ends, fails, times out or the client disconnects, the investigation and the assistant message are saved with status completed or failed, the text streamed so far and the tool executions. The save is shielded from cancellation.

SSE Message Schema​

Events in the order they can appear. Fields may be added later; clients should ignore unknown fields.

  • Info (first event): {"type": "info", "request_id": "<uuid>", "session_id": "<uuid or null>", "sources": [...]}
  • Iteration started: {"type": "agent_reasoning", "iteration": 1, "request_id": "<uuid>"}
  • Token Output: {"type": "token", "text": "<string>"}
  • Human-In-The-Loop Confirmation: {"type": "confirm_required", "confirm_id": "<uuid>", "name": "<tool>", "args": {...}, "tool_call_id": "<id>"}
  • Tool Call Started: {"type": "tool_start", "name": "<tool_name>", "args": {...}, "tool_call_id": "<id>"}
  • Tool Call Completed: {"type": "tool_end", "name": "<tool_name>", "tool_call_id": "<id>", "success": true, "duration_ms": 12.4, "result": "..."}; on failure or rejection success is false and error replaces result.
  • Error: {"type": "error", "message": "<text for the user>", "code": "<CODE>", "request_id": "<uuid>"}
  • Stream End: {"type": "done", "answer": "<final text>", "request_id": "<uuid>"}; done is always the last event.

tool_call_id is the model's tool call id, or a generated UUID when the model sent none. It is the same on every event of one tool call (confirm_required, tool_start, tool_end), so parallel calls of the same tool can be paired.

done.answer is the full text streamed in the turn. After an error it is the text streamed so far followed by the error message, so the final bubble always shows what happened. Error codes: DEADLINE_EXCEEDED, OPENROUTER_BAD_REQUEST, OPENROUTER_KEY_INVALID, OPENROUTER_OUT_OF_CREDITS, OPENROUTER_RATE_LIMITED, OPENROUTER_UNAVAILABLE, OPENROUTER_NOT_CONFIGURED, OPENROUTER_ERROR and INTERNAL_ERROR.

HITL Confirmation Flow​

  1. Before running a tool in CONFIRMATION_REQUIRED_TOOLS (today send_report_email), the stream stores a pending row in ai_tool_confirmations (confirm_id, session, user, tool, arguments) and sends confirm_required.
  2. The dashboard answers with POST /chat/confirm and {"confirm_id": "<uuid>", "approve": true}; approved is accepted as an alias of approve. Only the user who started the turn can decide: other users get 404, a decided or expired confirmation gives 409, a malformed id 422, and 503 means the confirmation store is unavailable.
  3. The stream polls its own row every HITL_POLL_INTERVAL_SECONDS (default 0.5). Because the decision is in the database, it works when the confirm request reaches another instance.
  4. With no decision within HITL_CONFIRMATION_TIMEOUT_SECONDS (default 120), or when the client disconnects, the row is marked expired and the tool is not run. A rejected or expired call ends with a tool_end event with success: false.

The non-streaming POST /chat and POST /api/tools/execute never run confirmation-required tools; they refuse them (/api/tools/execute with 403).

3. Go Backend Live Log Streams​

Both streams deliver new execution_logs rows as the same JSON object:

{"type":"log","id":123,"level":"info","message":"Cloning repository","repository":"octocat/Hello-World","timestamp":"2026-09-24T07:41:09.452381Z"}

Shared poller (src/backend/github-backup-api/livefeed)​

  • One poller serves every SSE and WebSocket subscriber; each subscriber has its own queue (256 events) and is dropped, not waited for, when it falls behind.
  • It only queries while at least one subscriber is connected, so an idle API sends no queries and Neon can scale to zero.
  • It polls every 2 s, backs off to 15 s after a minute without new rows, and returns to 2 s when rows arrive or a subscriber joins. A full batch (200 rows) is followed by another query at once.
  • When the first subscriber arrives it reads the newest log ID (retrying with backoff if that fails) and starts there: history is never replayed, including after an idle period.
  • All queries use the primary key (id > $1 ORDER BY id).

SSE: GET /api/logs/stream (used by the dashboard proxy)​

  • Under /api: rate limited, and authenticated when API_REQUIRE_AUTH=true.
  • Sends retry: 5000, then id: <log id> / event: log / data: <json> events, a : ping comment every 15 s, and closes after 5 minutes so clients reconnect and no connection outlives proxy or serverless limits.
  • Resumes from Last-Event-ID (preferred) or ?after_id=: the missed rows up to the moment the stream started (at most the newest 1,000) come first, then live rows, with no gap or duplicate.
  • Headers Cache-Control: no-cache and X-Accel-Buffering: no; each flush has a 10 s write deadline.

WebSocket: GET /ws/live​

  • Browser handshakes must come from an origin in WS_ALLOWED_ORIGINS (default: the dashboard origins); with API_REQUIRE_AUTH=true the handshake needs the internal secret header, which browsers cannot send, so browsers use the SSE endpoint instead.
  • Keepalive: server ping every 30 s, a 75 s read deadline extended by every pong, a 10 s deadline on every write. Dead clients are unregistered; slow ones are closed with code 1013, and shutdown closes with 1001.

Server timeouts and shutdown​

  • The Fiber server has a 15 s read timeout and a 2 min idle timeout but no write timeout: fasthttp applies it to the whole response and would cut the SSE stream. WebSockets are hijacked connections and are not affected by the server timeouts.
  • On SIGINT/SIGTERM the API closes the live feed first (ending every stream), then shuts the HTTP server down with a 10 s timeout.

4. Troubleshooting Real-Time Streams​

  • Buffering Issues: Ensure intermediate proxies (Vercel, Render, Cloudflare) have response buffering disabled (X-Accel-Buffering: no, which the API sets on the log stream).
  • Connection Drops: The API ends every log stream after 5 minutes on purpose. The live log view first lets the browser's EventSource reconnect (it resends Last-Event-ID), then reconnects with jittered exponential backoff from the last seen id, up to 8 attempts. Leaving the AI chat or starting a new question aborts the chat stream.
  • No events: a stream only carries rows written after it started (or after the resume ID). Check that the worker writes to the same database, and that the client is not being dropped as too slow.

5. Dashboard Proxy for Streams​

  • The browser opens both streams on the dashboard's own origin: POST /api/proxy/agent/chat/stream and GET /api/proxy/backup/api/logs/stream. The route handlers forward them to AGENT_API_URL and BACKUP_API_URL (read at request time) and pass the body through unchanged.
  • Proxied responses carry Cache-Control: no-cache, no-transform and X-Accel-Buffering: no, and the upstream is asked for an uncompressed body, so events are relayed as they arrive. text/event-stream responses are exempt from the proxy's 30 s timeout.
  • A browser disconnect aborts the upstream request and closes its connection; the Observatory's StreamingResponse then cancels the agent generator, and the Go API can end the log stream.
  • When the log stream endpoint answers 404 (Go API without it), the dashboard polls GET /api/logs?limit=50 every 5 s instead.