Streaming Events

The symfonic agent exposes three streaming HTTP endpoints, each serving a different trade-off between verbosity and ease of consumption. This page is the reference for the event taxonomy, ordering guarantees, error semantics, and client-side consumption patterns.

Endpoints

Endpoint	Payload	Best for
POST /api/v1/stream	StreamChunk (loose dict)	existing consumers, plug-and-play UI
POST /api/v1/stream/typed	StreamEvent (typed Pydantic)	structured UIs, graph animation

Both endpoints return text/event-stream on success. Auth, budget, and rate-limit failures return JSON (not SSE) with the matching 4xx/429/401 status code — clients must distinguish at the HTTP layer before starting the SSE parser.

Event Taxonomy (typed stream)

All events are frozen Pydantic models defined in symfonic.core.contracts.types. The wire format:

event: <ClassName>
data: <JSON payload>
\n

Event	Fires	Payload fields
ActivationEvent	once per activated memory node, BEFORE text	node_id, node_label, score, layer, source_node_id
ThinkingDeltaEvent	during reasoning (extended thinking)	text
TextDeltaEvent	each cleaned text chunk from the LLM	text
ToolCallStartEvent	LLM decides to call a tool	tool_name, tool_call_id
ToolResultEvent	tool returned a result	tool_call_id, result
MemoryExtractedEvent	once, after all text deltas	ops: list[dict]
ConsolidationScheduledEvent	once, penultimate	report, graph_mutations
GraphNodeCreatedEvent	during consolidation (synchronous path only)	node_id, label, layer
GraphEdgeCreatedEvent	during consolidation (synchronous path only)	source, target, relationship
ResponseCompleteEvent	terminal event on success	text, duration_ms
error	terminal event on failure	detail

Ordering Guarantees

For a successful request the first-occurrence order is:

1. Zero or more ActivationEvent
2. Zero or more ThinkingDeltaEvent interleaved with TextDeltaEvent
3. Zero or more TextDeltaEvent, ToolCallStartEvent, ToolResultEvent
4. At most one MemoryExtractedEvent (emitted only when ops were parsed)
5. Exactly one ConsolidationScheduledEvent
6. Zero or more GraphNodeCreatedEvent / GraphEdgeCreatedEvent (synchronous consolidation path only)
7. Exactly one ResponseCompleteEvent

Guarantees enforced by the Gate 4 integration tests in tests/agent/fastapi/test_sse_event_order.py:

• activation precedes text — ActivationEvent.first_index < TextDeltaEvent.first_index
• memory extraction after text, before consolidation — TextDeltaEvent.last < MemoryExtractedEvent < ConsolidationScheduledEvent
• consolidation is a singleton — exactly one per stream
• stream ends with ResponseCompleteEvent OR error — never both, never neither

Not guaranteed:

• Absolute timing between events. TextDeltaEvents may arrive with any pacing (typically 5–50ms apart for Anthropic; bursty otherwise).
• Graph node/edge events in the Celery path. When consolidation runs on a worker queue, the graph events arrive out-of-band via a separate notification bus and are NOT multiplexed into the same SSE response.

Error Semantics

Scenario	HTTP	Body
Missing Authorization	401	JSON {"detail": "..."}
Missing X-Tenant-ID	401	JSON
Forged tenant (not a member)	403	JSON
Budget exceeded	429	JSON with Retry-After header
Rate limited	429	JSON with Retry-After header
Runtime failure after stream opens	200 + event: error	SSE

The critical invariant: 4xx errors arrive before the SSE handshake. Once the server has written text/event-stream headers, it will only close the stream by emitting event: error\ndata: {...}\n\n.

Client Consumption

JavaScript (EventSource / fetch)

const resp = await fetch('/api/v1/stream/typed', {
  method: 'POST',
  headers: {
    'Authorization': `Bearer ${token}`,
    'X-Tenant-ID': tenantId,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({ query, tenant_id: tenantId }),
});
if (resp.status !== 200) {
  // 4xx/5xx — parse JSON body, surface to user
  const err = await resp.json();
  throw new Error(err.detail);
}
const reader = resp.body.getReader();
// ... parse SSE frames using a proper SSE parser (e.g. `eventsource-parser`).

Python (httpx)

async with httpx.AsyncClient() as c:
    async with c.stream(
        "POST", "/api/v1/stream/typed",
        headers={"Authorization": f"Bearer {token}", "X-Tenant-ID": tenant},
        json={"query": query, "tenant_id": tenant},
    ) as r:
        if r.status_code != 200:
            body = await r.aread()
            raise RuntimeError(body)
        async for line in r.aiter_lines():
            ...  # parse SSE frames

A wire-compliant Python parser ships with the test suite at tests/agent/fastapi/sse_test_harness.py (parse_sse_stream) and can be reused outside tests.

Known Gaps (as of v5.5.0)

• [semantic] / [episodic] / [procedural] citation tags may appear in mid-stream TextDeltaEvent payloads. They are stripped from ResponseCompleteEvent.text and from /chat responses but not from intermediate deltas. A follow-up CitationFilter (parallel to the existing ExtractionFilter) will fix this.
• Last-Event-ID reconnect is NOT supported. Every POST starts a new stream from the beginning.