Streaming

Three surfaces stream over Server-Sent Events (SSE): the models plane (/v1/chat/completions with stream:true), the workflows plane (/v1/workflows/execute-stream), and the agents plane (/v1/agents/:id/execute with stream:true). The models plane speaks the OpenAI delta wire; the workflows plane speaks the CNL engine's named-event wire; the agents plane re-shapes an engine run into the OpenAI delta wire. All three send Content-Type: text/event-stream and frame events as …\n\n.

4.1 Models-plane SSE

Code: routes/chat-completions.ts (stream branch) + streaming/sse-translate.ts + streaming/usage-meter.ts.

Wire format. Each event is one data: <json>\n\n line whose payload is a chat.completion.chunk, terminated by a literal data: [DONE]\n\n. The gateway injects no comment lines on /v1; under the OpenRouter drop-in base /api/v1 it sends : OPENROUTER PROCESSING until the first byte (5.0).

Role primer. Differs by provider shape:

Final usage chunk.

Ocean meters tokens on every stream from the upstream's own usage chunk (a passive usage-meter sits between the upstream and the shaper; Groq's x_groq.usage is read too); what you receive still depends on stream_options.

model and provider on stream chunks. Every chunk of every shape carries the served slug in model (also after a failover) and the canonical provider id in provider.

Response headers: content-type: text/event-stream, cache-control: no-cache, no-transform, x-accel-buffering: no, x-request-id.

Failover vs mid-stream failure. Failover between candidates is decided on the response status, before any SSE byte is read (runAttempts); once bytes are flowing there is no re-failover. A failure after that point travels in-band: the final chunk carries error: { message, type: "upstream_error", code: "upstream_error", metadata: { provider_name, raw } } and finish_reason: "error", followed by data: [DONE] — for an Anthropic error event, a Gemini error object, or an upstream connection that dies mid-stream. An openai-shape upstream that sends its own {error} event passes it through as sent.

Cancellation / abort. The client's AbortSignal is forwarded to the upstream fetch, so hanging up stops the provider. An aborted stream is billed for what it streamed — the completion tokens seen so far (a character estimate when the provider never sent usage) plus the prompt estimate — logged with status 499 / error_code client_abort, and the hold is settled at that amount, not released (a trial org is logged but never debited). The */5 sweeper is only a backstop for a Worker that dies. There is no gateway-imposed stream idle timeout on this plane beyond the Workers runtime and the upstream's own.

A streamed chat ends with [DONE] (live-verified 2026-10-04):

curl -s "$API_BASE/v1/chat/completions" \
  -H "Authorization: Bearer $PLUNGE_API_KEY" \
  -H "content-type: application/json" \
  -d '{"model":"openai/gpt-4o-mini","stream":true,"messages":[{"role":"user","content":"Reply OK"}],"max_tokens":8}'

4.2 Workflows-plane SSE

Code: routes/workflows.ts + lib/stream-relay.ts (workflowsEncoder) + engine orchestration/cnl-engine/context.ts.

Routes: POST /v1/workflows/execute-stream (inline YAML/JSON body), POST /v1/workflows/:id/execute-stream (saved workflow), and the legacy alias POST /v1/cnl/execute-stream. All require an ozk_ key (Authentication & keys). The relay also drains every provider sub-stream the run advertises and closes the execution row under waitUntil, so a client disconnect never aborts the run.

Wire format. Engine frames pass through verbatim as event: <name>\ndata: <json>\n\n. Provider token deltas are injected as event: token\ndata: {"task_id":"…","delta":"…"}\n\n. Token frames for a task are flushed before that run's terminal frame (ordering guarantee).

Engine event vocabulary (CNLEventType, orchestration/cnl-engine/context.ts):

request_received, workflow_loaded, workflow_started, task_dispatched, task_completed, task_error, task_skipped, streaming_started, streaming_completed, output_delivered, parallel_start, parallel_complete, sequential_start, sequential_complete, dynamic_start, dynamic_generated, dynamic_complete, batch_start, batch_item_start, batch_item_complete, batch_progress, batch_complete, debate_start, debate_round_complete, debate_judgment, debate_complete, condition_evaluated, workflow_yielded, workflow_result, workflow_completed, workflow_error, heartbeat.

Each data payload carries workflow_id, execution_id, timestamp, type, a nested data object, and elapsed_ms. The outcome frame is workflow_completed (carries final_task_id, execution_summary) or workflow_error (carries error); workflow_result is always the last frame, after the outcome — keep reading until it arrives.

Keepalive. The engine emits its own heartbeat events; the relay forwards them verbatim and additionally touches the execution row's updated_at every ~60s so a long run is not swept mid-flight. This plane does not inject comment-line keepalives (that is the agents plane, 4.3).

Reading the final result. After workflow_result (the last frame), fetch the stored output with GET /v1/workflows/results/:workflowId/:taskId (use the final_task_id) — see Workflows. While a run is in flight the results route returns 404 not_ready.

4.3 Agents-plane OpenAI-shaped stream

Code: routes/agents.ts (stream branch) + lib/stream-relay.ts (openaiEncoder). Requesting the stream: set "stream": true in the JSON body. This runs the same one-task engine envelope as the sync path but with force_streaming, and the openaiEncoder re-shapes the run into an OpenAI chat.completion.chunk stream — so an OpenAI SDK's streaming client works against an Ocean agent. (For the CNL-native event stream of the same agent, use the workflows plane instead.)

What the client sees — only data: lines and comments, never engine event: lines (they are dropped by the encoder):

• id: chatcmpl-<executionId hex, 24 chars> — derived from the execution id, stable across the stream.
• model: the served model — a caller-pinned provider/model, else the provider's default from its /health, else <provider>/default. Never the agent id.
• A primer chunk (delta:{role:"assistant"}) is emitted lazily on the first token.
• Content chunks: delta:{content:"…"}.
• Final chunk: delta:{} with finish_reason (stop or length) and, when the split is readable, a usage object; then data: [DONE]\n\n.
• Keepalive: a comment line : OCEAN PROCESSING\n\n roughly every 15s while idle (so proxies don't time out a slow first token).

Error after the first byte. Once SSE headers are committed, a run error is emitted as data: {"error":{"message":…,"type":"server_error","code":"engine_error"}} and the stream closes without [DONE] (matching OpenRouter's mid-stream-error behaviour). A rejection before the first byte is a normal JSON 502 engine_error instead.

Headers: content-type: text/event-stream; charset=utf-8, cache-control: no-cache, x-accel-buffering: no, x-request-id, and X-Execution-Id (the run id — reuse it to read the result later at GET /v1/agents/results/:workflowId/:taskId, Agents).

4.4 Verified examples

An agent OpenAI-shaped stream — the primer, content deltas, a final usage chunk, and [DONE]:

curl -s -N "$API_BASE/v1/agents/one-agent/execute" \
  -H "Authorization: Bearer $PLUNGE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt":"say hi in one word","stream":true,"max_tokens":16}'

A workflow event stream — engine frames pass through verbatim as named event: lines:

curl -s -N "$API_BASE/v1/workflows/execute-stream" \
  -H "Authorization: Bearer $PLUNGE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"workflow":"workflow:\n  name: doc-stream\n  tasks:\n    - type: task\n      id: t1\n      agent: one-agent\n      prompt: say hi"}'