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

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

// This is rejected by the contract type.
onChunk(chunk, ctx): Promise<ObserverVerdict> { /* … */ }

There is no Promise<ObserverVerdict> 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 verdict ladder is enumerated in the source as a discriminated union. The substrate exposes four widely-used verdicts plus two backpressure verdicts.

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)

Some providers emit XML-flavored tool-call markers (<tool name="search">{...}</tool>) 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.

const XML_TOOL_RE = /<tool\s+name=["']([^"']+)["']\s*>([\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/.

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)

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.

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/.

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

The when clause narrows which seam invocations fire the observer:

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

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:

// 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

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