# Stream events (/docs/stream-events)
`runtime.executeMessage()` is an async generator. Every yielded
value is a `StreamEvent` — a discriminated union keyed on `type`.
Browser UIs render from these; server consumers forward them over
SSE. See [SessionRuntime](/docs/session-runtime) for the
generator's signature and [Event log](/docs/event-log) for how
these shapes persist as durable rows.
```typescript
import type { StreamEvent } from "@pleach/core";
for await (const event of runtime.executeMessage(sessionId, prompt)) {
// event is a StreamEvent — switch on event.type
}
```
Every event also carries an optional `namespace: string[]` field —
empty for the root orchestrator, populated for events emitted from
inside a subagent. Use it to route subagent output into its own
UI surface.
LangGraph's `graph.stream(input, { streamMode: "updates" | "messages" | "values" | "custom" | "tools" | "debug" | "checkpoints" | "tasks" })` returns different shapes per mode. Pleach has a single discriminated union — `StreamEvent` — where every shape ships in every stream and consumers filter on `event.type`. The rough mapping:
| LangGraph `streamMode` | Pleach equivalent |
| ---------------------- | ------------------------------------------------------------------------------------ |
| `"updates"` | `session.updated` events |
| `"messages"` | `message.delta` + `message.complete` |
| `"values"` | Re-read the session via `runtime.sessions.resume(sessionId)` after `session.updated` |
| `"custom"` | Emit a `custom.event` via `runtime.events.emit()` from inside a node |
| `"tools"` | `tool.started` / `tool.completed` / `tool.failed` |
| `"debug"` | The full untouched stream; pleach has no filter |
| `"checkpoints"` | `checkpoint.created` |
| `"tasks"` | `task.scheduled` / `task.completed` (subagent + async) |
LangGraph's `streamMode: ["updates", "custom"]` tuple-yielding form has no pleach analog — the discriminated union plays the same role with one `switch (event.type)`. The array-tuple form trades one extra type-narrow step for the convenience of multi-mode iteration; we picked the discriminated-union shape because it composes better with downstream `flatMap` / `filter` / `tap` pipelines.
## Session lifecycle [#session-lifecycle]
## Message stream [#message-stream]
The main user-facing surface — what gets rendered into the chat
viewport.
| Type | Payload | When |
| ------------------- | --------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ |
| `message.added` | `{ message: Message }` | Full message appended (user or assistant) |
| `message.delta` | `{ delta: string }` | Streaming chunk; concatenate to render |
| `message.complete` | `{ message: Message }` | Final assembled assistant message |
| `message.citations` | `{ citations: MessageCitation[]; section: string }` | Citation set attached to a message section |
| `message.entities` | `{ toolCallId, toolName, entities: Entity[] }` | Structured entities extracted from a tool result |
| `reasoning.delta` | `{ delta: string }` | A chunk of the model's reasoning (chain-of-thought), surfaced from the provider as it arrives — **never answer content** |
| `thinking.delta` | `{ delta: string }` | Streaming reasoning trace (when the model emits one) |
| `thinking.complete` | `{ thinking: string }` | Full reasoning trace assembled |
Render `message.delta` and `thinking.delta` as they arrive;
treat `message.complete` as the authoritative final text.
### Reasoning traces (`reasoning.delta`) [#reasoning-traces-reasoningdelta]
Reasoning models (DeepSeek-R1, etc.) stream a chain-of-thought
*before* the answer. Pleach surfaces that as `reasoning.delta`
events — a provider-agnostic side-channel that is **never**
concatenated into the answer text. Two reasons it exists:
1. **Liveness.** A long reasoning phase (tens of seconds with no
answer token) would otherwise look like a stalled stream.
`reasoning.delta` keeps the no-content watchdog from
false-firing while the model thinks.
2. **UX.** Render a "Thinking…" indicator, or accumulate the
deltas into a collapsible trace next to the final answer.
This is **distinct from the `reasoning` call class** (a
type-level routing invariant on the seam — see
[Seams](/docs/seams)). `reasoning.delta` is *runtime trace
content*; the `reasoning` call class is *which seam a node
consumes*. The `@pleach/react` `useSessionMessageStream` hook
exposes `isReasoning` + `reasoningText` so you don't have to
accumulate the deltas yourself — see [React](/docs/react).
## Tool calls [#tool-calls]
| Type | Payload | When |
| ---------------- | ----------------------------------------- | ----------------------------------------------- |
| `tool.started` | `{ toolCall: ToolCall }` | Tool invocation dispatched |
| `tool.delta` | `{ toolCallId: string; delta: string }` | Streaming argument assembly (e.g. partial JSON) |
| `tool.completed` | `{ toolCall: ToolCall; result: unknown }` | Tool returned a result |
| `tool.failed` | `{ toolCall: ToolCall; error: string }` | Tool threw or rejected |
Pair `tool.started` with `tool.completed` / `tool.failed` on
`toolCall.id` to render per-tool spinners and result cards.
A worked sequence for a single `search_corpus` call in a
knowledge-base assistant:
1. `tool.started` — `{ toolCall: { id: "tc-018f-7a-1", name: "search_corpus", args: { query: "indexing strategies" } } }`
2. `tool.delta` × N — streaming arg assembly while the model emits JSON; concatenate into a per-`toolCallId` buffer
3. `tool.completed` — `{ toolCall: { id: "tc-018f-7a-1", … }, result: [{ id: "doc-abc123", score: 0.91 }, …] }`
4. `message.entities` — `{ toolCallId: "tc-018f-7a-1", toolName: "search_corpus", entities: [{ kind: "document", id: "doc-abc123" }] }`
5. `message.citations` — citations attached to the synthesized answer's body section once the LLM references the hits
Steps 1 and 3 are the spinner bracket; step 4 is what hydrates a
"sources" sidebar; step 5 is what underlines the cited sentence in
the rendered transcript.
## Async jobs [#async-jobs]
For tools that dispatch work to a queue and return later.
| Type | Payload | When |
| ---------------- | ------------------------------------------------------------------------------------ | -------------------------------------- |
| `job.dispatched` | `{ job: PendingJob }` | Job submitted to the dispatch endpoint |
| `job.progress` | `{ jobId, progress, phase?, phaseProgress?, statusMessage?, estimatedRemainingMs? }` | Worker progress update |
| `job.completed` | `{ job: CompletedJob }` | Worker finished successfully |
| `job.failed` | `{ jobId: string; error: string }` | Worker errored |
| `jobs.pending` | `{ jobs: PendingJob[] }` | Snapshot of all currently-pending jobs |
`progress` is 0–1. Use `phaseProgress` for multi-phase jobs (e.g.
`download` then `process` then `upload`).
## Interrupts (human-in-the-loop) [#interrupts-human-in-the-loop]
| Type | Payload | When |
| --------------------- | ----------------------------------------------------- | --------------------------------------------- |
| `interrupt.requested` | `{ interrupt: PendingInterrupt }` | Runtime is blocked waiting on a user decision |
| `interrupt.resolved` | `{ interruptId: string; decision: ApprovalDecision }` | Decision submitted; turn resumes |
The `interrupt.requested` event is when you'd surface an approval
modal in your UI; the runtime pauses until `decision` lands.
## Execution lifecycle [#execution-lifecycle]
Agnostic signals an external `@pleach/core` consumer subscribes to via
`useChat({ onEvent })` (or any `StreamEvent` driver) to observe *how* a
turn ran — stage transitions, recovery, retries, stream timing — without
inheriting the host's console-grep idiom. They fire from graph-internal
sites alongside the runtime's own diagnostics; the values reuse the
canonical core unions (`StageId`, `RecoveryDispatchArm`) and carry no
domain concepts.
| Type | Payload | When |
| ------------------------- | ------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- |
| `turn.started` | `{}` | A turn begins (brackets graph + imperative paths uniformly) |
| `turn.completed` | `{ outcome: "ok" \| "error" \| "interrupted"; durationMs: number }` | The turn ends; `outcome` is the terminal state |
| `stage.transition` | `{ from: StageId \| null; to: StageId }` | The compiled graph moves stage; `from` is `null` on graph entry |
| `stage.lattice_violation` | `{ from: StageId; to: StageId }` | A runtime transition was NOT in `ALLOWED_EDGE_PATTERNS` — a structural-conformance canary (fail-soft) |
| `stage.post_turn_fan_out` | `{ writer: string }` | A post-turn parallel writer fired (`"sessionMemoryWrite"` / `"costRollup"`); both per turn means the fan-out held |
| `recovery.fired` | `{ arm: RecoveryDispatchArm }` | A post-turn recovery filter dispatched; `arm` ∈ `"zeroToolRecovery"` \| `"allToolsFailedMissingParams"` \| `"maxStepsHit"` |
| `retry.attempted` | `{ reason: string; attempt: number }` | The turn re-entered the model loop (e.g. tool re-resolution) |
| `stream.first_chunk` | `{ ttfbMs: number }` | First content delta arrived; `ttfbMs` is time-to-first-byte from invoke |
| `stream.completed` | `{ chunks: number; durationMs: number }` | The model stream finished |
Use `stage.transition` to reconstruct the per-turn lattice walk
(`anchor-plan → tool-loop → synthesize → post-turn`); a
`stage.lattice_violation` is your signal that execution diverged from the
advertised lattice (see [Architecture](/docs/architecture) for the
nine-pattern lattice the same `ALLOWED_EDGE_PATTERNS` constant defines).
`stream.first_chunk` / `turn.completed` give you TTFB and total latency
without a separate metrics pipeline.
These nine are *stream* events (`executeMessage` / `useChat`). The
per-LLM-call cost/latency signal — `model.called`
(`{ provider, model, callClass, inputTokens, outputTokens, costUSD,
latencyMs }`) — fires on the durable `runtime.events` bus, not this
stream, because cost attribution is a long-lived concern. Wire it straight
into a destination with the observe bridge:
```typescript
import { observeSink } from "@pleach/observe";
runtime.events.on("model.called", observeSink({ destinations: [store] }));
```
`model.called` maps one-to-one to an `ObserveRow`. See
[Observe](/docs/observe) for destinations, redaction, and sampling.
## Artifacts [#artifacts]
| Type | Payload | When |
| ------------------ | --------------------------- | ------------------------------------------------------- |
| `artifact.created` | `{ artifact: ArtifactRef }` | New artifact (file, image, structured doc) materialized |
## Checkpoints and sync [#checkpoints-and-sync]
| Type | Payload | When |
| -------------------- | ------------------------------------------------------------ | ---------------------------------------------------------------------------- |
| `checkpoint.created` | `{ checkpoint: Checkpoint }` | Stage boundary crossed; snapshot written |
| `sync.started` | `{ sessionId: string }` | Sync coordinator begins push/pull cycle |
| `sync.completed` | `{ sessionId: string }` | Sync cycle landed cleanly |
| `sync.conflict` | `{ sessionId, resolution: "local" \| "remote" \| "merged" }` | Version-vector conflict surfaced; `resolution` reports what the merger chose |
## Subagents [#subagents]
When `enableSubagentConcurrency: true` and the orchestrator spawns
parallel workers.
| Type | Payload | When |
| -------------------- | ------------------------------------------------------ | ----------------------------------------------------------------------------------- |
| `subagent.spawned` | `{ subagentId, task, specName?, context? }` | New subagent created |
| `subagent.progress` | `{ subagentId, progress, message?, activeToolName? }` | Subagent progress update |
| `subagent.completed` | `{ subagentId, content, toolsUsed, toolCallDetails? }` | Subagent finished |
| `subagent.failed` | `{ subagentId, error, terminalStatus? }` | Subagent errored; `terminalStatus` discriminates `cancelled` / `failed` / `timeout` |
Subagent-emitted events carry `namespace` so a UI can route them
into a separate panel.
## Sandbox [#sandbox]
For tools that allocate a sandboxed execution environment. The
event shapes below are stable contracts in `@pleach/core`.
`@pleach/sandbox@0.1.0` ships the `SandboxProvider` canonical
adapter and `@pleach/coding-agent@0.1.0` consumes these events —
its `start()` / `stop()` / `executeStep()` bodies all land at
the `0.1.0` cut; `executeStep()` throws
`PACK_270_D3_EXECUTE_STEP_NOT_STARTED_MESSAGE` only when called
before `start()`. Hosts can implement their own sandbox adapter
against the same event surface.
| Type | Payload | When |
| -------------------- | ----------------------------------------------------------- | ------------------------------------------------------------------ |
| `sandbox.created` | `{ sandboxId, tunnelUrl }` | Sandbox booted; `tunnelUrl` is reachable for preview |
| `sandbox.terminated` | `{ sandboxId, exportedFiles }` | Sandbox torn down; `exportedFiles` is the count of files saved |
| `result.offloaded` | `{ toolName, filename, sandboxPath, sizeBytes, rowCount? }` | Large tool result written to sandbox filesystem instead of inlined |
## Stream control [#stream-control]
| Type | Payload | When |
| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- |
| `stream.truncated` | `{ reason: "entropy_collapse" \| "phrase_loop" \| "substring_repetition"; partialLength }` | Output was truncated by a degeneration guard |
| `stream.reconnecting` | `{ attempt, maxRetries, disconnectedAt }` | SSE reconnect cycle in progress |
| `stream.disconnected` | `{ maxRetries }` | SSE reconnect budget exhausted |
| `context.summarized` | `{ originalMessageCount, compressedTokens, historyPath? }` | History was compacted into a summary before the next call |
| `content.correction` | `{ correctedContent, reason, materializerPreviousLen? }` | Post-stream fabrication / leakage guard replaced already-streamed content |
| `content.reset` | `{ reason: "providerError" \| "garbledNarration" \| "synthesisReplacementRetry" \| "synthesisHeaderOverlap"; failedModel?; clearedDeltaLength? }` | Drop streamed-so-far content; `clearedDeltaLength` tail-slices instead of full-clears when present |
For `content.correction`, the renderer should treat
`correctedContent` as the authoritative final text — replace the
last assistant message wholesale. A common case: the model streamed
a sentence claiming it had called `fetch_document` when no such
tool call landed in the turn; the fabrication guard fires
`content.correction` with `reason: "phantom_tool_reference"` and a
`correctedContent` payload that drops the false claim. The UI's
message buffer for that `messageId` is replaced wholesale, not
appended to.
## Provider cascade and LLM turn [#provider-cascade-and-llm-turn]
Emitted by the fallback executor and the graph LLM node — surface
per-model progress, model transitions, and token usage.
| Type | Payload | When |
| ------------------------ | -------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------- |
| `provider.cascade` | `{ model?, provider?, error?, failedModels?, attempt?, isForceSynthesis? }` | A model in the cascade failed; UI shows "(N failed)" |
| `provider.cascade.reset` | – | New graph turn started; cascade counter resets |
| `llm.turn` | `{ model, provider?, input_tokens?, output_tokens?, tool_call_count?, depth?, modelSwitchReason?, requestedModel? }` | LLM turn completed; carries per-turn model + token metadata |
`modelSwitchReason` discriminates `"provider_fallback"`,
`"zero_tool_recovery"`, `"forceSynthesis_escalation"`,
`"forceSynthesis_garble_cascade"`. Use `llm.turn.depth` to attribute
spend across subagent depth levels — `depth: 0` is the root turn,
`depth: 1` is a direct subagent, `depth: 2` is a grandchild, and the
hard ceiling is `SUBAGENT_LIMITS.maxDepth = 3` (see
[Subagents](/docs/subagents#subagent-limits-and-depth-tracking)). A
`GROUP BY depth` on the per-call ledger gives you "how much did
fan-out cost relative to the root turn."
## Domain events [#domain-events]
| Type | Payload | When |
| -------------- | --------------------------------------- | ----------------------------------------------- |
| `domain.event` | `{ domainType, domain, kind, payload }` | Plugin-namespaced event yielded onto the stream |
Plugins emit `domain.event` to surface arbitrary structured
events to the consumer without inventing a new variant per
plugin. The `domain` namespaces the plugin (e.g.
`compliance.audit`, `gateway.cost`), `kind` is the event kind,
`payload` carries the data.
## Step lifecycle [#step-lifecycle]
| Type | Payload | When |
| ------------ | ------------------ | --------------------------------------------------------- |
| `step.start` | `{ step: string }` | Named step (`anchor-plan`, `tool-loop-iter`, etc.) begins |
| `step.end` | `{ step: string }` | Step completes |
## Errors [#errors]
| Type | Payload | When |
| ------- | ---------------------------------- | ------------------------------------------- |
| `error` | `{ error: string; code?: string }` | Recoverable or terminal error in the stream |
The `code` field is the structured error code (1xxx–7xxx) — see
[Error codes](/docs/error-codes) for the catalog.
## Forwarding over SSE [#forwarding-over-sse]
The stream maps directly onto Server-Sent Events. The reference
Next.js route in the package looks roughly like:
```typescript
export async function POST(req: Request) {
const { sessionId, content } = await req.json();
const stream = new ReadableStream({
async start(controller) {
for await (const event of runtime.executeMessage(sessionId, content)) {
controller.enqueue(`data: ${JSON.stringify(event)}\n\n`);
}
controller.close();
},
});
return new Response(stream, {
headers: { "Content-Type": "text/event-stream" },
});
}
```
Each event becomes one SSE `data:` line. Clients re-hydrate the
union by parsing `JSON.parse(line)` and switching on `type`.
## Mode-filtered subscription [#mode-filtered-subscription]
`SessionRuntime` also exposes a non-iterator subscription that
filters by `StreamMode`. Use this for cross-cutting subscribers
that don't own the turn loop.
```typescript
const unsubscribe = runtime.subscribeToStream("messages", (event) => {
// only message.delta / message.complete / thinking.* land here
});
```
`"values"`, `"updates"`, `"messages"`, `"custom"`, `"debug"`, and
`"all"` are the legal modes — see [SessionRuntime → Stream
subscription](/docs/session-runtime#stream-subscription) for the
per-mode event sets.
## `content.delta` [#contentdelta]
`content.delta` is the canonical streaming-chunk event type. The
substrate emits one per token-or-chunk during model output for a
turn.
It pairs with the chunk-loop that folds deltas into a terminal
message at end-of-turn. See [Event log](/docs/event-log) §
`content.delta streaming chunks` for the persisted row shape, and
[Event log projections](/docs/event-log-projections) for the fold
that reconstructs the final message.
Consumers subscribing to `content.delta` should treat each chunk
as forward-only. Don't try to "undo" a delta — the terminal
message is the canonical reconstructed state.
## tool.execution span emit [#toolexecution-span-emit]
The substrate emits a `tool.execution` OTel span per tool
dispatch. The span brackets the same lifecycle moment as the
`tool.completed` stream event above.
The two are paired surfaces for the same dispatch: the span feeds
traces and observability backends; the stream event feeds
in-process subscribers rendering the turn.
See [OTel observability](/docs/otel-observability) for the full
span surface and attribute catalog.
## Where to go next [#where-to-go-next]