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;