refactor(runtime): move chunk interpretation into the agent adapter

src/replay/adapter.test.ts

@@ -1,6 +1,7 @@
 import { Effect } from "effect";
 import { describe, expect, test } from "bun:test";
 import { buildAgentStepEffect } from "../runtime/agent-step";
+import type { AgentAdapterYield } from "../runtime/agent-adapter";
 import { createCorpusReplayAdapter, createSlowFakeAdapter, loadCorpusBlocks } from "./adapter";
 
 const FULL_ROUND_TRIP_CORPUS = `${import.meta.dir}/../../test/corpus/run-1789308170212.ndjson`;
@@ -65,6 +66,29 @@ describe("createCorpusReplayAdapter", () => {
     expect(outcome.finalText.length).toBeGreaterThan(0);
     expect(outcome.runError).toBeUndefined();
   });
+
+  test("yields AgentAdapterYield items with signals extracted from opencode chunks", async () => {
+    const adapter = createCorpusReplayAdapter(FULL_ROUND_TRIP_CORPUS);
+    const stream = adapter.stream({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      abortController: new AbortController(),
+    });
+
+    const yields: AgentAdapterYield[] = [];
+    for await (const y of stream) {
+      yields.push(y as AgentAdapterYield);
+    }
+
+    expect(yields.length).toBe(39);
+    const sessionIdYield = yields.find((y) => y.signal?._tag === "sessionId");
+    expect(sessionIdYield).toBeDefined();
+    expect(sessionIdYield?.signal?.value).toBe("ses_f64ec04acffeJ0tjsHSkjAEqZF");
+    const textYields = yields.filter((y) => y.signal === undefined);
+    expect(textYields.length).toBeGreaterThan(0);
+  });
 });
 
 describe("createSlowFakeAdapter", () => {

src/replay/adapter.ts

@@ -3,6 +3,12 @@
  * adapter a corpus replayer") — the runtime code path is identical whether
  * chunks come from here or from live opencode; only this module swaps.
  *
+ * ADR 0012 §2: adapters yield `AgentAdapterYield` items (opaque chunk +
+ * optional signal). The corpus replay adapter extracts signals from recorded
+ * opencode chunks using `extractOpencodeSignal`, so existing corpus traces
+ * replay without modification. A second adapter could supply signals without
+ * imitating opencode chunk shapes at all.
+ *
  * `createCorpusReplayAdapter` replays a recorded NDJSON trace
  * (`test/corpus/*.ndjson`, one `{step, chunk}` line each). The traces were
  * captured one workflow step at a time, so consecutive lines sharing a
@@ -13,7 +19,12 @@
  */
 
 import { readFileSync } from "node:fs";
-import type { AgentAdapter, AgentAdapterOptions } from "../runtime/agent-adapter";
+import type {
+  AgentAdapter,
+  AgentAdapterOptions,
+  AgentAdapterYield,
+} from "../runtime/agent-adapter";
+import { extractOpencodeSignal } from "../runtime/opencode-adapter";
 
 export interface CorpusStepBlock {
   readonly step: string;
@@ -54,7 +65,7 @@
   let cursor = 0;
 
   return {
-    stream(_options: AgentAdapterOptions): AsyncIterable<unknown> {
+    stream(_options: AgentAdapterOptions): AsyncIterable<AgentAdapterYield> {
       const index = cursor;
       cursor += 1;
       const block = blocks[index];
@@ -64,9 +75,10 @@
         );
       }
       return {
         async *[Symbol.asyncIterator]() {
           for (const chunk of block.chunks) {
-            yield chunk;
+            const signal = extractOpencodeSignal(chunk);
+            yield signal !== undefined ? { chunk, signal } : { chunk };
           }
         },
       };
@@ -83,12 +95,12 @@
  */
 export function createSlowFakeAdapter(chunks: ReadonlyArray<unknown>, delayMs = 20): AgentAdapter {
   return {
-    stream(_options: AgentAdapterOptions): AsyncIterable<unknown> {
+    stream(_options: AgentAdapterOptions): AsyncIterable<AgentAdapterYield> {
       return {
         async *[Symbol.asyncIterator]() {
           for (const chunk of chunks) {
             await new Promise<void>((resolve) => setTimeout(resolve, delayMs));
-            yield chunk;
+            yield { chunk };
           }
         },
       };

src/runtime/agent-adapter.ts

@@ -3,6 +3,12 @@
  * actually produces a chunk stream. Live opencode for real runs, corpus
  * replay for `bun test` (STATUS.md phase 1: "make the fake adapter a corpus
  * replayer") — same runtime code path either way, only this function swaps.
+ *
+ * ADR 0012 §2: the adapter interprets its own chunk stream and yields the
+ * opaque chunk together with a normalized signal drawn from a small closed
+ * union. The runtime consumes signals directly; it never string-matches a
+ * vendor event name. A member exists in the union only when Factory has a
+ * field for it (`AgentStepFinished.sessionId`, `.output`, `.error`).
  */
 
 export interface AgentAdapterOptions {
@@ -15,6 +21,29 @@ export interface AgentAdapterOptions {
   readonly abortController: AbortController;
 }
 
+/**
+ * The normalized signal union (ADR 0012 §2). Each member populates one
+ * field on `AgentStepFinished`:
+ * - `sessionId` → `AgentStepFinished.sessionId`
+ * - `structuredOutput` → `AgentStepFinished.output` (via `resolveOutput`)
+ * - `runError` → `AgentStepFinished.error`
+ */
+export type AgentSignal =
+  | { readonly _tag: "sessionId"; readonly value: string }
+  | { readonly _tag: "structuredOutput"; readonly value: unknown }
+  | { readonly _tag: "runError"; readonly value: string };
+
+/**
+ * One item yielded by an adapter: the opaque AG-UI chunk (unchanged) plus
+ * an optional signal. The chunk is forwarded verbatim to `AgentChunk` events
+ * and to the SPA's `StreamProcessor`. The signal is consumed by the runtime
+ * to populate `AgentStepFinished` fields.
+ */
+export interface AgentAdapterYield {
+  readonly chunk: unknown;
+  readonly signal?: AgentSignal;
+}
+
 export interface AgentAdapter {
-  stream(options: AgentAdapterOptions): AsyncIterable<unknown>;
+  stream(options: AgentAdapterOptions): AsyncIterable<AgentAdapterYield>;
 }

src/runtime/agent-step.test.ts

@@ -0,0 +1,137 @@
+import { Effect } from "effect";
+import { describe, expect, test } from "bun:test";
+import type { AgentAdapterYield, AgentSignal } from "./agent-adapter";
+import { buildAgentStepEffect } from "./agent-step";
+
+function makeYield(chunk: unknown, signal?: AgentSignal): AgentAdapterYield {
+  return signal !== undefined ? { chunk, signal } : { chunk };
+}
+
+function signalAdapter(yields: ReadonlyArray<AgentAdapterYield>) {
+  return {
+    stream() {
+      return {
+        async *[Symbol.asyncIterator]() {
+          for (const y of yields) yield y;
+        },
+      };
+    },
+  };
+}
+
+describe("buildAgentStepEffect signal extraction", () => {
+  test("extracts sessionId from a sessionId signal", async () => {
+    const adapter = signalAdapter([
+      makeYield({ type: "TEXT_MESSAGE_START" }),
+      makeYield({ type: "TEXT_MESSAGE_CONTENT", delta: "hello" }),
+      makeYield({ type: "TEXT_MESSAGE_END" }),
+      makeYield(
+        { type: "CUSTOM", name: "opencode.session-id", value: { sessionId: "ses_abc" } },
+        { _tag: "sessionId", value: "ses_abc" },
+      ),
+    ]);
+
+    const handle = buildAgentStepEffect({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      adapter,
+      onChunk: () => {},
+    });
+
+    const outcome = await Effect.runPromise(handle.effect);
+    expect(outcome.sessionId).toBe("ses_abc");
+    expect(outcome.finalText).toBe("hello");
+    expect(outcome.chunkCount).toBe(4);
+  });
+
+  test("extracts structuredOutput from a structuredOutput signal", async () => {
+    const outputObject = { title: "test", body: "content" };
+    const adapter = signalAdapter([
+      makeYield({ type: "TEXT_MESSAGE_START" }),
+      makeYield({ type: "TEXT_MESSAGE_CONTENT", delta: "done" }),
+      makeYield({ type: "TEXT_MESSAGE_END" }),
+      makeYield(
+        { type: "CUSTOM", name: "structured-output.complete", value: { object: outputObject } },
+        { _tag: "structuredOutput", value: outputObject },
+      ),
+    ]);
+
+    const handle = buildAgentStepEffect({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      adapter,
+      onChunk: () => {},
+    });
+
+    const outcome = await Effect.runPromise(handle.effect);
+    expect(outcome.structuredOutput).toEqual(outputObject);
+  });
+
+  test("extracts runError from a runError signal", async () => {
+    const adapter = signalAdapter([
+      makeYield(
+        { type: "RUN_ERROR", message: "something broke" },
+        { _tag: "runError", value: "something broke" },
+      ),
+    ]);
+
+    const handle = buildAgentStepEffect({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      adapter,
+      onChunk: () => {},
+    });
+
+    const outcome = await Effect.runPromise(handle.effect);
+    expect(outcome.runError).toBe("something broke");
+  });
+
+  test("onChunk receives the raw opaque chunk, not the signal wrapper", async () => {
+    const rawChunk = { type: "TEXT_MESSAGE_START" };
+    const adapter = signalAdapter([makeYield(rawChunk)]);
+
+    const chunks: Array<unknown> = [];
+    const handle = buildAgentStepEffect({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      adapter,
+      onChunk: (chunk) => chunks.push(chunk),
+    });
+
+    await Effect.runPromise(handle.effect);
+    expect(chunks).toHaveLength(1);
+    expect(chunks[0]).toBe(rawChunk);
+  });
+
+  test("yields without signals pass through without error", async () => {
+    const adapter = signalAdapter([
+      makeYield({ type: "TEXT_MESSAGE_START" }),
+      makeYield({ type: "TEXT_MESSAGE_CONTENT", delta: "no signals here" }),
+      makeYield({ type: "TEXT_MESSAGE_END" }),
+    ]);
+
+    const handle = buildAgentStepEffect({
+      threadId: "t",
+      dir: "/tmp",
+      model: "m",
+      prompt: "p",
+      adapter,
+      onChunk: () => {},
+    });
+
+    const outcome = await Effect.runPromise(handle.effect);
+    expect(outcome.chunkCount).toBe(3);
+    expect(outcome.finalText).toBe("no signals here");
+    expect(outcome.sessionId).toBeUndefined();
+    expect(outcome.structuredOutput).toBeUndefined();
+    expect(outcome.runError).toBeUndefined();
+  });
+});

src/runtime/agent-step.ts

@@ -5,6 +5,11 @@
  * §5, D17): the boundary wiring is unchanged, only the chunk source and the
  * bookkeeping surface (now `ctx.agent`'s granular result, ADR 0002 §2) moved.
  *
+ * ADR 0012 §2: the runtime consumes `AgentSignal`s from the adapter and never
+ * string-matches vendor event names. AG-UI standard types (`TEXT_MESSAGE_*`)
+ * are still interpreted here for `finalText` accumulation — these are part
+ * of the open AG-UI protocol, not vendor-specific.
+ *
  * WHY THE EXPLICIT `abortController.abort()` IS NEEDED (0a-1/0a-2 findings):
  * closing the IO stream does not terminate the opencode process; only an
  * explicit abort does, and even that is indirect — `abort()` fires the
@@ -18,7 +23,7 @@
 
 import { Effect, Schema, Stream } from "effect";
 import type { AgentStepUsage } from "../events";
-import type { AgentAdapter } from "./agent-adapter";
+import type { AgentAdapter, AgentAdapterYield } from "./agent-adapter";
 
 /**
  * Pull the four token counts out of a `RUN_FINISHED.usage` object.
@@ -120,28 +125,33 @@ export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStep
   // Plain closure mutation (not a `Ref`) is fine: this Effect never runs
   // concurrently with itself, and the callback always runs on the same
   // single-threaded event loop turn (mirrors the spike's finding exactly).
-  const processed = Stream.mapEffect(rawStream, (chunk) =>
+  const processed = Stream.mapEffect(rawStream, (yieldItem: AgentAdapterYield) =>
     Effect.sync(() => {
       partial.chunkCount += 1;
-      options.onChunk(chunk);
+      options.onChunk(yieldItem.chunk);
+
+      if (yieldItem.signal !== undefined) {
+        switch (yieldItem.signal._tag) {
+          case "sessionId":
+            partial.sessionId = yieldItem.signal.value;
+            break;
+          case "structuredOutput":
+            structuredOutput = yieldItem.signal.value;
+            break;
+          case "runError":
+            runError = yieldItem.signal.value;
+            break;
+        }
+      }
 
-      const record = chunk as {
+      const record = yieldItem.chunk as {
         type?: unknown;
-        name?: unknown;
-        value?: unknown;
         delta?: unknown;
-        message?: unknown;
         usage?: unknown;
       };
 
-      if (record.type === "CUSTOM" && typeof record.name === "string") {
-        if (record.name === "structured-output.complete") {
-          const value = record.value as { object?: unknown } | undefined;
-          structuredOutput = value?.object;
-        } else if (record.name === "opencode.session-id") {
-          const value = record.value as { sessionId?: unknown } | undefined;
-          if (typeof value?.sessionId === "string") partial.sessionId = value.sessionId;
-        }
+      if (record.type === "RUN_FINISHED") {
+        partial.usage = readUsage(record.usage);
       }
 
       if (record.type === "TEXT_MESSAGE_START") {
@@ -156,14 +166,9 @@ export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStep
           partial.finalText = currentMessageBuffer;
         }
         currentMessageBuffer = undefined;
-      } else if (record.type === "RUN_FINISHED") {
-        partial.usage = readUsage(record.usage);
-      } else if (record.type === "RUN_ERROR") {
-        const message = record.message;
-        runError = typeof message === "string" ? message : JSON.stringify(chunk);
       }
 
-      return chunk;
+      return yieldItem;
     }),
   );
 

src/runtime/agent-step.usage.test.ts

@@ -15,6 +15,7 @@
 import { describe, expect, test } from "bun:test";
 import { Effect } from "effect";
 import { agentStepContextTokens } from "../events";
+import type { AgentAdapterYield } from "./agent-adapter";
 import { loadCorpusBlocks } from "../replay/adapter";
 import { buildAgentStepEffect } from "./agent-step";
 
@@ -27,8 +28,10 @@ async function runBlock(chunks: ReadonlyArray<unknown>) {
     model: "model",
     prompt: "prompt",
     adapter: {
-      async *stream() {
-        yield* chunks;
+      async *stream(): AsyncGenerator<AgentAdapterYield> {
+        for (const chunk of chunks) {
+          yield { chunk };
+        }
       },
     },
     onChunk: () => {},