fix(opencode): recover turns after event stream drops (#3395)

* fix(opencode): recover turns after event stream drops

Own the global event transport for the lifetime of the OpenCode server generation instead of coupling it to individual turns. Reconcile scoped provider state after reconnect so transient EOF, errors, and stalls do not falsely fail healthy foreground or delegated work.

* fix(opencode): settle recovery lifecycle edges

Retry only failed post-gap snapshots, preserve terminal child evidence when status is unavailable, and treat provider process exit as an authoritative stop boundary.

* fix(opencode): keep gap repair snapshot-scoped

Refresh child and blocking-request state once per reconnect, retry only rejected root snapshot reads, and settle a successful snapshot without the active dispatch instead of polling transcripts.

* fix(opencode): keep live ingress ahead of repair retries

Schedule failed snapshot retries back onto the serialized session ingress so newly arrived live terminal events run first. Keep retry backoff scoped to root reads and invalidate stale retries across later reconnects.

* fix(opencode): invalidate stale gap repairs between turns

A delayed recovery retry could inspect a replacement turn after the original turn completed. Turn creation now invalidates older repair work, and the slash-command timeout regression drives the generation-owned event source used by sessions.

* fix(opencode): bind gap repair to its turn

A queued or in-flight repair could resume after its original turn ended and read a replacement turn's mutable dispatch state. Recovery now captures its boundary and revalidates the repair revision after provider awaits and while materializing snapshots.

* fix(opencode): preserve child status on failed recovery

packages/server/src/server/agent/providers/opencode-agent.real.e2e.test.ts

@@ -0,0 +1,515 @@
+import { execFileSync, spawn, type ChildProcess } from "node:child_process";
+import {
+  copyFileSync,
+  createWriteStream,
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  readdirSync,
+  rmSync,
+  writeFileSync,
+} from "node:fs";
+import { createServer, request as httpRequest, type ServerResponse } from "node:http";
+import net from "node:net";
+import os from "node:os";
+import path from "node:path";
+import pino from "pino";
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+
+import type { AgentStreamEvent } from "../agent-sdk-types.js";
+import { OpenCodeAgentClient } from "./opencode-agent.js";
+import { OpenCodeServerManager } from "./opencode/server-manager.js";
+
+const MODEL = "openai/gpt-5.4";
+const NORMAL_SENTINEL = "PASEO_NORMAL_RECOVERY_SENTINEL_7F31";
+const ROOT_SENTINEL = "PASEO_ROOT_RECOVERY_SENTINEL_8C42";
+const CHILD_SENTINEL = "PASEO_CHILD_RECOVERY_SENTINEL_5A19";
+const PARENT_SENTINEL = "PASEO_PARENT_RECOVERY_SENTINEL_6B20";
+const TIMEOUT_MS = 240_000;
+
+describe.sequential("OpenCode real event recovery", () => {
+  let harness: Awaited<ReturnType<typeof createRealHarness>>;
+
+  beforeAll(async () => {
+    harness = await createRealHarness();
+  }, 30_000);
+
+  afterAll(async () => {
+    await harness?.close();
+  }, 30_000);
+
+  test("removes disposable runtime state when auth setup fails", () => {
+    const observed = { runtimeDir: "" };
+    expect(() => createDisposableRuntime(failAuthCopy, recordRuntimeDirectory(observed))).toThrow(
+      "induced auth copy failure",
+    );
+    expect(observed.runtimeDir).not.toBe("");
+    expect(existsSync(observed.runtimeDir)).toBe(false);
+  });
+
+  test("removes disposable runtime state when the creation hook fails", () => {
+    const observed = { runtimeDir: "" };
+    expect(() => createDisposableRuntime(copyFileSync, failRuntimeCreation(observed))).toThrow(
+      "induced runtime creation failure",
+    );
+    expect(observed.runtimeDir).not.toBe("");
+    expect(existsSync(observed.runtimeDir)).toBe(false);
+  });
+
+  test(
+    "normal foreground completes once",
+    async () => {
+      const result = await harness.runScenario(
+        "normal",
+        `Respond with exactly ${NORMAL_SENTINEL} and no other text.`,
+      );
+
+      assertParentResult(result.events, NORMAL_SENTINEL);
+      harness.writeSummary("normal", summarize(result));
+    },
+    TIMEOUT_MS,
+  );
+
+  test(
+    "recovers a root turn after cutting the active global event stream",
+    async () => {
+      const result = await harness.runScenario(
+        "root-recovery",
+        [
+          "You must use the bash tool exactly once to run: sleep 4",
+          `After the tool completes, respond with exactly ${ROOT_SENTINEL} and no other text.`,
+        ].join("\n"),
+        (event) => isRunningToolEvent(event),
+      );
+
+      assertParentResult(result.events, ROOT_SENTINEL);
+      expect(result.activeCutTrigger).toBe(true);
+      expect(result.pidAfter).toBe(result.pidBefore);
+      expect(result.cuts).toBe(1);
+      expect(result.connectionsAfter - result.connectionsBefore).toBe(1);
+      harness.writeSummary("root-recovery", summarize(result));
+    },
+    TIMEOUT_MS,
+  );
+
+  test(
+    "recovers an active delegated child and its parent once",
+    async () => {
+      const result = await harness.runScenario(
+        "child-recovery",
+        [
+          "You must delegate this work using the task tool with subagent type general.",
+          `Tell the child to use bash to run sleep 4, then return exactly ${CHILD_SENTINEL}.`,
+          `After the child finishes, respond with exactly ${PARENT_SENTINEL} and no other text.`,
+        ].join("\n"),
+        (event) => isRunningChildEvent(event),
+      );
+
+      assertParentResult(result.events, PARENT_SENTINEL, true);
+      expect(result.activeCutTrigger).toBe(true);
+      expect(result.pidAfter).toBe(result.pidBefore);
+      expect(result.cuts).toBe(1);
+      expect(result.connectionsAfter - result.connectionsBefore).toBe(1);
+      const childOutput = result.events
+        .flatMap((event) => {
+          if (
+            event.type !== "provider_subagent" ||
+            event.event.type !== "timeline" ||
+            event.event.item.type !== "assistant_message"
+          ) {
+            return [];
+          }
+          return [event.event.item.text];
+        })
+        .join("");
+      expect(childOutput.trim()).toBe(CHILD_SENTINEL);
+      expect(childOutput.split(CHILD_SENTINEL)).toHaveLength(2);
+      expect(
+        result.events.filter(
+          (event) =>
+            event.type === "provider_subagent" &&
+            event.event.type === "upsert" &&
+            event.event.status === "completed",
+        ),
+      ).toHaveLength(1);
+      expect(result.events.some((event) => event.type === "provider_subagent")).toBe(true);
+      harness.writeSummary("child-recovery", summarize(result));
+    },
+    TIMEOUT_MS,
+  );
+});
+
+async function createRealHarness() {
+  const artifactDir = mkdtempSync(path.join(os.tmpdir(), "paseo-opencode-recovery-"));
+  const runtime = createDisposableRuntime();
+  const { runtimeDir, home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace } =
+    runtime;
+
+  let setupManager: OpenCodeServerManager | null = null;
+  let setupProxy: Awaited<ReturnType<typeof createCutProxy>> | null = null;
+  try {
+    const upstreamPort = await reservePort();
+    const proxy = await createCutProxy(upstreamPort, path.join(artifactDir, "proxy.jsonl"));
+    setupProxy = proxy;
+    const isolatedEnv = {
+      ...process.env,
+      HOME: home,
+      PASEO_HOME: paseoHome,
+      XDG_CONFIG_HOME: xdgConfig,
+      XDG_DATA_HOME: xdgData,
+      XDG_CACHE_HOME: xdgCache,
+      XDG_STATE_HOME: xdgState,
+      OPENCODE_DISABLE_AUTOUPDATE: "1",
+      OPENCODE_DISABLE_AUTO_UPDATE: "1",
+    };
+    const openCodeVersion = execFileSync("opencode", ["--version"], {
+      env: isolatedEnv,
+      encoding: "utf8",
+    }).trim();
+    expect(openCodeVersion).toBe("1.14.33");
+    const selectedModels = new Set<string>();
+    let child: ChildProcess | null = null;
+    const manager = new OpenCodeServerManager({
+      logger: pino({ level: "silent" }),
+      portAllocator: async () => proxy.port,
+      resolveCommandPrefix: async () => ({ command: "opencode", args: [] }),
+      resolveHomeDir: () => home,
+      spawnServerProcess: () => {
+        child = spawn("opencode", ["serve", "--port", String(upstreamPort)], {
+          cwd: workspace,
+          env: isolatedEnv,
+          stdio: ["ignore", "pipe", "pipe"],
+        });
+        child.stdout?.pipe(createWriteStream(path.join(artifactDir, "opencode.stdout.log")));
+        child.stderr?.pipe(createWriteStream(path.join(artifactDir, "opencode.stderr.log")));
+        return child;
+      },
+    });
+    setupManager = manager;
+    const lease = await manager.acquireCurrent();
+    const client = new OpenCodeAgentClient(pino({ level: "silent" }), undefined, {
+      serverManager: manager,
+    });
+
+    return {
+      artifactDir,
+      manager,
+      proxy,
+      writeSummary(name: string, summary: unknown) {
+        writeFileSync(
+          path.join(artifactDir, `${name}.summary.json`),
+          JSON.stringify(summary, null, 2),
+        );
+      },
+      async runScenario(
+        name: string,
+        prompt: string,
+        cutWhen?: (event: AgentStreamEvent) => boolean,
+      ) {
+        const sessionConfig = {
+          provider: "opencode",
+          cwd: workspace,
+          model: MODEL,
+          modeId: "build",
+          featureValues: { auto_accept: true },
+        } as const;
+        selectedModels.add(sessionConfig.model);
+        const session = await client.createSession(sessionConfig);
+        const events: AgentStreamEvent[] = [];
+        const eventPath = path.join(artifactDir, `${name}.events.jsonl`);
+        const pidBefore = child?.pid;
+        const connectionsBefore = proxy.connections;
+        const cutsBefore = proxy.cuts;
+        let activeCutTrigger = false;
+        session.subscribe((event) => {
+          events.push(event);
+          writeFileSync(eventPath, `${JSON.stringify(event)}\n`, { flag: "a" });
+          if (!activeCutTrigger && cutWhen?.(event)) {
+            activeCutTrigger = true;
+            proxy.cutActiveGlobalEvent();
+          }
+        });
+        try {
+          await session.startTurn(prompt);
+          await waitForTerminal(events, TIMEOUT_MS - 10_000);
+          return {
+            events,
+            activeCutTrigger,
+            pidBefore,
+            pidAfter: child?.pid,
+            cuts: proxy.cuts - cutsBefore,
+            connectionsBefore,
+            connectionsAfter: proxy.connections,
+          };
+        } finally {
+          await session.close().catch(() => undefined);
+        }
+      },
+      async close() {
+        try {
+          await lease.release().catch(() => undefined);
+          await manager.shutdown().catch(() => undefined);
+          await proxy.close();
+          writeFileSync(
+            path.join(artifactDir, "run.summary.json"),
+            JSON.stringify(
+              { pid: child?.pid, cuts: proxy.cuts, connections: proxy.connections },
+              null,
+              2,
+            ),
+          );
+          writeFileSync(
+            path.join(artifactDir, "run-metadata.summary.json"),
+            JSON.stringify(
+              {
+                openCodeVersion,
+                models: [...selectedModels],
+                pid: child?.pid,
+                scenarios: ["normal", "root-recovery", "child-recovery"],
+              },
+              null,
+              2,
+            ),
+          );
+        } finally {
+          rmSync(runtimeDir, { recursive: true, force: true });
+        }
+        const credentialScan = scanRetainedArtifacts(artifactDir);
+        writeFileSync(
+          path.join(artifactDir, "credential-scan.summary.json"),
+          JSON.stringify(credentialScan, null, 2),
+        );
+        expect(credentialScan.credentialPatternFiles).toBe(0);
+      },
+    };
+  } catch (error) {
+    await setupManager?.shutdown().catch(() => undefined);
+    await setupProxy?.close().catch(() => undefined);
+    rmSync(runtimeDir, { recursive: true, force: true });
+    throw error;
+  }
+}
+
+function createDisposableRuntime(
+  copyAuth: typeof copyFileSync = copyFileSync,
+  onCreate?: (runtimeDir: string) => void,
+) {
+  let runtimeDir: string | null = null;
+  try {
+    runtimeDir = mkdtempSync(path.join(os.tmpdir(), "paseo-opencode-runtime-"));
+    onCreate?.(runtimeDir);
+    const home = path.join(runtimeDir, "home");
+    const paseoHome = path.join(runtimeDir, "paseo-home");
+    const xdgConfig = path.join(runtimeDir, "xdg-config");
+    const xdgData = path.join(runtimeDir, "xdg-data");
+    const xdgCache = path.join(runtimeDir, "xdg-cache");
+    const xdgState = path.join(runtimeDir, "xdg-state");
+    const workspace = path.join(runtimeDir, "workspace");
+    for (const directory of [home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace]) {
+      mkdirSync(directory, { recursive: true });
+    }
+    const isolatedAuthDirectory = path.join(xdgData, "opencode");
+    mkdirSync(isolatedAuthDirectory, { recursive: true });
+    copyAuth(
+      path.join(os.homedir(), ".local", "share", "opencode", "auth.json"),
+      path.join(isolatedAuthDirectory, "auth.json"),
+    );
+    return { runtimeDir, home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace };
+  } catch (error) {
+    if (runtimeDir) {
+      rmSync(runtimeDir, { recursive: true, force: true });
+    }
+    throw error;
+  }
+}
+
+function failAuthCopy(): never {
+  throw new Error("induced auth copy failure");
+}
+
+function recordRuntimeDirectory(observed: { runtimeDir: string }) {
+  return (runtimeDir: string) => {
+    observed.runtimeDir = runtimeDir;
+  };
+}
+
+function failRuntimeCreation(observed: { runtimeDir: string }) {
+  return (runtimeDir: string): never => {
+    observed.runtimeDir = runtimeDir;
+    throw new Error("induced runtime creation failure");
+  };
+}
+
+function scanRetainedArtifacts(artifactDir: string) {
+  const credentialFilename = /(?:auth|credential|secret|token)/i;
+  const credentialContent =
+    /(?:authorization\s*[:=]\s*["']?(?:bearer|basic)|["'](?:access_token|refresh_token|api[_-]?key|client_secret)["']\s*:)/i;
+  const files = readdirSync(artifactDir, { withFileTypes: true })
+    .filter((entry) => entry.isFile())
+    .map((entry) => entry.name)
+    .filter(
+      (name) =>
+        credentialFilename.test(name) ||
+        credentialContent.test(readFileSync(path.join(artifactDir, name), "utf8")),
+    )
+    .sort();
+  return { credentialPatternFiles: files.length, files };
+}
+
+async function createCutProxy(upstreamPort: number, artifactPath: string) {
+  let connections = 0;
+  let cuts = 0;
+  let activeGlobal: { upstream: NodeJS.ReadableStream; downstream: ServerResponse } | null = null;
+  const record = (value: unknown) =>
+    writeFileSync(artifactPath, `${JSON.stringify({ at: Date.now(), ...value })}\n`, { flag: "a" });
+  const server = createServer((incoming, outgoing) => {
+    const globalEvent = incoming.url?.startsWith("/global/event") === true;
+    if (globalEvent) {
+      connections += 1;
+      record({ type: "global-connect", connection: connections });
+    }
+    const upstream = httpRequest(
+      {
+        host: "127.0.0.1",
+        port: upstreamPort,
+        method: incoming.method,
+        path: incoming.url,
+        headers: incoming.headers,
+      },
+      (response) => {
+        outgoing.writeHead(response.statusCode ?? 502, response.headers);
+        if (globalEvent) {
+          activeGlobal = { upstream: response, downstream: outgoing };
+          let buffer = "";
+          response.on("data", (chunk: Buffer) => {
+            buffer += chunk.toString();
+            for (;;) {
+              const boundary = buffer.indexOf("\n\n");
+              if (boundary < 0) break;
+              const raw = buffer.slice(0, boundary);
+              buffer = buffer.slice(boundary + 2);
+              record({ type: "global-record", connection: connections, raw });
+            }
+          });
+        }
+        response.pipe(outgoing);
+        response.on("close", () => {
+          if (activeGlobal?.upstream === response) activeGlobal = null;
+        });
+      },
+    );
+    upstream.on("error", (error) => {
+      record({ type: "proxy-upstream-error", message: error.message });
+      outgoing.destroy();
+    });
+    incoming.pipe(upstream);
+  });
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Proxy did not bind");
+  return {
+    port: address.port,
+    get connections() {
+      return connections;
+    },
+    get cuts() {
+      return cuts;
+    },
+    cutActiveGlobalEvent() {
+      if (!activeGlobal) throw new Error("No active global event connection to cut");
+      cuts += 1;
+      record({ type: "global-cut", connection: connections, cut: cuts });
+      activeGlobal.upstream.destroy();
+      activeGlobal.downstream.destroy();
+      activeGlobal = null;
+    },
+    close: async () => new Promise<void>((resolve) => server.close(() => resolve())),
+  };
+}
+
+function isRunningToolEvent(event: AgentStreamEvent): boolean {
+  return (
+    event.type === "timeline" &&
+    event.item.type === "tool_call" &&
+    (event.item.status === "running" || event.item.status === "pending")
+  );
+}
+
+function isRunningChildEvent(event: AgentStreamEvent): boolean {
+  return (
+    event.type === "provider_subagent" &&
+    event.event.type === "upsert" &&
+    event.event.status === "running"
+  );
+}
+
+function assertParentResult(
+  events: AgentStreamEvent[],
+  sentinel: string,
+  assertFinalMessageOnly = false,
+): void {
+  expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1);
+  expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1);
+  expect(events.filter((event) => event.type === "turn_failed")).toHaveLength(0);
+  expect(events.filter((event) => event.type === "turn_canceled")).toHaveLength(0);
+  const messages = new Map<string, string>();
+  for (const event of events) {
+    if (event.type !== "timeline" || event.item.type !== "assistant_message") continue;
+    messages.set(
+      event.item.messageId,
+      `${messages.get(event.item.messageId) ?? ""}${event.item.text}`,
+    );
+  }
+  const output = Array.from(messages.values()).join("");
+  const assertedOutput = assertFinalMessageOnly ? Array.from(messages.values()).at(-1) : output;
+  expect(assertedOutput?.trim()).toBe(sentinel);
+  expect(output.split(sentinel)).toHaveLength(2);
+}
+
+function summarize(result: {
+  events: AgentStreamEvent[];
+  pidBefore?: number;
+  pidAfter?: number;
+  cuts: number;
+  connectionsBefore: number;
+  connectionsAfter: number;
+}) {
+  return {
+    pidBefore: result.pidBefore,
+    pidAfter: result.pidAfter,
+    cuts: result.cuts,
+    connections: result.connectionsAfter - result.connectionsBefore,
+    started: result.events.filter((event) => event.type === "turn_started").length,
+    completed: result.events.filter((event) => event.type === "turn_completed").length,
+    failed: result.events.filter((event) => event.type === "turn_failed").length,
+    canceled: result.events.filter((event) => event.type === "turn_canceled").length,
+  };
+}
+
+async function waitForTerminal(events: AgentStreamEvent[], timeoutMs: number): Promise<void> {
+  const started = Date.now();
+  while (Date.now() - started < timeoutMs) {
+    if (
+      events.some(
+        (event) =>
+          event.type === "turn_completed" ||
+          event.type === "turn_failed" ||
+          event.type === "turn_canceled",
+      )
+    ) {
+      return;
+    }
+    await new Promise<void>((resolve) => setTimeout(resolve, 100));
+  }
+  throw new Error(`Timed out after ${timeoutMs}ms waiting for terminal event`);
+}
+
+async function reservePort(): Promise<number> {
+  const server = net.createServer();
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Port reservation failed");
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+  return address.port;
+}

packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts

@@ -0,0 +1,457 @@
+import type { ChildProcess } from "node:child_process";
+import { EventEmitter } from "node:events";
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import { createOpencodeClient } from "@opencode-ai/sdk/v2/client";
+import { expect, test } from "vitest";
+
+import { createTestLogger } from "../../../test-utils/test-logger.js";
+import type { AgentStreamEvent } from "../agent-sdk-types.js";
+import { OpenCodeAgentClient } from "./opencode-agent.js";
+import {
+  OpenCodeEventConsumer,
+  type OpenCodeEventConsumerTiming,
+} from "./opencode/event-consumer.js";
+import { OpenCodeServerManager } from "./opencode/server-manager.js";
+
+test("dispatches ascending OpenCode message identifiers through the public provider path", async () => {
+  const upstream = await createRecoveryUpstream();
+  const fixture = await createPublicRecoverySession(upstream, new RecoveryTiming());
+  const observed: AgentStreamEvent[] = [];
+  fixture.session.subscribe((event) => observed.push(event));
+  await upstream.connected(1);
+  upstream.send(0, connectedRecord());
+
+  await fixture.session.startTurn("first");
+  await upstream.dispatched(1);
+  upstream.send(0, idleRecord());
+  await eventually(() =>
+    expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1),
+  );
+  await fixture.session.startTurn("second");
+  const ids = await upstream.dispatched(2);
+
+  expect(ids).toHaveLength(2);
+  expect(ids[0]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/);
+  expect(ids[1]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/);
+  expect((ids[0] as string) < (ids[1] as string)).toBe(true);
+
+  await fixture.session.close();
+  await fixture.manager.shutdown();
+  await upstream.close();
+});
+
+test("recovers missed output after EOF without failing the active turn", async () => {
+  const upstream = await createRecoveryUpstream();
+  const timing = new RecoveryTiming();
+  const fixture = await createPublicRecoverySession(upstream, timing);
+  const { session } = fixture;
+  const observed: AgentStreamEvent[] = [];
+  session.subscribe((event) => observed.push(event));
+
+  await upstream.connected(1);
+  upstream.send(0, connectedRecord());
+  await session.startTurn("hello");
+  const [dispatchId] = await upstream.dispatched();
+  expect(upstream.messageReads()).toBe(0);
+  upstream.setRecoveredMessages(dispatchId as string);
+  upstream.end(0);
+  await timing.waiting();
+  timing.advance();
+  await upstream.connected(2);
+  upstream.send(1, connectedRecord());
+
+  await eventually(() => expect(observed.map((event) => event.type)).toContain("turn_completed"));
+  expect(observed.filter((event) => event.type === "turn_started")).toHaveLength(1);
+  expect(observed.filter((event) => event.type === "turn_failed")).toHaveLength(0);
+  const assistantTexts = observed.flatMap((event) =>
+    event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [],
+  );
+  expect(assistantTexts).toHaveLength(52);
+  expect(assistantTexts.at(-1)).toBe("recovered output");
+  expect(assistantTexts).not.toContain("before boundary");
+
+  await session.close();
+  await fixture.manager.shutdown();
+  await upstream.close();
+});
+
+test("orders queued live deltas behind an incomplete recovery snapshot and suppresses late output", async () => {
+  const upstream = await createRecoveryUpstream();
+  const timing = new RecoveryTiming();
+  const fixture = await createPublicRecoverySession(upstream, timing);
+  const { session } = fixture;
+  const observed: AgentStreamEvent[] = [];
+  session.subscribe((event) => observed.push(event));
+
+  await upstream.connected(1);
+  upstream.send(0, connectedRecord());
+  await session.startTurn("hello");
+  const [dispatchId] = await upstream.dispatched();
+  upstream.send(0, assistantMessageRecord("message-race"));
+  upstream.send(0, textDeltaRecord("message-race", "part-race", "Hello"));
+  await eventually(() => expect(assistantText(observed)).toEqual(["Hello"]));
+
+  upstream.setStatus("busy");
+  upstream.setMessages([
+    { info: { id: dispatchId as string, sessionID: "session-1", role: "user" }, parts: [] },
+    {
+      info: { id: "message-race", sessionID: "session-1", role: "assistant" },
+      parts: [
+        {
+          id: "part-race",
+          sessionID: "session-1",
+          messageID: "message-race",
+          type: "text",
+          text: "",
+          time: { start: 1 },
+        },
+      ],
+    },
+  ]);
+  upstream.end(0);
+  await timing.waiting();
+  timing.advance();
+  await upstream.connected(2);
+  upstream.send(1, connectedRecord());
+  upstream.send(1, textDeltaRecord("message-race", "part-race", " world"));
+  await eventually(() => expect(assistantText(observed)).toEqual(["Hello", " world"]));
+
+  upstream.setMessages([
+    {
+      info: {
+        id: "message-race",
+        sessionID: "session-1",
+        role: "assistant",
+        time: { created: 1, completed: 2 },
+      },
+      parts: [
+        {
+          id: "part-race",
+          sessionID: "session-1",
+          messageID: "message-race",
+          type: "text",
+          text: "Hello world!",
+          time: { start: 1, end: 2 },
+        },
+      ],
+    },
+  ]);
+  upstream.send(1, finalTextRecord("message-race", "part-race", "Hello world!"));
+  upstream.send(1, {
+    directory: "/workspace",
+    payload: {
+      type: "session.status",
+      properties: { sessionID: "session-1", status: { type: "idle" } },
+    },
+  });
+  upstream.send(1, textDeltaRecord("message-race", "part-race", " too late"));
+
+  await eventually(() =>
+    expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1),
+  );
+  expect(observed.filter((event) => event.type === "turn_started")).toHaveLength(1);
+  expect(observed.filter((event) => event.type === "turn_failed")).toHaveLength(0);
+  expect(observed.filter((event) => event.type === "turn_canceled")).toHaveLength(0);
+  expect(assistantText(observed)).toEqual(["Hello", " world", "!"]);
+  expect(assistantText(observed).join("")).toBe("Hello world!");
+  const ordered = observed.flatMap((event) => {
+    if (event.type === "turn_started" || event.type === "turn_completed") return [event.type];
+    if (event.type === "timeline" && event.item.type === "assistant_message") {
+      return [event.item.text];
+    }
+    return [];
+  });
+  expect(ordered).toEqual(["turn_started", "Hello", " world", "!", "turn_completed"]);
+
+  await session.close();
+  await fixture.manager.shutdown();
+  await upstream.close();
+});
+
+class RecoveryTiming implements OpenCodeEventConsumerTiming {
+  private resolveWait: (() => void) | null = null;
+  arm(): () => void {
+    return () => undefined;
+  }
+  wait(_delayMs: number, signal: AbortSignal): Promise<void> {
+    return new Promise((resolve, reject) => {
+      this.resolveWait = resolve;
+      signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+    });
+  }
+  async waiting(): Promise<void> {
+    await eventually(() => expect(this.resolveWait).not.toBeNull());
+  }
+  advance(): void {
+    this.resolveWait?.();
+    this.resolveWait = null;
+  }
+}
+
+async function createPublicRecoverySession(
+  upstream: Awaited<ReturnType<typeof createRecoveryUpstream>>,
+  timing: OpenCodeEventConsumerTiming,
+) {
+  let serverProcess: RecoveryServerProcess | null = null;
+  const manager = new OpenCodeServerManager({
+    logger: createTestLogger(),
+    portAllocator: async () => upstream.port,
+    resolveCommandPrefix: async () => ({ command: "opencode", args: [] }),
+    resolveHomeDir: () => process.cwd(),
+    spawnServerProcess: () => {
+      serverProcess = new RecoveryServerProcess();
+      return serverProcess as unknown as ChildProcess;
+    },
+    terminateProcess: async () => {
+      serverProcess?.exit();
+      return "terminated";
+    },
+    createEventSource: (options) => new OpenCodeEventConsumer({ ...options, timing }),
+  });
+  const client = new OpenCodeAgentClient(createTestLogger(), undefined, {
+    serverManager: manager,
+    createClient: ({ baseUrl, directory }) => createOpencodeClient({ baseUrl, directory }),
+  });
+  const session = await client.createSession({ provider: "opencode", cwd: "/workspace" });
+  return { manager, session };
+}
+
+class RecoveryServerProcess extends EventEmitter {
+  readonly stdout = new EventEmitter();
+  readonly stderr = new EventEmitter();
+  readonly pid = 42_001;
+  killed = false;
+  exitCode: number | null = null;
+  signalCode: NodeJS.Signals | null = null;
+
+  constructor() {
+    super();
+    queueMicrotask(() => this.stdout.emit("data", Buffer.from("listening on test server\n")));
+  }
+
+  exit(): void {
+    if (this.exitCode !== null) return;
+    this.exitCode = 0;
+    this.emit("exit", 0, null);
+  }
+}
+
+async function createRecoveryUpstream() {
+  const streams: ServerResponse[] = [];
+  const dispatchIds: string[] = [];
+  let messages: unknown[] = [];
+  let messageReadCount = 0;
+  let status: "busy" | "idle" = "idle";
+  const server = createServer(async (request, response) => {
+    const pathname = new URL(request.url ?? "/", "http://127.0.0.1").pathname;
+    if (request.url?.startsWith("/global/event")) {
+      response.writeHead(200, { "content-type": "text/event-stream" });
+      response.flushHeaders();
+      streams.push(response);
+      return;
+    }
+    if (request.method === "POST" && pathname === "/session") {
+      return json(response, { id: "session-1", directory: "/workspace" });
+    }
+    if (request.url?.includes("/prompt_async")) {
+      const body = JSON.parse(await readBody(request)) as { messageID: string };
+      dispatchIds.push(body.messageID);
+      return json(response, {});
+    }
+    if (request.url?.includes("/message")) {
+      messageReadCount += 1;
+      return json(response, messages);
+    }
+    if (request.url?.includes("/status"))
+      return json(response, status === "busy" ? { "session-1": { type: "busy" } } : {});
+    return json(response, {});
+  });
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Missing test server address");
+  return {
+    port: address.port,
+    url: `http://127.0.0.1:${address.port}`,
+    connected: async (count: number) => eventually(() => expect(streams).toHaveLength(count)),
+    send(index: number, value: unknown) {
+      streams[index]?.write(`data: ${JSON.stringify(value)}\n\n`);
+    },
+    end(index: number) {
+      streams[index]?.end();
+    },
+    dispatched: async (count = 1) => {
+      await eventually(() => expect(dispatchIds).toHaveLength(count));
+      return [...dispatchIds];
+    },
+    messageReads: () => messageReadCount,
+    setMessages(nextMessages: unknown[]) {
+      messages = nextMessages;
+    },
+    setStatus(nextStatus: "busy" | "idle") {
+      status = nextStatus;
+    },
+    setRecoveredMessages(messageId: string) {
+      messages = [
+        {
+          info: {
+            id: "msg_before_boundary",
+            sessionID: "session-1",
+            role: "assistant",
+            time: { created: 1, completed: 2 },
+          },
+          parts: [
+            {
+              id: "part-before",
+              sessionID: "session-1",
+              messageID: "msg_before_boundary",
+              type: "text",
+              text: "before boundary",
+              time: { start: 1, end: 2 },
+            },
+          ],
+        },
+        {
+          info: { id: messageId, sessionID: "session-1", role: "user" },
+          parts: [],
+        },
+        ...Array.from({ length: 51 }, (_, index) => ({
+          info: {
+            id: `msg_step_${index}`,
+            sessionID: "session-1",
+            role: "assistant",
+            time: { created: index + 3, completed: index + 4 },
+          },
+          parts: [
+            {
+              id: `part-step-${index}`,
+              sessionID: "session-1",
+              messageID: `msg_step_${index}`,
+              type: "text",
+              text: `step ${index}`,
+              time: { start: index + 3, end: index + 4 },
+            },
+          ],
+        })),
+        {
+          info: { id: "msg_compaction", sessionID: "session-1", role: "user" },
+          parts: [
+            {
+              id: "part-compaction",
+              sessionID: "session-1",
+              messageID: "msg_compaction",
+              type: "compaction",
+              auto: true,
+            },
+          ],
+        },
+        {
+          info: {
+            id: "msg_assistant",
+            sessionID: "session-1",
+            role: "assistant",
+            time: { created: 1, completed: 2 },
+          },
+          parts: [
+            {
+              id: "part-1",
+              sessionID: "session-1",
+              messageID: "msg_assistant",
+              type: "text",
+              text: "recovered output",
+              time: { start: 1, end: 2 },
+            },
+          ],
+        },
+      ];
+    },
+    close: async () => new Promise<void>((resolve) => server.close(() => resolve())),
+  };
+}
+
+function assistantMessageRecord(messageId: string) {
+  return {
+    directory: "/workspace",
+    payload: {
+      type: "message.updated",
+      properties: { info: { id: messageId, sessionID: "session-1", role: "assistant" } },
+    },
+  };
+}
+
+function textDeltaRecord(messageId: string, partId: string, delta: string) {
+  return {
+    directory: "/workspace",
+    payload: {
+      type: "message.part.delta",
+      properties: {
+        sessionID: "session-1",
+        messageID: messageId,
+        partID: partId,
+        field: "text",
+        delta,
+      },
+    },
+  };
+}
+
+function finalTextRecord(messageId: string, partId: string, text: string) {
+  return {
+    directory: "/workspace",
+    payload: {
+      type: "message.part.updated",
+      properties: {
+        part: {
+          id: partId,
+          sessionID: "session-1",
+          messageID: messageId,
+          type: "text",
+          text,
+          time: { start: 1, end: 2 },
+        },
+      },
+    },
+  };
+}
+
+function assistantText(events: AgentStreamEvent[]): string[] {
+  return events.flatMap((event) =>
+    event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [],
+  );
+}
+
+function connectedRecord() {
+  return { directory: "/workspace", payload: { type: "server.connected", properties: {} } };
+}
+
+function idleRecord() {
+  return {
+    directory: "/workspace",
+    payload: {
+      type: "session.status",
+      properties: { sessionID: "session-1", status: { type: "idle" } },
+    },
+  };
+}
+
+function json(response: ServerResponse, body: unknown): void {
+  response.writeHead(200, { "content-type": "application/json" });
+  response.end(JSON.stringify(body));
+}
+
+async function readBody(request: IncomingMessage): Promise<string> {
+  let body = "";
+  for await (const chunk of request) body += chunk;
+  return body;
+}
+
+async function eventually(assertion: () => void): Promise<void> {
+  for (let attempt = 0; attempt < 100; attempt += 1) {
+    try {
+      assertion();
+      return;
+    } catch {
+      await new Promise<void>((resolve) => setImmediate(resolve));
+    }
+  }
+  assertion();
+}

packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts

@@ -8,20 +8,6 @@ import {
   TestOpenCodeHarness,
 } from "./opencode/test-utils/test-opencode-harness.js";
 
-function createDeferred<T>(): {
-  promise: Promise<T>;
-  resolve: (value: T) => void;
-  reject: (error: unknown) => void;
-} {
-  let resolve!: (value: T) => void;
-  let reject!: (error: unknown) => void;
-  const promise = new Promise<T>((res, rej) => {
-    resolve = res;
-    reject = rej;
-  });
-  return { promise, resolve, reject };
-}
-
 describe("OpenCodeAgentSession slash command timeout handling", () => {
   test("lists only OpenCode built-in slash commands Paseo can execute", async () => {
     const runtime = new TestOpenCodeHarness();
@@ -75,20 +61,12 @@ describe("OpenCodeAgentSession slash command timeout handling", () => {
   });
 
   test("waits for SSE completion when slash commands hit a header timeout", async () => {
-    const idleEventGate = createDeferred<void>();
     const runtime = new TestOpenCodeHarness();
     const openCodeClient = createOpenCodeClientWithConnectedProvider();
     openCodeClient.sessionCommandError = new Error("fetch failed: Headers Timeout Error");
     openCodeClient.commandListResponse = {
       data: [{ name: "help", description: "Show help", hints: [] }],
     };
-    openCodeClient.eventStream = (async function* () {
-      await idleEventGate.promise;
-      yield {
-        type: "session.idle",
-        properties: { sessionID: "session-1" },
-      };
-    })();
     runtime.enqueueClient(openCodeClient);
 
     const client = new OpenCodeAgentClient(createTestLogger(), undefined, {
@@ -98,8 +76,17 @@ describe("OpenCodeAgentSession slash command timeout handling", () => {
     const session = await client.createSession({ provider: "opencode", cwd: "/tmp" });
 
     const runPromise = session.run("/help");
-    await Promise.resolve();
+    await nextTick();
-    idleEventGate.resolve();
+    expect(openCodeClient.calls.sessionCommand).toHaveLength(1);
+    let settled = false;
+    void runPromise.then(() => {
+      settled = true;
+      return undefined;
+    });
+    await nextTick();
+    expect(settled).toBe(false);
+
+    openCodeClient.emitEvent(idleEvent());
 
     await expect(runPromise).resolves.toMatchObject({
       sessionId: "session-1",

packages/server/src/server/agent/providers/opencode-server-manager.test.ts

@@ -1,6 +1,7 @@
 import type { ChildProcess } from "node:child_process";
 import { EventEmitter } from "node:events";
 import { mkdtempSync, rmSync } from "node:fs";
+import { createServer, type ServerResponse } from "node:http";
 import net from "node:net";
 import os from "node:os";
 import path from "node:path";
@@ -27,6 +28,38 @@ afterEach(() => {
 });
 
 describe("OpenCodeServerManager generations", () => {
+  test("shares one real SDK event stream across acquisitions until generation shutdown", async () => {
+    const responses: ServerResponse[] = [];
+    let requestCount = 0;
+    const upstream = createServer((_request, response) => {
+      requestCount += 1;
+      response.writeHead(200, { "content-type": "text/event-stream" });
+      response.flushHeaders();
+      responses.push(response);
+    });
+    await new Promise<void>((resolve) => upstream.listen(0, "127.0.0.1", resolve));
+    const address = upstream.address();
+    if (!address || typeof address === "string") throw new Error("Missing upstream address");
+    const { manager } = createTestManager([address.port]);
+
+    const first = await manager.acquireCurrent();
+    const second = await manager.acquireCurrent();
+    expect(first.events).toBe(second.events);
+    await vi.waitFor(() => expect(responses).toHaveLength(1));
+    responses[0]?.write(
+      `data: ${JSON.stringify({ directory: "/workspace", payload: { type: "server.connected", properties: {} } })}\n\n`,
+    );
+    await first.events.ready();
+    expect(requestCount).toBe(1);
+
+    await first.release();
+    expect(requestCount).toBe(1);
+    await second.release();
+    expect(requestCount).toBe(1);
+    await manager.shutdown();
+    await new Promise<void>((resolve) => upstream.close(() => resolve()));
+  });
+
   test("uses an explicit base environment for the server process", async () => {
     const baseEnv = { HOME: "/isolated/home", PATH: "/isolated/bin" };
     const { manager, runtime } = createTestManager([4091], { baseEnv });

packages/server/src/server/agent/providers/opencode/event-consumer.test.ts

@@ -0,0 +1,405 @@
+import { createServer, type ServerResponse } from "node:http";
+import { createOpencodeClient, type OpencodeClient } from "@opencode-ai/sdk/v2/client";
+import { afterEach, describe, expect, test } from "vitest";
+
+import {
+  OpenCodeEventConsumer,
+  type OpenCodeEventConsumerTiming,
+  type OpenCodeEventSourceInput,
+} from "./event-consumer.js";
+
+describe("OpenCodeEventConsumer", () => {
+  const cleanups: Array<() => Promise<void>> = [];
+  afterEach(async () => {
+    await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()));
+  });
+
+  test("settles when close wins immediately before an injected backoff wait", async () => {
+    const upstream = await createSseUpstream();
+    upstream.failNext(1);
+    let consumer!: OpenCodeEventConsumer;
+    let closePromise: Promise<void> | null = null;
+    let observedAlreadyAborted = false;
+    const timing: OpenCodeEventConsumerTiming = {
+      arm: () => () => undefined,
+      get wait() {
+        closePromise = consumer.close();
+        return async (_delayMs, signal) => {
+          observedAlreadyAborted = signal.aborted;
+          if (signal.aborted) throw signal.reason;
+        };
+      },
+    };
+    consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(upstream.close);
+
+    await eventually(() => expect(observedAlreadyAborted).toBe(true));
+    await expect(closePromise).resolves.toBeUndefined();
+  });
+
+  test("becomes ready on the first record and reconnects once after EOF", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const processExit = deferred<Error>();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: processExit.promise,
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe((input) => inputs.push(input));
+
+    await upstream.connected(1);
+    expect(await promiseState(consumer.ready())).toBe("pending");
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+    expect(inputs).toEqual([connectedRecord("/one")]);
+
+    upstream.end(0);
+    await timing.waiting();
+    expect(timing.waitDelays).toEqual([100]);
+    timing.advanceWait();
+    await upstream.connected(2);
+    upstream.send(1, connectedRecord("/two"));
+    await eventually(() => expect(inputs).toHaveLength(3));
+    expect(inputs).toEqual([
+      connectedRecord("/one"),
+      { type: "reconnected" },
+      connectedRecord("/two"),
+    ]);
+    expect(upstream.requests).toHaveLength(2);
+    expect(upstream.requests.map((request) => request.url)).toEqual([
+      "/global/event",
+      "/global/event",
+    ]);
+  });
+
+  test("reconnects a stalled response without publishing a terminal", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe((input) => inputs.push(input));
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+
+    timing.expireWatchdog();
+    expect(timing.watchdogDelays).toEqual([30_000, 30_000]);
+    await timing.waiting();
+    timing.advanceWait();
+    await upstream.connected(2);
+    upstream.send(1, connectedRecord("/two"));
+    await eventually(() => expect(inputs).toHaveLength(3));
+    expect(inputs[1]).toEqual({ type: "reconnected" });
+  });
+
+  test("publishes one terminal and stops reconnecting on process exit", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const processExit = deferred<Error>();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: processExit.promise,
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe((input) => inputs.push(input));
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+
+    processExit.resolve(new Error("process exited"));
+    await eventually(() => expect(inputs).toHaveLength(2));
+    expect(inputs[1]).toMatchObject({ type: "server-exited" });
+    expect(upstream.requests).toHaveLength(1);
+  });
+
+  test("backs off repeated zero-record EOFs exponentially up to the cap", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+
+    for (let connection = 0; connection < 8; connection += 1) {
+      await upstream.connected(connection + 1);
+      upstream.end(connection);
+      await timing.waiting();
+      timing.advanceWait();
+    }
+
+    expect(timing.waitDelays).toEqual([100, 200, 400, 800, 1_600, 3_200, 5_000, 5_000]);
+  });
+
+  test("isolates a throwing subscriber from the shared transport", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe(() => {
+      throw new Error("listener failed");
+    });
+    consumer.subscribe((input) => inputs.push(input));
+
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+    expect(inputs).toEqual([connectedRecord("/one")]);
+  });
+
+  test("reconnects after a socket error with a delivered-record backoff reset", async () => {
+    const upstream = await createSseUpstream();
+    const timing = new ControlledTiming();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+    upstream.destroy(0);
+    await timing.waiting();
+    expect(timing.waitDelays).toEqual([100]);
+  });
+
+  test("routes initial HTTP failures through consumer backoff before readiness", async () => {
+    const upstream = await createSseUpstream();
+    upstream.failNext(2);
+    const timing = new ControlledTiming();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      timing,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+
+    await timing.waiting();
+    expect(upstream.requests).toHaveLength(1);
+    expect(await promiseState(consumer.ready())).toBe("pending");
+    timing.advanceWait();
+    await timing.waiting();
+    expect(upstream.requests).toHaveLength(2);
+    timing.advanceWait();
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/ready"));
+    await consumer.ready();
+    expect(timing.waitDelays).toEqual([100, 200]);
+  });
+
+  test("disables SDK SSE retries at the real request seam", async () => {
+    const upstream = await createSseUpstream();
+    const realClient = createOpencodeClient({ baseUrl: upstream.url });
+    let requestOptions: unknown;
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+      createClient: () =>
+        ({
+          ...realClient,
+          global: {
+            ...realClient.global,
+            event: (options: unknown) => {
+              requestOptions = options;
+              return realClient.global.event(options as never);
+            },
+          },
+        }) as OpencodeClient,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    await upstream.connected(1);
+
+    expect(requestOptions).toMatchObject({ sseMaxRetryAttempts: 0 });
+  });
+
+  test("rejects readiness and publishes terminal when the process exits before a record", async () => {
+    const upstream = await createSseUpstream();
+    const processExit = deferred<Error>();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: processExit.promise,
+    });
+    cleanups.push(async () => {
+      await consumer.close();
+      await upstream.close();
+    });
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe((input) => inputs.push(input));
+    await upstream.connected(1);
+
+    processExit.resolve(new Error("process exited before ready"));
+
+    await expect(consumer.ready()).rejects.toThrow("process exited before ready");
+    expect(inputs).toEqual([expect.objectContaining({ type: "server-exited" })]);
+  });
+
+  test("intentional close does not publish a false terminal", async () => {
+    const upstream = await createSseUpstream();
+    const consumer = new OpenCodeEventConsumer({
+      serverUrl: upstream.url,
+      processExit: new Promise<Error>(() => undefined),
+    });
+    cleanups.push(upstream.close);
+    const inputs: OpenCodeEventSourceInput[] = [];
+    consumer.subscribe((input) => inputs.push(input));
+    await upstream.connected(1);
+    upstream.send(0, connectedRecord("/one"));
+    await consumer.ready();
+
+    await consumer.close();
+
+    expect(inputs).toEqual([connectedRecord("/one")]);
+  });
+});
+
+class ControlledTiming implements OpenCodeEventConsumerTiming {
+  private waits: Array<() => void> = [];
+  private watchdog: (() => void) | null = null;
+  readonly waitDelays: number[] = [];
+  readonly watchdogDelays: number[] = [];
+  arm(delayMs: number, callback: () => void): () => void {
+    this.watchdogDelays.push(delayMs);
+    this.watchdog = callback;
+    return () => {
+      if (this.watchdog === callback) this.watchdog = null;
+    };
+  }
+  wait(delayMs: number, signal: AbortSignal): Promise<void> {
+    this.waitDelays.push(delayMs);
+    if (signal.aborted) return Promise.reject(signal.reason);
+    return new Promise((resolve, reject) => {
+      this.waits.push(resolve);
+      signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+    });
+  }
+  async waiting(): Promise<void> {
+    await eventually(() => expect(this.waits).toHaveLength(1));
+  }
+  advanceWait(): void {
+    this.waits.shift()?.();
+  }
+  expireWatchdog(): void {
+    const watchdog = this.watchdog;
+    this.watchdog = null;
+    if (!watchdog) throw new Error("No watchdog is armed");
+    watchdog();
+  }
+}
+
+function connectedRecord(directory: string) {
+  return { directory, payload: { type: "server.connected", properties: {} } };
+}
+
+async function createSseUpstream() {
+  const responses: ServerResponse[] = [];
+  const requests: Array<{ url: string | undefined }> = [];
+  let failuresRemaining = 0;
+  const server = createServer((request, response) => {
+    requests.push({ url: request.url });
+    if (failuresRemaining > 0) {
+      failuresRemaining -= 1;
+      response.writeHead(503).end();
+      return;
+    }
+    response.writeHead(200, {
+      "content-type": "text/event-stream",
+      "cache-control": "no-cache",
+      connection: "keep-alive",
+    });
+    response.flushHeaders();
+    responses.push(response);
+  });
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Missing test server address");
+  return {
+    url: `http://127.0.0.1:${address.port}`,
+    requests,
+    connected: async (count: number) => eventually(() => expect(responses).toHaveLength(count)),
+    send(index: number, value: unknown) {
+      responses[index]?.write(`data: ${JSON.stringify(value)}\n\n`);
+    },
+    end(index: number) {
+      responses[index]?.end();
+    },
+    destroy(index: number) {
+      responses[index]?.destroy(new Error("socket failed"));
+    },
+    failNext(count: number) {
+      failuresRemaining = count;
+    },
+    close: async () => new Promise<void>((resolve) => server.close(() => resolve())),
+  };
+}
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((next) => {
+    resolve = next;
+  });
+  return { promise, resolve };
+}
+
+async function promiseState(promise: Promise<unknown>): Promise<"pending" | "settled"> {
+  return await Promise.race([
+    promise.then(() => "settled" as const),
+    new Promise<"pending">((resolve) => setImmediate(() => resolve("pending"))),
+  ]);
+}
+
+async function eventually(assertion: () => void): Promise<void> {
+  for (let attempt = 0; attempt < 100; attempt += 1) {
+    try {
+      assertion();
+      return;
+    } catch {
+      await new Promise<void>((resolve) => setImmediate(resolve));
+    }
+  }
+  assertion();
+}

packages/server/src/server/agent/providers/opencode/event-consumer.ts

@@ -0,0 +1,169 @@
+import {
+  createOpencodeClient,
+  type GlobalEvent,
+  type OpencodeClient,
+} from "@opencode-ai/sdk/v2/client";
+
+export type OpenCodeEventSourceInput =
+  | GlobalEvent
+  | { type: "reconnected" }
+  | { type: "server-exited"; error: Error };
+
+export interface OpenCodeEventSource {
+  ready(): Promise<void>;
+  subscribe(listener: (input: OpenCodeEventSourceInput) => void): () => void;
+}
+
+export interface OpenCodeEventConsumerTiming {
+  arm(delayMs: number, callback: () => void): () => void;
+  wait(delayMs: number, signal: AbortSignal): Promise<void>;
+}
+
+export interface OpenCodeEventConsumerOptions {
+  serverUrl: string;
+  processExit: Promise<Error>;
+  createClient?: (baseUrl: string) => OpencodeClient;
+  timing?: OpenCodeEventConsumerTiming;
+}
+
+const WATCHDOG_MS = 30_000;
+const MAX_BACKOFF_MS = 5_000;
+
+const systemTiming: OpenCodeEventConsumerTiming = {
+  arm(delayMs, callback) {
+    const handle = setTimeout(callback, delayMs);
+    return () => clearTimeout(handle);
+  },
+  wait(delayMs, signal) {
+    if (signal.aborted) return Promise.reject(signal.reason);
+    return new Promise((resolve, reject) => {
+      const handle = setTimeout(() => {
+        signal.removeEventListener("abort", onAbort);
+        resolve();
+      }, delayMs);
+      const onAbort = () => {
+        clearTimeout(handle);
+        reject(signal.reason);
+      };
+      signal.addEventListener("abort", onAbort, { once: true });
+    });
+  },
+};
+
+export class OpenCodeEventConsumer implements OpenCodeEventSource {
+  private readonly listeners = new Set<(input: OpenCodeEventSourceInput) => void>();
+  private readonly client: OpencodeClient;
+  private readonly timing: OpenCodeEventConsumerTiming;
+  private readonly readyPromise: Promise<void>;
+  private resolveReady!: () => void;
+  private rejectReady!: (error: Error) => void;
+  private connectionAbort = new AbortController();
+  private connectionTask: Promise<void>;
+  private connected = false;
+  private closed = false;
+
+  constructor(options: OpenCodeEventConsumerOptions) {
+    this.client =
+      options.createClient?.(options.serverUrl) ??
+      createOpencodeClient({ baseUrl: options.serverUrl });
+    this.timing = options.timing ?? systemTiming;
+    this.readyPromise = new Promise<void>((resolve, reject) => {
+      this.resolveReady = resolve;
+      this.rejectReady = reject;
+    });
+    void this.readyPromise.catch(() => undefined);
+    this.connectionTask = this.consume(options.processExit);
+    void this.connectionTask.catch(() => undefined);
+  }
+
+  ready(): Promise<void> {
+    return this.readyPromise;
+  }
+
+  subscribe(listener: (input: OpenCodeEventSourceInput) => void): () => void {
+    this.listeners.add(listener);
+    return () => this.listeners.delete(listener);
+  }
+
+  async close(): Promise<void> {
+    if (this.closed) return;
+    this.closed = true;
+    this.listeners.clear();
+    const error = new Error("OpenCode event source closed");
+    this.rejectReady(error);
+    this.connectionAbort.abort(error);
+    await this.connectionTask.catch(() => undefined);
+  }
+
+  private async consume(processExit: Promise<Error>): Promise<void> {
+    void processExit.then((error) => this.exit(error));
+    let reconnectAttempt = 0;
+    while (!this.closed) {
+      const delivered = await this.consumeConnection(this.connectionAbort.signal).catch(
+        () => false,
+      );
+      if (this.closed) return;
+      reconnectAttempt = delivered ? 0 : reconnectAttempt + 1;
+      const delayMs = Math.min(100 * 2 ** Math.max(0, reconnectAttempt - 1), MAX_BACKOFF_MS);
+      await this.timing.wait(delayMs, this.connectionAbort.signal).catch(() => undefined);
+    }
+  }
+
+  private async consumeConnection(signal: AbortSignal): Promise<boolean> {
+    const requestAbort = new AbortController();
+    const abortRequest = () => requestAbort.abort(signal.reason);
+    signal.addEventListener("abort", abortRequest, { once: true });
+    let cancelWatchdog: () => void = () => undefined;
+    let delivered = false;
+    try {
+      const result = await this.client.global.event({
+        signal: requestAbort.signal,
+        sseMaxRetryAttempts: 0,
+      });
+      const armWatchdog = () => {
+        cancelWatchdog();
+        cancelWatchdog = this.timing.arm(WATCHDOG_MS, () => requestAbort.abort());
+      };
+      armWatchdog();
+      for await (const event of result.stream) {
+        if (this.closed) return delivered;
+        armWatchdog();
+        if (!delivered) {
+          delivered = true;
+          if (this.connected) this.publish({ type: "reconnected" });
+          this.connected = true;
+          this.resolveReady();
+        }
+        this.publish(event);
+      }
+      return delivered;
+    } finally {
+      cancelWatchdog();
+      signal.removeEventListener("abort", abortRequest);
+      requestAbort.abort();
+    }
+  }
+
+  private exit(error: Error): void {
+    if (this.closed) return;
+    this.closed = true;
+    this.connectionAbort.abort(error);
+    if (!this.connected) this.rejectReady(error);
+    this.publish({ type: "server-exited", error });
+    this.listeners.clear();
+  }
+
+  private publish(input: OpenCodeEventSourceInput): void {
+    for (const listener of this.listeners) {
+      try {
+        listener(input);
+      } catch {
+        // A session callback cannot tear down the generation-owned transport.
+      }
+    }
+  }
+}
+
+export type OpenCodeEventConsumerFactory = (
+  options: Pick<OpenCodeEventConsumerOptions, "serverUrl" | "processExit">,
+) => OpenCodeEventConsumer;

packages/server/src/server/agent/providers/opencode/event-translator.test.ts

@@ -29,7 +29,7 @@ function createState(sessionId = "session-1"): OpenCodeEventTranslationState {
     sessionId,
     messageRoles: new Map(),
     accumulatedUsage: {},
-    streamedPartKeys: new Set(),
+    materializedParts: new Map(),
     emittedStructuredMessageIds: new Set(),
     compactionSummaryMessageIds: new Set(),
     emittedCompactionPartIds: new Set(),
@@ -38,6 +38,130 @@ function createState(sessionId = "session-1"): OpenCodeEventTranslationState {
 }
 
 describe("translateOpenCodeEvent", () => {
+  it("emits only the missing suffix from a final full text part", () => {
+    const state = createState();
+    translateOpenCodeEvent(
+      {
+        type: "message.updated",
+        properties: {
+          info: { id: "message-1", sessionID: "session-1", role: "assistant" },
+        },
+      },
+      state,
+    );
+    const first = translateOpenCodeEvent(
+      {
+        type: "message.part.delta",
+        properties: {
+          sessionID: "session-1",
+          messageID: "message-1",
+          partID: "part-1",
+          field: "text",
+          delta: "Hello",
+        },
+      },
+      state,
+    );
+    const incomplete = translateOpenCodeEvent(
+      {
+        type: "message.part.updated",
+        properties: {
+          part: {
+            id: "part-1",
+            sessionID: "session-1",
+            messageID: "message-1",
+            type: "text",
+            text: "",
+            time: { start: 1 },
+          },
+        },
+      },
+      state,
+    );
+    const second = translateOpenCodeEvent(
+      {
+        type: "message.part.delta",
+        properties: {
+          sessionID: "session-1",
+          messageID: "message-1",
+          partID: "part-1",
+          field: "text",
+          delta: " world",
+        },
+      },
+      state,
+    );
+    const final = translateOpenCodeEvent(
+      {
+        type: "message.part.updated",
+        properties: {
+          part: {
+            id: "part-1",
+            sessionID: "session-1",
+            messageID: "message-1",
+            type: "text",
+            text: "Hello world!",
+            time: { start: 1, end: 2 },
+          },
+        },
+      },
+      state,
+    );
+
+    expect(incomplete).toEqual([]);
+    expect([first, second, final].flatMap((events) => events)).toMatchObject([
+      { item: { text: "Hello" } },
+      { item: { text: " world" } },
+      { item: { text: "!" } },
+    ]);
+    expect(
+      translateOpenCodeEvent(
+        {
+          type: "message.part.delta",
+          properties: {
+            sessionID: "session-1",
+            messageID: "message-1",
+            partID: "part-1",
+            field: "text",
+            delta: " delayed",
+          },
+        },
+        state,
+      ),
+    ).toEqual([]);
+  });
+
+  it("diagnoses a non-prefix final text snapshot", () => {
+    const diagnostics: unknown[] = [];
+    const state = createState();
+    state.onMaterializationMismatch = (diagnostic) => diagnostics.push(diagnostic);
+    state.materializedParts.set("part-1", {
+      messageId: "message-1",
+      emittedText: "streamed text",
+      closed: false,
+    });
+
+    expect(
+      translateOpenCodeEvent(
+        {
+          type: "message.part.updated",
+          properties: {
+            part: {
+              id: "part-1",
+              sessionID: "session-1",
+              messageID: "message-1",
+              type: "text",
+              text: "mutated text",
+              time: { start: 1, end: 2 },
+            },
+          },
+        },
+        state,
+      ),
+    ).toEqual([]);
+    expect(diagnostics).toEqual([{ partId: "part-1", messageId: "message-1", kind: "text" }]);
+  });
+
   it("resolves context window max tokens from assistant message.updated model metadata", () => {
     const resolvedContextWindowMaxTokens: number[] = [];
     const state = createState();
@@ -1185,7 +1309,11 @@ describe("translateOpenCodeEvent", () => {
 
   it("emits turn_completed from session.status idle", () => {
     const state = createState();
-    state.streamedPartKeys.add("text:part-1");
+    state.materializedParts.set("part-1", {
+      messageId: "message-1",
+      emittedText: "partial",
+      closed: false,
+    });
     state.partTypes.set("part-1", "text");
 
     const result = translateOpenCodeEvent(
@@ -1206,13 +1334,17 @@ describe("translateOpenCodeEvent", () => {
         usage: undefined,
       },
     ]);
-    expect(state.streamedPartKeys.size).toBe(0);
+    expect(state.materializedParts.size).toBe(1);
     expect(state.partTypes.size).toBe(0);
   });
 
   it("forwards session.status retry as a non-terminal timeline error item", () => {
     const state = createState();
-    state.streamedPartKeys.add("text:part-1");
+    state.materializedParts.set("part-1", {
+      messageId: "message-1",
+      emittedText: "partial",
+      closed: false,
+    });
     state.partTypes.set("part-1", "text");
 
     const result = translateOpenCodeEvent(
@@ -1240,7 +1372,7 @@ describe("translateOpenCodeEvent", () => {
     ]);
     // Streaming state must NOT be reset — the turn is still alive, opencode
     // will eventually either succeed or emit session.idle / session.error.
-    expect(state.streamedPartKeys.size).toBe(1);
+    expect(state.materializedParts.size).toBe(1);
     expect(state.partTypes.size).toBe(1);
   });
 

packages/server/src/server/agent/providers/opencode/server-manager.ts

@@ -15,12 +15,18 @@ import {
   type ProviderRuntimeSettings,
 } from "../../provider-launch-config.js";
 import { resolveOpenCodeHomeDir } from "./paths.js";
+import {
+  OpenCodeEventConsumer,
+  type OpenCodeEventConsumerFactory,
+  type OpenCodeEventSource,
+} from "./event-consumer.js";
 
 const OPENCODE_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 5_000;
 const OPENCODE_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000;
 
 export interface OpenCodeServerAcquisition {
   server: { port: number; url: string };
+  events: OpenCodeEventSource;
   release: () => Promise<void>;
 }
 
@@ -39,6 +45,7 @@ export interface OpenCodeServerGeneration {
   refCount: number;
   retired: boolean;
   ready: Promise<void>;
+  events: OpenCodeEventConsumer;
   managedProcessId?: string;
   managedProcessRecord?: Promise<{ id: string } | null>;
 }
@@ -61,6 +68,7 @@ export interface OpenCodeServerManagerOptions {
   resolveCommandPrefix?: OpenCodeCommandPrefixResolver;
   resolveHomeDir?: () => string;
   spawnServerProcess?: OpenCodeServerProcessSpawner;
+  createEventSource?: OpenCodeEventConsumerFactory;
 }
 
 export class OpenCodeServerManager implements OpenCodeServerManagerLike {
@@ -80,6 +88,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
   private readonly resolveCommandPrefix: OpenCodeCommandPrefixResolver;
   private readonly resolveHomeDir: () => string;
   private readonly spawnServerProcess: OpenCodeServerProcessSpawner;
+  private readonly createEventSource: OpenCodeEventConsumerFactory;
 
   constructor(options: OpenCodeServerManagerOptions) {
     this.logger = options.logger;
@@ -94,6 +103,8 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
       (() => resolveProviderCommandPrefix(this.runtimeSettings?.command, resolveOpenCodeBinary));
     this.resolveHomeDir = options.resolveHomeDir ?? resolveOpenCodeHomeDir;
     this.spawnServerProcess = options.spawnServerProcess ?? spawnProcess;
+    this.createEventSource =
+      options.createEventSource ?? ((input) => new OpenCodeEventConsumer(input));
   }
 
   static getInstance(
@@ -191,6 +202,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
     let releasePromise: Promise<void> | null = null;
     return {
       server: { port: server.port, url: server.url },
+      events: server.events,
       release: async () => {
         if (releasePromise) {
           return releasePromise;
@@ -316,6 +328,10 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
       args: serverArgs,
       port,
     });
+    let resolveProcessExit!: (error: Error) => void;
+    const processExit = new Promise<Error>((resolve) => {
+      resolveProcessExit = resolve;
+    });
     const server: OpenCodeServerGeneration = {
       process: serverProcess,
       port,
@@ -323,6 +339,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
       refCount: 0,
       retired: false,
       ready: Promise.resolve(),
+      events: this.createEventSource({ serverUrl: url, processExit }),
       managedProcessRecord,
     };
     void managedProcessRecord.then((record) => {
@@ -396,6 +413,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
       });
 
       serverProcess.on("exit", (code) => {
+        resolveProcessExit(new Error(`OpenCode server exited with code ${code}`));
         this.removeManagedServerRecord(server);
         if (!started) {
           failStartup(
@@ -447,6 +465,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
   }
 
   private async killServer(server: OpenCodeServerGeneration): Promise<void> {
+    await server.events.close();
     if (
       (server.process.exitCode !== null && server.process.exitCode !== undefined) ||
       (server.process.signalCode !== null && server.process.signalCode !== undefined)

packages/server/src/server/agent/providers/opencode/test-server-manager.ts

@@ -1,5 +1,10 @@
 import type { OpenCodeServerAcquisition, OpenCodeServerManagerLike } from "./server-manager.js";
 
+const events = {
+  ready: async () => undefined,
+  subscribe: () => () => undefined,
+};
+
 export interface TestOpenCodeServerAcquisition {
   kind: "current" | "new" | "dedicated" | "existing";
   env?: Record<string, string>;
@@ -41,6 +46,7 @@ export class TestOpenCodeServerManager implements OpenCodeServerManagerLike {
     this.acquisitions.push(acquisition);
     return {
       server: this.server,
+      events,
       release: async () => {
         acquisition.released = true;
       },

packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts

@@ -1,5 +1,6 @@
 import type { OpencodeClient } from "@opencode-ai/sdk/v2/client";
 
+import type { OpenCodeEventSourceInput } from "../event-consumer.js";
 import type { OpenCodeServerAcquisition, OpenCodeServerManagerLike } from "../server-manager.js";
 
 interface OpenCodeResponse {
@@ -16,10 +17,21 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike {
   }> = [];
   readonly clientCreations: Array<{ baseUrl: string; directory: string }> = [];
   private readonly clients: TestOpenCodeClient[] = [];
+  private readonly eventListeners = new Set<(input: OpenCodeEventSourceInput) => void>();
+  readonly events = {
+    ready: async () => undefined,
+    subscribe: (listener: (input: OpenCodeEventSourceInput) => void) => {
+      this.eventListeners.add(listener);
+      return () => this.eventListeners.delete(listener);
+    },
+  };
 
   server = { port: 1234, url: "http://127.0.0.1:1234" };
 
   enqueueClient(client: TestOpenCodeClient): void {
+    client.observeEvents((event) => {
+      for (const listener of this.eventListeners) listener(event as OpenCodeEventSourceInput);
+    });
     this.clients.push(client);
   }
 
@@ -53,6 +65,7 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike {
     this.acquisitions.push(acquisition);
     return {
       server: this.server,
+      events: this.events,
       release: async () => {
         acquisition.releaseCount += 1;
       },
@@ -79,9 +92,11 @@ export class TestOpenCodeClient {
     mcpAdd: [] as unknown[],
     mcpConnect: [] as unknown[],
     permissionReply: [] as unknown[],
+    permissionList: [] as unknown[],
     providerList: [] as unknown[],
     providerListOptions: [] as unknown[],
     questionReject: [] as unknown[],
+    questionList: [] as unknown[],
     questionReply: [] as unknown[],
     sessionAbort: [] as unknown[],
     sessionCommand: [] as unknown[],
@@ -106,6 +121,10 @@ export class TestOpenCodeClient {
   mcpAddResponse: OpenCodeResponse = {};
   mcpConnectResponse: OpenCodeResponse = {};
   permissionReplyResponse: OpenCodeResponse = {};
+  permissionListResponse: OpenCodeResponse = { data: [] };
+  permissionListImplementation:
+    | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
+    | null = null;
   providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } };
   providerListImplementation:
     | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
@@ -114,6 +133,10 @@ export class TestOpenCodeClient {
     | ((options: unknown) => Promise<{ stream: AsyncIterable<unknown> }>)
     | null = null;
   questionRejectResponse: OpenCodeResponse = {};
+  questionListResponse: OpenCodeResponse = { data: [] };
+  questionListImplementation:
+    | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
+    | null = null;
   questionReplyResponse: OpenCodeResponse = {};
   sessionAbortResponse: OpenCodeResponse = {};
   sessionAbortImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null = null;
@@ -123,19 +146,29 @@ export class TestOpenCodeClient {
   sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } };
   sessionDeleteResponse: OpenCodeResponse = {};
   sessionChildrenResponses: OpenCodeResponse[] = [];
-  sessionChildrenImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null = null;
+  sessionChildrenImplementation:
+    | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
+    | null = null;
   sessionGetResponse: OpenCodeResponse = {
     data: { id: "session-1", directory: "/workspace/repo", title: null },
   };
   sessionMessagesResponse: OpenCodeResponse = { data: [] };
+  sessionMessagesImplementation:
+    | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
+    | null = null;
+  sessionPromptAsyncImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null =
+    null;
   sessionPromptAsyncEvents: unknown[] = [idleEvent()];
   sessionPromptAsyncResponse: OpenCodeResponse = {};
   sessionStatusResponse: OpenCodeResponse = { data: {} };
-  sessionStatusImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null = null;
+  sessionStatusImplementation:
+    | ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
+    | null = null;
   sessionSummarizeEvents: unknown[] = [idleEvent()];
   sessionSummarizeResponse: OpenCodeResponse = { data: {} };
   sessionUpdateResponse: OpenCodeResponse = {};
   private readonly queuedEventStream = createQueuedEventStream();
+  private eventObserver: ((event: unknown) => void) | null = null;
 
   constructor() {
     this.eventStream = this.queuedEventStream.stream;
@@ -143,6 +176,11 @@ export class TestOpenCodeClient {
 
   emitEvent(event: unknown): void {
     this.queuedEventStream.emit(event);
+    this.eventObserver?.(event);
+  }
+
+  observeEvents(observer: (event: unknown) => void): void {
+    this.eventObserver = observer;
   }
 
   asSdkClient(): OpencodeClient {
@@ -200,6 +238,12 @@ export class TestOpenCodeClient {
         },
       },
       permission: {
+        list: async (parameters: unknown, options: unknown) => {
+          this.calls.permissionList.push(parameters);
+          return this.permissionListImplementation
+            ? await this.permissionListImplementation(parameters, options)
+            : this.permissionListResponse;
+        },
         reply: async (parameters: unknown) => {
           this.calls.permissionReply.push(parameters);
           return this.permissionReplyResponse;
@@ -215,6 +259,12 @@ export class TestOpenCodeClient {
         },
       },
       question: {
+        list: async (parameters: unknown, options: unknown) => {
+          this.calls.questionList.push(parameters);
+          return this.questionListImplementation
+            ? await this.questionListImplementation(parameters, options)
+            : this.questionListResponse;
+        },
         reject: async (parameters: unknown) => {
           this.calls.questionReject.push(parameters);
           return this.questionRejectResponse;
@@ -249,10 +299,10 @@ export class TestOpenCodeClient {
           this.calls.sessionDelete.push(parameters);
           return this.sessionDeleteResponse;
         },
-        children: async (parameters: unknown) => {
+        children: async (parameters: unknown, options: unknown) => {
           this.calls.sessionChildren.push(parameters);
           if (this.sessionChildrenImplementation) {
-            return await this.sessionChildrenImplementation(parameters);
+            return await this.sessionChildrenImplementation(parameters, options);
           }
           return this.sessionChildrenResponses.shift() ?? { data: [] };
         },
@@ -260,21 +310,26 @@ export class TestOpenCodeClient {
           this.calls.sessionGet.push(parameters);
           return this.sessionGetResponse;
         },
-        messages: async (parameters: unknown) => {
+        messages: async (parameters: unknown, options: unknown) => {
           this.calls.sessionMessages.push(parameters);
-          return this.sessionMessagesResponse;
+          return this.sessionMessagesImplementation
+            ? await this.sessionMessagesImplementation(parameters, options)
+            : this.sessionMessagesResponse;
         },
         promptAsync: async (parameters: unknown) => {
           this.calls.sessionPromptAsync.push(parameters);
+          if (this.sessionPromptAsyncImplementation) {
+            return await this.sessionPromptAsyncImplementation(parameters);
+          }
           for (const event of this.sessionPromptAsyncEvents) {
             this.emitEvent(event);
           }
           return this.sessionPromptAsyncResponse;
         },
-        status: async (parameters: unknown) => {
+        status: async (parameters: unknown, options: unknown) => {
           this.calls.sessionStatus.push(parameters);
           return this.sessionStatusImplementation
-            ? await this.sessionStatusImplementation(parameters)
+            ? await this.sessionStatusImplementation(parameters, options)
             : this.sessionStatusResponse;
         },
         summarize: async (parameters: unknown) => {