fix(opencode): keep long v2 turns alive until execution completes (#5674)

The long-lived session wait hit Node’s headers timeout while OpenCode kept working. Reconcile execution events against active state and the durable log so transport observation failures do not fail healthy turns, including across event gaps and reconnects.

docs/providers.md

@@ -99,6 +99,8 @@ Pi RPC extension UI dialog requests (`select`, `input`, `editor`, `confirm`) are
 
 OpenCode adapters target v1.14.46 and v2.0.10. V2 rejects binaries older than the tested 2.0.10 SDK at runtime selection. Runtime selection uses the configured command and environment. A recognized version is cached until provider configuration reload; a failed, timed-out, or unrecognized probe retains the v1 path and retries detection on the next operation. Load v2 code and materialize its plugin only after positive v2 selection; v1 sessions remain undecorated. Keep upstream SDK types inside the version-specific adapter. OpenCode owns storage migration; a missing native session must fail resume rather than create a replacement. V2 has no native archive/unarchive operation: archiving affects Paseo only. V1 retains native archiving.
 
+Use OpenCode v2 execution events to trigger turn completion, with active-state and durable-log reconciliation after admission, reconnect, and while a turn remains active. Do not use `session.wait`: a healthy turn exceeding Node's HTTP headers deadline produces a transport error while OpenCode keeps working. The live event feed has no replay, and shutdown interruption preserves the previous idle outcome, so the session snapshot alone cannot recover missed execution events. Quiet streams are healthy: v2 heartbeats are SSE comments, not application events.
+
 V2.0.4 also removed the activation endpoint that gated a cold location, and a cold location registers its config-derived commands, skills, and providers asynchronously. Wait until `plugin.list` returns a populated inventory before reading the catalog or commands; an empty inventory means the location is still warming. Fail when the readiness deadline expires, including when an inventory request stalls.
 
 Paseo installs its OpenCode tool bridge through `OPENCODE_CONFIG_CONTENT`. V1 accepts a plugin file; v2 silently skips configured files and requires a package directory with a server entry point. Both versions use the daemon's private loopback bridge for caller-scoped tools. Bridge context lives only in daemon memory and is removed when the Paseo session closes. The content-addressed plugin artifacts contain no session data or secrets. V2 also needs this plugin when native Paseo tools are disabled: its prompt API has no structured-output format, so the plugin supplies a schema-validated final-answer tool.

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

@@ -3,6 +3,7 @@ import type {
   SessionInfo,
   SessionMessageInfo,
   SessionCreateInput,
+  SessionLogItem,
 } from "@opencode/client";
 import type { V2Api } from "../v2/api.js";
 import type { V2Connection } from "../v2/runtime.js";
@@ -28,17 +29,61 @@ export class V2Harness {
   readonly environments: Array<{ sessionID: string; variables: Record<string, string> }> = [];
   readonly mcpAdds: string[] = [];
   releases = 0;
+  active = false;
+  activeReads = 0;
+  autoComplete = true;
+  constructor(private sequence = 0) {}
+  private executions: SessionLogItem[] = [];
+  startExecution() {
+    this.active = true;
+    this.push({
+      id: `execution-${++this.sequence}`,
+      created: this.sequence,
+      type: "session.execution.started",
+      durable: { aggregateID: this.info.id, seq: this.sequence, version: 1 },
+      data: { sessionID: this.info.id },
+    });
+  }
+  finishExecution(outcome: "succeeded" | "interrupted" = "succeeded", publish = true) {
+    this.active = false;
+    this.info.outcome = outcome;
+    const common = {
+      id: `execution-${++this.sequence}`,
+      created: this.sequence,
+      durable: { aggregateID: this.info.id, seq: this.sequence, version: 1 as const },
+    };
+    this.push(
+      outcome === "succeeded"
+        ? { ...common, type: "session.execution.succeeded", data: { sessionID: this.info.id } }
+        : {
+            ...common,
+            type: "session.execution.interrupted",
+            data: { sessionID: this.info.id, reason: "user" },
+          },
+      publish,
+    );
+  }
   prompt: V2Api["session"]["prompt"] = async (input) => {
     this.prompts.push(input.text);
+    if (!this.autoComplete) this.startExecution();
   };
-  wait: V2Api["session"]["wait"] = async () => undefined;
   interrupt: V2Api["session"]["interrupt"] = async () => {
-    this.info.outcome = "interrupted";
+    this.finishExecution("interrupted");
     return { interrupted: true };
   };
   private pendingEvents: OpenCodeEvent[] = [];
   private notify: (() => void) | null = null;
-  push(event: OpenCodeEvent) {
+  push(event: OpenCodeEvent, publish = true) {
+    if (
+      event.type === "session.execution.started" ||
+      event.type === "session.execution.succeeded" ||
+      event.type === "session.execution.failed" ||
+      event.type === "session.execution.interrupted"
+    ) {
+      this.sequence = Math.max(this.sequence, event.durable.seq);
+      this.executions.push(event);
+    }
+    if (!publish) return;
     this.pendingEvents.push(event);
     this.notify?.();
   }
@@ -93,7 +138,10 @@ export class V2Harness {
       },
       get: async () => this.info,
       list: async () => ({ data: [], cursor: {} }),
-      active: async () => ({}),
+      active: async () => {
+        this.activeReads += 1;
+        return this.active ? { [this.info.id]: { type: "running" as const } } : {};
+      },
       remove: async () => undefined,
       switchAgent: unexpected,
       switchModel: unexpected,
@@ -101,13 +149,21 @@ export class V2Harness {
         this.environments.push(input);
       },
       instructions: { entry: { list: unexpected, put: unexpected, remove: unexpected } },
-      prompt: (input, options) => this.prompt(input, options),
-      wait: (input, options) => this.wait(input, options),
+      prompt: async (input, options) => {
+        const before = this.executions.length;
+        await this.prompt(input, options);
+        if (
+          this.autoComplete &&
+          this.executions.length === before &&
+          this.info.outcome !== "failed"
+        )
+          this.finishExecution();
+      },
       interrupt: (input, options) => this.interrupt(input, options),
       command: unexpected,
       compact: unexpected,
       revert: { stage: unexpected, clear: unexpected, commit: unexpected },
-      log: unexpected,
+      log: (input) => this.executionLog(input.after),
     },
     event: { subscribe: (options) => this.events(options?.signal) },
   };
@@ -121,6 +177,12 @@ export class V2Harness {
   };
   readonly runtime = { acquire: async () => this.connection, shutdown: async () => undefined };
 
+  private async *executionLog(after = 0) {
+    for (const event of this.executions) {
+      if ("durable" in event && event.durable.seq > after) yield event;
+    }
+  }
+
   private async *events(signal?: AbortSignal): AsyncGenerator<OpenCodeEvent> {
     yield { id: "connected", created: 1, type: "server.connected", data: {} };
     const wake = () => this.notify?.();

packages/server/src/server/agent/providers/opencode/v2/agent.test.ts

@@ -6,7 +6,289 @@ import type { AgentStreamEvent } from "../../../agent-sdk-types.js";
 import { OpenCodeV2AgentClient } from "./agent.js";
 import { V2Harness } from "../test-utils/v2-harness.js";
 
+function terminalEvents(events: AgentStreamEvent[]) {
+  return events.filter(
+    (event) =>
+      event.type === "turn_completed" ||
+      event.type === "turn_failed" ||
+      event.type === "turn_canceled",
+  );
+}
+
 describe("OpenCode v2 session lifecycle", () => {
+  test("completes from execution events without a long-lived HTTP wait", async () => {
+    const harness = new V2Harness();
+    harness.prompt = async (input) => {
+      harness.prompts.push(input.text);
+      harness.startExecution();
+    };
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    try {
+      await session.startTurn("long-running work");
+      await expect.poll(() => harness.prompts).toEqual(["long-running work"]);
+      expect(events.filter((event) => event.type === "turn_failed")).toEqual([]);
+      expect(events.filter((event) => event.type === "turn_completed")).toEqual([]);
+      harness.finishExecution();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+      expect(events.filter((event) => event.type === "turn_failed")).toEqual([]);
+    } finally {
+      await session.close();
+    }
+  });
+  test.each(["reconnect", "watchdog"])(
+    "recovers a missed completion through %s without reconnecting a quiet stream",
+    async (recovery) => {
+      const harness = new V2Harness();
+      harness.autoComplete = false;
+      let subscriptions = 0;
+      const subscribe = harness.api.event.subscribe;
+      harness.api.event.subscribe = (options) => {
+        subscriptions += 1;
+        return subscribe(options);
+      };
+      const client = new OpenCodeV2AgentClient({
+        logger: createTestLogger(),
+        runtime: harness.runtime,
+      });
+      const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+      const events: AgentStreamEvent[] = [];
+      session.subscribe((event) => events.push(event));
+      try {
+        await session.startTurn("finish while disconnected");
+        await expect.poll(() => harness.prompts).toHaveLength(1);
+        // Persisted completion without its event, as after an upstream stream gap.
+        harness.finishExecution("succeeded", false);
+        if (recovery === "reconnect")
+          harness.push({ id: "reconnect", created: 3, type: "server.connected", data: {} });
+        await expect.poll(() => terminalEvents(events), { timeout: 7_000 }).toHaveLength(1);
+        expect(terminalEvents(events)[0]).toMatchObject({ type: "turn_completed" });
+        expect(subscriptions).toBe(1);
+      } finally {
+        await session.close();
+      }
+    },
+    10_000,
+  );
+
+  test("does not finish from a previous outcome before prompt admission", async () => {
+    const harness = new V2Harness();
+    let accept!: () => void;
+    harness.prompt = () =>
+      new Promise<void>((resolve) => {
+        accept = resolve;
+      });
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    try {
+      await session.startTurn("fast execution");
+      const reads = harness.activeReads;
+      harness.push({ id: "reconnect", created: 3, type: "server.connected", data: {} });
+      await expect.poll(() => harness.activeReads).toBeGreaterThan(reads);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.startExecution();
+      harness.finishExecution();
+      accept();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+      expect(terminalEvents(events)[0]).toMatchObject({ type: "turn_completed" });
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("follows a queued successor until the session is idle", async () => {
+    const harness = new V2Harness();
+    harness.autoComplete = false;
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    try {
+      await session.startTurn("first and queued work");
+      await expect.poll(() => harness.prompts).toHaveLength(1);
+      harness.finishExecution();
+      harness.startExecution();
+      const reads = harness.activeReads;
+      harness.push({ id: "reconnect", created: 3, type: "server.connected", data: {} });
+      await expect.poll(() => harness.activeReads).toBeGreaterThan(reads);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.finishExecution();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("discards an idle snapshot overtaken by a new execution event", async () => {
+    const harness = new V2Harness();
+    harness.autoComplete = false;
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    let readSnapshot!: () => void;
+    let release!: () => void;
+    const reading = new Promise<void>((resolve) => {
+      readSnapshot = resolve;
+    });
+    const blocked = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    try {
+      await session.startTurn("work");
+      await expect.poll(() => harness.prompts).toHaveLength(1);
+      harness.api.session.get = async () => {
+        const snapshot = { ...harness.info };
+        readSnapshot();
+        await blocked;
+        return snapshot;
+      };
+      harness.finishExecution();
+      await reading;
+      const reads = harness.activeReads;
+      harness.startExecution();
+      release();
+      await expect.poll(() => harness.activeReads).toBeGreaterThan(reads);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.finishExecution();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+    } finally {
+      release();
+      await session.close();
+    }
+  });
+
+  test("does not treat shutdown interruption as a completed or canceled turn", async () => {
+    const harness = new V2Harness();
+    harness.autoComplete = false;
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    try {
+      await session.startTurn("resumable work");
+      await expect.poll(() => harness.prompts).toHaveLength(1);
+      harness.active = false;
+      // Shutdown does not project an idle outcome; the preceding run's outcome can remain.
+      harness.info.outcome = "succeeded";
+      harness.push({
+        id: "shutdown",
+        created: 3,
+        type: "session.execution.interrupted",
+        durable: { aggregateID: "session", seq: 3, version: 1 },
+        data: { sessionID: "session", reason: "shutdown" },
+      });
+      const reads = harness.activeReads;
+      harness.push({ id: "reconnect", created: 4, type: "server.connected", data: {} });
+      await expect.poll(() => harness.activeReads).toBeGreaterThan(reads);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.startExecution();
+      harness.finishExecution();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+      expect(terminalEvents(events)[0]).toMatchObject({ type: "turn_completed" });
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("recovers the latest execution from the durable log when shutdown and resume events were missed", async () => {
+    const harness = new V2Harness();
+    harness.autoComplete = false;
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    try {
+      await session.startTurn("resumable work");
+      await expect.poll(() => harness.prompts).toHaveLength(1);
+      harness.active = false;
+      harness.info.outcome = "succeeded";
+      harness.api.session.log = async function* () {
+        yield {
+          id: "shutdown",
+          created: 3,
+          type: "session.execution.interrupted",
+          durable: { aggregateID: "session", seq: 3, version: 1 },
+          data: { sessionID: "session", reason: "shutdown" },
+        };
+      };
+      const reads = harness.activeReads;
+      harness.push({ id: "reconnect", created: 4, type: "server.connected", data: {} });
+      await expect.poll(() => harness.activeReads).toBeGreaterThan(reads);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.api.session.log = async function* () {
+        yield {
+          id: "done",
+          created: 6,
+          type: "session.execution.succeeded",
+          durable: { aggregateID: "session", seq: 6, version: 1 },
+          data: { sessionID: "session" },
+        };
+      };
+      harness.push({
+        id: "reconnect-after-resume",
+        created: 7,
+        type: "server.connected",
+        data: {},
+      });
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+      expect(terminalEvents(events)[0]).toMatchObject({ type: "turn_completed" });
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("keeps the turn alive through an observation transport failure", async () => {
+    const harness = new V2Harness();
+    harness.autoComplete = false;
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    const active = harness.api.session.active;
+    let attempted = false;
+    harness.api.session.active = async () => {
+      attempted = true;
+      throw new Error("fetch failed");
+    };
+    try {
+      await session.startTurn("work");
+      await expect.poll(() => attempted).toBe(true);
+      expect(terminalEvents(events)).toEqual([]);
+      harness.api.session.active = active;
+      harness.finishExecution();
+      await expect.poll(() => terminalEvents(events)).toHaveLength(1);
+      expect(terminalEvents(events)[0]).toMatchObject({ type: "turn_completed" });
+    } finally {
+      await session.close();
+    }
+  });
+
   test.each([
     { variants: ["medium", "custom"], selected: "medium", expected: "medium" },
     { variants: ["custom"], selected: "custom", expected: "custom" },
@@ -169,7 +451,7 @@ describe("OpenCode v2 session lifecycle", () => {
 
   test("reconnects after a helper exits and restores session configuration on the next turn", async () => {
     const first = new V2Harness();
-    const second = new V2Harness();
+    const second = new V2Harness(1);
     let exit!: (error: Error) => void;
     const exited = new Promise<Error>((resolve) => {
       exit = resolve;
@@ -185,7 +467,7 @@ describe("OpenCode v2 session lifecycle", () => {
         shutdown: async () => undefined,
       },
     });
-    first.wait = () => new Promise<void>(() => undefined);
+    first.autoComplete = false;
     second.prompt = async (input) => {
       second.history.push({
         id: "answer",
@@ -361,11 +643,8 @@ describe("OpenCode v2 session lifecycle", () => {
 
   test("reconciles reconnect snapshots without replaying text or losing repeated chunks", async () => {
     const harness = new V2Harness();
-    let finish!: () => void;
-    harness.wait = () =>
-      new Promise<void>((resolve) => {
-        finish = resolve;
-      });
+    harness.autoComplete = false;
+    const finish = () => harness.finishExecution();
     const client = new OpenCodeV2AgentClient({
       logger: createTestLogger(),
       runtime: harness.runtime,
@@ -378,7 +657,7 @@ describe("OpenCode v2 session lifecycle", () => {
     });
     try {
       const running = session.run("laugh");
-      await expect.poll(() => typeof finish).toBe("function");
+      await expect.poll(() => harness.prompts).toEqual(["laugh"]);
       const answer = {
         id: "answer",
         type: "assistant",
@@ -401,7 +680,7 @@ describe("OpenCode v2 session lifecycle", () => {
     }
   });
 
-  test("reports a terminal failure when reading the execution error fails", async () => {
+  test("retries observation failures without replacing the execution error", async () => {
     const harness = new V2Harness();
     harness.info.outcome = "failed";
     const client = new OpenCodeV2AgentClient({
@@ -412,9 +691,27 @@ describe("OpenCode v2 session lifecycle", () => {
     const events: AgentStreamEvent[] = [];
     session.subscribe((event) => events.push(event));
     try {
+      let attempted = false;
+      harness.api.session.log = () => {
+        attempted = true;
+        throw new Error("log unavailable");
+      };
       await session.startTurn("hello");
       const failures = () => events.filter((event) => event.type === "turn_failed");
+      await expect.poll(() => attempted).toBe(true);
+      expect(failures()).toEqual([]);
+      harness.api.session.log = async function* () {
+        yield {
+          id: "failure",
+          created: 2,
+          durable: { aggregateID: "session", seq: 2, version: 1 },
+          type: "session.execution.failed",
+          data: { sessionID: "session", error: { type: "provider", message: "quota exhausted" } },
+        };
+      };
+      harness.push({ id: "reconnect", created: 3, type: "server.connected", data: {} });
       await expect.poll(failures).toHaveLength(1);
+      expect(failures()[0]).toMatchObject({ error: "quota exhausted" });
       expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(0);
     } finally {
       await session.close();
@@ -424,16 +721,13 @@ describe("OpenCode v2 session lifecycle", () => {
   test("waits for prompt acceptance before interrupting", async () => {
     const harness = new V2Harness();
     let accept!: () => void;
-    let settle!: () => void;
+    const settle = () => harness.finishExecution("interrupted");
     let interruptions = 0;
     harness.prompt = () =>
       new Promise<void>((resolve) => {
         accept = resolve;
       });
-    harness.wait = () =>
-      new Promise<void>((resolve) => {
-        settle = resolve;
-      });
+
     harness.interrupt = async () => {
       interruptions += 1;
       harness.info.outcome = "interrupted";
@@ -494,12 +788,9 @@ describe("OpenCode v2 session lifecycle", () => {
 
   test("waits for interruption settlement before submitting replacement work", async () => {
     const harness = new V2Harness();
-    let finishFirst!: () => void;
+    harness.autoComplete = false;
+    const finishFirst = () => harness.finishExecution("interrupted");
     let finishStop!: () => void;
-    harness.wait = () =>
-      new Promise<void>((resolve) => {
-        finishFirst = resolve;
-      });
     harness.interrupt = () =>
       new Promise((resolve) => {
         finishStop = () => {
@@ -515,12 +806,12 @@ describe("OpenCode v2 session lifecycle", () => {
     const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
     try {
       await session.startTurn("first");
-      await expect.poll(() => typeof finishFirst).toBe("function");
+      await expect.poll(() => harness.prompts).toEqual(["first"]);
       const stopping = session.interrupt();
       const replacement = session.startTurn("second");
       await Promise.resolve();
       expect(harness.prompts).toEqual(["first"]);
-      harness.wait = async () => undefined;
+      harness.autoComplete = true;
       finishStop();
       await stopping;
       await replacement;
@@ -587,11 +878,8 @@ describe("OpenCode v2 session lifecycle", () => {
 
   test("refuses replacement work after a failed stop until Stop succeeds", async () => {
     const harness = new V2Harness();
-    let finishFirst!: () => void;
-    harness.wait = () =>
-      new Promise<void>((resolve) => {
-        finishFirst = resolve;
-      });
+    harness.autoComplete = false;
+    const finishFirst = () => harness.finishExecution("interrupted");
     harness.interrupt = async () => {
       throw new Error("stop failed");
     };
@@ -602,7 +890,7 @@ describe("OpenCode v2 session lifecycle", () => {
     const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
     try {
       await session.startTurn("first");
-      await expect.poll(() => typeof finishFirst).toBe("function");
+      await expect.poll(() => harness.prompts).toEqual(["first"]);
       await expect(session.interrupt()).rejects.toThrow("stop failed");
       await expect(session.startTurn("unsafe replacement")).rejects.toThrow("stop failed");
       expect(harness.prompts).toEqual(["first"]);
@@ -611,7 +899,7 @@ describe("OpenCode v2 session lifecycle", () => {
         return { interrupted: true };
       };
       await session.interrupt();
-      harness.wait = async () => undefined;
+      harness.autoComplete = true;
       await session.startTurn("retry");
       await expect.poll(() => harness.prompts).toEqual(["first", "retry"]);
     } finally {

packages/server/src/server/agent/providers/opencode/v2/api.ts

@@ -16,7 +16,6 @@ export interface V2Api {
     | "prompt"
     | "command"
     | "compact"
-    | "wait"
     | "interrupt"
     | "revert"
     | "log"

packages/server/src/server/agent/providers/opencode/v2/session.ts

@@ -91,6 +91,11 @@ export class OpenCodeV2Session implements AgentSession {
         await this.reconcile();
         return { info: this.info, history: this.history };
       },
+      reportReconciliationError: (error) =>
+        this.logger.warn(
+          { error: toDiagnosticErrorMessage(error) },
+          "OpenCode turn reconciliation failed; retrying",
+        ),
       clearPermissions: async () => {
         for (const request of this.permissions.list())
           await this.permissions.respondToPermission(request.id, { behavior: "deny" });
@@ -417,8 +422,7 @@ export class OpenCodeV2Session implements AgentSession {
       await this.client.session.environment({ sessionID: this.id, variables: this.launchEnv });
     await this.reconcile();
     await this.children.reconcile(this.id);
-    const active = await this.client.session.active();
-    if (active[this.id]) this.turns.observeActiveTurn();
+    await this.turns.reconcile();
   }
   private scheduleReconcile() {
     if (this.refreshTimer || this.closed) return;
@@ -461,11 +465,8 @@ export class OpenCodeV2Session implements AgentSession {
       return;
     }
     this.permissions.observe(event);
-    if (event.type === "session.execution.failed")
-      this.turns.executionError = event.data.error.message;
+    this.turns.observe(event);
     this.scheduleReconcile();
-    if (event.type !== "session.execution.started" || this.turns.id) return;
-    this.turns.observeActiveTurn();
   }
   private async consume(ready: () => void, fail: (error: unknown) => void) {
     const signal = AbortSignal.any([this.abort.signal, this.streamAbort.signal]);
@@ -502,6 +503,7 @@ export class OpenCodeV2Session implements AgentSession {
     if (this.closed) return;
     this.closed = true;
     this.abort.abort();
+    this.turns.close();
     this.streamAbort.abort();
     if (this.refreshTimer) clearTimeout(this.refreshTimer);
     await this.stream;

packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts

@@ -121,11 +121,8 @@ describe("OpenCode v2 token streaming", () => {
 
   test("withholds streamed prose during a structured-output turn", async () => {
     const harness = new V2Harness();
-    let settle!: () => void;
-    harness.wait = () =>
-      new Promise<void>((resolve) => {
-        settle = resolve;
-      });
+    harness.autoComplete = false;
+    const settle = () => harness.finishExecution();
     const client = new OpenCodeV2AgentClient({
       logger: createTestLogger(),
       runtime: harness.runtime,

packages/server/src/server/agent/providers/opencode/v2/turns.ts

@@ -1,5 +1,5 @@
 import { structuredOutput } from "./structured-output.js";
-import type { SessionInfo, SessionMessageInfo } from "@opencode/client";
+import type { OpenCodeEvent, SessionInfo, SessionMessageInfo } from "@opencode/client";
 
 import { randomUUID } from "node:crypto";
 
@@ -20,6 +20,25 @@ import { usageFromV2 } from "./mapping.js";
 import { commands } from "./commands.js";
 
 import type { V2Api } from "./api.js";
+type ExecutionEvent = Extract<
+  OpenCodeEvent,
+  {
+    type:
+      | "session.execution.started"
+      | "session.execution.succeeded"
+      | "session.execution.failed"
+      | "session.execution.interrupted";
+  }
+>;
+function isExecutionEvent(event: { type: string }): event is ExecutionEvent {
+  return (
+    event.type === "session.execution.started" ||
+    event.type === "session.execution.succeeded" ||
+    event.type === "session.execution.failed" ||
+    event.type === "session.execution.interrupted"
+  );
+}
+
 interface TurnSnapshot {
   info: SessionInfo;
   history: SessionMessageInfo[];
@@ -32,19 +51,27 @@ interface TurnOptions {
   emit(event: AgentStreamEvent): void;
   reconcile(): Promise<TurnSnapshot>;
   clearPermissions(): Promise<void>;
+  reportReconciliationError(error: unknown): void;
 }
 interface Turn {
   output?: ReturnType<typeof structuredOutput>;
   id: string;
   submitted: Promise<void>;
   completion: Promise<void>;
+  settle(): void;
+  accepted: boolean;
 }
 
 export class SessionTurns {
   private turn: Turn | null = null;
   private stopping: Promise<void> | null = null;
   private stopFailed = false;
-  executionError: string | null = null;
+  private revision = 0;
+  private execution: ExecutionEvent | null = null;
+  private sequence = 0;
+  private refresh: Promise<void> | null = null;
+  private dirty = false;
+  private retryTimer: ReturnType<typeof setTimeout> | null = null;
   constructor(private readonly options: TurnOptions) {}
   get id() {
     return this.turn?.id;
@@ -54,21 +81,49 @@ export class SessionTurns {
   }
   fail(error: Error) {
     const turn = this.turn;
+    if (!turn) return;
+    this.release(turn);
+    this.options.emit({
+      type: "turn_failed",
+      provider: "opencode",
+      turnId: turn.id,
+      error: toDiagnosticErrorMessage(error),
+    });
+  }
+  private release(turn: Turn) {
     this.turn = null;
-    if (turn)
-      this.options.emit({
-        type: "turn_failed",
-        provider: "opencode",
-        turnId: turn.id,
-        error: error.message,
-      });
+    this.revision += 1;
+    if (this.retryTimer) clearTimeout(this.retryTimer);
+    this.retryTimer = null;
+    turn.settle();
+  }
+  close() {
+    if (this.turn) this.release(this.turn);
+  }
+  private createTurn(submitted: Promise<void>, accepted: boolean, output?: Turn["output"]): Turn {
+    let settle!: () => void;
+    const completion = new Promise<void>((resolve) => {
+      settle = resolve;
+    });
+    const turn: Turn = {
+      id: randomUUID(),
+      submitted,
+      completion,
+      settle,
+      accepted,
+      output,
+    };
+    this.turn = turn;
+    this.execution = null;
+    this.revision += 1;
+    this.options.emit({ type: "turn_started", provider: "opencode", turnId: turn.id });
+    this.scheduleRecovery();
+    return turn;
   }
   async startTurn(prompt: AgentPromptInput, options?: AgentRunOptions) {
     await this.stopping;
     if (this.options.signal.aborted) throw new Error("OpenCode session is closed");
     if (this.turn) throw new Error("OpenCode session already has an active turn");
-    this.executionError = null;
-    const id = randomUUID();
     const input = this.promptInput(prompt);
     let accept!: () => void;
     const submitted = new Promise<void>((resolve) => {
@@ -76,38 +131,106 @@ export class SessionTurns {
     });
     const output =
       options?.outputSchema === undefined ? undefined : structuredOutput(options.outputSchema);
-    const completion = this.submit(id, input, accept, options, output);
-    this.turn = { id, submitted, completion, output };
-    return { turnId: id };
+    const turn = this.createTurn(submitted, false, output);
+    void this.submit(turn, input, accept, options);
+    return { turnId: turn.id };
   }
   private async submit(
-    id: string,
+    turn: Turn,
     input: ReturnType<SessionTurns["promptInput"]>,
     accept: () => void,
     options?: AgentRunOptions,
-    output?: ReturnType<typeof structuredOutput>,
   ) {
-    // Defer until startTurn publishes ownership, including for immediately resolved test transports.
-    await Promise.resolve();
-    this.options.emit({ type: "turn_started", provider: "opencode", turnId: id });
     try {
-      await this.dispatch(input, options, output);
-      accept();
-      await this.finish(id);
+      await this.dispatch(input, options, turn.output);
+      if (this.turn !== turn) return;
+      turn.accepted = true;
+      this.revision += 1;
+      // Covers a fast execution whose terminal event preceded the admission response.
+      this.requestReconciliation();
     } catch (error) {
-      if (this.turn?.id === id) {
-        this.turn = null;
-        this.options.emit({
-          type: "turn_failed",
-          provider: "opencode",
-          turnId: id,
-          error: toDiagnosticErrorMessage(error),
-        });
-      }
+      if (this.turn === turn) this.fail(new Error(toDiagnosticErrorMessage(error)));
     } finally {
       accept();
     }
   }
+  observe(event: OpenCodeEvent) {
+    if (!isExecutionEvent(event) || event.durable.seq <= this.sequence) return;
+    this.revision += 1;
+    if (event.type === "session.execution.started") this.observeActiveTurn();
+    this.execution = event;
+    this.sequence = event.durable.seq;
+    this.requestReconciliation();
+  }
+  private requestReconciliation() {
+    void this.reconcile().catch((error: unknown) => this.options.reportReconciliationError(error));
+  }
+  // Recovery reads state; a quiet stream (including SSE comment heartbeats) is not a reason to reconnect.
+  private scheduleRecovery() {
+    if (this.retryTimer || !this.turn || this.options.signal.aborted) return;
+    this.retryTimer = setTimeout(() => {
+      this.retryTimer = null;
+      this.requestReconciliation();
+    }, 5_000);
+    this.retryTimer.unref();
+  }
+  reconcile(): Promise<void> {
+    this.dirty = true;
+    if (this.refresh) return this.refresh;
+    this.refresh = Promise.resolve()
+      .then(() => this.drainReconciliation())
+      .finally(() => {
+        this.refresh = null;
+        this.scheduleRecovery();
+      });
+    return this.refresh;
+  }
+  private async drainReconciliation() {
+    while (this.dirty && !this.options.signal.aborted) {
+      this.dirty = false;
+      await this.reconcileTurn();
+    }
+  }
+  private async reconcileTurn() {
+    const revision = this.revision;
+    const turn = this.turn;
+    const active = await this.options.client().session.active({ signal: this.options.signal });
+    if (revision !== this.revision) {
+      this.dirty = true;
+      return;
+    }
+    if (active[this.options.id]) {
+      this.observeActiveTurn();
+      return;
+    }
+    if (!turn?.accepted) return;
+    // Idle outcome survives shutdown, and /event has no replay. Recover the latest
+    // durable execution even when its start or terminal was lost across reconnect.
+    const execution = await this.readExecution();
+    if (revision !== this.revision) {
+      this.dirty = true;
+      return;
+    }
+    if (!execution) return;
+    this.execution = execution;
+    this.sequence = execution.durable.seq;
+    if (
+      execution.type === "session.execution.started" ||
+      (execution.type === "session.execution.interrupted" && execution.data.reason === "shutdown")
+    )
+      return;
+    // Read the final history only after observing idle. Never turn an observation transport error into an execution failure.
+    const snapshot = await this.options.reconcile();
+    if (revision !== this.revision) {
+      this.dirty = true;
+      return;
+    }
+    try {
+      this.finish(turn, snapshot, execution);
+    } catch (error) {
+      if (this.turn === turn) this.fail(new Error(toDiagnosticErrorMessage(error)));
+    }
+  }
   private async dispatch(
     input: ReturnType<SessionTurns["promptInput"]>,
     options?: AgentRunOptions,
@@ -147,51 +270,43 @@ export class SessionTurns {
       },
     });
   }
-  private async finish(id: string) {
-    await this.options
-      .client()
-      .session.wait({ sessionID: this.options.id }, { signal: this.options.signal });
-    const { info, history } = await this.options.reconcile();
-    if (this.turn?.id !== id) return;
-    if (info.outcome !== "failed" && info.outcome !== "interrupted")
-      this.turn.output?.assert(history);
-    const failure =
-      info.outcome === "failed" ? (this.executionError ?? (await this.readExecutionError())) : null;
-    if (this.turn?.id !== id) return;
-    this.turn = null;
-    if (info.outcome === "interrupted")
+  private finish(turn: Turn, { info, history }: TurnSnapshot, execution: ExecutionEvent) {
+    if (execution.type === "session.execution.succeeded") turn.output?.assert(history);
+    this.release(turn);
+    if (execution.type === "session.execution.interrupted")
       this.options.emit({
         type: "turn_canceled",
         provider: "opencode",
-        turnId: id,
+        turnId: turn.id,
         reason: "OpenCode interrupted execution",
       });
-    else if (info.outcome === "failed")
+    else if (execution.type === "session.execution.failed")
       this.options.emit({
         type: "turn_failed",
         provider: "opencode",
-        turnId: id,
-        error: failure ?? "OpenCode execution failed",
+        turnId: turn.id,
+        error: execution.data.error.message,
       });
     else
       this.options.emit({
         type: "turn_completed",
         provider: "opencode",
-        turnId: id,
+        turnId: turn.id,
         usage: usageFromV2(info),
       });
   }
-  private async readExecutionError(): Promise<string> {
-    let message = "OpenCode execution failed";
+  private async readExecution(): Promise<ExecutionEvent | null> {
+    let execution = this.execution;
     for await (const event of this.options
       .client()
       .session.log(
-        { sessionID: this.options.id, follow: false },
+        { sessionID: this.options.id, follow: false, after: this.sequence },
         { signal: this.options.signal },
       )) {
-      if (event.type === "session.execution.failed") message = event.data.error.message;
+      if (isExecutionEvent(event) && event.durable.seq > (execution?.durable.seq ?? this.sequence))
+        execution = event;
     }
-    return message;
+    return execution;
   }
   private promptInput(prompt: AgentPromptInput) {
     if (typeof prompt === "string") return { text: prompt, files: [] };
@@ -229,6 +344,7 @@ export class SessionTurns {
         // A stop sent before the queued prompt is accepted would interrupt an idle session.
         await turn?.submitted;
         await this.options.client().session.interrupt({ sessionID: this.options.id });
+        await this.reconcile();
         await turn?.completion;
       })();
       this.stopping = stop;
@@ -241,22 +357,8 @@ export class SessionTurns {
       }
     } else await this.stopping;
   }
-  observeActiveTurn() {
-    if (this.turn) return;
-    const id = randomUUID();
-    const completion = Promise.resolve()
-      .then(() => this.finish(id))
-      .catch((error: unknown) => {
-        if (this.turn?.id !== id) return;
-        this.turn = null;
-        this.options.emit({
-          type: "turn_failed",
-          provider: "opencode",
-          turnId: id,
-          error: toDiagnosticErrorMessage(error),
-        });
-      });
-    this.turn = { id, submitted: Promise.resolve(), completion };
-    this.options.emit({ type: "turn_started", provider: "opencode", turnId: id });
+  private observeActiveTurn() {
+    if (this.turn || this.options.signal.aborted) return;
+    this.createTurn(Promise.resolve(), true);
   }
 }