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(