fix(server): address OpenCode recovery review findings (#904)

* fix(server): pin OpenCode SDK version

* fix(server): tolerate transient OpenCode recovery poll errors

* fix(server): address OpenCode recovery findings 2 and 4

* fix(server): address OpenCode recovery finding 1

* fix(server): address OpenCode recovery finding 3

package-lock.json

@@ -39157,7 +39157,7 @@
         "@mariozechner/pi-ai": "^0.70.2",
         "@mariozechner/pi-coding-agent": "^0.70.2",
         "@modelcontextprotocol/sdk": "^1.20.1",
-        "@opencode-ai/sdk": "^1.14.46",
+        "@opencode-ai/sdk": "1.14.46",
         "@sctg/sentencepiece-js": "^1.1.0",
         "@xterm/headless": "^6.0.0",
         "ai": "5.0.78",

packages/server/package.json

@@ -65,7 +65,7 @@
     "@mariozechner/pi-ai": "^0.70.2",
     "@mariozechner/pi-coding-agent": "^0.70.2",
     "@modelcontextprotocol/sdk": "^1.20.1",
-    "@opencode-ai/sdk": "^1.14.46",
+    "@opencode-ai/sdk": "1.14.46",
     "@sctg/sentencepiece-js": "^1.1.0",
     "@xterm/headless": "^6.0.0",
     "ai": "5.0.78",

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

@@ -664,6 +664,182 @@ describe("OpenCode adapter startTurn error handling", () => {
     rmSync(storageRoot, { recursive: true, force: true });
   });
 
+  test("continues SSE EOF recovery when one messages API poll rejects", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    let messagesCallCount = 0;
+    const completedMessagesResponse = {
+      data: [
+        {
+          info: buildAssistantMessageInfo({
+            id: "msg_assistant",
+            createdAt: 2100,
+            completedAt: 2500,
+            tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 }, total: 15 },
+          }),
+          parts: [buildTextPart("msg_assistant", "prt_text", "Recovered after retry")],
+        },
+      ],
+      error: undefined,
+    };
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockImplementation(async () => {
+          messagesCallCount += 1;
+          if (messagesCallCount === 1) {
+            throw new Error("transient messages failure");
+          }
+          return completedMessagesResponse;
+        }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(turn.assistantMessages.map((message) => message.text).join("")).toBe(
+      "Recovered after retry",
+    );
+    expect(messagesCallCount).toBe(2);
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
+  test("recovers structured-only assistant completions from messages API", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const startedAt = Date.now();
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const completedMessagesResponse = {
+      data: [
+        {
+          info: {
+            ...(buildAssistantMessageInfo({
+              id: "msg_assistant",
+              createdAt: startedAt + 1,
+              completedAt: startedAt + 500,
+              tokens: {
+                input: 10,
+                output: 5,
+                reasoning: 0,
+                cache: { read: 0, write: 0 },
+                total: 15,
+              },
+            }) as Record<string, unknown>),
+            structured: { status: "ok", files: ["README.md"] },
+          },
+          parts: [],
+        },
+      ],
+      error: undefined,
+    };
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockResolvedValue(completedMessagesResponse),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 1, pollIntervalMs: 1, livenessMs: 1 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(turn.assistantMessages).toEqual([
+      { type: "assistant_message", text: '{"status":"ok","files":["README.md"]}' },
+    ]);
+
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
   test("keeps SSE EOF as turn_failed when messages API never returns a completion before the cap", async () => {
     const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
     const cwd = "/tmp/test";
@@ -727,6 +903,95 @@ describe("OpenCode adapter startTurn error handling", () => {
     rmSync(storageRoot, { recursive: true, force: true });
   });
 
+  test("aborts the OpenCode session when recovery caps an in-progress assistant message", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const abort = vi.fn().mockResolvedValue({ data: true, error: undefined });
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockResolvedValue({
+          data: [
+            {
+              info: {
+                ...(buildAssistantMessageInfo({
+                  id: "msg_assistant",
+                  createdAt: 2100,
+                  completedAt: 2500,
+                  tokens: {
+                    input: 0,
+                    output: 0,
+                    reasoning: 0,
+                    cache: { read: 0, write: 0 },
+                    total: 0,
+                  },
+                }) as Record<string, unknown>),
+                time: { created: 2100 },
+              },
+              parts: [],
+            },
+          ],
+          error: undefined,
+        }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort,
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 0, pollIntervalMs: 1, livenessMs: 1_000 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(false);
+    expect(turn.turnFailed).toBe(true);
+    expect(abort).toHaveBeenCalledWith({
+      sessionID: "ses_unit_test",
+      directory: cwd,
+    });
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
   test("ignores assistant messages that completed before the turn started", async () => {
     const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
     const cwd = "/tmp/test";
@@ -1196,6 +1461,311 @@ describe("OpenCode adapter startTurn error handling", () => {
     rmSync(storageRoot, { recursive: true, force: true });
   });
 
+  test("emits only new reasoning text when recovered reasoning parts grow across polls", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    let messagesCallCount = 0;
+    const inProgressAssistantInfo = {
+      ...(buildAssistantMessageInfo({
+        id: "msg_assistant",
+        createdAt: 2100,
+        completedAt: 2500,
+        tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 },
+      }) as Record<string, unknown>),
+      time: { created: 2100 },
+    };
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockImplementation(async () => {
+          messagesCallCount += 1;
+          if (messagesCallCount === 1) {
+            return {
+              data: [
+                {
+                  info: inProgressAssistantInfo,
+                  parts: [
+                    {
+                      id: "prt_reasoning",
+                      sessionID: "ses_unit_test",
+                      messageID: "msg_assistant",
+                      type: "reasoning",
+                      text: "Thinking",
+                    },
+                  ],
+                },
+              ],
+              error: undefined,
+            };
+          }
+          if (messagesCallCount === 2) {
+            return {
+              data: [
+                {
+                  info: inProgressAssistantInfo,
+                  parts: [
+                    {
+                      id: "prt_reasoning",
+                      sessionID: "ses_unit_test",
+                      messageID: "msg_assistant",
+                      type: "reasoning",
+                      text: "Thinking more",
+                    },
+                  ],
+                },
+              ],
+              error: undefined,
+            };
+          }
+          return {
+            data: [
+              {
+                info: buildAssistantMessageInfo({
+                  id: "msg_assistant",
+                  createdAt: 2100,
+                  completedAt: 2700,
+                  tokens: {
+                    input: 10,
+                    output: 5,
+                    reasoning: 0,
+                    cache: { read: 0, write: 0 },
+                    total: 15,
+                  },
+                }),
+                parts: [
+                  {
+                    id: "prt_reasoning",
+                    sessionID: "ses_unit_test",
+                    messageID: "msg_assistant",
+                    type: "reasoning",
+                    text: "Thinking more",
+                  },
+                  buildTextPart("msg_assistant", "prt_text", "Done."),
+                ],
+              },
+            ],
+            error: undefined,
+          };
+        }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(
+      turn.allTimelineItems.filter((item) => item.type === "reasoning").map((item) => item.text),
+    ).toEqual(["Thinking", " more"]);
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
+  test("emits running tool calls again when recovered tool input changes", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    let messagesCallCount = 0;
+    const inProgressAssistantInfo = {
+      ...(buildAssistantMessageInfo({
+        id: "msg_assistant",
+        createdAt: 2100,
+        completedAt: 2500,
+        tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 },
+      }) as Record<string, unknown>),
+      time: { created: 2100 },
+    };
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockImplementation(async () => {
+          messagesCallCount += 1;
+          if (messagesCallCount === 1) {
+            return {
+              data: [
+                {
+                  info: inProgressAssistantInfo,
+                  parts: [
+                    {
+                      id: "prt_tool",
+                      sessionID: "ses_unit_test",
+                      messageID: "msg_assistant",
+                      type: "tool",
+                      tool: "bash",
+                      callID: "call_long",
+                      state: { status: "running", input: { command: "echo one" } },
+                    },
+                  ],
+                },
+              ],
+              error: undefined,
+            };
+          }
+          if (messagesCallCount === 2) {
+            return {
+              data: [
+                {
+                  info: inProgressAssistantInfo,
+                  parts: [
+                    {
+                      id: "prt_tool",
+                      sessionID: "ses_unit_test",
+                      messageID: "msg_assistant",
+                      type: "tool",
+                      tool: "bash",
+                      callID: "call_long",
+                      state: { status: "running", input: { command: "echo two" } },
+                    },
+                  ],
+                },
+              ],
+              error: undefined,
+            };
+          }
+          return {
+            data: [
+              {
+                info: buildAssistantMessageInfo({
+                  id: "msg_assistant",
+                  createdAt: 2100,
+                  completedAt: 2700,
+                  tokens: {
+                    input: 10,
+                    output: 5,
+                    reasoning: 0,
+                    cache: { read: 0, write: 0 },
+                    total: 15,
+                  },
+                }),
+                parts: [
+                  {
+                    id: "prt_tool",
+                    sessionID: "ses_unit_test",
+                    messageID: "msg_assistant",
+                    type: "tool",
+                    tool: "bash",
+                    callID: "call_long",
+                    state: {
+                      status: "completed",
+                      input: { command: "echo two" },
+                      output: "two",
+                    },
+                  },
+                  buildTextPart("msg_assistant", "prt_text", "Done."),
+                ],
+              },
+            ],
+            error: undefined,
+          };
+        }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(
+      turn.toolCalls
+        .filter((toolCall) => toolCall.status === "running")
+        .map((toolCall) => toolCall.detail),
+    ).toEqual([
+      { type: "shell", command: "echo one" },
+      { type: "shell", command: "echo two" },
+    ]);
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
   test("emits permission_requested for pending OpenCode questions while the turn is still in progress", async () => {
     const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
     const cwd = "/tmp/test";
@@ -1352,6 +1922,134 @@ describe("OpenCode adapter startTurn error handling", () => {
     rmSync(storageRoot, { recursive: true, force: true });
   });
 
+  test("keeps recovering past the completion cap while a question is pending", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    let messagesCallCount = 0;
+    const inProgressAssistantInfo = {
+      ...(buildAssistantMessageInfo({
+        id: "msg_assistant",
+        createdAt: 2100,
+        completedAt: 2500,
+        tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 },
+      }) as Record<string, unknown>),
+      time: { created: 2100 },
+    };
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      question: {
+        list: vi.fn().mockResolvedValue({
+          data: [
+            {
+              id: "qst_1",
+              sessionID: "ses_unit_test",
+              questions: [
+                {
+                  question: "Pick a color",
+                  header: "Color",
+                  options: [{ label: "red" }, { label: "blue" }],
+                },
+              ],
+            },
+          ],
+          error: undefined,
+        }),
+      },
+      permission: {
+        list: vi.fn().mockResolvedValue({ data: [], error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        messages: vi.fn().mockImplementation(async () => {
+          messagesCallCount += 1;
+          if (messagesCallCount < 3) {
+            return {
+              data: [
+                {
+                  info: inProgressAssistantInfo,
+                  parts: [],
+                },
+              ],
+              error: undefined,
+            };
+          }
+          return {
+            data: [
+              {
+                info: buildAssistantMessageInfo({
+                  id: "msg_assistant",
+                  createdAt: 2100,
+                  completedAt: 2700,
+                  tokens: {
+                    input: 10,
+                    output: 5,
+                    reasoning: 0,
+                    cache: { read: 0, write: 0 },
+                    total: 15,
+                  },
+                }),
+                parts: [buildTextPart("msg_assistant", "prt_text", "Done")],
+              },
+            ],
+            error: undefined,
+          };
+        }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      recovery: { timeoutMs: 0, pollIntervalMs: 1, livenessMs: 1_000 },
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(turn.assistantMessages.map((message) => message.text).join("")).toBe("Done");
+    expect(turn.events.filter((event) => event.type === "permission_requested")).toHaveLength(1);
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
   test("emits tool calls and reasoning from the recovered assistant message parts", async () => {
     const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
     const cwd = "/tmp/test";
@@ -1659,6 +2357,59 @@ describe("OpenCode adapter startTurn error handling", () => {
       expect(failed.error).toContain("boom: synchronous throw");
     }
   });
+
+  test("delays the next prompt until a slow interrupt abort settles", async () => {
+    vi.useFakeTimers();
+    const abortDeferred = createTestDeferred<{ data: boolean; error: undefined }>();
+    const promptAsync = vi.fn().mockResolvedValue({ data: {}, error: undefined });
+    const abort = vi
+      .fn()
+      .mockReturnValueOnce(abortDeferred.promise)
+      .mockResolvedValue({ data: true, error: undefined });
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockImplementation(
+          async (
+            _params: { directory: string },
+            options: { signal: AbortSignal },
+          ): Promise<{ stream: AsyncIterable<OpenCodeEvent> }> => ({
+            stream: abortableOpenCodeStream(options.signal),
+          }),
+        ),
+      },
+      session: {
+        promptAsync,
+        abort,
+      },
+    } as never;
+
+    const session = new __openCodeInternals.OpenCodeAgentSession(
+      { provider: "opencode", cwd: "/tmp/test" },
+      fakeClient,
+      "ses_unit_test",
+      createTestLogger(),
+      "/tmp/opencode-storage",
+    );
+
+    await session.startTurn("first");
+    expect(promptAsync).toHaveBeenCalledTimes(1);
+
+    const interruptPromise = session.interrupt();
+    await vi.advanceTimersByTimeAsync(2_000);
+    await interruptPromise;
+    expect(abort).toHaveBeenCalledTimes(1);
+
+    const secondTurnPromise = session.startTurn("second");
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(promptAsync).toHaveBeenCalledTimes(1);
+
+    abortDeferred.resolve({ data: true, error: undefined });
+    await secondTurnPromise;
+    expect(promptAsync).toHaveBeenCalledTimes(2);
+
+    await session.interrupt();
+    vi.useRealTimers();
+  });
 });
 
 describe("OpenCode persisted sessions", () => {
@@ -1782,3 +2533,34 @@ function buildTextPart(messageId: string, partId: string, text: string): unknown
     text,
   };
 }
+
+function createTestDeferred<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 };
+}
+
+function abortableOpenCodeStream(signal: AbortSignal): AsyncIterable<OpenCodeEvent> {
+  return {
+    [Symbol.asyncIterator]: () => ({
+      next: () =>
+        new Promise<IteratorResult<OpenCodeEvent>>((resolve) => {
+          if (signal.aborted) {
+            resolve({ done: true, value: undefined });
+            return;
+          }
+          signal.addEventListener("abort", () => resolve({ done: true, value: undefined }), {
+            once: true,
+          });
+        }),
+    }),
+  };
+}

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

@@ -83,6 +83,8 @@ const OPENCODE_STORAGE_SESSION_LIMIT = 200;
 // requests; SSE path broken).
 const OPENCODE_EOF_RECOVERY_TIMEOUT_MS = 5 * 60 * 1000;
 const OPENCODE_EOF_RECOVERY_POLL_INTERVAL_MS = 1_000;
+const OPENCODE_RECOVERY_ABORT_TIMEOUT_MS = 2_000;
+const OPENCODE_PENDING_ABORT_START_TIMEOUT_MS = 10_000;
 // If OpenCode silently rejects the prompt (invalid model/mode/auth), no assistant
 // message is ever persisted. Bound the wait so the turn fails in seconds instead
 // of hanging until the completion cap. Valid models normally persist their first
@@ -2206,6 +2208,7 @@ class OpenCodeAgentSession implements AgentSession {
   private currentMode: string = "default";
   private pendingPermissions = new Map<string, AgentPermissionRequest>();
   private abortController: AbortController | null = null;
+  private pendingAbortPromise: Promise<void> | null = null;
   private accumulatedUsage: AgentUsage = {};
   private mcpConfigured = false;
   private mcpSetupPromise: Promise<void> | null = null;
@@ -2235,8 +2238,8 @@ class OpenCodeAgentSession implements AgentSession {
   private foregroundKnownMessageIds = new Set<string>();
   private foregroundEmittedQuestionIds = new Set<string>();
   private foregroundEmittedPermissionIds = new Set<string>();
-  private foregroundEmittedReasoningPartIds = new Set<string>();
-  private foregroundEmittedToolCallStatusByCallId = new Map<string, string>();
+  private foregroundEmittedReasoningTextLengthByPartId = new Map<string, number>();
+  private foregroundEmittedToolCallSignatureByCallId = new Map<string, string>();
   private foregroundTurnStartedAt: number | null = null;
   private readonly recovery: OpenCodeRecoveryOptions;
   constructor(
@@ -2319,17 +2322,7 @@ class OpenCodeAgentSession implements AgentSession {
     // quickly while still giving OpenCode a chance to confirm the abort
     // cleanly. Drop the timeout once upstream returns abort acknowledgement
     // before tool teardown.
-    const abortPromise = this.client.session
-      .abort({
-        sessionID: this.sessionId,
-        directory: this.config.cwd,
-      })
-      .catch((error) => {
-        this.logger.warn(
-          { err: error, sessionId: this.sessionId, turnId },
-          "OpenCode session.abort rejected",
-        );
-      });
+    const abortPromise = this.beginSessionAbort(turnId, "interrupt");
     await withTimeout(abortPromise, 2_000, "OpenCode session.abort").catch((error) => {
       this.logger.warn(
         { err: error, sessionId: this.sessionId, turnId },
@@ -2344,6 +2337,46 @@ class OpenCodeAgentSession implements AgentSession {
     }
   }
 
+  private beginSessionAbort(turnId: string | null, reason: string): Promise<void> {
+    const abortPromise = this.client.session
+      .abort({
+        sessionID: this.sessionId,
+        directory: this.config.cwd,
+      })
+      .then(() => undefined)
+      .catch((error) => {
+        this.logger.warn(
+          { err: error, sessionId: this.sessionId, turnId, reason },
+          "OpenCode session.abort rejected",
+        );
+      });
+    const trackedAbortPromise = abortPromise.finally(() => {
+      if (this.pendingAbortPromise === trackedAbortPromise) {
+        this.pendingAbortPromise = null;
+      }
+    });
+    this.pendingAbortPromise = trackedAbortPromise;
+    return trackedAbortPromise;
+  }
+
+  private async awaitPendingAbortBeforeStartingTurn(): Promise<void> {
+    const pendingAbortPromise = this.pendingAbortPromise;
+    if (!pendingAbortPromise) {
+      return;
+    }
+
+    await withTimeout(
+      pendingAbortPromise,
+      OPENCODE_PENDING_ABORT_START_TIMEOUT_MS,
+      "OpenCode pending session.abort",
+    ).catch((error) => {
+      this.logger.warn(
+        { err: error, sessionId: this.sessionId },
+        "OpenCode session.abort was still pending before starting the next turn",
+      );
+    });
+  }
+
   async startTurn(
     prompt: AgentPromptInput,
     options?: AgentRunOptions,
@@ -2351,6 +2384,7 @@ class OpenCodeAgentSession implements AgentSession {
     if (this.activeForegroundTurnId) {
       throw new Error("A foreground turn is already active");
     }
+    await this.awaitPendingAbortBeforeStartingTurn();
 
     this.foregroundTurnStartedAt = Date.now();
     this.runningToolCalls.clear();
@@ -2362,8 +2396,8 @@ class OpenCodeAgentSession implements AgentSession {
     this.foregroundUsageUpdated = false;
     this.foregroundEmittedQuestionIds.clear();
     this.foregroundEmittedPermissionIds.clear();
-    this.foregroundEmittedReasoningPartIds.clear();
-    this.foregroundEmittedToolCallStatusByCallId.clear();
+    this.foregroundEmittedReasoningTextLengthByPartId.clear();
+    this.foregroundEmittedToolCallSignatureByCallId.clear();
     this.foregroundKnownMessageIds = await this.readPersistedSessionMessageIds();
     const turnAbortController = new AbortController();
     this.abortController = turnAbortController;
@@ -2746,30 +2780,116 @@ class OpenCodeAgentSession implements AgentSession {
 
       const now = Date.now();
       if (!observedActivity && now >= livenessDeadline) {
-        traceOpenCode("recovery.liveness-exhausted", { turnId, attempt });
-        return false;
+        const deferred = await this.deferForPendingPermissionOrFailRecoveredTurnAfterCap(
+          turnId,
+          attempt,
+          "liveness",
+        );
+        if (deferred) {
+          continue;
+        }
+        return true;
       }
       if (now >= completionDeadline) {
-        traceOpenCode("recovery.exhausted", { turnId, attempt });
-        return false;
+        const deferred = await this.deferForPendingPermissionOrFailRecoveredTurnAfterCap(
+          turnId,
+          attempt,
+          "completion",
+        );
+        if (deferred) {
+          continue;
+        }
+        return true;
       }
       const waitMs = Math.min(this.recovery.pollIntervalMs, completionDeadline - now);
-      if (waitMs <= 0) {
-        traceOpenCode("recovery.exhausted", { turnId, attempt });
-        return false;
-      }
       await new Promise<void>((resolve) => setTimeout(resolve, waitMs));
     }
   }
 
+  private async deferForPendingPermissionOrFailRecoveredTurnAfterCap(
+    turnId: string,
+    attempt: number,
+    cap: "liveness" | "completion",
+  ): Promise<boolean> {
+    if (this.pendingPermissions.size > 0) {
+      // A pending OpenCode question/permission means the turn is blocked on
+      // user input, not dead. Keep polling until the user response lets the
+      // assistant finish or the turn is canceled.
+      traceOpenCode(`recovery.${cap}-deferred-for-permission`, {
+        turnId,
+        attempt,
+        pendingPermissionIds: Array.from(this.pendingPermissions.keys()),
+      });
+      await new Promise<void>((resolve) => setTimeout(resolve, this.recovery.pollIntervalMs));
+      return true;
+    }
+
+    traceOpenCode(cap === "liveness" ? "recovery.liveness-exhausted" : "recovery.exhausted", {
+      turnId,
+      attempt,
+    });
+    await this.failRecoveredTurnAfterCap(turnId, cap);
+    return false;
+  }
+
+  private async failRecoveredTurnAfterCap(
+    turnId: string,
+    cap: "liveness" | "completion",
+  ): Promise<void> {
+    await this.abortOpenCodeSessionAfterRecoveryCap(turnId, cap);
+    this.finishForegroundTurn(
+      {
+        type: "turn_failed",
+        provider: "opencode",
+        error: "OpenCode event stream ended before the turn reached a terminal state",
+      },
+      turnId,
+    );
+  }
+
+  private async abortOpenCodeSessionAfterRecoveryCap(
+    turnId: string,
+    cap: "liveness" | "completion",
+  ): Promise<void> {
+    const abortPromise = this.beginSessionAbort(turnId, `recovery-${cap}`);
+    await withTimeout(
+      abortPromise,
+      OPENCODE_RECOVERY_ABORT_TIMEOUT_MS,
+      "OpenCode session.abort",
+    ).catch((error) => {
+      this.logger.warn(
+        { err: error, sessionId: this.sessionId, turnId, cap },
+        "OpenCode session.abort exceeded the EOF recovery cap",
+      );
+    });
+  }
+
   private async pollPendingQuestionsAndPermissions(turnId: string): Promise<number> {
     const [questionsResponse, permissionsResponse] = await Promise.all([
       Promise.resolve()
         .then(() => this.client.question.list({ directory: this.config.cwd }))
-        .catch(() => null),
+        .catch((error) => {
+          traceOpenCode("recovery.question-list.throw", {
+            turnId,
+            error:
+              error instanceof Error
+                ? { name: error.name, message: error.message, stack: error.stack }
+                : String(error),
+          });
+          return null;
+        }),
       Promise.resolve()
         .then(() => this.client.permission.list({ directory: this.config.cwd }))
-        .catch(() => null),
+        .catch((error) => {
+          traceOpenCode("recovery.permission-list.throw", {
+            turnId,
+            error:
+              error instanceof Error
+                ? { name: error.name, message: error.message, stack: error.stack }
+                : String(error),
+          });
+          return null;
+        }),
     ]);
 
     if (this.activeForegroundTurnId !== turnId) return 0;
@@ -2821,10 +2941,25 @@ class OpenCodeAgentSession implements AgentSession {
     | { kind: "in-progress"; messageId: string; parts: readonly OpenCodePart[] }
     | null
   > {
-    const response = await this.client.session.messages({
-      sessionID: this.sessionId,
-      directory: this.config.cwd,
-    });
+    const response = await Promise.resolve()
+      .then(() =>
+        this.client.session.messages({
+          sessionID: this.sessionId,
+          directory: this.config.cwd,
+        }),
+      )
+      .catch((error) => {
+        traceOpenCode("recovery.messages.throw", {
+          error:
+            error instanceof Error
+              ? { name: error.name, message: error.message, stack: error.stack }
+              : String(error),
+        });
+        return null;
+      });
+    if (response === null) {
+      return null;
+    }
     if (response.error || !response.data) {
       return null;
     }
@@ -2849,12 +2984,15 @@ class OpenCodeAgentSession implements AgentSession {
         return { kind: "in-progress", messageId: info.id, parts: item.parts };
       }
 
-      const text = item.parts
+      let text = item.parts
         .filter((part): part is Extract<OpenCodePart, { type: "text" }> => part.type === "text")
         .map((part) => (part.text ?? "").trim())
         .filter((part) => part.length > 0)
         .join("\n\n");
 
+      if (!text) {
+        text = stringifyStructuredAssistantMessage(info.structured) ?? "";
+      }
       if (!text) continue;
 
       const usage: AgentUsage = {};
@@ -2868,13 +3006,16 @@ class OpenCodeAgentSession implements AgentSession {
   private emitIncrementalAssistantParts(parts: readonly OpenCodePart[], turnId: string): void {
     for (const part of parts) {
       if (part.type === "reasoning" && part.text) {
-        if (this.foregroundEmittedReasoningPartIds.has(part.id)) continue;
-        this.foregroundEmittedReasoningPartIds.add(part.id);
+        const emittedTextLength =
+          this.foregroundEmittedReasoningTextLengthByPartId.get(part.id) ?? 0;
+        if (part.text.length <= emittedTextLength) continue;
+        const text = part.text.slice(emittedTextLength);
+        this.foregroundEmittedReasoningTextLengthByPartId.set(part.id, part.text.length);
         this.notifySubscribers(
           {
             type: "timeline",
             provider: "opencode",
-            item: { type: "reasoning", text: part.text },
+            item: { type: "reasoning", text },
           },
           turnId,
         );
@@ -2884,10 +3025,10 @@ class OpenCodeAgentSession implements AgentSession {
       const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part);
       if (!parsedToolPart.success || !parsedToolPart.data) continue;
       const callId = parsedToolPart.data.callId;
-      const status = parsedToolPart.data.status;
-      const lastStatus = this.foregroundEmittedToolCallStatusByCallId.get(callId);
-      if (lastStatus === status) continue;
-      this.foregroundEmittedToolCallStatusByCallId.set(callId, status);
+      const signature = this.createRecoveredToolCallSignature(part, parsedToolPart.data);
+      const lastSignature = this.foregroundEmittedToolCallSignatureByCallId.get(callId);
+      if (lastSignature === signature) continue;
+      this.foregroundEmittedToolCallSignatureByCallId.set(callId, signature);
       this.trackToolCall(parsedToolPart.data);
       this.notifySubscribers(
         {
@@ -2900,6 +3041,21 @@ class OpenCodeAgentSession implements AgentSession {
     }
   }
 
+  private createRecoveredToolCallSignature(
+    part: Extract<OpenCodePart, { type: "tool" }>,
+    item: ToolCallTimelineItem,
+  ): string {
+    const state = (part as { state?: { input?: unknown; output?: unknown; error?: unknown } })
+      .state;
+    return JSON.stringify([
+      item.callId,
+      item.status,
+      state?.input ?? null,
+      state?.output ?? null,
+      state?.error ?? null,
+    ]);
+  }
+
   private applyRecoveredAssistantCompletion(
     completion: {
       messageId: string;