SSE Streaming

AEGIS uses Server-Sent Events (SSE) for real-time streaming of agent execution progress to the frontend. This page documents the EventSource implementation, event types, the streaming protocol, and fallback behavior.

Overview

When a user sends a message in the conversations page, the frontend opens an SSE connection to the orchestration engine. As the agent processes the request through the LangGraph pipeline, the server emits events at each pipeline node completion. The frontend parses these events to show real-time progress indicators, tool execution status, and the final assistant response.

SSE Connection

The Conversations Canvas (rendered by /conversations and the bookmarkable /conversations/[conversationId] routes) speaks SSE through a dedicated data layer in src/lib/conversations/:

• turn-stream.ts — the TurnStream class wraps EventSource with the reconnect/fallback lifecycle (Last-Event-ID resume, bounded reconnects, the streamedAny gate that decides whether a sync retry is safe).
• turn-events.ts — pure payload parsers (fromNodeEnd, fromDone, …), fixture-tested against recorded transcripts in __tests__/__fixtures__/.
• store.ts — the Zustand store that applies parsed updates to conversation state; components subscribe via granular selectors.

(The pre-R40.5 legacy view and its NEXT_PUBLIC_CONV_LEGACY rollback flag were deleted in R40.5 P5 — the Canvas is the only conversations surface.)

The underlying connection is the browser’s EventSource API, opened against a same-origin /api/v1/* URL (R42c). TurnStream builds the URL from a basePath of /api/v1:

// turnStreamUrl("/api/v1", convId, message, agentId, agentType, …)
const streamUrl = `${basePath}/conversations/${conversationId}/stream`
  + `?message=${encodeURIComponent(message)}`
  + `&agent_id=${agentId}`
  + `&agent_type=${agentType}`;
 
const eventSource = new EventSource(streamUrl);

The connection is same-origin and routes through the API Gateway (Caddy on the box, the next.config.ts rewrite in dev), so the browser attaches the httpOnly aegis_token cookie automatically. The message, agent ID, and agent type are passed as URL query parameters.

Message attachments (R40.5 P3) ride the same GET as attachment_ids — a comma-separated list of previously uploaded attachment ids (only the ids; the extracted text stays server-side). The server splices each file’s text into the STATE user message — so the model and turn_snapshots see it — while persist_turn stores the user’s typed text with pill refs in the user row’s metadata. Reloads therefore show clean bubbles with attachment pills, never raw extractions.

Event Types

The server emits two named SSE event types:

node_end

Fired each time a LangGraph pipeline node completes execution. The payload contains:

{
  "node": "initial_llm_call",
  "data": {
    "message": {
      "content": "Here is the analysis...",
      "tool_calls": [
        { "function": { "name": "spacing_calculation" } }
      ]
    }
  }
}

The frontend maps node names to human-readable status labels:

When a node_end event fires for the LLM call node (initial_llm_call):

• If data.message.tool_calls is present, tool status indicators are added to the chat.
• If data.message.content is present, the assistant response is displayed (or updated if already visible).

done

Fired when the entire pipeline completes:

{
  "data": {
    "status": "completed",
    "execution_id": "a1b2c3d4-...",
    "tokens_used": 18400,
    "cost_usd": 0.11,
    "skills_injected": ["flaring_watch"],
    "thinking_mode": "adaptive",
    "referenced_entities": [
      { "entity_id": "e5f6...", "name": "Mitchell Ranch 1H", "entity_type": "Well" }
    ]
  }
}

If status is "awaiting_hitl", a HITL review banner is displayed with a link to the filings page, and the context panel auto-opens scoped to the paused run.

The run-detail fields (tokens_used, cost_usd, skills_injected, thinking_mode — the effective post-clamp mode — and referenced_entities) land in the store’s runs map keyed by execution_id. They power the context panel (src/components/conversations/ContextPanel.tsx) and the run chip in each answer’s metadata row. The map is session-memory only: reloaded conversations have no entry and the panel shows ”—” (the turn_snapshots-backed fetch is a planned v1.5 follow-up).

Artifact events on the conversation channel

Rich tool outputs (Tier 1 artifacts, R43 agent charts) ride the same stream as tool_result payloads: the orchestration engine attaches the tool envelope to the persisted tool row’s metadata.tool_result, and the frontend renderer registry (including ChartArtifact for render_chart specs) renders it in the canvas. Because the payload is persisted, a reload re-renders exactly what streamed.

Structured Workspace Events (assessment stream)

The entity compliance workspace (/compliance/[entityId]/[domain]) uses a separate SSE channel from conversations. Its assessment stream emits typed, structured events — not conversational markdown — defined by the WorkspaceEventType enum in services/orchestration-engine/src/orchestration/compliance/workspace/events.py:

These power the three-panel workspace’s live progress (assessment phase → item-by-item interaction).

Tool Status Display

When tool calls are detected in an LLM response, the frontend displays animated status indicators with human-readable labels:

const TOOL_LABELS: Record<string, string> = {
  spacing_calculation:      "Calculating spacing distances...",
  offset_well_analysis:     "Querying knowledge graph for offset wells...",
  rule37_filing_assembly:   "Assembling Form W-1 filing package...",
  good_cause_narrative:     "Drafting good cause narrative...",
  flaring_volume_calc:      "Calculating flaring volumes...",
  gas_analysis:             "Analyzing gas composition...",
  rule32_filing_assembly:   "Assembling Form R-32 filing package...",
  emissions_estimate:       "Estimating CO2e emissions...",
};

Each tool status appears as a small inline indicator with a pulsing blue dot and the label text.

Payload Parsing

The SSE data can arrive in two formats due to how sse-starlette serializes payloads:

1. Standard JSON: data: {"node": "...", "data": {...}}
2. Double-encoded string: data: "{\"node\": \"...\"}"

The frontend handles both:

const parsePayload = (raw: string): any => {
  let parsed = JSON.parse(raw);
  // sse-starlette may double-encode: data is a JSON string instead of object
  if (typeof parsed === "string") {
    parsed = JSON.parse(parsed);
  }
  return parsed;
};

Event Listener Registration

Events are handled via two mechanisms for maximum compatibility:

Named event listeners (primary path)

eventSource.addEventListener("node_end", (e) => {
  handleNodeEnd(parsePayload(e.data));
});
 
eventSource.addEventListener("done", (e) => {
  handleDone(parsePayload(e.data));
});

Generic message handler (fallback)

For servers that dispatch events without the event: field, the onmessage handler inspects the parsed payload:

eventSource.onmessage = (e) => {
  const payload = parsePayload(e.data);
  if (payload.event === "node_end") {
    handleNodeEnd(payload);
  } else if (payload.event === "done") {
    handleDone(payload);
  }
};

Streaming UI Indicators

While streaming is active, the UI shows:

• An animated bouncing dots indicator (three blue dots with staggered y animation).
• A text label showing the current pipeline stage (e.g., “Thinking…”, “Executing tools…”).
• The input field is disabled to prevent duplicate submissions.

{streaming && (
  <div className="flex items-center gap-2 pl-2">
    <div className="flex gap-1">
      {[0, 1, 2].map((i) => (
        <motion.div
          key={i}
          className="w-1.5 h-1.5 rounded-full bg-blue-400"
          animate={{ y: [0, -4, 0] }}
          transition={{ repeat: Infinity, duration: 0.6, delay: i * 0.15 }}
        />
      ))}
    </div>
    <span className="text-xs text-muted-foreground">{streamStatus}</span>
  </div>
)}

Message Types

The conversation page handles five message types:

HITL Banner

When an agent execution results in status: "awaiting_hitl", a banner is added to the chat:

<div className="border border-amber-800/40 bg-amber-950/30 px-4 py-3">
  <p>Filing package ready for review</p>
  <p>pre_filing checkpoint -- requires human review</p>
  <Link href="/filings">
    <Button>Review</Button>
  </Link>
</div>

The conversation status is also updated to "awaiting_hitl" and a badge is shown in the conversation list.

Agent Handoff

Agent handoff messages suggest switching to a different agent. They display:

• The suggestion message content.
• The target agent’s color indicator, name, and description.
• A “Continue with {agent}” button that creates a new conversation with the suggested agent and pre-populates context from the handoff.

Fallback: Synchronous Execution

If the stream never starts (the TurnStream never_started outcome), the frontend falls back to synchronous execution via POST /api/v1/execute:

const res = await fetch(`/api/v1/execute`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    conversation_id: conv.id,
    agent_id: conv.agent.id,
    agent_type: conv.agent.type,
    message: msg,
    metadata: {},
  }),
});

This returns the complete result in a single response. Tool calls and assistant messages are extracted from the response payload and displayed in sequence.

Auth for SSE

Because the stream is opened same-origin (/api/v1/conversations/{id}/stream), the browser attaches the httpOnly aegis_token cookie automatically — EventSource cannot set custom headers, but it does send cookies for same-origin requests. The gateway validates the cookie exactly as it does for fetch calls. There is no JavaScript cookie-reading or Authorization-header construction (the cookie is httpOnly and not JS-readable since R42b); any older code that parsed document.cookie for a bearer token is gone.

Conversation Persistence

Conversations and messages are persisted to the backend:

• Load on mount: GET /api/v1/conversations (same-origin, through the gateway).
• Load messages: GET /api/v1/conversations/{id}/messages when selecting a conversation.
• Create: POST /api/v1/conversations with agent_id and conversation_type.
• Update title: PATCH /api/v1/conversations/{id} after the first message.