# Stream observers (/docs/plugins/stream-observers) `contributeStreamObservers` is the plugin slot for per-chunk inspection of provider seam output. Every plugin that registers an observer gets each inbound chunk in registration order, and each observer returns a verdict that the seam acts on before the next observer runs (and before the chunk reaches downstream graph nodes). It's the substrate's most-reached-for extension hook — redaction, dialect normalization, refusal detection, mid-stream metrics, prompt-leak detection, off-topic detection all fit the shape. ## The contract [#the-contract] ```typescript contributeStreamObservers?(): readonly StreamObserverRegistration[] interface StreamObserverRegistration { when: { callClass: CallClass | "*" }; factory: (factoryContext: ObserverFactoryContext) => StreamObserver; } interface StreamObserver { observerId: string; onChunk(chunk: StreamChunk, ctx: ObserverContext): ObserverVerdict; onEndInvocation?(result: InvocationResult, ctx: ObserverContext): void; } ``` **The factory is called fresh per seam invocation.** Per- invocation state (counters, rolling buffers, dedup sets) belongs inside the factory closure and does not leak across concurrent seam invocations on the same runtime. This is the seam-side half of the replay-determinism contract. ## Honest scope-limit: `onChunk` is SYNC ONLY [#honest-scope-limit-onchunk-is-sync-only] ```typescript // This is rejected by the contract type. onChunk(chunk, ctx): Promise { /* … */ } ``` There is no `Promise` overload. The reason is the replay-determinism story: an observer that returns a promise introduces non-determinism into the stream, and replay determinism is the load-bearing property the `@pleach/eval@0.1.0` and `@pleach/replay@0.1.0` SKUs are built on. If you need async work (an API call, a model call, a side-effect ship), use the `emit` verdict to fan the structured chunk out to a named channel where the consumer runs async out-of-band. Same goes for `onEndInvocation` — sync only, by the same property. ## The 6 verdicts [#the-6-verdicts] The verdict ladder is enumerated in the source as a discriminated union. The substrate exposes four widely-used verdicts plus two backpressure verdicts. | Verdict | Effect | | ---------------------------------------------------------------- | ----------------------------------------------------------- | | `{ kind: "continue" }` | Pass through unchanged. The observer-default. | | `{ kind: "amend", chunk }` | Replace chunk content 1:1 — strict, no multiplex. | | `{ kind: "emit", events }` | Pass through AND fan out a side-event onto a named channel. | | `{ kind: "stop", reason, chunkIndex, accumulatedContentLength }` | Stop the stream. Downstream reads the stop sentinel. | | `{ kind: "buffer" }` | Hold the chunk for later release (backpressure). | | `{ kind: "release", chunks }` | Emit previously buffered chunks. | `amend` being 1:1 is deliberate. A one-chunk-in, many-chunks-out observer would break the byte-replay property. Plugins that need fan-out emit named envelopes on a channel via `emit`, not extra stream chunks. Concretely: a redaction observer that replaces a span with `[REDACTED]` returns an `amend` verdict whose `chunk.text` is the cleaned string; the chunk count stays the same. A metrics observer that wants to ship a structured side-effect returns an `emit` verdict with an envelope on a named channel (e.g. `metrics`), and the main stream sees the original chunk pass through unchanged. ## Worked example: dialect normalization (`amend`) [#worked-example-dialect-normalization-amend] Some providers emit XML-flavored tool-call markers (`{...}`) or tuple-keyed JSON (`{0:"search", 1:{...}}`) instead of the harness's canonical `tool_call_delta` shape. A normalization observer detects the variants and substitutes a canonicalized chunk so downstream graph nodes see one shape regardless of provider quirks. ```javascript const XML_TOOL_RE = /([\s\S]*?)<\/tool>/g; const TUPLE_TOOL_RE = /\{\s*0:\s*["']([^"']+)["']\s*,\s*1:\s*(\{[^}]*\})\s*\}/g; function makeCanonicalToolCallDelta({ name, argsJson, observerId }) { return { kind: "tool_call_delta", payload: { name, args: argsJson, _amendedBy: observerId }, }; } function makeObserver(_factoryContext) { let chunkCount = 0; return { observerId: "example-dialect-normalization", onChunk(chunk, _ctx) { chunkCount += 1; if (chunk.kind !== "content_delta") return { kind: "continue" }; const text = extractText(chunk.payload); if (text === null) return { kind: "continue" }; const xmlMatch = XML_TOOL_RE.exec(text); XML_TOOL_RE.lastIndex = 0; if (xmlMatch !== null) { return { kind: "amend", chunk: makeCanonicalToolCallDelta({ name: xmlMatch[1], argsJson: xmlMatch[2].trim(), observerId: "example-dialect-normalization", }), }; } // …tuple fallback… return { kind: "continue" }; }, }; } export const dialectNormalizationPlugin = { name: "example-dialect-normalization", version: "1.0.0", contributeStreamObservers: () => [ { when: { callClass: "*" }, factory: makeObserver }, ], }; ``` The runnable version (with the tuple fallback, the duck-typed text extractor, and a `node --test` smoke driving it with synthetic chunks) lives at [`examples/observers/dialect-normalization/`](https://github.com/pleachhq/core/tree/main/examples/observers/dialect-normalization). **Things to notice:** * `factory: makeObserver` — the seam calls this once per invocation. `chunkCount` is per-invocation closure state. * `when: { callClass: "*" }` — fire on every call class. Narrow with `{ callClass: "synthesize" }` if your normalization is synthesis-specific. * The `_amendedBy: observerId` field tags the amended chunk for downstream debugging. The substrate doesn't require it; it's a load-bearing diagnostic in production observers. ## Worked example: stop on refusal pattern (`stop`) [#worked-example-stop-on-refusal-pattern-stop] Cancel a streaming generation as early as possible when the model emits a generic refusal so the host can fail-over to another provider, rewrite the prompt, or surface the refusal to the user without paying for the rest of the wasted tokens. ```javascript const DEFAULT_REFUSAL_RE = /\b(?:I cannot|I'm not able to|I am unable to|Sorry, I can'?t|I won'?t be able to)\s+(?:help|assist|comply|provide)\b/i; const ACCUMULATOR_CAP_BYTES = 16 * 1024; export function makeHaltObserver(options = {}) { const pattern = options.pattern ?? DEFAULT_REFUSAL_RE; const observerId = options.observerId ?? "example-halt-on-pattern"; return (_factoryContext) => { let chunkCount = 0; let accumulator = ""; return { observerId, onChunk(chunk, _ctx) { chunkCount += 1; if (chunk.kind !== "content_delta") return { kind: "continue" }; const text = extractText(chunk.payload); if (text === null) return { kind: "continue" }; accumulator += text; if (accumulator.length > ACCUMULATOR_CAP_BYTES) { accumulator = accumulator.slice(-ACCUMULATOR_CAP_BYTES); } const match = pattern.exec(accumulator); pattern.lastIndex = 0; if (match !== null) { return { kind: "stop", reason: `pattern matched: ${JSON.stringify(match[0])}`, chunkIndex: chunkCount, accumulatedContentLength: accumulator.length, }; } return { kind: "continue" }; }, }; }; } export const haltOnPatternPlugin = { name: "example-halt-on-pattern", version: "1.0.0", contributeStreamObservers: () => [ { when: { callClass: "synthesize" }, factory: makeHaltObserver() }, ], }; ``` The runnable version lives at [`examples/observers/halt-on-pattern/`](https://github.com/pleachhq/core/tree/main/examples/observers/halt-on-pattern). **Things to notice:** * **Memory cap** — `ACCUMULATOR_CAP_BYTES = 16 * 1024`. Generic refusals fire in the first few hundred bytes; longer windows blow memory if the stream degenerates. Production observers cap any rolling buffer they maintain. * **Pattern is configurable.** The default is a generic refusal detector. Swap for your own — off-topic detection, jailbreak detection, prompt-leak detection, etc. * **The seam owns abort semantics.** The observer just signals `stop`; the seam handles the cancel. Downstream recovery handlers position themselves at the stop point using `chunkIndex` + `accumulatedContentLength`. ## Per-`callClass` registration [#per-callclass-registration] The `when` clause narrows which seam invocations fire the observer: | `when.callClass` | Fires for | | ---------------- | ------------------------------------------------------------- | | `"synthesize"` | The final assistant-message synthesis seam | | `"reasoning"` | Mid-turn reasoning calls (chain-of-thought style) | | `"utility"` | Lightweight provider calls (re-ranking, classification, etc.) | | `"converse"` | Multi-turn conversational chunks | | `"*"` | Every seam — broad match | Narrowing to one call class is a real performance win when the observer's pattern only applies in one context. The dialect- normalization example uses `"*"` because canonicalization is universal; the halt-on-refusal example uses `"synthesize"` because that's where refusals surface. ## Per-invocation freshness — what the seam guarantees [#per-invocation-freshness--what-the-seam-guarantees] ```typescript factory: (factoryContext: ObserverFactoryContext) => StreamObserver; ``` The seam calls `factory(factoryContext)` once per seam invocation. The returned `StreamObserver` lives only for that invocation. Any state inside the factory closure — counters, rolling buffers, dedup sets, partial matches — is per-invocation by construction. Two concurrent invocations on the same runtime get two separate observer instances with two separate closures, no shared state. Authors should NOT capture state at plugin construction time and expect it to be per-invocation: ```typescript // WRONG — `chunkCount` is shared across every invocation. let chunkCount = 0; export const myPlugin = { contributeStreamObservers: () => [ { factory: () => ({ onChunk(chunk) { chunkCount += 1; /* … */ }, }), }, ], }; // RIGHT — `chunkCount` is per-invocation. export const myPlugin = { contributeStreamObservers: () => [ { factory: () => { let chunkCount = 0; return { onChunk(chunk) { chunkCount += 1; /* … */ }, }; }, }, ], }; ``` ## Composing multiple observers [#composing-multiple-observers] Multiple plugins (or one plugin registering multiple `StreamObserverRegistration` entries) compose in registration order. The first observer sees the original chunk; the second observer sees whatever the first amended (or skipped); and so on. A `stop` verdict from any observer terminates the stream for all subsequent observers. The canonical non-commuting pair is a PII-redaction observer and a metrics observer that records chunk lengths: register redaction first and metrics sees the redacted length; register metrics first and it sees the original length. Both orderings are legal; the construction site is the place to make the choice visible. ## Where to go next [#where-to-go-next]