From ef892bd27dd1ea18d3bb5c973aa72643c7f762b1 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 18:28:53 +0700 Subject: [PATCH] Revert "fix(server): wait for opencode completion after EOF" This reverts commit 2a84b081298f904936f5ad61dca2388b48835e0e. --- .../agent/providers/opencode-agent.test.ts | 308 ------------------ .../server/agent/providers/opencode-agent.ts | 55 +--- 2 files changed, 6 insertions(+), 357 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 7b9d937ed..c2205220a 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -13,7 +13,6 @@ import { } from "./opencode-agent.js"; import { streamSession } from "./test-utils/session-stream-adapter.js"; import type { - AgentSession, AgentSessionConfig, AgentStreamEvent, ToolCallTimelineItem, @@ -43,8 +42,6 @@ interface TurnResult { error?: string; } -const FAST_OPENCODE_EOF_RECOVERY_POLICY = { maxAttempts: 5, delayMs: 1 }; - async function collectTurnEvents(iterator: AsyncGenerator): Promise { const result: TurnResult = { events: [], @@ -81,140 +78,6 @@ 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(); @@ -808,7 +671,6 @@ 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 }); @@ -834,126 +696,6 @@ 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"; @@ -1021,7 +763,6 @@ 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 }); @@ -1126,7 +867,6 @@ 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 }); @@ -1318,51 +1058,3 @@ 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 da935d5a5..236c1cbd7 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -74,10 +74,6 @@ 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[] = [ { @@ -148,11 +144,6 @@ interface OpenCodePersistedAssistantCompletion { usage?: AgentUsage; } -interface OpenCodeEofRecoveryPolicy { - maxAttempts: number; - delayMs: number; -} - type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" }; type OpenCodeMessageRole = "user" | "assistant"; @@ -913,10 +904,6 @@ async function readOpenCodePersistedAssistantCompletion( return null; } -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - function hasStrongPersistedCompletionEvidence( message: OpenCodeStoredMessage, parts: OpenCodeStoredPart[], @@ -1056,7 +1043,6 @@ export const __openCodeInternals = { interface OpenCodeAgentClientDeps { runtime?: OpenCodeRuntime; - eofRecoveryPolicy?: OpenCodeEofRecoveryPolicy; } class ProductionOpenCodeRuntime implements OpenCodeRuntime { @@ -1088,7 +1074,6 @@ 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, @@ -1099,7 +1084,6 @@ 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( @@ -1147,7 +1131,6 @@ export class OpenCodeAgentClient implements AgentClient { new Map(this.modelContextWindows), acquisition.release, options?.persistSession, - this.eofRecoveryPolicy, ); } catch (error) { acquisition.release(); @@ -1189,8 +1172,6 @@ export class OpenCodeAgentClient implements AgentClient { this.storageRoot, new Map(this.modelContextWindows), acquisition.release, - undefined, - this.eofRecoveryPolicy, ); } catch (error) { acquisition.release(); @@ -2321,7 +2302,6 @@ 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 = ""; @@ -2337,7 +2317,6 @@ 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; @@ -2348,7 +2327,6 @@ 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, ); @@ -2692,7 +2670,12 @@ class OpenCodeAgentSession implements AgentSession { return false; } - const completion = await this.waitForPersistedAssistantCompletion(turnId); + const completion = await readOpenCodePersistedAssistantCompletion( + this.storageRoot, + this.sessionId, + this.foregroundKnownMessageIds, + this.foregroundTurnStartedAt, + ); if (!completion || this.activeForegroundTurnId !== turnId) { return false; } @@ -2749,32 +2732,6 @@ 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;