From 84f36d2e20af5e2971a6199a67c6c3d8cd9946c7 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 22:43:25 +0800 Subject: [PATCH] fix(server): recover OpenCode turns when 1.14.42+ SSE drops early (#902) OpenCode 1.14.42+ closes the /event SSE stream cleanly right after server.connected, breaking the entire turn lifecycle: prompts get queued and run, but Paseo never sees session.idle / message deltas / tool calls / questions. Every turn either fails with a generic stream EOF or hangs waiting for events that never arrive. Switch the post-EOF recovery to the canonical messages REST endpoint (via the upgraded SDK) and poll incrementally so tool calls and clarifying questions surface live during the SSE gap. Bound the wait with a completion cap and a separate liveness cap so silent rejections fail fast instead of hanging until the cap. Cap session.abort similarly so explicit cancels land within seconds. The recovery path is gated on the for-await loop exiting without a terminal event, so healthy turns never enter it - if upstream restores SSE delivery, this code becomes dead and is mechanically removable via the COMPAT(opencodeEofRecovery) and COMPAT(opencodeSlowAbort) tags. Also tighten the type of projectSettingsRoute on WorktreeSetupCalloutPolicy so router.navigate accepts it (was widened to string, broke app typecheck on main). Refs: getpaseo/paseo#861, anomalyco/opencode#26697, anomalyco/opencode#26635 --- package-lock.json | 17 +- .../worktree-setup-callout-policy.ts | 2 +- packages/server/package.json | 2 +- .../agent/providers/opencode-agent.test.ts | 1182 +++++++++++++---- .../server/agent/providers/opencode-agent.ts | 528 ++++++-- ...encode-plan-and-questions.real.e2e.test.ts | 14 +- .../opencode-send-interrupt.real.e2e.test.ts | 25 +- 7 files changed, 1355 insertions(+), 415 deletions(-) diff --git a/package-lock.json b/package-lock.json index b55ae2335..fdac0b59e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9397,12 +9397,6 @@ "node": ">=20.0" } }, - "node_modules/@opencode-ai/sdk": { - "version": "1.2.6", - "resolved": "https://registry.npmjs.org/@opencode-ai/sdk/-/sdk-1.2.6.tgz", - "integrity": "sha512-dWMF8Aku4h7fh8sw5tQ2FtbqRLbIFT8FcsukpxTird49ax7oUXP+gzqxM/VdxHjfksQvzLBjLZyMdDStc5g7xA==", - "license": "MIT" - }, "node_modules/@opentelemetry/api": { "version": "1.9.0", "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.0.tgz", @@ -39163,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.2.6", + "@opencode-ai/sdk": "^1.14.46", "@sctg/sentencepiece-js": "^1.1.0", "@xterm/headless": "^6.0.0", "ai": "5.0.78", @@ -39344,6 +39338,15 @@ "url": "https://opencollective.com/express" } }, + "packages/server/node_modules/@opencode-ai/sdk": { + "version": "1.14.46", + "resolved": "https://registry.npmjs.org/@opencode-ai/sdk/-/sdk-1.14.46.tgz", + "integrity": "sha512-7KOMuoCkNI+bLOw3GCg0nWZ5m7A/MzNsyLfTbZYmE/DIaUqkV2LNRULtrW6PHL1WtYVmJEFPws4dbw/4dVxjzA==", + "license": "MIT", + "dependencies": { + "cross-spawn": "7.0.6" + } + }, "packages/server/node_modules/accepts": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/accepts/-/accepts-2.0.0.tgz", diff --git a/packages/app/src/components/worktree-setup-callout-policy.ts b/packages/app/src/components/worktree-setup-callout-policy.ts index fc3dd6211..29b8d2a9b 100644 --- a/packages/app/src/components/worktree-setup-callout-policy.ts +++ b/packages/app/src/components/worktree-setup-callout-policy.ts @@ -30,7 +30,7 @@ export interface WorktreeSetupCalloutPolicy { title: string; description: string; actionLabel: string; - projectSettingsRoute: string; + projectSettingsRoute: ReturnType; testID: string; } diff --git a/packages/server/package.json b/packages/server/package.json index c1a7adaf5..706ce731c 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.2.6", + "@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 c2205220a..20b0e45bb 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -566,21 +566,926 @@ describe("OpenCode adapter context-window normalization", () => { }); describe("OpenCode adapter startTurn error handling", () => { - test("recovers SSE EOF into turn_completed when persisted assistant completion exists", async () => { + test("recovers SSE EOF into turn_completed when messages API returns a completed assistant after a delay", async () => { const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); const cwd = "/tmp/test"; const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", { - id: "ses_unit_test", - directory: cwd, - time: { created: 1000, updated: 3000 }, - }); 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 assistant reply")], + }, + ], + 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 < 3) { + return { data: [], error: undefined }; + } + 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 assistant reply", + ); + expect(turn.events).toContainEqual( + expect.objectContaining({ + type: "turn_completed", + usage: expect.objectContaining({ + contextWindowUsedTokens: 15, + inputTokens: 10, + outputTokens: 5, + }), + }), + ); + expect(messagesCallCount).toBeGreaterThanOrEqual(3); + + dateNowSpy.mockRestore(); + 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"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + let releaseStream!: () => void; + const streamMayEnd = new Promise((resolve) => { + releaseStream = resolve; + }); + + 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: [], 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: 0 }, + }); + + 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(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state"); + + 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"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + let releaseStream!: () => void; + const streamMayEnd = new Promise((resolve) => { + releaseStream = resolve; + }); + + const staleMessages = { + data: [ + { + info: buildAssistantMessageInfo({ + id: "msg_assistant_old", + createdAt: 1500, + completedAt: 1600, + tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 }, total: 15 }, + }), + parts: [buildTextPart("msg_assistant_old", "prt_text_old", "Old assistant reply")], + }, + ], + 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(staleMessages), + 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: 0 }, + }); + + 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(turn.assistantMessages).toHaveLength(0); + expect(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state"); + + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + }); + + test("fails fast when no assistant message ever appears before the liveness cap (silent rejection)", 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 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: [], 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: 60_000, pollIntervalMs: 5, livenessMs: 0 }, + }); + + 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(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state"); + + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + }); + + test("keeps polling for completion past the liveness cap once an assistant message appears", 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 completedAssistantInfo = buildAssistantMessageInfo({ + id: "msg_assistant", + createdAt: 2100, + completedAt: 2500, + tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 }, total: 15 }, + }); + const inProgressAssistantInfo = { + ...(completedAssistantInfo 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 < 4) { + return { + data: [ + { + info: inProgressAssistantInfo, + parts: [], + }, + ], + error: undefined, + }; + } + return { + data: [ + { + info: completedAssistantInfo, + parts: [buildTextPart("msg_assistant", "prt_text", "Done after waiting")], + }, + ], + 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: 0 }, + }); + + 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 after waiting", + ); + + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + }); + + test("fails fast when messages API surfaces an assistant message with provider error", 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 erroredMessages = { + data: [ + { + info: { + id: "msg_assistant_failed", + sessionID: "ses_unit_test", + role: "assistant", + time: { created: 2100 }, + parentID: "msg_user", + modelID: "test-model", + providerID: "test-provider", + mode: "build", + agent: "build", + path: { cwd: "/tmp/test", root: "/tmp/test" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + error: { + name: "ProviderAuthError", + data: { providerID: "openai", message: "Insufficient balance for openai/gpt-5-nano" }, + }, + }, + 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(erroredMessages), + 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(false); + expect(turn.turnFailed).toBe(true); + expect(turn.error).toBe("Insufficient balance for openai/gpt-5-nano"); + + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + }); + + test("emits running tool calls incrementally while the assistant message is in progress", 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 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 < 3) { + return { + 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: [ + { + id: "prt_tool", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "tool", + tool: "bash", + callID: "call_long", + state: { status: "running", input: { command: "sleep 60" } }, + }, + ], + }, + ], + 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: "sleep 60" }, + output: "", + }, + }, + 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: 5_000, pollIntervalMs: 5, livenessMs: 5_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); + const toolCallStatuses = turn.toolCalls.map((toolCall) => toolCall.status); + expect(toolCallStatuses).toContain("running"); + expect(toolCallStatuses).toContain("completed"); + + 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"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + let releaseStream!: () => void; + const streamMayEnd = new Promise((resolve) => { + releaseStream = resolve; + }); + + let questionListCount = 0; + let messagesCallCount = 0; + 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().mockImplementation(async () => { + questionListCount += 1; + if (questionListCount < 2) { + return { data: [], error: undefined }; + } + return { + 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 < 5) { + return { + data: [ + { + info: { + ...(buildAssistantMessageInfo({ + id: "msg_assistant", + createdAt: 2100, + completedAt: 2500, + tokens: { + input: 10, + output: 5, + reasoning: 0, + cache: { read: 0, write: 0 }, + total: 15, + }, + }) as Record), + time: { created: 2100 }, + }, + 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: 5_000, pollIntervalMs: 5, livenessMs: 5_000 }, + }); + + const session = await client.createSession({ provider: "opencode", cwd }); + const events: AgentStreamEvent[] = []; + const terminalReached = new Promise((resolve) => { + session.subscribe((event) => { + events.push(event); + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + resolve(); + } + }); + }); + await session.startTurn("hello"); + await terminalReached; + + const permissionEvents = events.filter((event) => event.type === "permission_requested"); + expect(permissionEvents).toHaveLength(1); + expect(permissionEvents[0]).toMatchObject({ + type: "permission_requested", + request: { id: "qst_1", kind: "question" }, + }); + + 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"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + let releaseStream!: () => void; + const streamMayEnd = new Promise((resolve) => { + releaseStream = resolve; + }); + + const completedMessages = { + 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: [ + { + id: "prt_reasoning", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "reasoning", + text: "Thinking through the request", + }, + { + id: "prt_tool", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "tool", + tool: "bash", + callID: "call_1", + state: { + status: "completed", + input: { command: "echo hi" }, + output: "hi", + }, + }, + buildTextPart("msg_assistant", "prt_text", "Done."), + ], + }, + ], + 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(completedMessages), + 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).toHaveLength(1); + expect(turn.toolCalls[0]).toMatchObject({ + type: "tool_call", + callId: "call_1", + status: "completed", + }); + const reasoningItems = turn.allTimelineItems.filter((item) => item.type === "reasoning"); + expect(reasoningItems).toHaveLength(1); + expect(reasoningItems[0]).toMatchObject({ text: "Thinking through the request" }); + expect(turn.assistantMessages.map((message) => message.text).join("")).toBe("Done."); + + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + }); + + test("dedups partial-streamed assistant text against recovered completion", 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 completedMessages = { + 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", "Hello, world!")], + }, + ], + error: undefined, + }; + const fakeClient = { event: { subscribe: vi.fn().mockResolvedValue({ @@ -605,11 +1510,10 @@ describe("OpenCode adapter startTurn error handling", () => { messageID: "msg_assistant", partID: "prt_text", field: "text", - delta: "Recovered ", + delta: "Hello, ", }, } as OpenCodeEvent, ]; - return { next: async () => { if (index < events.length) { @@ -628,30 +1532,8 @@ describe("OpenCode adapter startTurn error handling", () => { }, session: { create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), + messages: vi.fn().mockResolvedValue(completedMessages), promptAsync: vi.fn().mockImplementation(async () => { - writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant.json", { - id: "msg_assistant", - sessionID: "ses_unit_test", - role: "assistant", - finish: "stop", - time: { created: 2000, completed: 2500 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_text.json", { - id: "prt_text", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "text", - text: "Recovered assistant reply", - time: { start: 2100, end: 2400 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_finish.json", { - id: "prt_finish", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "step-finish", - time: { start: 2400, end: 2500 }, - tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } }, - }); releaseStream(); return { data: {}, error: undefined }; }), @@ -671,6 +1553,7 @@ describe("OpenCode adapter startTurn error handling", () => { 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 }); @@ -678,204 +1561,7 @@ describe("OpenCode adapter startTurn error handling", () => { expect(turn.turnCompleted).toBe(true); expect(turn.turnFailed).toBe(false); - expect(turn.assistantMessages.map((message) => message.text).join("")).toBe( - "Recovered assistant reply", - ); - expect(turn.events).toContainEqual( - expect.objectContaining({ - type: "turn_completed", - usage: expect.objectContaining({ - contextWindowUsedTokens: 15, - inputTokens: 10, - outputTokens: 5, - }), - }), - ); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("keeps SSE EOF as turn_failed without persisted completion evidence", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", { - id: "ses_unit_test", - directory: cwd, - time: { created: 1000, updated: 3000 }, - }); - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - 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 }), - promptAsync: vi.fn().mockImplementation(async () => { - writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant.json", { - id: "msg_assistant", - sessionID: "ses_unit_test", - role: "assistant", - time: { created: 2000 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_text.json", { - id: "prt_text", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "text", - text: "Incomplete assistant reply", - time: { start: 2100 }, - }); - 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), - }, - }); - - 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(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state"); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("does not recover a previous turn's completed assistant reply when the current turn only persists incomplete output", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", { - id: "ses_unit_test", - directory: cwd, - time: { created: 1000, updated: 3000 }, - }); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - 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 }), - promptAsync: vi.fn().mockImplementation(async () => { - writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant_old.json", { - id: "msg_assistant_old", - sessionID: "ses_unit_test", - role: "assistant", - finish: "stop", - time: { created: 1500, completed: 1600 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant_old/prt_text.json", { - id: "prt_text_old", - sessionID: "ses_unit_test", - messageID: "msg_assistant_old", - type: "text", - text: "Old assistant reply", - time: { start: 1500, end: 1550 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant_old/prt_finish.json", { - id: "prt_finish_old", - sessionID: "ses_unit_test", - messageID: "msg_assistant_old", - type: "step-finish", - time: { start: 1550, end: 1600 }, - tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } }, - }); - writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant_current.json", { - id: "msg_assistant_current", - sessionID: "ses_unit_test", - role: "assistant", - time: { created: 2100 }, - }); - writeOpenCodeJson(storageRoot, "part/msg_assistant_current/prt_text.json", { - id: "prt_text_current", - sessionID: "ses_unit_test", - messageID: "msg_assistant_current", - type: "text", - text: "Incomplete assistant reply", - time: { start: 2100 }, - }); - 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), - }, - }); - - 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(turn.assistantMessages).toHaveLength(0); - expect(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state"); + expect(turn.assistantMessages.map((message) => message.text).join("")).toBe("Hello, world!"); dateNowSpy.mockRestore(); rmSync(storageRoot, { recursive: true, force: true }); @@ -1058,3 +1744,41 @@ function writeOpenCodeJson(storageRoot: string, relativePath: string, value: unk mkdirSync(path.dirname(filePath), { recursive: true }); writeFileSync(filePath, JSON.stringify(value), "utf8"); } + +function buildAssistantMessageInfo(args: { + id: string; + createdAt: number; + completedAt: number; + tokens: { + input: number; + output: number; + reasoning: number; + cache: { read: number; write: number }; + total: number; + }; +}): unknown { + return { + id: args.id, + sessionID: "ses_unit_test", + role: "assistant", + time: { created: args.createdAt, completed: args.completedAt }, + parentID: "msg_user", + modelID: "test-model", + providerID: "test-provider", + mode: "build", + agent: "build", + path: { cwd: "/tmp/test", root: "/tmp/test" }, + cost: 0, + tokens: args.tokens, + }; +} + +function buildTextPart(messageId: string, partId: string, text: string): unknown { + return { + id: partId, + sessionID: "ses_unit_test", + messageID: messageId, + type: "text", + text, + }; +} diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 236c1cbd7..9ef89c205 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -74,6 +74,20 @@ const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { const OPENCODE_BUILD_MODE_ID = "build"; const OPENCODE_FULL_ACCESS_MODE_ID = "full-access"; const OPENCODE_STORAGE_SESSION_LIMIT = 200; +// COMPAT(opencodeEofRecovery): added in v0.1.73 to compensate for OpenCode 1.14.42+ +// closing the /event SSE stream cleanly after `server.connected`. Drop this whole +// recovery path once OpenCode upstream restores live event delivery and the floor +// version reflects that. +// Upstream: anomalyco/opencode#26697 (SSE /event closes immediately after +// server.connected) and anomalyco/opencode#26635 (prompt_async silently discards +// requests; SSE path broken). +const OPENCODE_EOF_RECOVERY_TIMEOUT_MS = 5 * 60 * 1000; +const OPENCODE_EOF_RECOVERY_POLL_INTERVAL_MS = 1_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 +// message within a second of LLM start, so 10s leaves comfortable headroom. +const OPENCODE_EOF_RECOVERY_LIVENESS_MS = 10_000; const DEFAULT_MODES: AgentMode[] = [ { @@ -138,12 +152,6 @@ type OpenCodeStoredSession = z.infer; type OpenCodeStoredMessage = z.infer; type OpenCodeStoredPart = z.infer; -interface OpenCodePersistedAssistantCompletion { - messageId: string; - text: string; - usage?: AgentUsage; -} - type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" }; type OpenCodeMessageRole = "user" | "assistant"; @@ -657,6 +665,19 @@ function mergeOpenCodeStepFinishUsage( } } +function formatOpenCodeAssistantErrorMessage( + error: NonNullable, +): string { + const data = (error as { data?: unknown }).data; + if (data && typeof data === "object" && "message" in data) { + const message = (data as { message?: unknown }).message; + if (typeof message === "string" && message.trim().length > 0) { + return message.trim(); + } + } + return error.name; +} + function hasNormalizedOpenCodeUsage(usage: AgentUsage): boolean { return [ usage.inputTokens, @@ -855,102 +876,6 @@ function readOpenCodeTextFromParts(parts: OpenCodeStoredPart[]): string { .join("\n\n"); } -async function readOpenCodePersistedAssistantCompletion( - storageRoot: string, - sessionId: string, - knownMessageIds: ReadonlySet, - turnStartedAt: number, -): Promise { - const messageRoot = path.join(storageRoot, "message", sessionId); - const messageFiles = await findJsonFiles(messageRoot); - const messages: OpenCodeStoredMessage[] = []; - - for (const file of messageFiles) { - const parsed = await readJsonFile(file, OpenCodeStoredMessageSchema); - if ( - parsed?.sessionID === sessionId && - parsed.role === "assistant" && - !knownMessageIds.has(parsed.id) - ) { - messages.push(parsed); - } - } - - const candidates = messages.sort( - (left, right) => getOpenCodeMessageTimestamp(right) - getOpenCodeMessageTimestamp(left), - ); - - for (const message of candidates) { - const parts = (await readOpenCodeStoredParts(storageRoot, message.id)).filter((part) => - isOpenCodePartAtOrAfterTurnStart(part, turnStartedAt), - ); - if (!isOpenCodeMessageAtOrAfterTurnStart(message, turnStartedAt) && parts.length === 0) { - continue; - } - - const text = readOpenCodeTextFromParts(parts); - if (!text || !hasStrongPersistedCompletionEvidence(message, parts, turnStartedAt)) { - continue; - } - const usage = readPersistedStepFinishUsage(parts); - - return { - messageId: message.id, - text, - ...(hasNormalizedOpenCodeUsage(usage) ? { usage } : {}), - }; - } - - return null; -} - -function hasStrongPersistedCompletionEvidence( - message: OpenCodeStoredMessage, - parts: OpenCodeStoredPart[], - turnStartedAt: number, -): boolean { - const messageRecord = readOpenCodeRecord(message); - const infoRecord = readOpenCodeRecord(messageRecord?.["info"]); - const finish = - readNonEmptyString(messageRecord?.["finish"]) ?? readNonEmptyString(infoRecord?.["finish"]); - const hasCompletedMessage = isOpenCodeTimestampAtOrAfter(message.time?.completed, turnStartedAt); - const hasCompletedTextPart = parts.some( - (part) => part.type === "text" && typeof part.time?.end === "number", - ); - const hasStepFinish = parts.some((part) => part.type === "step-finish"); - - return finish === "stop" && (hasCompletedMessage || hasCompletedTextPart || hasStepFinish); -} - -function readPersistedStepFinishUsage(parts: OpenCodeStoredPart[]): AgentUsage { - const usage: AgentUsage = {}; - - for (const part of parts.filter((candidate) => candidate.type === "step-finish")) { - const partRecord = readOpenCodeRecord(part); - const tokensRecord = readOpenCodeRecord(partRecord?.["tokens"]); - const cacheRecord = readOpenCodeRecord(tokensRecord?.["cache"]); - mergeOpenCodeStepFinishUsage(usage, { - cost: partRecord?.["cost"], - tokens: tokensRecord - ? { - input: tokensRecord["input"], - output: tokensRecord["output"], - reasoning: tokensRecord["reasoning"], - total: tokensRecord["total"], - cache: cacheRecord - ? { - read: cacheRecord["read"], - write: cacheRecord["write"], - } - : undefined, - } - : undefined, - }); - } - - return usage; -} - async function findJsonFiles(root: string): Promise { let entries; try { @@ -996,33 +921,6 @@ function getOpenCodePartTimestamp(part: OpenCodeStoredPart): number { return part.time?.start ?? part.time?.end ?? 0; } -function isOpenCodeTimestampAtOrAfter( - timestamp: number | undefined, - turnStartedAt: number, -): boolean { - return typeof timestamp === "number" && timestamp >= turnStartedAt; -} - -function isOpenCodeMessageAtOrAfterTurnStart( - message: OpenCodeStoredMessage, - turnStartedAt: number, -): boolean { - return ( - isOpenCodeTimestampAtOrAfter(message.time?.created, turnStartedAt) || - isOpenCodeTimestampAtOrAfter(message.time?.completed, turnStartedAt) - ); -} - -function isOpenCodePartAtOrAfterTurnStart( - part: OpenCodeStoredPart, - turnStartedAt: number, -): boolean { - return ( - isOpenCodeTimestampAtOrAfter(part.time?.start, turnStartedAt) || - isOpenCodeTimestampAtOrAfter(part.time?.end, turnStartedAt) - ); -} - export const __openCodeInternals = { buildOpenCodePromptParts, buildOpenCodeModelContextWindowLookup, @@ -1041,8 +939,15 @@ export const __openCodeInternals = { }, }; +interface OpenCodeRecoveryOptions { + timeoutMs: number; + pollIntervalMs: number; + livenessMs: number; +} + interface OpenCodeAgentClientDeps { runtime?: OpenCodeRuntime; + recovery?: OpenCodeRecoveryOptions; } class ProductionOpenCodeRuntime implements OpenCodeRuntime { @@ -1074,6 +979,7 @@ export class OpenCodeAgentClient implements AgentClient { private readonly runtimeSettings?: ProviderRuntimeSettings; private readonly modelContextWindows = new Map(); private readonly storageRoot: string; + private readonly recovery: OpenCodeRecoveryOptions; constructor( logger: Logger, @@ -1084,6 +990,11 @@ export class OpenCodeAgentClient implements AgentClient { this.logger = logger.child({ module: "agent", provider: "opencode" }); this.runtimeSettings = runtimeSettings; this.storageRoot = storageRoot ?? resolveOpenCodeStorageRoot(); + this.recovery = deps.recovery ?? { + timeoutMs: OPENCODE_EOF_RECOVERY_TIMEOUT_MS, + pollIntervalMs: OPENCODE_EOF_RECOVERY_POLL_INTERVAL_MS, + livenessMs: OPENCODE_EOF_RECOVERY_LIVENESS_MS, + }; this.runtime = deps.runtime ?? new ProductionOpenCodeRuntime( @@ -1131,6 +1042,7 @@ export class OpenCodeAgentClient implements AgentClient { new Map(this.modelContextWindows), acquisition.release, options?.persistSession, + this.recovery, ); } catch (error) { acquisition.release(); @@ -1172,6 +1084,8 @@ export class OpenCodeAgentClient implements AgentClient { this.storageRoot, new Map(this.modelContextWindows), acquisition.release, + undefined, + this.recovery, ); } catch (error) { acquisition.release(); @@ -2267,6 +2181,18 @@ function createDeferred(): Deferred { return { promise, resolve, reject }; } +const OPENCODE_TRACE_ENABLED = process.env.PASEO_OPENCODE_TRACE === "1"; + +function traceOpenCode(tag: string, data: Record = {}): void { + if (!OPENCODE_TRACE_ENABLED) return; + const line = JSON.stringify({ ts: new Date().toISOString(), tag, ...data }, (_k, v) => { + if (v instanceof Error) return { name: v.name, message: v.message, stack: v.stack }; + if (typeof v === "bigint") return v.toString(); + return v; + }); + process.stderr.write(`[opencode-trace] ${line}\n`); +} + class OpenCodeAgentSession implements AgentSession { readonly provider = "opencode" as const; readonly capabilities = OPENCODE_CAPABILITIES; @@ -2307,7 +2233,12 @@ class OpenCodeAgentSession implements AgentSession { private foregroundAssistantText = ""; private foregroundUsageUpdated = false; private foregroundKnownMessageIds = new Set(); + private foregroundEmittedQuestionIds = new Set(); + private foregroundEmittedPermissionIds = new Set(); + private foregroundEmittedReasoningPartIds = new Set(); + private foregroundEmittedToolCallStatusByCallId = new Map(); private foregroundTurnStartedAt: number | null = null; + private readonly recovery: OpenCodeRecoveryOptions; constructor( config: OpenCodeAgentConfig, client: OpencodeClient, @@ -2317,6 +2248,11 @@ class OpenCodeAgentSession implements AgentSession { modelContextWindowsByModelKey: ReadonlyMap = new Map(), releaseServer?: () => void, persistSession = true, + recovery: OpenCodeRecoveryOptions = { + timeoutMs: OPENCODE_EOF_RECOVERY_TIMEOUT_MS, + pollIntervalMs: OPENCODE_EOF_RECOVERY_POLL_INTERVAL_MS, + livenessMs: OPENCODE_EOF_RECOVERY_LIVENESS_MS, + }, ) { this.config = config; this.client = client; @@ -2327,6 +2263,7 @@ class OpenCodeAgentSession implements AgentSession { this.currentMode = normalizeOpenCodeModeId(config.modeId); this.releaseServer = releaseServer ?? null; this.persistSession = persistSession; + this.recovery = recovery; this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( config.model, ); @@ -2376,9 +2313,28 @@ class OpenCodeAgentSession implements AgentSession { const turnId = this.activeForegroundTurnId; const turnAbortController = this.abortController; turnAbortController?.abort(); - await this.client.session.abort({ - sessionID: this.sessionId, - directory: this.config.cwd, + // COMPAT(opencodeSlowAbort): OpenCode 1.14.42+ blocks session.abort until + // the running tool actually stops, which can be tens of seconds for + // long-running tools. Cap the wait so the user-visible cancel lands + // 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", + ); + }); + await withTimeout(abortPromise, 2_000, "OpenCode session.abort").catch((error) => { + this.logger.warn( + { err: error, sessionId: this.sessionId, turnId }, + "OpenCode session.abort exceeded the cancel cap; proceeding with local cancel", + ); }); if (turnId) { this.finishForegroundTurn( @@ -2404,6 +2360,10 @@ class OpenCodeAgentSession implements AgentSession { this.foregroundAssistantMessageEmitted = false; this.foregroundAssistantText = ""; this.foregroundUsageUpdated = false; + this.foregroundEmittedQuestionIds.clear(); + this.foregroundEmittedPermissionIds.clear(); + this.foregroundEmittedReasoningPartIds.clear(); + this.foregroundEmittedToolCallStatusByCallId.clear(); this.foregroundKnownMessageIds = await this.readPersistedSessionMessageIds(); const turnAbortController = new AbortController(); this.abortController = turnAbortController; @@ -2538,6 +2498,14 @@ class OpenCodeAgentSession implements AgentSession { // SDK input validation) is caught alongside async rejections. A plain // `.then().catch()` chain would let a sync throw escape unhandled. void (async () => { + traceOpenCode("promptAsync.start", { + turnId, + sessionId: this.sessionId, + model, + effectiveMode, + effectiveVariant, + partTypes: parts.map((p) => p.type), + }); try { const promptResponse = await this.client.session.promptAsync({ sessionID: this.sessionId, @@ -2556,6 +2524,12 @@ class OpenCodeAgentSession implements AgentSession { ...(effectiveMode ? { agent: effectiveMode } : {}), ...(effectiveVariant ? { variant: effectiveVariant } : {}), }); + traceOpenCode("promptAsync.response", { + turnId, + hasError: promptResponse.error !== undefined, + error: promptResponse.error, + data: promptResponse.data, + }); if (promptResponse.error) { this.finishForegroundTurn( { @@ -2567,6 +2541,13 @@ class OpenCodeAgentSession implements AgentSession { ); } } catch (error) { + traceOpenCode("promptAsync.throw", { + turnId, + error: + error instanceof Error + ? { name: error.name, message: error.message, stack: error.stack } + : String(error), + }); this.finishForegroundTurn( { type: "turn_failed", @@ -2594,21 +2575,44 @@ class OpenCodeAgentSession implements AgentSession { turnAbortController: AbortController, subscriptionReady: Deferred, ): Promise { + traceOpenCode("subscribe.start", { turnId, sessionId: this.sessionId, cwd: this.config.cwd }); try { const result = await this.client.event.subscribe( { directory: this.config.cwd }, - { signal: turnAbortController.signal, sseMaxRetryAttempts: 0 }, + { signal: turnAbortController.signal }, ); + traceOpenCode("subscribe.ready", { turnId, sessionId: this.sessionId }); subscriptionReady.resolve(); + let eventCount = 0; for await (const event of result.stream) { + eventCount += 1; + traceOpenCode("event.raw", { + turnId, + n: eventCount, + type: (event as { type?: string }).type, + properties: (event as { properties?: unknown }).properties, + }); if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) { + traceOpenCode("event.skip", { + turnId, + n: eventCount, + aborted: turnAbortController.signal.aborted, + activeTurnId: this.activeForegroundTurnId, + }); break; } const translated = await this.translateEvent(event); + traceOpenCode("event.translated", { + turnId, + n: eventCount, + count: translated.length, + types: translated.map((t) => t.type), + }); for (const e of translated) { if (this.activeForegroundTurnId !== turnId) { + traceOpenCode("event.translated.skip-active", { turnId, type: e.type }); return; } if (e.type === "timeline" && e.item.type === "tool_call") { @@ -2616,6 +2620,7 @@ class OpenCodeAgentSession implements AgentSession { } const terminalEvent = toTerminalTurnEvent(e); if (terminalEvent) { + traceOpenCode("event.terminal", { turnId, type: terminalEvent.type }); this.finishForegroundTurn(terminalEvent, turnId); return; } @@ -2623,10 +2628,20 @@ class OpenCodeAgentSession implements AgentSession { } } + traceOpenCode("stream.eof", { + turnId, + eventCount, + aborted: turnAbortController.signal.aborted, + stillActive: this.activeForegroundTurnId === turnId, + }); + if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { - if (await this.recoverTurnFromPersistedCompletion(turnId)) { + const recovered = await this.recoverTurnFromPersistedCompletion(turnId); + traceOpenCode("recovery.result", { turnId, recovered }); + if (recovered) { return; } + traceOpenCode("turn.fail.eof", { turnId, eventCount }); this.finishForegroundTurn( { type: "turn_failed", @@ -2637,6 +2652,11 @@ class OpenCodeAgentSession implements AgentSession { ); } } catch (error) { + traceOpenCode("subscribe.error", { + turnId, + error: + error instanceof Error ? { name: error.name, message: error.message } : String(error), + }); subscriptionReady.reject(error); if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { this.finishForegroundTurn( @@ -2666,24 +2686,237 @@ class OpenCodeAgentSession implements AgentSession { } private async recoverTurnFromPersistedCompletion(turnId: string): Promise { - if (this.foregroundTurnStartedAt === null) { + traceOpenCode("recovery.start", { + turnId, + foregroundTurnStartedAt: this.foregroundTurnStartedAt, + knownMessageIds: Array.from(this.foregroundKnownMessageIds), + sessionId: this.sessionId, + timeoutMs: this.recovery.timeoutMs, + pollIntervalMs: this.recovery.pollIntervalMs, + }); + const startedAt = this.foregroundTurnStartedAt; + if (startedAt === null) { + traceOpenCode("recovery.no-start-time", { turnId }); return false; } - const completion = await readOpenCodePersistedAssistantCompletion( - this.storageRoot, - this.sessionId, - this.foregroundKnownMessageIds, - this.foregroundTurnStartedAt, - ); - if (!completion || this.activeForegroundTurnId !== turnId) { + const completionDeadline = Date.now() + this.recovery.timeoutMs; + const livenessDeadline = Date.now() + this.recovery.livenessMs; + let attempt = 0; + let observedActivity = false; + while (true) { + if (this.activeForegroundTurnId !== turnId) { + traceOpenCode("recovery.cancelled", { turnId, attempt }); + return true; + } + + attempt += 1; + const emittedPromptIds = await this.pollPendingQuestionsAndPermissions(turnId); + if (emittedPromptIds > 0) { + observedActivity = true; + } + const outcome = await this.fetchAssistantOutcomeFromMessagesApi(startedAt); + traceOpenCode("recovery.poll", { + turnId, + attempt, + kind: outcome?.kind ?? "none", + messageId: outcome?.messageId, + emittedPromptIds, + }); + if (outcome?.kind === "failure") { + this.foregroundKnownMessageIds.add(outcome.messageId); + this.finishForegroundTurn( + { + type: "turn_failed", + provider: "opencode", + error: outcome.error, + }, + turnId, + ); + return true; + } + if (outcome?.kind === "completion") { + this.emitIncrementalAssistantParts(outcome.parts, turnId); + return this.applyRecoveredAssistantCompletion(outcome, turnId); + } + if (outcome?.kind === "in-progress") { + observedActivity = true; + this.emitIncrementalAssistantParts(outcome.parts, turnId); + } + + const now = Date.now(); + if (!observedActivity && now >= livenessDeadline) { + traceOpenCode("recovery.liveness-exhausted", { turnId, attempt }); + return false; + } + if (now >= completionDeadline) { + traceOpenCode("recovery.exhausted", { turnId, attempt }); + return false; + } + 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 pollPendingQuestionsAndPermissions(turnId: string): Promise { + const [questionsResponse, permissionsResponse] = await Promise.all([ + Promise.resolve() + .then(() => this.client.question.list({ directory: this.config.cwd })) + .catch(() => null), + Promise.resolve() + .then(() => this.client.permission.list({ directory: this.config.cwd })) + .catch(() => null), + ]); + + if (this.activeForegroundTurnId !== turnId) return 0; + + let emitted = 0; + for (const question of questionsResponse?.data ?? []) { + if (question.sessionID !== this.sessionId) continue; + if (this.foregroundEmittedQuestionIds.has(question.id)) continue; + this.foregroundEmittedQuestionIds.add(question.id); + emitted += 1; + const synthetic = { + id: question.id, + type: "question.asked", + properties: question, + } as unknown as OpenCodeEvent; + const events = await this.translateEvent(synthetic); + for (const event of events) { + this.notifySubscribers(event, turnId); + } + } + + for (const permission of permissionsResponse?.data ?? []) { + if (permission.sessionID !== this.sessionId) continue; + if (this.foregroundEmittedPermissionIds.has(permission.id)) continue; + this.foregroundEmittedPermissionIds.add(permission.id); + emitted += 1; + const synthetic = { + id: permission.id, + type: "permission.asked", + properties: permission, + } as unknown as OpenCodeEvent; + const events = await this.translateEvent(synthetic); + for (const event of events) { + this.notifySubscribers(event, turnId); + } + } + return emitted; + } + + private async fetchAssistantOutcomeFromMessagesApi(startedAt: number): Promise< + | { + kind: "completion"; + messageId: string; + text: string; + parts: readonly OpenCodePart[]; + usage: AgentUsage; + } + | { kind: "failure"; messageId: string; error: string } + | { kind: "in-progress"; messageId: string; parts: readonly OpenCodePart[] } + | null + > { + const response = await this.client.session.messages({ + sessionID: this.sessionId, + directory: this.config.cwd, + }); + if (response.error || !response.data) { + return null; + } + + for (let index = response.data.length - 1; index >= 0; index -= 1) { + const item = response.data[index]; + if (!item) continue; + const info = item.info; + if (info.role !== "assistant") continue; + if (this.foregroundKnownMessageIds.has(info.id)) continue; + if (typeof info.time?.created === "number" && info.time.created < startedAt) continue; + + if (info.error) { + return { + kind: "failure", + messageId: info.id, + error: formatOpenCodeAssistantErrorMessage(info.error), + }; + } + + if (typeof info.time?.completed !== "number") { + return { kind: "in-progress", messageId: info.id, parts: item.parts }; + } + + const 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) continue; + + const usage: AgentUsage = {}; + mergeOpenCodeStepFinishUsage(usage, { cost: info.cost, tokens: info.tokens }); + + return { kind: "completion", messageId: info.id, text, parts: item.parts, usage }; + } + return null; + } + + 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); + this.notifySubscribers( + { + type: "timeline", + provider: "opencode", + item: { type: "reasoning", text: part.text }, + }, + turnId, + ); + continue; + } + if (part.type !== "tool") continue; + 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); + this.trackToolCall(parsedToolPart.data); + this.notifySubscribers( + { + type: "timeline", + provider: "opencode", + item: parsedToolPart.data, + }, + turnId, + ); + } + } + + private applyRecoveredAssistantCompletion( + completion: { + messageId: string; + text: string; + parts: readonly OpenCodePart[]; + usage: AgentUsage; + }, + turnId: string, + ): boolean { + if (this.activeForegroundTurnId !== turnId) { return false; } this.foregroundKnownMessageIds.add(completion.messageId); this.logger.warn( { sessionId: this.sessionId, turnId }, - "Recovered OpenCode turn completion from persisted session state after SSE EOF", + "Recovered OpenCode turn completion via messages API after SSE EOF", ); const recoveryText = this.resolvePersistedAssistantRecoveryText(completion.text); @@ -2703,7 +2936,7 @@ class OpenCodeAgentSession implements AgentSession { this.foregroundAssistantMessageEmitted = true; } - if (completion.usage && !this.foregroundUsageUpdated) { + if (hasNormalizedOpenCodeUsage(completion.usage) && !this.foregroundUsageUpdated) { this.accumulatedUsage = { ...this.accumulatedUsage, ...completion.usage, @@ -2765,6 +2998,13 @@ class OpenCodeAgentSession implements AgentSession { event: Extract, turnId: string, ): void { + traceOpenCode("finishForegroundTurn", { + turnId, + activeTurnId: this.activeForegroundTurnId, + type: event.type, + error: event.type === "turn_failed" ? event.error : undefined, + reason: event.type === "turn_canceled" ? event.reason : undefined, + }); if (this.activeForegroundTurnId !== turnId) { return; } diff --git a/packages/server/src/server/daemon-e2e/opencode-plan-and-questions.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-plan-and-questions.real.e2e.test.ts index 4307ff29f..07c8f3d0d 100644 --- a/packages/server/src/server/daemon-e2e/opencode-plan-and-questions.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/opencode-plan-and-questions.real.e2e.test.ts @@ -13,16 +13,6 @@ function tmpCwd(): string { return mkdtempSync(path.join(tmpdir(), "daemon-real-opencode-")); } -function pickOpenCodeModel( - models: Array<{ id: string }>, - preferences: string[] = ["gpt-5-nano", "gpt-4.1-nano", "mini", "free"], -): string { - const preferred = models.find((model) => - preferences.some((fragment) => model.id.includes(fragment)), - ); - return preferred?.id ?? models[0].id; -} - async function createHarness(): Promise<{ client: DaemonClient; daemon: Awaited>; @@ -63,7 +53,7 @@ describe("daemon E2E (real opencode) - plan mode and clarifying questions", () = provider: "opencode", cwd, title: "OpenCode question regression", - model: pickOpenCodeModel(modelList.models, ["minimax-m2.5-free", "minimax", "free"]), + model: "opencode/big-pickle", modeId: "plan", }); @@ -111,7 +101,7 @@ describe("daemon E2E (real opencode) - plan mode and clarifying questions", () = provider: "opencode", cwd, title: "OpenCode plan mode regression", - model: pickOpenCodeModel(modelList.models), + model: "opencode/big-pickle", modeId: "plan", }); diff --git a/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts index a991c1f1d..15f264592 100644 --- a/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts @@ -22,23 +22,6 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function pickOpenCodeModel( - models: Array<{ id: string }>, - preferences: string[] = [ - "minimax-m2.5-free", - "kimi-k2.5-free", - "glm-5-free", - "free", - "mini", - "gpt-5-nano", - ], -): string { - const preferred = models.find((model) => - preferences.some((fragment) => model.id.includes(fragment)), - ); - return preferred?.id ?? models[0].id; -} - function hasRunningBashToolCall(messages: SessionOutboundMessage[], agentId: string): boolean { return messages.some( (message) => @@ -305,8 +288,8 @@ describe("daemon E2E (real opencode) - send while working and interrupt", () => provider: "opencode", cwd, title: "OpenCode send while working", - model: pickOpenCodeModel(modelList.models), - modeId: "default", + model: "opencode/big-pickle", + modeId: "build", }); await client.sendMessage( @@ -365,8 +348,8 @@ describe("daemon E2E (real opencode) - send while working and interrupt", () => provider: "opencode", cwd, title: "OpenCode explicit interrupt", - model: pickOpenCodeModel(modelList.models), - modeId: "default", + model: "opencode/big-pickle", + modeId: "build", }); await client.sendMessage(