From 2a84b081298f904936f5ad61dca2388b48835e0e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 18:08:38 +0700 Subject: [PATCH] fix(server): wait for opencode completion after EOF The original EOF recovery only checked OpenCode storage once after the SSE stream ended. That missed the real failure mode from #861: OpenCode can drop /event while the turn continues behind the scenes, then persist the assistant completion a little later. Poll the persisted session for the active turn before failing EOF, while still requiring strong completion evidence and ignoring messages that predate the turn. This preserves failure behavior when there is no persisted completion. Add behavioral coverage for delayed completion after EOF, partial streamed text plus persisted completion without duplication, stale old completions, and the no-evidence failure path. --- .../agent/providers/opencode-agent.test.ts | 308 ++++++++++++++++++ .../server/agent/providers/opencode-agent.ts | 55 +++- 2 files changed, 357 insertions(+), 6 deletions(-) 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 c2205220a1..7b9d937ede 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -13,6 +13,7 @@ import { } from "./opencode-agent.js"; import { streamSession } from "./test-utils/session-stream-adapter.js"; import type { + AgentSession, AgentSessionConfig, AgentStreamEvent, ToolCallTimelineItem, @@ -42,6 +43,8 @@ interface TurnResult { error?: string; } +const FAST_OPENCODE_EOF_RECOVERY_POLICY = { maxAttempts: 5, delayMs: 1 }; + async function collectTurnEvents(iterator: AsyncGenerator): Promise { const result: TurnResult = { events: [], @@ -78,6 +81,140 @@ async function collectTurnEvents(iterator: AsyncGenerator): Pr return result; } +interface FakeOpenCodeStream { + stream: AsyncIterable; + close: () => void; + closed: Promise; +} + +function createFakeOpenCodeStream(events: OpenCodeEvent[] = []): FakeOpenCodeStream { + let releaseStream!: () => void; + const canEnd = new Promise((resolve) => { + releaseStream = resolve; + }); + let notifyEnded!: () => void; + const closed = new Promise((resolve) => { + notifyEnded = resolve; + }); + + return { + close: releaseStream, + closed, + stream: { + [Symbol.asyncIterator]: () => { + let index = 0; + return { + next: async () => { + if (index < events.length) { + return { done: false, value: events[index++] }; + } + await canEnd; + notifyEnded(); + return { done: true, value: undefined }; + }, + }; + }, + }, + }; +} + +function createFakeOpenCode(stream: FakeOpenCodeStream, onPrompt: () => void | Promise) { + return { + event: { + subscribe: async () => ({ stream: stream.stream }), + }, + provider: { + list: async () => ({ data: { connected: [], all: [] }, error: undefined }), + }, + session: { + create: async () => ({ data: { id: "ses_unit_test" }, error: undefined }), + promptAsync: async () => { + await onPrompt(); + return { data: {}, error: undefined }; + }, + abort: async () => ({ data: true, error: undefined }), + update: async () => ({ data: true, error: undefined }), + delete: async () => ({ data: true, error: undefined }), + }, + } as never; +} + +async function createSessionWithFakeOpenCode(params: { + storageRoot: string; + cwd: string; + stream: FakeOpenCodeStream; + onPrompt: () => void | Promise; +}): Promise { + const fakeClient = createFakeOpenCode(params.stream, params.onPrompt); + const client = new OpenCodeAgentClient(createTestLogger(), undefined, params.storageRoot, { + runtime: { + acquireServer: async () => ({ + server: { port: 0, url: "http://localhost" }, + release: () => {}, + }), + ensureServerRunning: async () => ({ port: 0, url: "http://localhost" }), + createClient: () => fakeClient, + shutdown: async () => undefined, + }, + eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY, + }); + + return client.createSession({ provider: "opencode", cwd: params.cwd }); +} + +function observeTurn(session: AgentSession) { + const events: AgentStreamEvent[] = []; + let resolveTerminal!: (event: AgentStreamEvent) => void; + const terminal = new Promise((resolve) => { + resolveTerminal = resolve; + }); + + session.subscribe((event) => { + events.push(event); + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + resolveTerminal(event); + } + }); + + return { + terminal, + assistantMessages: () => + events + .flatMap((event) => (event.type === "timeline" ? [event.item] : [])) + .filter((item): item is AssistantMessageTimelineItem => item.type === "assistant_message"), + }; +} + +function openCodeAssistantStarted(messageId: string): OpenCodeEvent { + return { + type: "message.updated", + properties: { + info: { + id: messageId, + sessionID: "ses_unit_test", + role: "assistant", + }, + }, + } as OpenCodeEvent; +} + +function openCodeTextDelta(messageId: string, partId: string, delta: string): OpenCodeEvent { + return { + type: "message.part.delta", + properties: { + sessionID: "ses_unit_test", + messageID: messageId, + partID: partId, + field: "text", + delta, + }, + } as OpenCodeEvent; +} + function isBinaryInstalled(binary: string): boolean { try { const out = execFileSync("which", [binary], { encoding: "utf8" }).trim(); @@ -671,6 +808,7 @@ describe("OpenCode adapter startTurn error handling", () => { createClient: vi.fn().mockReturnValue(fakeClient), shutdown: vi.fn().mockResolvedValue(undefined), }, + eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY, }); const session = await client.createSession({ provider: "opencode", cwd }); @@ -696,6 +834,126 @@ describe("OpenCode adapter startTurn error handling", () => { rmSync(storageRoot, { recursive: true, force: true }); }); + test("recovers SSE EOF when persisted assistant completion appears after the stream closes", async () => { + const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); + const cwd = "/tmp/test"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + try { + writeOpenCodeStoredSession(storageRoot, cwd); + const stream = createFakeOpenCodeStream(); + const session = await createSessionWithFakeOpenCode({ + storageRoot, + cwd, + stream, + onPrompt: stream.close, + }); + const turn = observeTurn(session); + + await session.startTurn("hello"); + await stream.closed; + writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply"); + + expect((await turn.terminal).type).toBe("turn_completed"); + expect( + turn + .assistantMessages() + .map((message) => message.text) + .join(""), + ).toBe("Recovered assistant reply"); + } finally { + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + } + }); + + test("recovers delayed persisted completion without duplicating text already streamed before EOF", async () => { + const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); + const cwd = "/tmp/test"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + let completionWriteSettled: Promise = Promise.resolve(); + const writeDelayedCompletion = async () => { + await nextMacrotask(); + writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply"); + }; + + try { + writeOpenCodeStoredSession(storageRoot, cwd); + const stream = createFakeOpenCodeStream([ + openCodeAssistantStarted("msg_assistant"), + openCodeTextDelta("msg_assistant", "prt_text", "Recovered "), + ]); + const session = await createSessionWithFakeOpenCode({ + storageRoot, + cwd, + stream, + onPrompt: async () => { + stream.close(); + completionWriteSettled = writeDelayedCompletion(); + await completionWriteSettled; + }, + }); + const turn = observeTurn(session); + + await session.startTurn("hello"); + await stream.closed; + + expect((await turn.terminal).type).toBe("turn_completed"); + const assistantMessages = turn.assistantMessages(); + expect(assistantMessages.map((message) => message.text).join("")).toBe( + "Recovered assistant reply", + ); + expect(assistantMessages.map((message) => message.text)).toEqual([ + "Recovered ", + "assistant reply", + ]); + } finally { + await completionWriteSettled; + dateNowSpy.mockRestore(); + rmSync(storageRoot, { recursive: true, force: true }); + } + }); + + test("ignores old completed assistant messages while waiting for delayed current-turn completion", async () => { + const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-")); + const cwd = "/tmp/test"; + const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000); + + try { + writeOpenCodeStoredSession(storageRoot, cwd); + writeOpenCodeAssistantCompletion(storageRoot, "msg_assistant_old", "Old assistant reply", { + created: 1500, + completed: 1600, + partStart: 1500, + partEnd: 1550, + }); + const stream = createFakeOpenCodeStream(); + const session = await createSessionWithFakeOpenCode({ + storageRoot, + cwd, + stream, + onPrompt: stream.close, + }); + const turn = observeTurn(session); + + await session.startTurn("hello"); + await stream.closed; + writeRecoveredAssistantCompletion( + storageRoot, + "msg_assistant_current", + "Current assistant reply", + ); + + expect((await turn.terminal).type).toBe("turn_completed"); + expect(turn.assistantMessages().map((message) => message.text)).toEqual([ + "Current assistant reply", + ]); + } finally { + 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"; @@ -763,6 +1021,7 @@ describe("OpenCode adapter startTurn error handling", () => { createClient: vi.fn().mockReturnValue(fakeClient), shutdown: vi.fn().mockResolvedValue(undefined), }, + eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY, }); const session = await client.createSession({ provider: "opencode", cwd }); @@ -867,6 +1126,7 @@ describe("OpenCode adapter startTurn error handling", () => { createClient: vi.fn().mockReturnValue(fakeClient), shutdown: vi.fn().mockResolvedValue(undefined), }, + eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY, }); const session = await client.createSession({ provider: "opencode", cwd }); @@ -1058,3 +1318,51 @@ function writeOpenCodeJson(storageRoot: string, relativePath: string, value: unk mkdirSync(path.dirname(filePath), { recursive: true }); writeFileSync(filePath, JSON.stringify(value), "utf8"); } + +function writeOpenCodeStoredSession(storageRoot: string, cwd: string): void { + writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", { + id: "ses_unit_test", + directory: cwd, + time: { created: 1000, updated: 3000 }, + }); +} + +function nextMacrotask(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +function writeRecoveredAssistantCompletion( + storageRoot: string, + messageId: string, + text: string, +): void { + writeOpenCodeAssistantCompletion(storageRoot, messageId, text, { + created: 2000, + completed: 2500, + partStart: 2100, + partEnd: 2400, + }); +} + +function writeOpenCodeAssistantCompletion( + storageRoot: string, + messageId: string, + text: string, + time: { created: number; completed: number; partStart: number; partEnd: number }, +): void { + writeOpenCodeJson(storageRoot, `message/ses_unit_test/${messageId}.json`, { + id: messageId, + sessionID: "ses_unit_test", + role: "assistant", + finish: "stop", + time: { created: time.created, completed: time.completed }, + }); + writeOpenCodeJson(storageRoot, `part/${messageId}/prt_text.json`, { + id: "prt_text", + sessionID: "ses_unit_test", + messageID: messageId, + type: "text", + text, + time: { start: time.partStart, end: time.partEnd }, + }); +} diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 236c1cbd77..da935d5a5d 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -74,6 +74,10 @@ const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { const OPENCODE_BUILD_MODE_ID = "build"; const OPENCODE_FULL_ACCESS_MODE_ID = "full-access"; const OPENCODE_STORAGE_SESSION_LIMIT = 200; +const DEFAULT_OPENCODE_EOF_RECOVERY_POLICY = { + maxAttempts: 600, + delayMs: 1_000, +}; const DEFAULT_MODES: AgentMode[] = [ { @@ -144,6 +148,11 @@ interface OpenCodePersistedAssistantCompletion { usage?: AgentUsage; } +interface OpenCodeEofRecoveryPolicy { + maxAttempts: number; + delayMs: number; +} + type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" }; type OpenCodeMessageRole = "user" | "assistant"; @@ -904,6 +913,10 @@ async function readOpenCodePersistedAssistantCompletion( return null; } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function hasStrongPersistedCompletionEvidence( message: OpenCodeStoredMessage, parts: OpenCodeStoredPart[], @@ -1043,6 +1056,7 @@ export const __openCodeInternals = { interface OpenCodeAgentClientDeps { runtime?: OpenCodeRuntime; + eofRecoveryPolicy?: OpenCodeEofRecoveryPolicy; } class ProductionOpenCodeRuntime implements OpenCodeRuntime { @@ -1074,6 +1088,7 @@ export class OpenCodeAgentClient implements AgentClient { private readonly runtimeSettings?: ProviderRuntimeSettings; private readonly modelContextWindows = new Map(); private readonly storageRoot: string; + private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy; constructor( logger: Logger, @@ -1084,6 +1099,7 @@ export class OpenCodeAgentClient implements AgentClient { this.logger = logger.child({ module: "agent", provider: "opencode" }); this.runtimeSettings = runtimeSettings; this.storageRoot = storageRoot ?? resolveOpenCodeStorageRoot(); + this.eofRecoveryPolicy = deps.eofRecoveryPolicy ?? DEFAULT_OPENCODE_EOF_RECOVERY_POLICY; this.runtime = deps.runtime ?? new ProductionOpenCodeRuntime( @@ -1131,6 +1147,7 @@ export class OpenCodeAgentClient implements AgentClient { new Map(this.modelContextWindows), acquisition.release, options?.persistSession, + this.eofRecoveryPolicy, ); } catch (error) { acquisition.release(); @@ -1172,6 +1189,8 @@ export class OpenCodeAgentClient implements AgentClient { this.storageRoot, new Map(this.modelContextWindows), acquisition.release, + undefined, + this.eofRecoveryPolicy, ); } catch (error) { acquisition.release(); @@ -2302,6 +2321,7 @@ class OpenCodeAgentSession implements AgentSession { private selectedModelContextWindowMaxTokens: number | undefined; private releaseServer: (() => void) | null; private readonly persistSession: boolean; + private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy; private deletedFromProvider = false; private foregroundAssistantMessageEmitted = false; private foregroundAssistantText = ""; @@ -2317,6 +2337,7 @@ class OpenCodeAgentSession implements AgentSession { modelContextWindowsByModelKey: ReadonlyMap = new Map(), releaseServer?: () => void, persistSession = true, + eofRecoveryPolicy: OpenCodeEofRecoveryPolicy = DEFAULT_OPENCODE_EOF_RECOVERY_POLICY, ) { this.config = config; this.client = client; @@ -2327,6 +2348,7 @@ class OpenCodeAgentSession implements AgentSession { this.currentMode = normalizeOpenCodeModeId(config.modeId); this.releaseServer = releaseServer ?? null; this.persistSession = persistSession; + this.eofRecoveryPolicy = eofRecoveryPolicy; this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( config.model, ); @@ -2670,12 +2692,7 @@ class OpenCodeAgentSession implements AgentSession { return false; } - const completion = await readOpenCodePersistedAssistantCompletion( - this.storageRoot, - this.sessionId, - this.foregroundKnownMessageIds, - this.foregroundTurnStartedAt, - ); + const completion = await this.waitForPersistedAssistantCompletion(turnId); if (!completion || this.activeForegroundTurnId !== turnId) { return false; } @@ -2732,6 +2749,32 @@ class OpenCodeAgentSession implements AgentSession { return true; } + private async waitForPersistedAssistantCompletion( + turnId: string, + ): Promise { + for (let attempt = 0; attempt < this.eofRecoveryPolicy.maxAttempts; attempt += 1) { + if (this.foregroundTurnStartedAt === null || this.activeForegroundTurnId !== turnId) { + return null; + } + + const completion = await readOpenCodePersistedAssistantCompletion( + this.storageRoot, + this.sessionId, + this.foregroundKnownMessageIds, + this.foregroundTurnStartedAt, + ); + if (completion) { + return completion; + } + + if (attempt < this.eofRecoveryPolicy.maxAttempts - 1) { + await sleep(this.eofRecoveryPolicy.delayMs); + } + } + + return null; + } + private resolvePersistedAssistantRecoveryText(completedText: string): string | null { if (!this.foregroundAssistantMessageEmitted) { return completedText;