> ## Documentation Index
> Fetch the complete documentation index at: https://www.plungeai.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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 …


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](/one-api/openrouter-compat#50-drop-in-mode)).

**Role primer.** Differs by provider shape:

| Shape | First chunk | How |
|---|---|---|
| openai-shape (openai, groq, mistral, deepseek, xai, together, fireworks, cerebras, perplexity, openrouter, moonshot) | whatever the upstream sends (typically `delta:{role:"assistant"}`) | each upstream `data:` chunk is re-emitted by `shapeOpenAIStream` with the served `model` and `provider`; comment lines and non-JSON lines pass through unchanged. |
| anthropic | `delta:{role:"assistant"}` injected by Ocean | `translateAnthropic` emits a primer in `start()`. |
| google, cohere | `delta:{role:"assistant"}` injected by Ocean | emitted on the first upstream frame. |

**Final `usage` chunk.**

| Shape | Final `usage` present? |
|---|---|
| anthropic, google, cohere | **always** — the translator emits a final chunk carrying `finish_reason` + `usage` (anthropic adds `prompt_tokens_details` when cached). |
| openai-shape | **only when you send `stream_options:{"include_usage":true}`** (or `usage:{"include":true}`) — the gateway itself asks OpenAI and DeepSeek for the usage chunk so billing never depends on you, and removes the usage-only chunk from the stream you receive when you did not ask. |

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):

```bash
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](/one-api/authentication)). 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](#43-agents-plane-openai-shaped-stream)).

**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](/one-api/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](/one-api/agents)).

## 4.4 Verified examples

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

```bash
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:

```bash
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"}'
```