From 152b07b5999ae5298f9412c7135ab3676b06ded0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 23:40:34 +0800 Subject: [PATCH] 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 | 2 +- packages/server/package.json | 2 +- .../agent/providers/opencode-agent.test.ts | 782 ++++++++++++++++++ .../server/agent/providers/opencode-agent.ts | 230 +++++- 4 files changed, 977 insertions(+), 39 deletions(-) diff --git a/package-lock.json b/package-lock.json index fdac0b59e..1d4ef605e 100644 --- a/package-lock.json +++ b/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", diff --git a/packages/server/package.json b/packages/server/package.json index 706ce731c..6df2450e2 100644 --- a/packages/server/package.json +++ b/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", diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index 20b0e45bb..767b3664a 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/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((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((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), + 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((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), + 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((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), + 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((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), + 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((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), + 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 }> => ({ + 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(): { + promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +} { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function abortableOpenCodeStream(signal: AbortSignal): AsyncIterable { + return { + [Symbol.asyncIterator]: () => ({ + next: () => + new Promise>((resolve) => { + if (signal.aborted) { + resolve({ done: true, value: undefined }); + return; + } + signal.addEventListener("abort", () => resolve({ done: true, value: undefined }), { + once: true, + }); + }), + }), + }; +} diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 9ef89c205..3de3e60ff 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/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(); private abortController: AbortController | null = null; + private pendingAbortPromise: Promise | null = null; private accumulatedUsage: AgentUsage = {}; private mcpConfigured = false; private mcpSetupPromise: Promise | null = null; @@ -2235,8 +2238,8 @@ class OpenCodeAgentSession implements AgentSession { private foregroundKnownMessageIds = new Set(); private foregroundEmittedQuestionIds = new Set(); private foregroundEmittedPermissionIds = new Set(); - private foregroundEmittedReasoningPartIds = new Set(); - private foregroundEmittedToolCallStatusByCallId = new Map(); + private foregroundEmittedReasoningTextLengthByPartId = new Map(); + private foregroundEmittedToolCallSignatureByCallId = new Map(); 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 { + 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 { + 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((resolve) => setTimeout(resolve, waitMs)); } } + private async deferForPendingPermissionOrFailRecoveredTurnAfterCap( + turnId: string, + attempt: number, + cap: "liveness" | "completion", + ): Promise { + 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((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 { + 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 { + 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 { 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 => 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, + 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;