From 40ab9e3f2073a50ff71be189e7236074f581d490 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 11 May 2026 17:17:00 +0800 Subject: [PATCH] Use OpenCode global event stream (#916) * Use OpenCode global event stream * Use stable OpenCode model in initial prompt e2e * Clean up OpenCode verification notes --- docs/opencode-global-event-baseline.md | 49 + .../agent/providers/opencode-agent.test.ts | 1893 ++--------------- .../server/agent/providers/opencode-agent.ts | 666 ++---- .../test-utils/test-opencode-runtime.ts | 7 + ...ncode-initial-prompt-wait.real.e2e.test.ts | 37 +- 5 files changed, 352 insertions(+), 2300 deletions(-) create mode 100644 docs/opencode-global-event-baseline.md diff --git a/docs/opencode-global-event-baseline.md b/docs/opencode-global-event-baseline.md new file mode 100644 index 000000000..5c49ab294 --- /dev/null +++ b/docs/opencode-global-event-baseline.md @@ -0,0 +1,49 @@ +# OpenCode Global Event Verification + +Date: 2026-05-11 + +## Objective + +Replace the OpenCode provider's per-directory `/event` stream with OpenCode's `/global/event` stream and remove the EOF polling recovery path that was added for the `/event` regression. + +## Environment + +- `opencode --version`: `1.14.46` +- `which opencode`: `/Users/moboudra/.asdf/installs/nodejs/22.20.0/bin/opencode` +- `node --version`: `v22.20.0` +- `npm --version`: `10.9.3` + +Each OpenCode test file was run independently with: + +```bash +/opt/homebrew/bin/timeout 420s npx vitest run --maxWorkers=1 --minWorkers=1 +``` + +## Baseline + +Before the provider change, the OpenCode matrix had 16 passing files and 4 failing files: + +- `packages/cli/tests/e2e/opencode-invalid-model.test.ts`: Vitest reports "No test suite found in file". +- `packages/server/src/server/agent/providers/opencode-agent.test.ts`: `plan mode blocks edits while build mode can write files` did not observe a completed tool call. +- `packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts`: brittle unavailable-model assertion received an auth failure from the upstream API. +- `packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts`: timed out waiting for an interrupted sleep tool call, even though the recent bash tool call status was `failed`. + +## Post-Change Result + +After switching to `/global/event`, removing polling recovery, and replacing the brittle initial-prompt model case with `opencode/big-pickle`, the OpenCode matrix had 18 passing files and 2 baseline-equivalent failing files: + +- `packages/cli/tests/e2e/opencode-invalid-model.test.ts`: unchanged; Vitest still reports "No test suite found in file". +- `packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts`: unchanged; still times out after the interrupted sleep tool call is already marked `failed`. + +The previously failing provider unit file now passes, and `packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts` passes with `opencode/big-pickle`. + +One live reasoning-dedup matrix run returned no reasoning content; an immediate targeted rerun passed. This appears model-output dependent rather than related to the event-stream change. + +## Focused Verification + +- `npm run typecheck` +- `npm run lint` +- `git diff --check` +- `npx vitest run packages/server/src/server/agent/providers/opencode-agent.test.ts --maxWorkers=1 --minWorkers=1` +- `npx vitest run packages/server/src/server/agent/providers/opencode-agent.error-handling.real.e2e.test.ts --maxWorkers=1 --minWorkers=1` +- `npx vitest run packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts --maxWorkers=1 --minWorkers=1` 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 767b3664a..63d29e370 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -78,6 +78,14 @@ async function collectTurnEvents(iterator: AsyncGenerator): Pr return result; } +function createAsyncIterable(items: T[]): AsyncIterable { + return (async function* () { + for (const item of items) { + yield item; + } + })(); +} + function isBinaryInstalled(binary: string): boolean { try { const out = execFileSync("which", [binary], { encoding: "utf8" }).trim(); @@ -566,1703 +574,158 @@ describe("OpenCode adapter context-window normalization", () => { }); describe("OpenCode adapter startTurn error handling", () => { - 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); - - 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")], + test("unwraps OpenCode global event payloads during a turn", async () => { + const globalEvents = [ + { + payload: { + type: "server.connected", + properties: {}, }, - ], - error: undefined, - }; - - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), + }, + { + directory: "/tmp/other", + payload: { + type: "message.part.delta", + properties: { + sessionID: "other-session", + messageID: "msg_other", + partID: "prt_other", + field: "text", + delta: "ignore me", }, - }), - }, - 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("continues SSE EOF recovery when one messages API poll rejects", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - let messagesCallCount = 0; - const completedMessagesResponse = { - data: [ - { - info: buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2500, - tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 }, total: 15 }, - }), - parts: [buildTextPart("msg_assistant", "prt_text", "Recovered after retry")], }, - ], - error: undefined, - }; - - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockImplementation(async () => { - messagesCallCount += 1; - if (messagesCallCount === 1) { - throw new Error("transient messages failure"); - } - return completedMessagesResponse; - }), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort: vi.fn().mockResolvedValue({ data: true, error: undefined }), - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(true); - expect(turn.turnFailed).toBe(false); - expect(turn.assistantMessages.map((message) => message.text).join("")).toBe( - "Recovered after retry", - ); - expect(messagesCallCount).toBe(2); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("recovers structured-only assistant completions from messages API", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const startedAt = Date.now(); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - const completedMessagesResponse = { - data: [ - { - info: { - ...(buildAssistantMessageInfo({ + { + directory: "/tmp/test", + payload: { + type: "message.updated", + properties: { + info: { id: "msg_assistant", - createdAt: startedAt + 1, - completedAt: startedAt + 500, - tokens: { - input: 10, - output: 5, - reasoning: 0, - cache: { read: 0, write: 0 }, - total: 15, - }, - }) as Record), - structured: { status: "ok", files: ["README.md"] }, - }, - parts: [], - }, - ], - error: undefined, - }; - - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), - }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockResolvedValue(completedMessagesResponse), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort: vi.fn().mockResolvedValue({ data: true, error: undefined }), - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 1, pollIntervalMs: 1, livenessMs: 1 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(true); - expect(turn.turnFailed).toBe(false); - expect(turn.assistantMessages).toEqual([ - { type: "assistant_message", text: '{"status":"ok","files":["README.md"]}' }, - ]); - - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("keeps SSE EOF as turn_failed when messages API never returns a completion before the cap", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - 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("aborts the OpenCode session when recovery caps an in-progress assistant message", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - const abort = vi.fn().mockResolvedValue({ data: true, error: undefined }); - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), - }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockResolvedValue({ - data: [ - { - info: { - ...(buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2500, - tokens: { - input: 0, - output: 0, - reasoning: 0, - cache: { read: 0, write: 0 }, - total: 0, - }, - }) as Record), - time: { created: 2100 }, - }, - parts: [], + sessionID: "ses_unit_test", + role: "assistant", }, - ], - error: undefined, - }), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort, - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 0, pollIntervalMs: 1, livenessMs: 1_000 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(false); - expect(turn.turnFailed).toBe(true); - expect(abort).toHaveBeenCalledWith({ - sessionID: "ses_unit_test", - directory: cwd, - }); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("ignores assistant messages that completed before the turn started", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - 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, - }; - + }, + { + directory: "/tmp/test", + payload: { + type: "message.part.delta", + properties: { + sessionID: "ses_unit_test", + messageID: "msg_assistant", + partID: "prt_text", + field: "text", + delta: "Hello from global", + }, + }, + }, + { + directory: "/tmp/test", + payload: { + type: "session.status", + properties: { + sessionID: "ses_unit_test", + status: { type: "idle" }, + }, + }, + }, + ]; const fakeClient = { event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), + subscribe: vi.fn(), }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), + global: { + event: vi.fn().mockResolvedValue({ stream: createAsyncIterable(globalEvents) }), }, 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 }), + promptAsync: vi.fn().mockResolvedValue({ data: {}, 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 = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/tmp/test" }, + fakeClient, + "ses_unit_test", + createTestLogger(), + "/tmp/opencode-storage", + ); - 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; + expect(fakeClient.global.event).toHaveBeenCalledWith({ + signal: expect.any(AbortSignal), + sseMaxRetryAttempts: 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().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(fakeClient.event.subscribe).not.toHaveBeenCalled(); expect(turn.turnCompleted).toBe(true); expect(turn.turnFailed).toBe(false); expect(turn.assistantMessages.map((message) => message.text).join("")).toBe( - "Done after waiting", + "Hello from global", ); - - 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 only new reasoning text when recovered reasoning parts grow across polls", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - let messagesCallCount = 0; - const inProgressAssistantInfo = { - ...(buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2500, - tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 }, - }) as Record), - time: { created: 2100 }, - }; - - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), - }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockImplementation(async () => { - messagesCallCount += 1; - if (messagesCallCount === 1) { - return { - data: [ - { - info: inProgressAssistantInfo, - parts: [ - { - id: "prt_reasoning", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "reasoning", - text: "Thinking", - }, - ], - }, - ], - error: undefined, - }; - } - if (messagesCallCount === 2) { - return { - data: [ - { - info: inProgressAssistantInfo, - parts: [ - { - id: "prt_reasoning", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "reasoning", - text: "Thinking more", - }, - ], - }, - ], - error: undefined, - }; - } - return { - data: [ - { - info: buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2700, - tokens: { - input: 10, - output: 5, - reasoning: 0, - cache: { read: 0, write: 0 }, - total: 15, - }, - }), - parts: [ - { - id: "prt_reasoning", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "reasoning", - text: "Thinking more", - }, - buildTextPart("msg_assistant", "prt_text", "Done."), - ], - }, - ], - error: undefined, - }; - }), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort: vi.fn().mockResolvedValue({ data: true, error: undefined }), - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(true); - expect(turn.turnFailed).toBe(false); - expect( - turn.allTimelineItems.filter((item) => item.type === "reasoning").map((item) => item.text), - ).toEqual(["Thinking", " more"]); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("emits running tool calls again when recovered tool input changes", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - let messagesCallCount = 0; - const inProgressAssistantInfo = { - ...(buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2500, - tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 }, - }) as Record), - time: { created: 2100 }, - }; - - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), - }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockImplementation(async () => { - messagesCallCount += 1; - if (messagesCallCount === 1) { - return { - data: [ - { - info: inProgressAssistantInfo, - parts: [ - { - id: "prt_tool", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "tool", - tool: "bash", - callID: "call_long", - state: { status: "running", input: { command: "echo one" } }, - }, - ], - }, - ], - error: undefined, - }; - } - if (messagesCallCount === 2) { - return { - data: [ - { - info: inProgressAssistantInfo, - parts: [ - { - id: "prt_tool", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "tool", - tool: "bash", - callID: "call_long", - state: { status: "running", input: { command: "echo two" } }, - }, - ], - }, - ], - error: undefined, - }; - } - return { - data: [ - { - info: buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2700, - tokens: { - input: 10, - output: 5, - reasoning: 0, - cache: { read: 0, write: 0 }, - total: 15, - }, - }), - parts: [ - { - id: "prt_tool", - sessionID: "ses_unit_test", - messageID: "msg_assistant", - type: "tool", - tool: "bash", - callID: "call_long", - state: { - status: "completed", - input: { command: "echo two" }, - output: "two", - }, - }, - buildTextPart("msg_assistant", "prt_text", "Done."), - ], - }, - ], - error: undefined, - }; - }), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort: vi.fn().mockResolvedValue({ data: true, error: undefined }), - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 1_000, pollIntervalMs: 5, livenessMs: 1_000 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(true); - expect(turn.turnFailed).toBe(false); - expect( - turn.toolCalls - .filter((toolCall) => toolCall.status === "running") - .map((toolCall) => toolCall.detail), - ).toEqual([ - { type: "shell", command: "echo one" }, - { type: "shell", command: "echo two" }, - ]); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("emits permission_requested for pending OpenCode questions while the turn is still in progress", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - 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("keeps recovering past the completion cap while a question is pending", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); - - let releaseStream!: () => void; - const streamMayEnd = new Promise((resolve) => { - releaseStream = resolve; - }); - - let messagesCallCount = 0; - const inProgressAssistantInfo = { - ...(buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2500, - tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 }, total: 0 }, - }) as Record), - time: { created: 2100 }, - }; - const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - await streamMayEnd; - return { done: true, value: undefined }; - }, - }), - }, - }), - }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), - }, - question: { - list: vi.fn().mockResolvedValue({ - data: [ - { - id: "qst_1", - sessionID: "ses_unit_test", - questions: [ - { - question: "Pick a color", - header: "Color", - options: [{ label: "red" }, { label: "blue" }], - }, - ], - }, - ], - error: undefined, - }), - }, - permission: { - list: vi.fn().mockResolvedValue({ data: [], error: undefined }), - }, - session: { - create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }), - messages: vi.fn().mockImplementation(async () => { - messagesCallCount += 1; - if (messagesCallCount < 3) { - return { - data: [ - { - info: inProgressAssistantInfo, - parts: [], - }, - ], - error: undefined, - }; - } - return { - data: [ - { - info: buildAssistantMessageInfo({ - id: "msg_assistant", - createdAt: 2100, - completedAt: 2700, - tokens: { - input: 10, - output: 5, - reasoning: 0, - cache: { read: 0, write: 0 }, - total: 15, - }, - }), - parts: [buildTextPart("msg_assistant", "prt_text", "Done")], - }, - ], - error: undefined, - }; - }), - promptAsync: vi.fn().mockImplementation(async () => { - releaseStream(); - return { data: {}, error: undefined }; - }), - abort: vi.fn().mockResolvedValue({ data: true, error: undefined }), - update: vi.fn().mockResolvedValue({ data: true, error: undefined }), - delete: vi.fn().mockResolvedValue({ data: true, error: undefined }), - }, - } as never; - - const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, { - runtime: { - acquireServer: vi.fn().mockResolvedValue({ - server: { port: 0, url: "http://localhost" }, - release: () => {}, - }), - ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }), - createClient: vi.fn().mockReturnValue(fakeClient), - shutdown: vi.fn().mockResolvedValue(undefined), - }, - recovery: { timeoutMs: 0, pollIntervalMs: 1, livenessMs: 1_000 }, - }); - - const session = await client.createSession({ provider: "opencode", cwd }); - const turn = await collectTurnEvents(streamSession(session, "hello")); - - expect(turn.turnCompleted).toBe(true); - expect(turn.turnFailed).toBe(false); - expect(turn.assistantMessages.map((message) => message.text).join("")).toBe("Done"); - expect(turn.events.filter((event) => event.type === "permission_requested")).toHaveLength(1); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); - }); - - test("emits tool calls and reasoning from the recovered assistant message parts", async () => { - const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); - const cwd = "/tmp/test"; - 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({ - stream: { - [Symbol.asyncIterator]: () => { - let index = 0; - const events: OpenCodeEvent[] = [ - { - type: "message.updated", - properties: { - info: { - id: "msg_assistant", - sessionID: "ses_unit_test", - role: "assistant", - }, - }, - } as OpenCodeEvent, - { - type: "message.part.delta", - properties: { - sessionID: "ses_unit_test", - messageID: "msg_assistant", - partID: "prt_text", - field: "text", - delta: "Hello, ", - }, - } as OpenCodeEvent, - ]; + test("fails a turn when OpenCode retry status does not recover", async () => { + vi.useFakeTimers(); + const retryStream: AsyncIterable = { + [Symbol.asyncIterator]: () => { + let emitted = false; + return { + next: async () => { + if (!emitted) { + emitted = true; return { - next: async () => { - if (index < events.length) { - return { done: false, value: events[index++] }; - } - await streamMayEnd; - return { done: true, value: undefined }; + done: false, + value: { + payload: { + type: "session.status", + properties: { + sessionID: "ses_unit_test", + status: { + type: "retry", + attempt: 1, + message: "model does not exist", + }, + }, + }, }, }; - }, + } + return new Promise(() => {}); }, - }), + }; }, - provider: { - list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }), + }; + const fakeClient = { + global: { + event: vi.fn().mockResolvedValue({ stream: retryStream }), }, 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 }), + promptAsync: vi.fn().mockResolvedValue({ data: {}, 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 = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/tmp/test" }, + fakeClient, + "ses_unit_test", + createTestLogger(), + "/tmp/opencode-storage", + ); + + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + await session.startTurn("hello"); + await vi.advanceTimersByTimeAsync(10_000); + + const failed = events.find((event) => event.type === "turn_failed"); + expect(failed).toMatchObject({ + type: "turn_failed", + error: expect.stringContaining("model does not exist"), }); - - 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("Hello, world!"); - - dateNowSpy.mockRestore(); - rmSync(storageRoot, { recursive: true, force: true }); + vi.useRealTimers(); }); test("deletes provider session on close when persistence is disabled", async () => { @@ -2316,19 +779,29 @@ describe("OpenCode adapter startTurn error handling", () => { }); test("emits turn_failed when client.session.promptAsync throws synchronously", async () => { - // Async iterable that never yields and never resolves. The IIFE in - // startTurn synchronously hits the promptAsync throw and finishes the - // turn before this iterator is ever pulled, so the never-resolving - // promise inside next() is fine and gets garbage-collected. + // Yield the server-connected event, then park forever. The adapter waits + // for that first event before sending the prompt. const neverYieldingStream: AsyncIterable = { - [Symbol.asyncIterator]: () => ({ - next: () => new Promise(() => {}), - }), + [Symbol.asyncIterator]: () => { + let emittedConnected = false; + return { + next: () => { + if (!emittedConnected) { + emittedConnected = true; + return Promise.resolve({ + done: false, + value: { type: "server.connected", properties: {} } as OpenCodeEvent, + }); + } + return new Promise(() => {}); + }, + }; + }, }; const fakeClient = { - event: { - subscribe: vi.fn().mockResolvedValue({ stream: neverYieldingStream }), + global: { + event: vi.fn().mockResolvedValue({ stream: neverYieldingStream }), }, session: { promptAsync: vi.fn(() => { @@ -2367,12 +840,11 @@ describe("OpenCode adapter startTurn error handling", () => { .mockReturnValueOnce(abortDeferred.promise) .mockResolvedValue({ data: true, error: undefined }); const fakeClient = { - event: { - subscribe: vi.fn().mockImplementation( - async ( - _params: { directory: string }, - options: { signal: AbortSignal }, - ): Promise<{ stream: AsyncIterable }> => ({ + global: { + event: vi.fn().mockImplementation( + async (options: { + signal: AbortSignal; + }): Promise<{ stream: AsyncIterable }> => ({ stream: abortableOpenCodeStream(options.signal), }), ), @@ -2496,44 +968,6 @@ function writeOpenCodeJson(storageRoot: string, relativePath: string, value: unk 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, - }; -} - function createTestDeferred(): { promise: Promise; resolve: (value: T) => void; @@ -2550,17 +984,28 @@ function createTestDeferred(): { function abortableOpenCodeStream(signal: AbortSignal): AsyncIterable { return { - [Symbol.asyncIterator]: () => ({ - next: () => - new Promise>((resolve) => { - if (signal.aborted) { - resolve({ done: true, value: undefined }); - return; + [Symbol.asyncIterator]: () => { + let emittedConnected = false; + return { + next: () => { + if (!emittedConnected) { + emittedConnected = true; + return Promise.resolve({ + done: false, + value: { type: "server.connected", properties: {} } as OpenCodeEvent, + }); } - signal.addEventListener("abort", () => resolve({ done: true, value: undefined }), { - once: true, + return new Promise>((resolve) => { + if (signal.aborted) { + resolve({ done: true, value: undefined }); + return; + } + signal.addEventListener("abort", () => resolve({ done: true, value: undefined }), { + once: true, + }); }); - }), - }), + }, + }; + }, }; } diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 3de3e60ff..83d27c3af 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -74,22 +74,8 @@ 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; -const OPENCODE_RECOVERY_ABORT_TIMEOUT_MS = 2_000; const OPENCODE_PENDING_ABORT_START_TIMEOUT_MS = 10_000; -// If OpenCode silently rejects the prompt (invalid model/mode/auth), no assistant -// message is ever persisted. Bound the wait so the turn fails in seconds instead -// of hanging until the completion cap. Valid models normally persist their first -// message within a second of LLM start, so 10s leaves comfortable headroom. -const OPENCODE_EOF_RECOVERY_LIVENESS_MS = 10_000; +const OPENCODE_RETRY_STATUS_FAILURE_MS = 10_000; const DEFAULT_MODES: AgentMode[] = [ { @@ -667,19 +653,6 @@ 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, @@ -941,15 +914,8 @@ export const __openCodeInternals = { }, }; -interface OpenCodeRecoveryOptions { - timeoutMs: number; - pollIntervalMs: number; - livenessMs: number; -} - interface OpenCodeAgentClientDeps { runtime?: OpenCodeRuntime; - recovery?: OpenCodeRecoveryOptions; } class ProductionOpenCodeRuntime implements OpenCodeRuntime { @@ -981,7 +947,6 @@ 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, @@ -992,11 +957,6 @@ 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( @@ -1044,7 +1004,6 @@ export class OpenCodeAgentClient implements AgentClient { new Map(this.modelContextWindows), acquisition.release, options?.persistSession, - this.recovery, ); } catch (error) { acquisition.release(); @@ -1087,7 +1046,6 @@ export class OpenCodeAgentClient implements AgentClient { new Map(this.modelContextWindows), acquisition.release, undefined, - this.recovery, ); } catch (error) { acquisition.release(); @@ -2195,6 +2153,24 @@ function traceOpenCode(tag: string, data: Record = {}): void { process.stderr.write(`[opencode-trace] ${line}\n`); } +function unwrapOpenCodeGlobalEvent(event: unknown): OpenCodeEvent | null { + const record = readOpenCodeRecord(event); + if (!record) { + return null; + } + + const payload = readOpenCodeRecord(record.payload); + if (typeof payload?.type === "string") { + return payload as unknown as OpenCodeEvent; + } + + if (typeof record.type === "string") { + return record as unknown as OpenCodeEvent; + } + + return null; +} + class OpenCodeAgentSession implements AgentSession { readonly provider = "opencode" as const; readonly capabilities = OPENCODE_CAPABILITIES; @@ -2203,7 +2179,6 @@ class OpenCodeAgentSession implements AgentSession { private readonly client: OpencodeClient; private readonly sessionId: string; private readonly logger: Logger; - private readonly storageRoot: string; private readonly modelContextWindowsByModelKey: ReadonlyMap; private currentMode: string = "default"; private pendingPermissions = new Map(); @@ -2232,41 +2207,25 @@ class OpenCodeAgentSession implements AgentSession { private releaseServer: (() => void) | null; private readonly persistSession: boolean; private deletedFromProvider = false; - private foregroundAssistantMessageEmitted = false; - private foregroundAssistantText = ""; - private foregroundUsageUpdated = false; - private foregroundKnownMessageIds = new Set(); - private foregroundEmittedQuestionIds = new Set(); - private foregroundEmittedPermissionIds = new Set(); - private foregroundEmittedReasoningTextLengthByPartId = new Map(); - private foregroundEmittedToolCallSignatureByCallId = new Map(); - private foregroundTurnStartedAt: number | null = null; - private readonly recovery: OpenCodeRecoveryOptions; + private retryFailureTimer: ReturnType | null = null; constructor( config: OpenCodeAgentConfig, client: OpencodeClient, sessionId: string, logger: Logger, - storageRoot: string, + _storageRoot: string, 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; this.sessionId = sessionId; this.logger = logger; - this.storageRoot = storageRoot; this.modelContextWindowsByModelKey = modelContextWindowsByModelKey; this.currentMode = normalizeOpenCodeModeId(config.modeId); this.releaseServer = releaseServer ?? null; this.persistSession = persistSession; - this.recovery = recovery; this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( config.model, ); @@ -2386,19 +2345,11 @@ class OpenCodeAgentSession implements AgentSession { } await this.awaitPendingAbortBeforeStartingTurn(); - this.foregroundTurnStartedAt = Date.now(); this.runningToolCalls.clear(); this.subAgentsByCallId.clear(); this.subAgentCallIdByChildSessionId.clear(); this.pendingChildToolPartsBySessionId.clear(); - this.foregroundAssistantMessageEmitted = false; - this.foregroundAssistantText = ""; - this.foregroundUsageUpdated = false; - this.foregroundEmittedQuestionIds.clear(); - this.foregroundEmittedPermissionIds.clear(); - this.foregroundEmittedReasoningTextLengthByPartId.clear(); - this.foregroundEmittedToolCallSignatureByCallId.clear(); - this.foregroundKnownMessageIds = await this.readPersistedSessionMessageIds(); + this.clearRetryFailureTimer(); const turnAbortController = new AbortController(); this.abortController = turnAbortController; await this.ensureMcpServersConfigured(); @@ -2611,54 +2562,27 @@ class OpenCodeAgentSession implements AgentSession { ): 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 }, - ); - traceOpenCode("subscribe.ready", { turnId, sessionId: this.sessionId }); - subscriptionReady.resolve(); - + const result = await this.client.global.event({ + signal: turnAbortController.signal, + sseMaxRetryAttempts: 0, + }); let eventCount = 0; - for await (const event of result.stream) { + let subscriptionReadyResolved = false; + for await (const rawEvent 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; + if (!subscriptionReadyResolved) { + subscriptionReadyResolved = true; + traceOpenCode("subscribe.ready", { turnId, sessionId: this.sessionId }); + subscriptionReady.resolve(); } - - const translated = await this.translateEvent(event); - traceOpenCode("event.translated", { + const shouldContinue = await this.consumeOpenCodeStreamEvent({ + rawEvent, + eventCount, turnId, - n: eventCount, - count: translated.length, - types: translated.map((t) => t.type), + turnAbortController, }); - 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") { - this.trackToolCall(e.item); - } - const terminalEvent = toTerminalTurnEvent(e); - if (terminalEvent) { - traceOpenCode("event.terminal", { turnId, type: terminalEvent.type }); - this.finishForegroundTurn(terminalEvent, turnId); - return; - } - this.notifySubscribers(e, turnId); + if (!shouldContinue) { + return; } } @@ -2670,12 +2594,10 @@ class OpenCodeAgentSession implements AgentSession { }); if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { - const recovered = await this.recoverTurnFromPersistedCompletion(turnId); - traceOpenCode("recovery.result", { turnId, recovered }); - if (recovered) { - return; - } traceOpenCode("turn.fail.eof", { turnId, eventCount }); + if (!subscriptionReadyResolved) { + subscriptionReady.reject(new Error("OpenCode event stream ended before it became ready")); + } this.finishForegroundTurn( { type: "turn_failed", @@ -2719,435 +2641,62 @@ class OpenCodeAgentSession implements AgentSession { } } - private async recoverTurnFromPersistedCompletion(turnId: string): Promise { - traceOpenCode("recovery.start", { + private async consumeOpenCodeStreamEvent(params: { + rawEvent: unknown; + eventCount: number; + turnId: string; + turnAbortController: AbortController; + }): Promise { + const { rawEvent, eventCount, turnId, turnAbortController } = params; + const event = unwrapOpenCodeGlobalEvent(rawEvent); + traceOpenCode("event.raw", { turnId, - foregroundTurnStartedAt: this.foregroundTurnStartedAt, - knownMessageIds: Array.from(this.foregroundKnownMessageIds), - sessionId: this.sessionId, - timeoutMs: this.recovery.timeoutMs, - pollIntervalMs: this.recovery.pollIntervalMs, + n: eventCount, + type: event?.type, + rawType: readOpenCodeRecord(rawEvent)?.type, + directory: readOpenCodeRecord(rawEvent)?.directory, + properties: event ? (event as { properties?: unknown }).properties : undefined, }); - const startedAt = this.foregroundTurnStartedAt; - if (startedAt === null) { - traceOpenCode("recovery.no-start-time", { turnId }); - return false; - } - - 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) { - const deferred = await this.deferForPendingPermissionOrFailRecoveredTurnAfterCap( - turnId, - attempt, - "liveness", - ); - if (deferred) { - continue; - } - return true; - } - if (now >= completionDeadline) { - const deferred = await this.deferForPendingPermissionOrFailRecoveredTurnAfterCap( - turnId, - attempt, - "completion", - ); - if (deferred) { - continue; - } - return true; - } - const waitMs = Math.min(this.recovery.pollIntervalMs, completionDeadline - now); - await new Promise((resolve) => setTimeout(resolve, waitMs)); - } - } - - private async deferForPendingPermissionOrFailRecoveredTurnAfterCap( - turnId: string, - attempt: number, - cap: "liveness" | "completion", - ): Promise { - if (this.pendingPermissions.size > 0) { - // A pending OpenCode question/permission means the turn is blocked on - // user input, not dead. Keep polling until the user response lets the - // assistant finish or the turn is canceled. - traceOpenCode(`recovery.${cap}-deferred-for-permission`, { - turnId, - attempt, - pendingPermissionIds: Array.from(this.pendingPermissions.keys()), - }); - await new Promise((resolve) => setTimeout(resolve, this.recovery.pollIntervalMs)); + if (!event) { return true; } - - traceOpenCode(cap === "liveness" ? "recovery.liveness-exhausted" : "recovery.exhausted", { - turnId, - attempt, - }); - await this.failRecoveredTurnAfterCap(turnId, cap); - return false; - } - - private async failRecoveredTurnAfterCap( - turnId: string, - cap: "liveness" | "completion", - ): Promise { - await this.abortOpenCodeSessionAfterRecoveryCap(turnId, cap); - this.finishForegroundTurn( - { - type: "turn_failed", - provider: "opencode", - error: "OpenCode event stream ended before the turn reached a terminal state", - }, - turnId, - ); - } - - private async abortOpenCodeSessionAfterRecoveryCap( - turnId: string, - cap: "liveness" | "completion", - ): Promise { - const abortPromise = this.beginSessionAbort(turnId, `recovery-${cap}`); - await withTimeout( - abortPromise, - OPENCODE_RECOVERY_ABORT_TIMEOUT_MS, - "OpenCode session.abort", - ).catch((error) => { - this.logger.warn( - { err: error, sessionId: this.sessionId, turnId, cap }, - "OpenCode session.abort exceeded the EOF recovery cap", - ); - }); - } - - private async pollPendingQuestionsAndPermissions(turnId: string): Promise { - const [questionsResponse, permissionsResponse] = await Promise.all([ - Promise.resolve() - .then(() => this.client.question.list({ directory: this.config.cwd })) - .catch((error) => { - traceOpenCode("recovery.question-list.throw", { - turnId, - error: - error instanceof Error - ? { name: error.name, message: error.message, stack: error.stack } - : String(error), - }); - return null; - }), - Promise.resolve() - .then(() => this.client.permission.list({ directory: this.config.cwd })) - .catch((error) => { - traceOpenCode("recovery.permission-list.throw", { - turnId, - error: - error instanceof Error - ? { name: error.name, message: error.message, stack: error.stack } - : String(error), - }); - return null; - }), - ]); - - if (this.activeForegroundTurnId !== turnId) return 0; - - 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 Promise.resolve() - .then(() => - this.client.session.messages({ - sessionID: this.sessionId, - directory: this.config.cwd, - }), - ) - .catch((error) => { - traceOpenCode("recovery.messages.throw", { - error: - error instanceof Error - ? { name: error.name, message: error.message, stack: error.stack } - : String(error), - }); - return null; + if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) { + traceOpenCode("event.skip", { + turnId, + n: eventCount, + aborted: turnAbortController.signal.aborted, + activeTurnId: this.activeForegroundTurnId, }); - if (response === null) { - return null; - } - 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 }; - } - - let text = item.parts - .filter((part): part is Extract => part.type === "text") - .map((part) => (part.text ?? "").trim()) - .filter((part) => part.length > 0) - .join("\n\n"); - - if (!text) { - text = stringifyStructuredAssistantMessage(info.structured) ?? ""; - } - if (!text) continue; - - const usage: AgentUsage = {}; - 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) { - const emittedTextLength = - this.foregroundEmittedReasoningTextLengthByPartId.get(part.id) ?? 0; - if (part.text.length <= emittedTextLength) continue; - const text = part.text.slice(emittedTextLength); - this.foregroundEmittedReasoningTextLengthByPartId.set(part.id, part.text.length); - this.notifySubscribers( - { - type: "timeline", - provider: "opencode", - item: { type: "reasoning", text }, - }, - turnId, - ); - continue; - } - if (part.type !== "tool") continue; - const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part); - if (!parsedToolPart.success || !parsedToolPart.data) continue; - const callId = parsedToolPart.data.callId; - const signature = this.createRecoveredToolCallSignature(part, parsedToolPart.data); - const lastSignature = this.foregroundEmittedToolCallSignatureByCallId.get(callId); - if (lastSignature === signature) continue; - this.foregroundEmittedToolCallSignatureByCallId.set(callId, signature); - this.trackToolCall(parsedToolPart.data); - this.notifySubscribers( - { - type: "timeline", - provider: "opencode", - item: parsedToolPart.data, - }, - turnId, - ); - } - } - - private createRecoveredToolCallSignature( - part: Extract, - item: ToolCallTimelineItem, - ): string { - const state = (part as { state?: { input?: unknown; output?: unknown; error?: unknown } }) - .state; - return JSON.stringify([ - item.callId, - item.status, - state?.input ?? null, - state?.output ?? null, - state?.error ?? null, - ]); - } - - private applyRecoveredAssistantCompletion( - completion: { - messageId: string; - 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 via messages API after SSE EOF", - ); - - const recoveryText = this.resolvePersistedAssistantRecoveryText(completion.text); - if (recoveryText === null) { return false; } - if (recoveryText.length > 0) { - this.notifySubscribers( - { - type: "timeline", - provider: "opencode", - item: { type: "assistant_message", text: recoveryText }, - }, - turnId, - ); - this.foregroundAssistantMessageEmitted = true; - } - - if (hasNormalizedOpenCodeUsage(completion.usage) && !this.foregroundUsageUpdated) { - this.accumulatedUsage = { - ...this.accumulatedUsage, - ...completion.usage, - }; - this.notifySubscribers( - { - type: "usage_updated", - provider: "opencode", - usage: { ...this.accumulatedUsage }, - }, - turnId, - ); - this.foregroundUsageUpdated = true; - } - - this.finishForegroundTurn( - { - type: "turn_completed", - provider: "opencode", - usage: hasNormalizedOpenCodeUsage(this.accumulatedUsage) - ? { ...this.accumulatedUsage } - : undefined, - }, + this.armRetryFailureTimerForStatus(event, turnId); + const translated = await this.translateEvent(event); + traceOpenCode("event.translated", { turnId, - ); - return true; - } + n: eventCount, + count: translated.length, + types: translated.map((t) => t.type), + }); - private resolvePersistedAssistantRecoveryText(completedText: string): string | null { - if (!this.foregroundAssistantMessageEmitted) { - return completedText; - } - - if (completedText === this.foregroundAssistantText) { - return ""; - } - - return completedText.startsWith(this.foregroundAssistantText) - ? completedText.slice(this.foregroundAssistantText.length) - : null; - } - - private async readPersistedSessionMessageIds(): Promise> { - const messageRoot = path.join(this.storageRoot, "message", this.sessionId); - const messageFiles = await findJsonFiles(messageRoot); - const messageIds = new Set(); - - for (const file of messageFiles) { - const parsed = await readJsonFile(file, OpenCodeStoredMessageSchema); - if (parsed?.sessionID === this.sessionId) { - messageIds.add(parsed.id); + for (const e of translated) { + if (this.activeForegroundTurnId !== turnId) { + traceOpenCode("event.translated.skip-active", { turnId, type: e.type }); + return false; } + if (e.type === "timeline" && e.item.type === "tool_call") { + this.trackToolCall(e.item); + } + const terminalEvent = toTerminalTurnEvent(e); + if (terminalEvent) { + traceOpenCode("event.terminal", { turnId, type: terminalEvent.type }); + this.finishForegroundTurn(terminalEvent, turnId); + return false; + } + this.notifySubscribers(e, turnId); } - return messageIds; + return true; } private finishForegroundTurn( @@ -3169,7 +2718,7 @@ class OpenCodeAgentSession implements AgentSession { } else { this.runningToolCalls.clear(); } - this.foregroundTurnStartedAt = null; + this.clearRetryFailureTimer(); this.activeForegroundTurnId = null; // Abort the SSE connection so the SDK tears down the underlying fetch. this.abortController?.abort(); @@ -3185,6 +2734,44 @@ class OpenCodeAgentSession implements AgentSession { this.runningToolCalls.delete(item.callId); } + private armRetryFailureTimerForStatus(event: OpenCodeEvent, turnId: string): void { + if (this.retryFailureTimer || event.type !== "session.status") { + return; + } + if (event.properties.sessionID !== this.sessionId || event.properties.status.type !== "retry") { + return; + } + + const retry = event.properties.status; + const message = typeof retry.message === "string" ? retry.message.trim() : ""; + const error = message + ? `OpenCode provider retry did not recover: ${message}` + : "OpenCode provider retry did not recover"; + + this.retryFailureTimer = setTimeout(() => { + this.retryFailureTimer = null; + if (this.activeForegroundTurnId !== turnId) { + return; + } + this.finishForegroundTurn( + { + type: "turn_failed", + provider: "opencode", + error, + }, + turnId, + ); + }, OPENCODE_RETRY_STATUS_FAILURE_MS); + } + + private clearRetryFailureTimer(): void { + if (!this.retryFailureTimer) { + return; + } + clearTimeout(this.retryFailureTimer); + this.retryFailureTimer = null; + } + private synthesizeInterruptedToolCalls(turnId: string): void { for (const item of this.runningToolCalls.values()) { const error = { message: "Tool execution aborted" }; @@ -3215,13 +2802,6 @@ class OpenCodeAgentSession implements AgentSession { private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void { const turnId = turnIdOverride ?? this.activeForegroundTurnId; - if (event.type === "timeline" && event.item.type === "assistant_message") { - this.foregroundAssistantMessageEmitted = true; - this.foregroundAssistantText += event.item.text; - } - if (event.type === "usage_updated") { - this.foregroundUsageUpdated = true; - } const tagged = turnId ? { ...event, turnId } : event; for (const callback of this.subscribers) { try { diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts index c410ffa03..5d695ebaa 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts @@ -47,6 +47,7 @@ export class TestOpenCodeClient { appAgents: [] as unknown[], commandList: [] as unknown[], eventSubscribe: [] as unknown[], + globalEvent: [] as unknown[], permissionReply: [] as unknown[], providerList: [] as unknown[], questionReject: [] as unknown[], @@ -99,6 +100,12 @@ export class TestOpenCodeClient { return { stream: this.eventStream }; }, }, + global: { + event: async (options: unknown) => { + this.calls.globalEvent.push(options); + return { stream: this.eventStream }; + }, + }, mcp: { add: async () => ({}), connect: async () => ({}), diff --git a/packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts index 879cb61ee..b71a7f03d 100644 --- a/packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/opencode-initial-prompt-wait.real.e2e.test.ts @@ -9,6 +9,8 @@ import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { isProviderAvailable } from "./agent-configs.js"; +const OPENCODE_REAL_TEST_MODEL = "opencode/big-pickle"; + function tmpCwd(): string { return mkdtempSync(path.join(tmpdir(), "daemon-real-opencode-init-prompt-")); } @@ -47,13 +49,13 @@ describe("daemon E2E (real opencode) - initial prompt wait", () => { try { const models = await client.listProviderModels("opencode"); - expect(models.models.some((model) => model.id === "zai/glm-5.1")).toBe(true); + expect(models.models.some((model) => model.id === OPENCODE_REAL_TEST_MODEL)).toBe(true); const agent = await client.createAgent({ provider: "opencode", cwd, title: "OpenCode initial prompt wait regression", - model: "opencode/big-pickle", + model: OPENCODE_REAL_TEST_MODEL, initialPrompt: "Reply with exactly: BIG_PICKLE_OK", }); @@ -83,35 +85,4 @@ describe("daemon E2E (real opencode) - initial prompt wait", () => { rmSync(cwd, { recursive: true, force: true }); } }, 90_000); - - test("waitForFinish surfaces a terminal error when zai/glm-5.1 enters a fatal retry loop", async () => { - const cwd = tmpCwd(); - const { client, daemon } = await createHarness(); - - try { - const models = await client.listProviderModels("opencode"); - expect(models.models.some((model) => model.id === "zai/glm-5.1")).toBe(true); - - const agent = await client.createAgent({ - provider: "opencode", - cwd, - title: "OpenCode zai fatal retry regression", - model: "zai/glm-5.1", - initialPrompt: "Reply with exactly: GLM_51_OK", - }); - - const finish = await client.waitForFinish(agent.id, 60_000); - expect(finish.status).toBe("error"); - expect((finish.error ?? "").toLowerCase()).toMatch( - /insufficient balance|resource package|recharge/, - ); - - const snapshot = await client.fetchAgent(agent.id); - expect(snapshot.agent?.status).toBe("error"); - } finally { - await client.close().catch(() => undefined); - await daemon.close(); - rmSync(cwd, { recursive: true, force: true }); - } - }, 90_000); });