diff --git a/packages/server/src/server/agent/providers/opencode-agent.real.e2e.test.ts b/packages/server/src/server/agent/providers/opencode-agent.real.e2e.test.ts new file mode 100644 index 000000000..d6aecffec --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode-agent.real.e2e.test.ts @@ -0,0 +1,515 @@ +import { execFileSync, spawn, type ChildProcess } from "node:child_process"; +import { + copyFileSync, + createWriteStream, + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + readdirSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { createServer, request as httpRequest, type ServerResponse } from "node:http"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; + +import type { AgentStreamEvent } from "../agent-sdk-types.js"; +import { OpenCodeAgentClient } from "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; + +const MODEL = "openai/gpt-5.4"; +const NORMAL_SENTINEL = "PASEO_NORMAL_RECOVERY_SENTINEL_7F31"; +const ROOT_SENTINEL = "PASEO_ROOT_RECOVERY_SENTINEL_8C42"; +const CHILD_SENTINEL = "PASEO_CHILD_RECOVERY_SENTINEL_5A19"; +const PARENT_SENTINEL = "PASEO_PARENT_RECOVERY_SENTINEL_6B20"; +const TIMEOUT_MS = 240_000; + +describe.sequential("OpenCode real event recovery", () => { + let harness: Awaited>; + + beforeAll(async () => { + harness = await createRealHarness(); + }, 30_000); + + afterAll(async () => { + await harness?.close(); + }, 30_000); + + test("removes disposable runtime state when auth setup fails", () => { + const observed = { runtimeDir: "" }; + expect(() => createDisposableRuntime(failAuthCopy, recordRuntimeDirectory(observed))).toThrow( + "induced auth copy failure", + ); + expect(observed.runtimeDir).not.toBe(""); + expect(existsSync(observed.runtimeDir)).toBe(false); + }); + + test("removes disposable runtime state when the creation hook fails", () => { + const observed = { runtimeDir: "" }; + expect(() => createDisposableRuntime(copyFileSync, failRuntimeCreation(observed))).toThrow( + "induced runtime creation failure", + ); + expect(observed.runtimeDir).not.toBe(""); + expect(existsSync(observed.runtimeDir)).toBe(false); + }); + + test( + "normal foreground completes once", + async () => { + const result = await harness.runScenario( + "normal", + `Respond with exactly ${NORMAL_SENTINEL} and no other text.`, + ); + + assertParentResult(result.events, NORMAL_SENTINEL); + harness.writeSummary("normal", summarize(result)); + }, + TIMEOUT_MS, + ); + + test( + "recovers a root turn after cutting the active global event stream", + async () => { + const result = await harness.runScenario( + "root-recovery", + [ + "You must use the bash tool exactly once to run: sleep 4", + `After the tool completes, respond with exactly ${ROOT_SENTINEL} and no other text.`, + ].join("\n"), + (event) => isRunningToolEvent(event), + ); + + assertParentResult(result.events, ROOT_SENTINEL); + expect(result.activeCutTrigger).toBe(true); + expect(result.pidAfter).toBe(result.pidBefore); + expect(result.cuts).toBe(1); + expect(result.connectionsAfter - result.connectionsBefore).toBe(1); + harness.writeSummary("root-recovery", summarize(result)); + }, + TIMEOUT_MS, + ); + + test( + "recovers an active delegated child and its parent once", + async () => { + const result = await harness.runScenario( + "child-recovery", + [ + "You must delegate this work using the task tool with subagent type general.", + `Tell the child to use bash to run sleep 4, then return exactly ${CHILD_SENTINEL}.`, + `After the child finishes, respond with exactly ${PARENT_SENTINEL} and no other text.`, + ].join("\n"), + (event) => isRunningChildEvent(event), + ); + + assertParentResult(result.events, PARENT_SENTINEL, true); + expect(result.activeCutTrigger).toBe(true); + expect(result.pidAfter).toBe(result.pidBefore); + expect(result.cuts).toBe(1); + expect(result.connectionsAfter - result.connectionsBefore).toBe(1); + const childOutput = result.events + .flatMap((event) => { + if ( + event.type !== "provider_subagent" || + event.event.type !== "timeline" || + event.event.item.type !== "assistant_message" + ) { + return []; + } + return [event.event.item.text]; + }) + .join(""); + expect(childOutput.trim()).toBe(CHILD_SENTINEL); + expect(childOutput.split(CHILD_SENTINEL)).toHaveLength(2); + expect( + result.events.filter( + (event) => + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.status === "completed", + ), + ).toHaveLength(1); + expect(result.events.some((event) => event.type === "provider_subagent")).toBe(true); + harness.writeSummary("child-recovery", summarize(result)); + }, + TIMEOUT_MS, + ); +}); + +async function createRealHarness() { + const artifactDir = mkdtempSync(path.join(os.tmpdir(), "paseo-opencode-recovery-")); + const runtime = createDisposableRuntime(); + const { runtimeDir, home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace } = + runtime; + + let setupManager: OpenCodeServerManager | null = null; + let setupProxy: Awaited> | null = null; + try { + const upstreamPort = await reservePort(); + const proxy = await createCutProxy(upstreamPort, path.join(artifactDir, "proxy.jsonl")); + setupProxy = proxy; + const isolatedEnv = { + ...process.env, + HOME: home, + PASEO_HOME: paseoHome, + XDG_CONFIG_HOME: xdgConfig, + XDG_DATA_HOME: xdgData, + XDG_CACHE_HOME: xdgCache, + XDG_STATE_HOME: xdgState, + OPENCODE_DISABLE_AUTOUPDATE: "1", + OPENCODE_DISABLE_AUTO_UPDATE: "1", + }; + const openCodeVersion = execFileSync("opencode", ["--version"], { + env: isolatedEnv, + encoding: "utf8", + }).trim(); + expect(openCodeVersion).toBe("1.14.33"); + const selectedModels = new Set(); + let child: ChildProcess | null = null; + const manager = new OpenCodeServerManager({ + logger: pino({ level: "silent" }), + portAllocator: async () => proxy.port, + resolveCommandPrefix: async () => ({ command: "opencode", args: [] }), + resolveHomeDir: () => home, + spawnServerProcess: () => { + child = spawn("opencode", ["serve", "--port", String(upstreamPort)], { + cwd: workspace, + env: isolatedEnv, + stdio: ["ignore", "pipe", "pipe"], + }); + child.stdout?.pipe(createWriteStream(path.join(artifactDir, "opencode.stdout.log"))); + child.stderr?.pipe(createWriteStream(path.join(artifactDir, "opencode.stderr.log"))); + return child; + }, + }); + setupManager = manager; + const lease = await manager.acquireCurrent(); + const client = new OpenCodeAgentClient(pino({ level: "silent" }), undefined, { + serverManager: manager, + }); + + return { + artifactDir, + manager, + proxy, + writeSummary(name: string, summary: unknown) { + writeFileSync( + path.join(artifactDir, `${name}.summary.json`), + JSON.stringify(summary, null, 2), + ); + }, + async runScenario( + name: string, + prompt: string, + cutWhen?: (event: AgentStreamEvent) => boolean, + ) { + const sessionConfig = { + provider: "opencode", + cwd: workspace, + model: MODEL, + modeId: "build", + featureValues: { auto_accept: true }, + } as const; + selectedModels.add(sessionConfig.model); + const session = await client.createSession(sessionConfig); + const events: AgentStreamEvent[] = []; + const eventPath = path.join(artifactDir, `${name}.events.jsonl`); + const pidBefore = child?.pid; + const connectionsBefore = proxy.connections; + const cutsBefore = proxy.cuts; + let activeCutTrigger = false; + session.subscribe((event) => { + events.push(event); + writeFileSync(eventPath, `${JSON.stringify(event)}\n`, { flag: "a" }); + if (!activeCutTrigger && cutWhen?.(event)) { + activeCutTrigger = true; + proxy.cutActiveGlobalEvent(); + } + }); + try { + await session.startTurn(prompt); + await waitForTerminal(events, TIMEOUT_MS - 10_000); + return { + events, + activeCutTrigger, + pidBefore, + pidAfter: child?.pid, + cuts: proxy.cuts - cutsBefore, + connectionsBefore, + connectionsAfter: proxy.connections, + }; + } finally { + await session.close().catch(() => undefined); + } + }, + async close() { + try { + await lease.release().catch(() => undefined); + await manager.shutdown().catch(() => undefined); + await proxy.close(); + writeFileSync( + path.join(artifactDir, "run.summary.json"), + JSON.stringify( + { pid: child?.pid, cuts: proxy.cuts, connections: proxy.connections }, + null, + 2, + ), + ); + writeFileSync( + path.join(artifactDir, "run-metadata.summary.json"), + JSON.stringify( + { + openCodeVersion, + models: [...selectedModels], + pid: child?.pid, + scenarios: ["normal", "root-recovery", "child-recovery"], + }, + null, + 2, + ), + ); + } finally { + rmSync(runtimeDir, { recursive: true, force: true }); + } + const credentialScan = scanRetainedArtifacts(artifactDir); + writeFileSync( + path.join(artifactDir, "credential-scan.summary.json"), + JSON.stringify(credentialScan, null, 2), + ); + expect(credentialScan.credentialPatternFiles).toBe(0); + }, + }; + } catch (error) { + await setupManager?.shutdown().catch(() => undefined); + await setupProxy?.close().catch(() => undefined); + rmSync(runtimeDir, { recursive: true, force: true }); + throw error; + } +} + +function createDisposableRuntime( + copyAuth: typeof copyFileSync = copyFileSync, + onCreate?: (runtimeDir: string) => void, +) { + let runtimeDir: string | null = null; + try { + runtimeDir = mkdtempSync(path.join(os.tmpdir(), "paseo-opencode-runtime-")); + onCreate?.(runtimeDir); + const home = path.join(runtimeDir, "home"); + const paseoHome = path.join(runtimeDir, "paseo-home"); + const xdgConfig = path.join(runtimeDir, "xdg-config"); + const xdgData = path.join(runtimeDir, "xdg-data"); + const xdgCache = path.join(runtimeDir, "xdg-cache"); + const xdgState = path.join(runtimeDir, "xdg-state"); + const workspace = path.join(runtimeDir, "workspace"); + for (const directory of [home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace]) { + mkdirSync(directory, { recursive: true }); + } + const isolatedAuthDirectory = path.join(xdgData, "opencode"); + mkdirSync(isolatedAuthDirectory, { recursive: true }); + copyAuth( + path.join(os.homedir(), ".local", "share", "opencode", "auth.json"), + path.join(isolatedAuthDirectory, "auth.json"), + ); + return { runtimeDir, home, paseoHome, xdgConfig, xdgData, xdgCache, xdgState, workspace }; + } catch (error) { + if (runtimeDir) { + rmSync(runtimeDir, { recursive: true, force: true }); + } + throw error; + } +} + +function failAuthCopy(): never { + throw new Error("induced auth copy failure"); +} + +function recordRuntimeDirectory(observed: { runtimeDir: string }) { + return (runtimeDir: string) => { + observed.runtimeDir = runtimeDir; + }; +} + +function failRuntimeCreation(observed: { runtimeDir: string }) { + return (runtimeDir: string): never => { + observed.runtimeDir = runtimeDir; + throw new Error("induced runtime creation failure"); + }; +} + +function scanRetainedArtifacts(artifactDir: string) { + const credentialFilename = /(?:auth|credential|secret|token)/i; + const credentialContent = + /(?:authorization\s*[:=]\s*["']?(?:bearer|basic)|["'](?:access_token|refresh_token|api[_-]?key|client_secret)["']\s*:)/i; + const files = readdirSync(artifactDir, { withFileTypes: true }) + .filter((entry) => entry.isFile()) + .map((entry) => entry.name) + .filter( + (name) => + credentialFilename.test(name) || + credentialContent.test(readFileSync(path.join(artifactDir, name), "utf8")), + ) + .sort(); + return { credentialPatternFiles: files.length, files }; +} + +async function createCutProxy(upstreamPort: number, artifactPath: string) { + let connections = 0; + let cuts = 0; + let activeGlobal: { upstream: NodeJS.ReadableStream; downstream: ServerResponse } | null = null; + const record = (value: unknown) => + writeFileSync(artifactPath, `${JSON.stringify({ at: Date.now(), ...value })}\n`, { flag: "a" }); + const server = createServer((incoming, outgoing) => { + const globalEvent = incoming.url?.startsWith("/global/event") === true; + if (globalEvent) { + connections += 1; + record({ type: "global-connect", connection: connections }); + } + const upstream = httpRequest( + { + host: "127.0.0.1", + port: upstreamPort, + method: incoming.method, + path: incoming.url, + headers: incoming.headers, + }, + (response) => { + outgoing.writeHead(response.statusCode ?? 502, response.headers); + if (globalEvent) { + activeGlobal = { upstream: response, downstream: outgoing }; + let buffer = ""; + response.on("data", (chunk: Buffer) => { + buffer += chunk.toString(); + for (;;) { + const boundary = buffer.indexOf("\n\n"); + if (boundary < 0) break; + const raw = buffer.slice(0, boundary); + buffer = buffer.slice(boundary + 2); + record({ type: "global-record", connection: connections, raw }); + } + }); + } + response.pipe(outgoing); + response.on("close", () => { + if (activeGlobal?.upstream === response) activeGlobal = null; + }); + }, + ); + upstream.on("error", (error) => { + record({ type: "proxy-upstream-error", message: error.message }); + outgoing.destroy(); + }); + incoming.pipe(upstream); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Proxy did not bind"); + return { + port: address.port, + get connections() { + return connections; + }, + get cuts() { + return cuts; + }, + cutActiveGlobalEvent() { + if (!activeGlobal) throw new Error("No active global event connection to cut"); + cuts += 1; + record({ type: "global-cut", connection: connections, cut: cuts }); + activeGlobal.upstream.destroy(); + activeGlobal.downstream.destroy(); + activeGlobal = null; + }, + close: async () => new Promise((resolve) => server.close(() => resolve())), + }; +} + +function isRunningToolEvent(event: AgentStreamEvent): boolean { + return ( + event.type === "timeline" && + event.item.type === "tool_call" && + (event.item.status === "running" || event.item.status === "pending") + ); +} + +function isRunningChildEvent(event: AgentStreamEvent): boolean { + return ( + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.status === "running" + ); +} + +function assertParentResult( + events: AgentStreamEvent[], + sentinel: string, + assertFinalMessageOnly = false, +): void { + expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1); + expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); + expect(events.filter((event) => event.type === "turn_failed")).toHaveLength(0); + expect(events.filter((event) => event.type === "turn_canceled")).toHaveLength(0); + const messages = new Map(); + for (const event of events) { + if (event.type !== "timeline" || event.item.type !== "assistant_message") continue; + messages.set( + event.item.messageId, + `${messages.get(event.item.messageId) ?? ""}${event.item.text}`, + ); + } + const output = Array.from(messages.values()).join(""); + const assertedOutput = assertFinalMessageOnly ? Array.from(messages.values()).at(-1) : output; + expect(assertedOutput?.trim()).toBe(sentinel); + expect(output.split(sentinel)).toHaveLength(2); +} + +function summarize(result: { + events: AgentStreamEvent[]; + pidBefore?: number; + pidAfter?: number; + cuts: number; + connectionsBefore: number; + connectionsAfter: number; +}) { + return { + pidBefore: result.pidBefore, + pidAfter: result.pidAfter, + cuts: result.cuts, + connections: result.connectionsAfter - result.connectionsBefore, + started: result.events.filter((event) => event.type === "turn_started").length, + completed: result.events.filter((event) => event.type === "turn_completed").length, + failed: result.events.filter((event) => event.type === "turn_failed").length, + canceled: result.events.filter((event) => event.type === "turn_canceled").length, + }; +} + +async function waitForTerminal(events: AgentStreamEvent[], timeoutMs: number): Promise { + const started = Date.now(); + while (Date.now() - started < timeoutMs) { + if ( + events.some( + (event) => + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled", + ) + ) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`Timed out after ${timeoutMs}ms waiting for terminal event`); +} + +async function reservePort(): Promise { + const server = net.createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Port reservation failed"); + await new Promise((resolve) => server.close(() => resolve())); + return address.port; +} diff --git a/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts b/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts new file mode 100644 index 000000000..f31a03738 --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts @@ -0,0 +1,457 @@ +import type { ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { createOpencodeClient } from "@opencode-ai/sdk/v2/client"; +import { expect, test } from "vitest"; + +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import type { AgentStreamEvent } from "../agent-sdk-types.js"; +import { OpenCodeAgentClient } from "./opencode-agent.js"; +import { + OpenCodeEventConsumer, + type OpenCodeEventConsumerTiming, +} from "./opencode/event-consumer.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; + +test("dispatches ascending OpenCode message identifiers through the public provider path", async () => { + const upstream = await createRecoveryUpstream(); + const fixture = await createPublicRecoverySession(upstream, new RecoveryTiming()); + const observed: AgentStreamEvent[] = []; + fixture.session.subscribe((event) => observed.push(event)); + await upstream.connected(1); + upstream.send(0, connectedRecord()); + + await fixture.session.startTurn("first"); + await upstream.dispatched(1); + upstream.send(0, idleRecord()); + await eventually(() => + expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1), + ); + await fixture.session.startTurn("second"); + const ids = await upstream.dispatched(2); + + expect(ids).toHaveLength(2); + expect(ids[0]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); + expect(ids[1]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); + expect((ids[0] as string) < (ids[1] as string)).toBe(true); + + await fixture.session.close(); + await fixture.manager.shutdown(); + await upstream.close(); +}); + +test("recovers missed output after EOF without failing the active turn", async () => { + const upstream = await createRecoveryUpstream(); + const timing = new RecoveryTiming(); + const fixture = await createPublicRecoverySession(upstream, timing); + const { session } = fixture; + const observed: AgentStreamEvent[] = []; + session.subscribe((event) => observed.push(event)); + + await upstream.connected(1); + upstream.send(0, connectedRecord()); + await session.startTurn("hello"); + const [dispatchId] = await upstream.dispatched(); + expect(upstream.messageReads()).toBe(0); + upstream.setRecoveredMessages(dispatchId as string); + upstream.end(0); + await timing.waiting(); + timing.advance(); + await upstream.connected(2); + upstream.send(1, connectedRecord()); + + await eventually(() => expect(observed.map((event) => event.type)).toContain("turn_completed")); + expect(observed.filter((event) => event.type === "turn_started")).toHaveLength(1); + expect(observed.filter((event) => event.type === "turn_failed")).toHaveLength(0); + const assistantTexts = observed.flatMap((event) => + event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [], + ); + expect(assistantTexts).toHaveLength(52); + expect(assistantTexts.at(-1)).toBe("recovered output"); + expect(assistantTexts).not.toContain("before boundary"); + + await session.close(); + await fixture.manager.shutdown(); + await upstream.close(); +}); + +test("orders queued live deltas behind an incomplete recovery snapshot and suppresses late output", async () => { + const upstream = await createRecoveryUpstream(); + const timing = new RecoveryTiming(); + const fixture = await createPublicRecoverySession(upstream, timing); + const { session } = fixture; + const observed: AgentStreamEvent[] = []; + session.subscribe((event) => observed.push(event)); + + await upstream.connected(1); + upstream.send(0, connectedRecord()); + await session.startTurn("hello"); + const [dispatchId] = await upstream.dispatched(); + upstream.send(0, assistantMessageRecord("message-race")); + upstream.send(0, textDeltaRecord("message-race", "part-race", "Hello")); + await eventually(() => expect(assistantText(observed)).toEqual(["Hello"])); + + upstream.setStatus("busy"); + upstream.setMessages([ + { info: { id: dispatchId as string, sessionID: "session-1", role: "user" }, parts: [] }, + { + info: { id: "message-race", sessionID: "session-1", role: "assistant" }, + parts: [ + { + id: "part-race", + sessionID: "session-1", + messageID: "message-race", + type: "text", + text: "", + time: { start: 1 }, + }, + ], + }, + ]); + upstream.end(0); + await timing.waiting(); + timing.advance(); + await upstream.connected(2); + upstream.send(1, connectedRecord()); + upstream.send(1, textDeltaRecord("message-race", "part-race", " world")); + await eventually(() => expect(assistantText(observed)).toEqual(["Hello", " world"])); + + upstream.setMessages([ + { + info: { + id: "message-race", + sessionID: "session-1", + role: "assistant", + time: { created: 1, completed: 2 }, + }, + parts: [ + { + id: "part-race", + sessionID: "session-1", + messageID: "message-race", + type: "text", + text: "Hello world!", + time: { start: 1, end: 2 }, + }, + ], + }, + ]); + upstream.send(1, finalTextRecord("message-race", "part-race", "Hello world!")); + upstream.send(1, { + directory: "/workspace", + payload: { + type: "session.status", + properties: { sessionID: "session-1", status: { type: "idle" } }, + }, + }); + upstream.send(1, textDeltaRecord("message-race", "part-race", " too late")); + + await eventually(() => + expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1), + ); + expect(observed.filter((event) => event.type === "turn_started")).toHaveLength(1); + expect(observed.filter((event) => event.type === "turn_failed")).toHaveLength(0); + expect(observed.filter((event) => event.type === "turn_canceled")).toHaveLength(0); + expect(assistantText(observed)).toEqual(["Hello", " world", "!"]); + expect(assistantText(observed).join("")).toBe("Hello world!"); + const ordered = observed.flatMap((event) => { + if (event.type === "turn_started" || event.type === "turn_completed") return [event.type]; + if (event.type === "timeline" && event.item.type === "assistant_message") { + return [event.item.text]; + } + return []; + }); + expect(ordered).toEqual(["turn_started", "Hello", " world", "!", "turn_completed"]); + + await session.close(); + await fixture.manager.shutdown(); + await upstream.close(); +}); + +class RecoveryTiming implements OpenCodeEventConsumerTiming { + private resolveWait: (() => void) | null = null; + arm(): () => void { + return () => undefined; + } + wait(_delayMs: number, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + this.resolveWait = resolve; + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + } + async waiting(): Promise { + await eventually(() => expect(this.resolveWait).not.toBeNull()); + } + advance(): void { + this.resolveWait?.(); + this.resolveWait = null; + } +} + +async function createPublicRecoverySession( + upstream: Awaited>, + timing: OpenCodeEventConsumerTiming, +) { + let serverProcess: RecoveryServerProcess | null = null; + const manager = new OpenCodeServerManager({ + logger: createTestLogger(), + portAllocator: async () => upstream.port, + resolveCommandPrefix: async () => ({ command: "opencode", args: [] }), + resolveHomeDir: () => process.cwd(), + spawnServerProcess: () => { + serverProcess = new RecoveryServerProcess(); + return serverProcess as unknown as ChildProcess; + }, + terminateProcess: async () => { + serverProcess?.exit(); + return "terminated"; + }, + createEventSource: (options) => new OpenCodeEventConsumer({ ...options, timing }), + }); + const client = new OpenCodeAgentClient(createTestLogger(), undefined, { + serverManager: manager, + createClient: ({ baseUrl, directory }) => createOpencodeClient({ baseUrl, directory }), + }); + const session = await client.createSession({ provider: "opencode", cwd: "/workspace" }); + return { manager, session }; +} + +class RecoveryServerProcess extends EventEmitter { + readonly stdout = new EventEmitter(); + readonly stderr = new EventEmitter(); + readonly pid = 42_001; + killed = false; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + + constructor() { + super(); + queueMicrotask(() => this.stdout.emit("data", Buffer.from("listening on test server\n"))); + } + + exit(): void { + if (this.exitCode !== null) return; + this.exitCode = 0; + this.emit("exit", 0, null); + } +} + +async function createRecoveryUpstream() { + const streams: ServerResponse[] = []; + const dispatchIds: string[] = []; + let messages: unknown[] = []; + let messageReadCount = 0; + let status: "busy" | "idle" = "idle"; + const server = createServer(async (request, response) => { + const pathname = new URL(request.url ?? "/", "http://127.0.0.1").pathname; + if (request.url?.startsWith("/global/event")) { + response.writeHead(200, { "content-type": "text/event-stream" }); + response.flushHeaders(); + streams.push(response); + return; + } + if (request.method === "POST" && pathname === "/session") { + return json(response, { id: "session-1", directory: "/workspace" }); + } + if (request.url?.includes("/prompt_async")) { + const body = JSON.parse(await readBody(request)) as { messageID: string }; + dispatchIds.push(body.messageID); + return json(response, {}); + } + if (request.url?.includes("/message")) { + messageReadCount += 1; + return json(response, messages); + } + if (request.url?.includes("/status")) + return json(response, status === "busy" ? { "session-1": { type: "busy" } } : {}); + return json(response, {}); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test server address"); + return { + port: address.port, + url: `http://127.0.0.1:${address.port}`, + connected: async (count: number) => eventually(() => expect(streams).toHaveLength(count)), + send(index: number, value: unknown) { + streams[index]?.write(`data: ${JSON.stringify(value)}\n\n`); + }, + end(index: number) { + streams[index]?.end(); + }, + dispatched: async (count = 1) => { + await eventually(() => expect(dispatchIds).toHaveLength(count)); + return [...dispatchIds]; + }, + messageReads: () => messageReadCount, + setMessages(nextMessages: unknown[]) { + messages = nextMessages; + }, + setStatus(nextStatus: "busy" | "idle") { + status = nextStatus; + }, + setRecoveredMessages(messageId: string) { + messages = [ + { + info: { + id: "msg_before_boundary", + sessionID: "session-1", + role: "assistant", + time: { created: 1, completed: 2 }, + }, + parts: [ + { + id: "part-before", + sessionID: "session-1", + messageID: "msg_before_boundary", + type: "text", + text: "before boundary", + time: { start: 1, end: 2 }, + }, + ], + }, + { + info: { id: messageId, sessionID: "session-1", role: "user" }, + parts: [], + }, + ...Array.from({ length: 51 }, (_, index) => ({ + info: { + id: `msg_step_${index}`, + sessionID: "session-1", + role: "assistant", + time: { created: index + 3, completed: index + 4 }, + }, + parts: [ + { + id: `part-step-${index}`, + sessionID: "session-1", + messageID: `msg_step_${index}`, + type: "text", + text: `step ${index}`, + time: { start: index + 3, end: index + 4 }, + }, + ], + })), + { + info: { id: "msg_compaction", sessionID: "session-1", role: "user" }, + parts: [ + { + id: "part-compaction", + sessionID: "session-1", + messageID: "msg_compaction", + type: "compaction", + auto: true, + }, + ], + }, + { + info: { + id: "msg_assistant", + sessionID: "session-1", + role: "assistant", + time: { created: 1, completed: 2 }, + }, + parts: [ + { + id: "part-1", + sessionID: "session-1", + messageID: "msg_assistant", + type: "text", + text: "recovered output", + time: { start: 1, end: 2 }, + }, + ], + }, + ]; + }, + close: async () => new Promise((resolve) => server.close(() => resolve())), + }; +} + +function assistantMessageRecord(messageId: string) { + return { + directory: "/workspace", + payload: { + type: "message.updated", + properties: { info: { id: messageId, sessionID: "session-1", role: "assistant" } }, + }, + }; +} + +function textDeltaRecord(messageId: string, partId: string, delta: string) { + return { + directory: "/workspace", + payload: { + type: "message.part.delta", + properties: { + sessionID: "session-1", + messageID: messageId, + partID: partId, + field: "text", + delta, + }, + }, + }; +} + +function finalTextRecord(messageId: string, partId: string, text: string) { + return { + directory: "/workspace", + payload: { + type: "message.part.updated", + properties: { + part: { + id: partId, + sessionID: "session-1", + messageID: messageId, + type: "text", + text, + time: { start: 1, end: 2 }, + }, + }, + }, + }; +} + +function assistantText(events: AgentStreamEvent[]): string[] { + return events.flatMap((event) => + event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [], + ); +} + +function connectedRecord() { + return { directory: "/workspace", payload: { type: "server.connected", properties: {} } }; +} + +function idleRecord() { + return { + directory: "/workspace", + payload: { + type: "session.status", + properties: { sessionID: "session-1", status: { type: "idle" } }, + }, + }; +} + +function json(response: ServerResponse, body: unknown): void { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify(body)); +} + +async function readBody(request: IncomingMessage): Promise { + let body = ""; + for await (const chunk of request) body += chunk; + return body; +} + +async function eventually(assertion: () => void): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + try { + assertion(); + return; + } catch { + await new Promise((resolve) => setImmediate(resolve)); + } + } + assertion(); +} diff --git a/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts b/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts index 11465fe83..3dd2d45cc 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts @@ -8,20 +8,6 @@ import { TestOpenCodeHarness, } from "./opencode/test-utils/test-opencode-harness.js"; -function createDeferred(): { - promise: Promise; - resolve: (value: T) => void; - reject: (error: unknown) => void; -} { - let resolve!: (value: T) => void; - let reject!: (error: unknown) => void; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - return { promise, resolve, reject }; -} - describe("OpenCodeAgentSession slash command timeout handling", () => { test("lists only OpenCode built-in slash commands Paseo can execute", async () => { const runtime = new TestOpenCodeHarness(); @@ -75,20 +61,12 @@ describe("OpenCodeAgentSession slash command timeout handling", () => { }); test("waits for SSE completion when slash commands hit a header timeout", async () => { - const idleEventGate = createDeferred(); const runtime = new TestOpenCodeHarness(); const openCodeClient = createOpenCodeClientWithConnectedProvider(); openCodeClient.sessionCommandError = new Error("fetch failed: Headers Timeout Error"); openCodeClient.commandListResponse = { data: [{ name: "help", description: "Show help", hints: [] }], }; - openCodeClient.eventStream = (async function* () { - await idleEventGate.promise; - yield { - type: "session.idle", - properties: { sessionID: "session-1" }, - }; - })(); runtime.enqueueClient(openCodeClient); const client = new OpenCodeAgentClient(createTestLogger(), undefined, { @@ -98,8 +76,17 @@ describe("OpenCodeAgentSession slash command timeout handling", () => { const session = await client.createSession({ provider: "opencode", cwd: "/tmp" }); const runPromise = session.run("/help"); - await Promise.resolve(); - idleEventGate.resolve(); + await nextTick(); + expect(openCodeClient.calls.sessionCommand).toHaveLength(1); + let settled = false; + void runPromise.then(() => { + settled = true; + return undefined; + }); + await nextTick(); + expect(settled).toBe(false); + + openCodeClient.emitEvent(idleEvent()); await expect(runPromise).resolves.toMatchObject({ sessionId: "session-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 3931fe035..60ce52a0f 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -5,6 +5,8 @@ import path from "node:path"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import type { Event as OpenCodeEvent } from "@opencode-ai/sdk/v2/client"; +import type { OpencodeClient } from "@opencode-ai/sdk/v2/client"; +import type { OpenCodeEventSource } from "./opencode/event-consumer.js"; import { __openCodeInternals, OpenCodeAgentClient, @@ -33,8 +35,61 @@ function tmpCwd(): string { } } +function countEvents(events: AgentStreamEvent[], type: AgentStreamEvent["type"]): number { + return events.filter((event) => event.type === type).length; +} + +function countProviderSubagentUpserts(events: AgentStreamEvent[]): number { + return events.filter( + (event) => event.type === "provider_subagent" && event.event.type === "upsert", + ).length; +} + +function pendingPermissionIds(session: { + getPendingPermissions(): Array<{ id: string }>; +}): string[] { + return session + .getPendingPermissions() + .map((request) => request.id) + .sort(); +} + +function countChildStatuses( + events: AgentStreamEvent[], + status?: "running" | "completed" | "failed", + id?: string, +): number { + return events.filter( + (event) => + event.type === "provider_subagent" && + event.event.type === "upsert" && + (status === undefined || event.event.status === status) && + (id === undefined || event.event.id === id), + ).length; +} + const TEST_MODEL = "opencode/big-pickle"; +function createDirectEventSource(client: OpencodeClient): OpenCodeEventSource { + const listeners = new Set<(input: never) => void>(); + const abort = new AbortController(); + void client.global + .event({ signal: abort.signal, sseMaxRetryAttempts: 0 }) + .then(async ({ stream }) => { + for await (const event of stream) { + for (const listener of listeners) listener(event as never); + } + return undefined; + }); + return { + ready: async () => undefined, + subscribe(listener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }; +} + interface TurnResult { events: AgentStreamEvent[]; assistantMessages: AssistantMessageTimelineItem[]; @@ -1058,7 +1113,7 @@ describe("OpenCode adapter normalization", () => { sessionId: "session-1", messageRoles: new Map(), accumulatedUsage: usage, - streamedPartKeys: new Set(), + materializedParts: new Map(), emittedStructuredMessageIds: new Set(), partTypes: new Map(), modelContextWindowsByModelKey: new Map([["openai/gpt-5", 400_000]]), @@ -1293,6 +1348,8 @@ describe("OpenCode adapter startTurn error handling", () => { fakeClient, "ses_unit_test", createTestLogger(), + new Map(), + createDirectEventSource(fakeClient), ); const turn = await collectTurnEvents( @@ -1401,6 +1458,8 @@ describe("OpenCode adapter startTurn error handling", () => { fakeClient, "ses_unit_test", createTestLogger(), + new Map(), + createDirectEventSource(fakeClient), ); const turn = await collectTurnEvents(streamSession(session, "hello")); @@ -1483,6 +1542,8 @@ describe("OpenCode adapter startTurn error handling", () => { fakeClient, "ses_unit_test", createTestLogger(), + new Map(), + createDirectEventSource(fakeClient), ); const events: AgentStreamEvent[] = []; @@ -1524,6 +1585,7 @@ describe("OpenCode adapter startTurn error handling", () => { createTestLogger(), new Map(), undefined, + undefined, false, ); @@ -1556,26 +1618,13 @@ describe("OpenCode adapter startTurn error handling", () => { expect(fakeClient.session.delete).not.toHaveBeenCalled(); }); - test("waits for the OpenCode event stream to finish after close aborts it", async () => { - const streamAborted = createTestDeferred(); - const finishStreamCleanup = createTestDeferred(); + test("unsubscribes from the shared event source without closing it", async () => { + const unsubscribe = vi.fn(); + const events = { + ready: async () => undefined, + subscribe: vi.fn(() => unsubscribe), + }; const fakeClient = { - global: { - event: vi.fn().mockImplementation(async ({ signal }: { signal: AbortSignal }) => ({ - stream: { - [Symbol.asyncIterator]: () => ({ - next: async () => { - if (!signal.aborted) { - await waitForAbort(signal); - } - streamAborted.resolve(); - await finishStreamCleanup.promise; - return { done: true, value: undefined }; - }, - }), - }, - })), - }, session: { abort: vi.fn().mockResolvedValue({ error: null }), update: vi.fn().mockResolvedValue({ error: null }), @@ -1586,20 +1635,13 @@ describe("OpenCode adapter startTurn error handling", () => { fakeClient, "ses_unit_test", createTestLogger(), + new Map(), + events, ); - let closeSettled = false; + await session.close(); - const closePromise = session.close().then(() => { - closeSettled = true; - return undefined; - }); - await streamAborted.promise; - await new Promise((resolve) => setImmediate(resolve)); - - expect(closeSettled).toBe(false); - - finishStreamCleanup.resolve(); - await closePromise; + expect(events.subscribe).toHaveBeenCalledOnce(); + expect(unsubscribe).toHaveBeenCalledOnce(); }); test("streamHistory preserves OpenCode replay timestamps from message and part times", async () => { @@ -2651,6 +2693,9 @@ describe("OpenCode adapter startTurn error handling", () => { vi.useFakeTimers(); const { parent: session, openCode } = await createParentSession("ses_unit_test"); openCode.sessionPromptAsyncEvents = []; + openCode.sessionStatusResponse = { + data: { ses_unit_test: { type: "busy" } }, + }; try { await session.startTurn("first"); @@ -2668,37 +2713,698 @@ describe("OpenCode adapter startTurn error handling", () => { } }); - test("keeps waiting for the stop terminal when the reconnect status probe fails", async () => { - const firstStreamEnd = createTestDeferred(); + test("settles summarize from provider status after reconnect", async () => { + const { parent: session, openCode } = await createParentSession("ses_summarize_recovery"); + openCode.sessionSummarizeEvents = []; + openCode.sessionStatusResponse = { data: {} }; + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + try { + await session.startTurn("/summarize"); + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(countEvents(events, "turn_completed")).toBe(1)); + + expect(openCode.calls.sessionMessages).toHaveLength(0); + expect(events.filter((event) => event.type === "timeline")).toHaveLength(0); + } finally { + await session.close(); + } + }); + + test.each(["status", "messages"] as const)( + "retries a failed foreground %s snapshot after one reconnect", + async (failedRequest) => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession( + `ses_retry_${failedRequest}_snapshot`, + ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.sessionPromptAsyncEvents = []; + let statusAttempts = 0; + let messageAttempts = 0; + + try { + await vi.waitFor(() => expect(openCode.calls.sessionChildren).toHaveLength(1)); + await session.startTurn("recover me"); + const dispatch = openCode.calls.sessionPromptAsync[0] as { messageID: string }; + const recoveredMessages = { + data: [ + { + info: { + id: dispatch.messageID, + sessionID: session.id, + role: "user", + }, + parts: [], + }, + { + info: { + id: "msg_recovered_assistant", + sessionID: session.id, + role: "assistant", + time: { created: 1, completed: 2 }, + }, + parts: [ + { + id: "part_recovered_text", + sessionID: session.id, + messageID: "msg_recovered_assistant", + type: "text", + text: "recovered after retry", + time: { start: 1, end: 2 }, + }, + ], + }, + ], + }; + openCode.sessionStatusImplementation = async () => { + statusAttempts += 1; + if (failedRequest === "status" && statusAttempts === 1) { + throw new Error("transient status failure"); + } + return { data: {} }; + }; + openCode.sessionMessagesImplementation = async () => { + messageAttempts += 1; + if (failedRequest === "messages" && messageAttempts === 1) { + throw new Error("transient messages failure"); + } + return recoveredMessages; + }; + + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => + expect(failedRequest === "status" ? statusAttempts : messageAttempts).toBe(1), + ); + await vi.advanceTimersByTimeAsync(100); + await vi.waitFor(() => expect(countEvents(events, "turn_completed")).toBe(1)); + + expect(events).toContainEqual( + expect.objectContaining({ + type: "timeline", + item: expect.objectContaining({ + type: "assistant_message", + text: "recovered after retry", + }), + }), + ); + expect(countEvents(events, "turn_failed")).toBe(0); + expect(failedRequest === "status" ? statusAttempts : messageAttempts).toBe(2); + expect(openCode.calls.sessionChildren).toHaveLength(2); + } finally { + vi.useRealTimers(); + await session.close(); + } + }, + ); + + test("does not poll a successful snapshot without the active dispatch", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession("ses_missing_dispatch"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionStatusResponse = { data: {} }; + openCode.sessionMessagesResponse = { data: [] }; + + try { + await session.startTurn("missing dispatch"); + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(countEvents(events, "turn_failed")).toBe(1)); + await vi.advanceTimersByTimeAsync(5_000); + + expect(openCode.calls.sessionStatus).toHaveLength(1); + expect(openCode.calls.sessionMessages).toHaveLength(1); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("lets live terminal events overtake a failed snapshot retry", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession("ses_live_overtakes_retry"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionStatusImplementation = async () => { + throw new Error("snapshot unavailable"); + }; + + try { + await session.startTurn("keep live ingress moving"); + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(openCode.calls.sessionStatus).toHaveLength(1)); + openCode.emitEvent({ type: "session.idle", properties: { sessionID: session.id } }); + await vi.waitFor(() => expect(countEvents(events, "turn_completed")).toBe(1)); + await vi.advanceTimersByTimeAsync(100); + + expect(openCode.calls.sessionStatus).toHaveLength(1); + expect(countEvents(events, "turn_failed")).toBe(0); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("does not apply a failed snapshot retry to the next turn", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession("ses_stale_retry"); + const events: AgentStreamEvent[] = []; + const firstCompletion = createTestDeferred(); + session.subscribe((event) => { + events.push(event); + if (event.type === "turn_completed") firstCompletion.resolve(); + }); + openCode.sessionPromptAsyncEvents = []; + let statusAttempts = 0; + const firstStatus = createTestDeferred(); + openCode.sessionStatusImplementation = async () => { + statusAttempts += 1; + if (statusAttempts === 1) { + firstStatus.resolve(); + throw new Error("snapshot unavailable"); + } + return { data: {} }; + }; + openCode.sessionMessagesResponse = { data: [] }; + + try { + await session.startTurn("first turn"); + openCode.emitEvent({ type: "reconnected" }); + await firstStatus.promise; + openCode.emitEvent({ type: "session.idle", properties: { sessionID: session.id } }); + await firstCompletion.promise; + + await session.startTurn("second turn"); + await vi.advanceTimersByTimeAsync(100); + + expect(statusAttempts).toBe(1); + expect(countEvents(events, "turn_started")).toBe(2); + expect(countEvents(events, "turn_failed")).toBe(0); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("abandons a repair when its turn fails during a snapshot read", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession("ses_replaced_repair"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.sessionPromptAsyncEvents = []; + const firstDispatch = createTestDeferred<{ data: Record }>(); + let dispatchAttempts = 0; + openCode.sessionPromptAsyncImplementation = async () => { + dispatchAttempts += 1; + return dispatchAttempts === 1 ? firstDispatch.promise : { data: {} }; + }; + const retryStatus = createTestDeferred<{ data: Record }>(); + const retryStarted = createTestDeferred(); + let statusAttempts = 0; + openCode.sessionStatusImplementation = async () => { + statusAttempts += 1; + if (statusAttempts === 1) throw new Error("snapshot unavailable"); + retryStarted.resolve(); + return retryStatus.promise; + }; + openCode.sessionMessagesResponse = { data: [] }; + + try { + await session.startTurn("first turn"); + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(statusAttempts).toBe(1)); + await vi.advanceTimersByTimeAsync(100); + await retryStarted.promise; + + firstDispatch.reject(new Error("first dispatch failed")); + await vi.waitFor(() => expect(countEvents(events, "turn_failed")).toBe(1)); + await session.startTurn("second turn"); + retryStatus.resolve({ data: {} }); + await vi.advanceTimersByTimeAsync(0); + + expect(openCode.calls.sessionMessages).toHaveLength(0); + expect(countEvents(events, "turn_started")).toBe(2); + expect(countEvents(events, "turn_failed")).toBe(1); + } finally { + firstDispatch.resolve({ data: {} }); + retryStatus.resolve({ data: {} }); + vi.useRealTimers(); + await session.close(); + } + }); + + test("preserves failed blocking-request snapshots and resolves successful empty ones", async () => { + const { parent: session, openCode } = await createParentSession("ses_request_recovery"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.emitEvent({ + type: "permission.asked", + properties: { + id: "permission-1", + sessionID: "ses_request_recovery", + permission: "bash", + patterns: ["npm test"], + metadata: { command: "npm test", cwd: "/workspace/repo" }, + }, + }); + await vi.waitFor(() => expect(session.getPendingPermissions()).toHaveLength(1)); + + openCode.permissionListResponse = { error: new Error("snapshot failed") }; + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(openCode.calls.permissionList).toHaveLength(1)); + expect(session.getPendingPermissions()).toHaveLength(1); + + openCode.permissionListResponse = { data: [] }; + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(session.getPendingPermissions()).toHaveLength(0)); + expect(events).toContainEqual( + expect.objectContaining({ type: "permission_resolved", requestId: "permission-1" }), + ); + await session.close(); + }); + + test("recovers child status and directory-scoped blocking requests", async () => { + const cases = [ + { name: "same active", directory: "/workspace/repo", status: "running", busy: true }, + { name: "same completed", directory: "/workspace/repo", status: "completed", busy: false }, + { name: "cross active", directory: "/workspace/child", status: "running", busy: true }, + { + name: "cross failed", + directory: "/workspace/child", + status: "failed", + busy: false, + failed: true, + }, + ] as const; + + for (const recoveryCase of cases) { + const suffix = recoveryCase.name.replaceAll(" ", "_"); + const parentId = `ses_parent_${suffix}`; + const childId = `ses_child_${suffix}`; + const permissionId = `permission_${suffix}`; + const questionId = `question_${suffix}`; + const { parent, openCode } = await createParentSession(parentId); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + await vi.waitFor(() => expect(openCode.calls.sessionChildren).toHaveLength(1)); + + openCode.sessionChildrenResponses = [ + { + data: [ + { + id: childId, + parentID: parentId, + title: recoveryCase.name, + directory: recoveryCase.directory, + }, + ], + }, + { data: [] }, + ]; + openCode.sessionStatusImplementation = async (parameters) => ({ + data: + recoveryCase.busy && + (parameters as { directory?: string }).directory === recoveryCase.directory + ? { [childId]: { type: "busy" } } + : {}, + }); + openCode.sessionMessagesResponse = recoveryCase.failed + ? { + data: [ + { + info: { + id: `message_${suffix}`, + sessionID: childId, + role: "assistant", + error: { name: "ProviderError", data: { message: "child failed" } }, + }, + parts: [], + }, + ], + } + : { data: [] }; + openCode.permissionListResponse = { + data: [ + { + id: permissionId, + sessionID: childId, + permission: "bash", + patterns: ["npm test"], + metadata: { command: "npm test" }, + }, + ], + }; + openCode.questionListResponse = { + data: [ + { + id: questionId, + sessionID: childId, + questions: [ + { + question: "Continue?", + header: "Continue", + options: [{ label: "Yes", description: "Continue" }], + }, + ], + }, + ], + }; + + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => + expect(countChildStatuses(events, recoveryCase.status, childId)).toBeGreaterThanOrEqual(1), + ); + await vi.waitFor(() => + expect(pendingPermissionIds(parent)).toEqual([permissionId, questionId].sort()), + ); + + await parent.respondToPermission(permissionId, { behavior: "allow" }); + await parent.respondToPermission(questionId, { + behavior: "allow", + updatedInput: { answers: { Continue: "Yes" } }, + }); + expect(openCode.calls.permissionReply).toContainEqual( + expect.objectContaining({ requestID: permissionId, directory: recoveryCase.directory }), + ); + expect(openCode.calls.questionReply).toContainEqual( + expect.objectContaining({ requestID: questionId, directory: recoveryCase.directory }), + ); + + openCode.permissionListResponse = { error: new Error("permission snapshot failed") }; + openCode.questionListResponse = { error: new Error("question snapshot failed") }; + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(openCode.calls.permissionList.length).toBeGreaterThan(1)); + expect(parent.getPendingPermissions()).toEqual([]); + + openCode.emitEvent({ + type: "permission.asked", + properties: { + id: permissionId, + sessionID: childId, + permission: "bash", + patterns: ["npm test"], + metadata: { command: "npm test" }, + }, + }); + openCode.emitEvent({ + type: "question.asked", + properties: { + id: questionId, + sessionID: childId, + questions: [{ question: "Continue?", header: "Continue", options: [] }], + }, + }); + await vi.waitFor(() => expect(parent.getPendingPermissions()).toHaveLength(2)); + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(openCode.calls.permissionList.length).toBeGreaterThan(2)); + expect(parent.getPendingPermissions()).toHaveLength(2); + + openCode.permissionListResponse = { data: [] }; + openCode.questionListResponse = { data: [] }; + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(parent.getPendingPermissions()).toHaveLength(0)); + await parent.close(); + } + }); + + test("preserves a failed child terminal when its status snapshot fails", async () => { + const parentId = "ses_parent_failed_child_status"; + const childId = "ses_child_failed_status"; + const { parent, openCode } = await createParentSession(parentId); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + await vi.waitFor(() => expect(openCode.calls.sessionChildren).toHaveLength(1)); + openCode.sessionChildrenResponses = [ + { + data: [ + { + id: childId, + parentID: parentId, + title: "failed child", + directory: "/workspace/child", + }, + ], + }, + { data: [] }, + ]; + openCode.sessionStatusImplementation = async () => { + throw new Error("child status unavailable"); + }; + openCode.sessionMessagesResponse = { + data: [ + { + info: { + id: "msg_failed_child", + sessionID: childId, + role: "assistant", + time: { created: 1, completed: 2 }, + error: { name: "ProviderError", data: { message: "child failed" } }, + }, + parts: [], + }, + ], + }; + + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(countChildStatuses(events, "failed", childId)).toBe(1)); + const recoveredStatuses = events.flatMap((event) => + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.id === childId && + event.event.status + ? [event.event.status] + : [], + ); + expect(recoveredStatuses.at(-1)).toBe("failed"); + await parent.close(); + }); + + test("preserves the last child status when both recovery snapshots fail", async () => { + const parentId = "ses_parent_unknown_child_recovery"; + const childId = "ses_child_unknown_recovery"; + const { parent, openCode } = await createParentSession(parentId); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + await vi.waitFor(() => expect(openCode.calls.sessionChildren).toHaveLength(1)); + + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: childId, + parentID: parentId, + title: "completed child", + directory: "/workspace/child", + }, + }, + }); + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: childId }, + }); + await vi.waitFor(() => expect(countChildStatuses(events, "completed", childId)).toBe(1)); + + openCode.sessionChildrenResponses = [ + { + data: [ + { + id: childId, + parentID: parentId, + title: "completed child", + directory: "/workspace/child", + }, + ], + }, + { data: [] }, + ]; + openCode.sessionStatusImplementation = async () => { + throw new Error("child status unavailable"); + }; + openCode.sessionMessagesImplementation = async () => { + throw new Error("child messages unavailable"); + }; + + openCode.emitEvent({ type: "reconnected" }); + await vi.waitFor(() => expect(openCode.calls.sessionMessages).toHaveLength(1)); + const recoveredStatuses = events.flatMap((event) => + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.id === childId && + event.event.status + ? [event.event.status] + : [], + ); + expect(recoveredStatuses).toEqual(["running", "completed"]); + await parent.close(); + }); + + test("settles stopping state when the OpenCode process exits", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession("ses_exit_while_stopping"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.sessionPromptAsyncEvents = []; + + try { + await session.startTurn("first"); + await session.interrupt(); + openCode.emitEvent({ type: "server-exited", error: new Error("OpenCode exited") }); + await vi.advanceTimersByTimeAsync(0); + + openCode.sessionPromptAsyncEvents = [ + { type: "session.idle", properties: { sessionID: session.id } }, + ]; + const replacement = session.startTurn("second"); + await vi.advanceTimersByTimeAsync(10_000); + await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" }); + expect(countEvents(events, "turn_canceled")).toBe(1); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("continues ordered ingress after one event callback rejects", async () => { + const { parent: session, openCode } = await createParentSession("ses_ingress_recovery"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + openCode.emitEvent({ type: "message.part.updated" }); + await new Promise((resolve) => setImmediate(resolve)); + openCode.emitEvent({ + type: "permission.asked", + properties: { + id: "permission-after-failure", + sessionID: "ses_ingress_recovery", + permission: "bash", + patterns: ["npm test"], + metadata: { command: "npm test", cwd: "/workspace/repo" }, + }, + }); + + await vi.waitFor(() => + expect(session.getPendingPermissions()).toEqual([ + expect.objectContaining({ id: "permission-after-failure" }), + ]), + ); + expect(events).toContainEqual( + expect.objectContaining({ + type: "permission_requested", + request: expect.objectContaining({ id: "permission-after-failure" }), + }), + ); + await session.close(); + }); + + test("bounds dispatch readiness at ten seconds without sending a prompt", async () => { + vi.useFakeTimers(); + const openCode = new TestOpenCodeClient(); + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/workspace/repo" }, + openCode.asSdkClient(), + "ses_readiness_timeout", + createTestLogger(), + new Map(), + { + ready: () => new Promise(() => undefined), + subscribe: () => () => undefined, + }, + ); + try { + const dispatch = session.startTurn("wait for transport"); + const rejection = expect(dispatch).rejects.toThrow("OpenCode event stream first record"); + await vi.advanceTimersByTimeAsync(9_999); + expect(openCode.calls.sessionPromptAsync).toHaveLength(0); + await vi.advanceTimersByTimeAsync(1); + await rejection; + expect(openCode.calls.sessionPromptAsync).toHaveLength(0); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + + test("close aborts every in-flight reconciliation request class", async () => { + const cases = [ + "child discovery", + "child status", + "child messages", + "permission snapshot", + "question snapshot", + "root status", + "root messages", + ] as const; + for (const requestClass of cases) { + const { parent: session, openCode } = await createParentSession( + `ses_close_${requestClass.replaceAll(" ", "_")}`, + ); + const blocked = createTestDeferred(); + const waitForRequestAbort = async (_parameters: unknown, options: unknown) => { + blocked.resolve(); + const signal = (options as { signal: AbortSignal }).signal; + await new Promise((_, reject) => { + if (signal.aborted) reject(signal.reason); + else signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + return { data: [] }; + }; + const child = { + id: "ses_child", + parentID: session.id, + directory: "/workspace/repo", + title: "child", + }; + if (requestClass === "child discovery") + openCode.sessionChildrenImplementation = waitForRequestAbort; + if (requestClass === "child status") { + openCode.sessionChildrenResponses = [{ data: [child] }, { data: [] }]; + openCode.sessionStatusImplementation = waitForRequestAbort; + } + if (requestClass === "child messages") { + openCode.sessionChildrenResponses = [{ data: [child] }, { data: [] }]; + openCode.sessionMessagesImplementation = waitForRequestAbort; + } + if (requestClass === "permission snapshot") + openCode.permissionListImplementation = waitForRequestAbort; + if (requestClass === "question snapshot") + openCode.questionListImplementation = waitForRequestAbort; + if (requestClass === "root status" || requestClass === "root messages") { + openCode.sessionPromptAsyncEvents = []; + if (requestClass === "root status") + openCode.sessionStatusImplementation = waitForRequestAbort; + else { + openCode.sessionStatusResponse = { data: { [session.id ?? ""]: { type: "busy" } } }; + openCode.sessionMessagesImplementation = waitForRequestAbort; + } + await session.startTurn("keep running"); + } + + openCode.emitEvent({ type: "reconnected" }); + await Promise.race([ + blocked.promise, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`request did not start: ${requestClass}`)), 500), + ), + ]); + await expect(session.close()).resolves.toBeUndefined(); + } + }); + + test("keeps probing provider status while stopping across a permanent disconnect", async () => { const settleAbort = createTestDeferred(); - const releaseTerminal = createTestDeferred(); - let subscriptionCount = 0; + const releaseIdle = createTestDeferred(); + let statusAttempt = 0; const { parent: session, openCode } = await createParentSession( "ses_stop_status_probe_failure", (client) => { - client.globalEventImplementation = async (options) => { - subscriptionCount += 1; - const signal = (options as { signal: AbortSignal }).signal; - return { - stream: { - async *[Symbol.asyncIterator]() { - yield { type: "server.connected", properties: {} }; - if (subscriptionCount === 1) { - await firstStreamEnd.promise; - return; - } - await releaseTerminal.promise; - yield { - type: "session.idle", - properties: { sessionID: "ses_stop_status_probe_failure" }, - }; - await waitForAbort(signal); - }, - }, - }; - }; client.sessionStatusImplementation = async () => { - throw new Error("provider status unavailable"); + statusAttempt += 1; + if (statusAttempt === 1) throw new Error("provider status unavailable"); + await releaseIdle.promise; + return { data: {} }; }; }, ); @@ -2712,27 +3418,58 @@ describe("OpenCode adapter startTurn error handling", () => { await session.startTurn("first"); const interrupt = session.interrupt(); const replacement = session.startTurn("second"); - firstStreamEnd.resolve(); await vi.waitFor(() => expect(openCode.calls.sessionStatus).toHaveLength(1)); settleAbort.resolve(); await interrupt; await new Promise((resolve) => setImmediate(resolve)); - expect(openCode.calls.globalEvent).toHaveLength(2); expect(openCode.calls.sessionStatus).toHaveLength(1); - releaseTerminal.resolve(); + releaseIdle.resolve(); await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" }); expect(openCode.calls.sessionPromptAsync).toHaveLength(2); - expect(openCode.calls.globalEvent).toHaveLength(2); + expect(openCode.calls.sessionStatus.length).toBeGreaterThanOrEqual(2); } finally { settleAbort.resolve(); - releaseTerminal.resolve(); + releaseIdle.resolve(); await session.close(); } }, 15_000); + test("backs off stop status observation within the ten second bound", async () => { + vi.useFakeTimers(); + const { parent: session, openCode } = await createParentSession( + "ses_stop_backoff", + (client) => { + client.sessionStatusImplementation = async () => { + throw new Error("transport unavailable"); + }; + }, + ); + openCode.sessionPromptAsyncEvents = []; + try { + await session.startTurn("first"); + await session.interrupt(); + const replacement = session.startTurn("second"); + const rejection = expect(replacement).rejects.toThrow("OpenCode previous turn to stop"); + await vi.advanceTimersByTimeAsync(99); + expect(openCode.calls.sessionStatus).toHaveLength(0); + await vi.advanceTimersByTimeAsync(1); + expect(openCode.calls.sessionStatus).toHaveLength(1); + await vi.advanceTimersByTimeAsync(199); + expect(openCode.calls.sessionStatus).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1); + expect(openCode.calls.sessionStatus).toHaveLength(2); + await vi.advanceTimersByTimeAsync(9_700); + await rejection; + expect(openCode.calls.sessionStatus.length).toBeLessThanOrEqual(13); + } finally { + vi.useRealTimers(); + await session.close(); + } + }); + test("does not reconnect the stop observer while closing", async () => { const firstStreamEnd = createTestDeferred(); const streamSignals: AbortSignal[] = []; @@ -2770,7 +3507,7 @@ describe("OpenCode adapter startTurn error handling", () => { await session.close(); await expect(replacement).rejects.toThrow("OpenCode session is closed"); - expect(openCode.calls.globalEvent).toHaveLength(1); + expect(openCode.calls.globalEvent).toHaveLength(0); } finally { for (const signal of streamSignals) { if (!signal.aborted) { @@ -4063,6 +4800,130 @@ describe("OpenCode provider subagent contract", () => { await parent.close(); }); + test("emits the same child status when descriptor semantics change", async () => { + const { parent, openCode } = await createParentSession("ses_parent_child_changed_status"); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_child_changed_status", + parentID: "ses_parent_child_changed_status", + title: "Inspect recovery", + agent: "general", + model: { providerID: "openai", id: "gpt-5.4", variant: "high" }, + directory: "/workspace/repo", + }, + }, + }); + await vi.waitFor(() => expect(countProviderSubagentUpserts(events)).toBe(1)); + + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_child_changed_status", status: { type: "busy" } }, + }); + await vi.waitFor(() => expect(countChildStatuses(events, "running")).toBe(2)); + expect( + events.filter( + (event) => + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.status === "running", + ), + ).toEqual([ + expect.objectContaining({ + event: expect.objectContaining({ + title: "general", + description: "Inspect recovery", + subtitle: expect.stringContaining("gpt-5.4"), + cwd: "/workspace/repo", + }), + }), + expect.objectContaining({ + event: { type: "upsert", id: "ses_child_changed_status", status: "running" }, + }), + ]); + await parent.close(); + }); + + test("suppresses an exact duplicate status-bearing child upsert", async () => { + const { parent, openCode } = await createParentSession("ses_parent_child_exact_status"); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + openCode.emitEvent({ + type: "session.created", + properties: { + info: { id: "ses_child_exact_status", parentID: "ses_parent_child_exact_status" }, + }, + }); + await vi.waitFor(() => expect(countChildStatuses(events, "running")).toBe(1)); + for (let count = 0; count < 2; count += 1) { + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_child_exact_status", status: { type: "busy" } }, + }); + } + await new Promise((resolve) => setImmediate(resolve)); + expect(countChildStatuses(events, "running")).toBe(1); + await parent.close(); + }); + + test("emits each child status transition", async () => { + const { parent, openCode } = await createParentSession("ses_parent_child_status_transitions"); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_child_status_transitions", + parentID: "ses_parent_child_status_transitions", + }, + }, + }); + await vi.waitFor(() => expect(countChildStatuses(events, "running")).toBe(1)); + for (let count = 0; count < 2; count += 1) { + openCode.emitEvent({ + type: "session.status", + properties: { sessionID: "ses_child_status_transitions", status: { type: "idle" } }, + }); + } + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + expect(countChildStatuses(events, "completed")).toBe(1); + + openCode.emitEvent({ + type: "session.error", + properties: { + sessionID: "ses_child_status_transitions", + error: { name: "ProviderError", message: "late failure" }, + }, + }); + await vi.waitFor(() => expect(countChildStatuses(events, "failed")).toBe(1)); + await parent.close(); + }); + + test("clears child status dedupe state when the child is removed", async () => { + const { parent, openCode } = await createParentSession("ses_parent_child_status_cleanup"); + const events: AgentStreamEvent[] = []; + parent.subscribe((event) => events.push(event)); + const childInfo = { + id: "ses_child_status_cleanup", + parentID: "ses_parent_child_status_cleanup", + directory: "/workspace/repo", + }; + openCode.emitEvent({ type: "session.created", properties: { info: childInfo } }); + await vi.waitFor(() => expect(countChildStatuses(events, "running")).toBe(1)); + openCode.emitEvent({ + type: "session.deleted", + properties: { sessionID: childInfo.id }, + }); + openCode.emitEvent({ type: "session.created", properties: { info: childInfo } }); + await vi.waitFor(() => expect(countChildStatuses(events, "running")).toBe(2)); + await parent.close(); + }); + test("emits a provider subagent for a child created while the parent has no active turn", async () => { const releaseChildEvent = createTestDeferred(); const childConsumed = createTestDeferred(); @@ -4123,12 +4984,15 @@ describe("OpenCode provider subagent contract", () => { fakeClient, "ses_parent", createTestLogger(), + new Map(), + createDirectEventSource(fakeClient), ); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); releaseChildEvent.resolve(); await childConsumed.promise; + await vi.waitFor(() => expect(events.length).toBeGreaterThan(1)); await session.close(); expect(events).toContainEqual({ @@ -4276,12 +5140,15 @@ describe("OpenCode provider subagent contract", () => { fakeClient, "ses_parent", createTestLogger(), + new Map(), + createDirectEventSource(fakeClient), ); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); releaseEvents.resolve(); await eventsConsumed.promise; + await vi.waitFor(() => expect(countProviderSubagentUpserts(events)).toBeGreaterThanOrEqual(3)); await session.close(); const subtitleUpserts = events.flatMap((event) => @@ -4767,10 +5634,10 @@ describe("OpenCode provider subagent contract", () => { await session.close(); expect(openCodeClient.calls.sessionChildren).toEqual([ - { path: { id: "ses_parent" } }, - { path: { id: "ses_child_a" } }, - { path: { id: "ses_child_b" } }, - { path: { id: "ses_grandchild_a" } }, + { sessionID: "ses_parent", directory: "/workspace/repo" }, + { sessionID: "ses_child_a", directory: "/workspace/repo" }, + { sessionID: "ses_child_b", directory: "/workspace/repo" }, + { sessionID: "ses_grandchild_a", directory: "/workspace/repo" }, ]); expect(events).toEqual([ { @@ -5314,7 +6181,7 @@ function createOpenCodeTranslationState(sessionId: string): OpenCodeEventTransla cwd: "/workspace/repo", messageRoles: new Map(), accumulatedUsage: {}, - streamedPartKeys: new Set(), + materializedParts: new Map(), emittedStructuredMessageIds: new Set(), compactionSummaryMessageIds: new Set(), emittedCompactionPartIds: new Set(), diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index d2eb6730f..ad3c88729 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -12,6 +12,8 @@ import { type TextPartInput as OpenCodeTextPartInput, } from "@opencode-ai/sdk/v2/client"; import fs from "node:fs/promises"; +import { randomBytes } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { createPathEquivalenceMatcher } from "../../../utils/path.js"; import pLimit from "p-limit"; import type { Logger } from "pino"; @@ -76,6 +78,11 @@ import { type OpenCodeServerAcquisition, type OpenCodeServerManagerLike, } from "./opencode/server-manager.js"; +import { + OpenCodeEventConsumer, + type OpenCodeEventSource, + type OpenCodeEventSourceInput, +} from "./opencode/event-consumer.js"; import { resolveOpenCodeHomeDir } from "./opencode/paths.js"; import { formatProviderDiagnostic, @@ -118,10 +125,53 @@ const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { const OPENCODE_BUILD_MODE_ID = "build"; const OPENCODE_LEGACY_FULL_ACCESS_MODE_ID = "full-access"; const OPENCODE_DEFAULT_VARIANT_ID = "default"; +const EMPTY_OPENCODE_EVENT_SOURCE: OpenCodeEventSource = { + ready: async () => undefined, + subscribe: () => () => undefined, +}; +const OPENCODE_ID_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"; +let lastOpenCodeMessageTimestamp = -1; +let openCodeMessageCounter = 0; + +function createOpenCodeMessageId( + now = Date.now(), + random = (length: number) => + Array.from(randomBytes(length), (value) => OPENCODE_ID_ALPHABET[value % 62]).join(""), +): string { + if (now !== lastOpenCodeMessageTimestamp) openCodeMessageCounter = 0; + lastOpenCodeMessageTimestamp = now; + openCodeMessageCounter += 1; + const ascending = (BigInt(now) * 0x1000n + BigInt(openCodeMessageCounter)) + .toString(16) + .padStart(12, "0") + .slice(-12); + return `msg_${ascending}${random(14)}`; +} const OPENCODE_AUTO_ACCEPT_FEATURE_ID = "auto_accept"; const OPENCODE_PERSISTED_SESSION_LIMIT = 200; const OPENCODE_PENDING_ABORT_START_TIMEOUT_MS = 10_000; const OPENCODE_CHILD_SESSION_HYDRATION_LIMIT = 100; +const OPENCODE_STOP_STATUS_MAX_DELAY_MS = 1_000; + +function waitForOpenCodeStopProbe(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const onAbort = () => reject(signal.reason); + setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, delayMs).unref(); + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +function recoverChildStatus(snapshot: boolean, type: unknown, assistant?: OpenCodeSessionMessage) { + if (type === "busy" || type === "retry") return "running"; + if (assistant && "error" in assistant.info && assistant.info.error) return "failed"; + const time = assistant?.info.time; + if (snapshot || (time && "completed" in time && time.completed !== undefined)) return "completed"; +} + const OPENCODE_CHILD_SESSION_SERVER_REGISTRY_LIMIT = 500; const OPENCODE_PERMISSION_ACTION_ALLOW_ONCE = "allow_once"; const OPENCODE_PERMISSION_ACTION_ALLOW_ALWAYS = "allow_always"; @@ -549,19 +599,6 @@ function readOpenCodeMcpOperationError(data: unknown, name: string): unknown { return entry.error ?? `OpenCode reported MCP server '${name}' failed`; } -function resolvePartDedupeKey( - part: { id: string; messageID: string }, - partType: "text" | "reasoning", -): string | null { - if (part.id.trim().length > 0) { - return `${partType}:${part.id}`; - } - if (part.messageID.trim().length > 0) { - return `${partType}:message:${part.messageID}`; - } - return null; -} - function matchesHydratedFingerprint( fingerprints: Map | undefined, id: string, @@ -1117,11 +1154,15 @@ function findOpenCodeCompactionPart( async function readOpenCodeSessionMessagesFromSdk( client: Pick, session: OpenCodePersistedSession, + signal?: AbortSignal, ): Promise { - const response = await client.session.messages({ - sessionID: session.id, - directory: session.directory, - }); + const response = await client.session.messages( + { + sessionID: session.id, + directory: session.directory, + }, + signal ? { signal } : undefined, + ); if (response.error || !response.data) { return []; @@ -1314,13 +1355,19 @@ export class OpenCodeAgentClient implements AgentClient { ) { this.logger = logger.child({ module: "agent", provider: "opencode" }); this.runtimeSettings = runtimeSettings; + this.createOpenCodeClient = deps.createClient ?? createSdkOpenCodeClient; this.serverManager = deps.serverManager ?? OpenCodeServerManager.getInstance(this.logger, runtimeSettings, { managedProcesses: deps.managedProcesses, resolveHomeDir: deps.resolveHomeDir, + createEventSource: ({ serverUrl, processExit }) => + new OpenCodeEventConsumer({ + serverUrl, + processExit, + createClient: (baseUrl) => this.createOpenCodeClient({ baseUrl, directory: "" }), + }), }); - this.createOpenCodeClient = deps.createClient ?? createSdkOpenCodeClient; this.resolveHomeDir = deps.resolveHomeDir ?? resolveOpenCodeHomeDir; } @@ -1363,6 +1410,7 @@ export class OpenCodeAgentClient implements AgentClient { session.id, this.logger, new Map(this.modelContextWindows), + acquisition.events, acquisition.release, options?.persistSession, launchContext?.agentId, @@ -1416,6 +1464,7 @@ export class OpenCodeAgentClient implements AgentClient { handle.sessionId, this.logger, new Map(this.modelContextWindows), + acquisition.events, acquisition.release, undefined, launchContext?.agentId, @@ -1775,14 +1824,13 @@ export interface OpenCodeEventTranslationState { emittedUserMessageIds?: Set; accumulatedUsage: AgentUsage; sessionTotalCostUsd?: number; - streamedPartKeys: Set; + materializedParts: Map; emittedStructuredMessageIds: Set; compactionSummaryMessageIds: Set; emittedCompactionPartIds: Set; hydratedMessageFingerprints?: Map; hydratedPartFingerprints?: Map; suppressAssistantMessagesUntilIdle?: { active: boolean }; - /** Tracks the type of each part by ID, learned from message.part.updated events. */ partTypes: Map; subAgentsByCallId?: Map; subAgentCallIdByChildSessionId?: Map; @@ -1790,6 +1838,11 @@ export interface OpenCodeEventTranslationState { subagentPresentationByChildId?: Map; modelContextWindowsByModelKey?: ReadonlyMap; onAssistantModelContextWindowResolved?: (contextWindowMaxTokens: number) => void; + onMaterializationMismatch?: (diagnostic: { + partId: string; + messageId: string; + kind: "text" | "reasoning"; + }) => void; } interface OpenCodeTraceData { @@ -2149,7 +2202,6 @@ export function translateOpenCodeEvent( } function resetOpenCodeTurnTrackingState(state: OpenCodeEventTranslationState): void { - state.streamedPartKeys.clear(); state.partTypes.clear(); state.compactionSummaryMessageIds.clear(); state.emittedCompactionPartIds.clear(); @@ -2246,7 +2298,7 @@ function appendOpenCodeChildSessionDetected( child: OpenCodeChildSessionInfo, state: OpenCodeEventTranslationState, events: AgentStreamEvent[], - status: "running" | "completed" = "running", + status: "running" | "completed" | null = "running", ): boolean { if ( child.id === state.sessionId || @@ -2281,7 +2333,7 @@ function appendOpenCodeChildSessionDetected( id: child.id, ...(title ? { title } : {}), ...(child.title && !presentation.descriptionFromLink ? { description: child.title } : {}), - status, + ...(status ? { status } : {}), ...(child.directory ? { cwd: child.directory } : {}), ...(subtitle ? { subtitle } : {}), }, @@ -2677,15 +2729,28 @@ function appendOpenCodeTextPart( if (!part.time?.end) { return; } - const partKey = resolvePartDedupeKey(part, "text"); - if (partKey && state.streamedPartKeys.delete(partKey)) { + const materialized = state.materializedParts.get(part.id); + if (materialized?.closed) return; + const emittedText = materialized?.messageId === part.messageID ? materialized.emittedText : ""; + if (!part.text.startsWith(emittedText)) { + state.onMaterializationMismatch?.({ + partId: part.id, + messageId: part.messageID, + kind: "text", + }); return; } - if (part.text) { + const suffix = part.text.slice(emittedText.length); + state.materializedParts.set(part.id, { + messageId: part.messageID, + emittedText: part.text, + closed: true, + }); + if (suffix) { events.push({ type: "timeline", provider: "opencode", - item: { type: "assistant_message", text: part.text, messageId: part.messageID }, + item: { type: "assistant_message", text: suffix, messageId: part.messageID }, }); } } @@ -2701,15 +2766,28 @@ function appendOpenCodeReasoningPart( if (!part.time.end) { return; } - const partKey = resolvePartDedupeKey(part, "reasoning"); - if (partKey && state.streamedPartKeys.delete(partKey)) { + const materialized = state.materializedParts.get(part.id); + if (materialized?.closed) return; + const emittedText = materialized?.messageId === part.messageID ? materialized.emittedText : ""; + if (!part.text.startsWith(emittedText)) { + state.onMaterializationMismatch?.({ + partId: part.id, + messageId: part.messageID, + kind: "reasoning", + }); return; } - if (part.text) { + const suffix = part.text.slice(emittedText.length); + state.materializedParts.set(part.id, { + messageId: part.messageID, + emittedText: part.text, + closed: true, + }); + if (suffix) { events.push({ type: "timeline", provider: "opencode", - item: { type: "reasoning", text: part.text }, + item: { type: "reasoning", text: suffix }, }); } } @@ -2735,9 +2813,7 @@ function appendOpenCodeMessagePartDelta( } if (isReasoning) { - if (partID) { - state.streamedPartKeys.add(`reasoning:${partID}`); - } + if (!appendOpenCodeMaterializedDelta(state, partID, messageID, delta)) return; events.push({ type: "timeline", provider: "opencode", @@ -2759,9 +2835,7 @@ function appendOpenCodeMessagePartDelta( state.compactionSummaryMessageIds.add(assistantMessageId); return; } - if (partID) { - state.streamedPartKeys.add(`text:${partID}`); - } + if (!appendOpenCodeMaterializedDelta(state, partID, assistantMessageId, delta)) return; events.push({ type: "timeline", provider: "opencode", @@ -2773,6 +2847,23 @@ function appendOpenCodeMessagePartDelta( }); } +function appendOpenCodeMaterializedDelta( + state: OpenCodeEventTranslationState, + partId: string | undefined, + messageId: string, + delta: string, +): boolean { + if (!partId) return true; + const previous = state.materializedParts.get(partId); + if (previous?.closed) return false; + state.materializedParts.set(partId, { + messageId, + emittedText: `${previous?.emittedText ?? ""}${delta}`, + closed: false, + }); + return true; +} + function appendOpenCodePermissionAsked( event: Extract, state: OpenCodeEventTranslationState, @@ -2926,7 +3017,6 @@ function appendOpenCodeSessionStatus( }); return; } - // "busy" is transient — no terminal event, no surfaced activity. } interface Deferred { @@ -3092,25 +3182,12 @@ async function listOpenCodeChildSessions( client: OpencodeClient, sessionId: string, directory: string, + signal?: AbortSignal, ): Promise { - try { - const pathResponse: unknown = await Reflect.apply(client.session.children, client.session, [ - { path: { id: sessionId } }, - ]); - const pathChildren = readOpenCodeChildSessionInfosFromResponse(pathResponse); - if (pathChildren) { - return pathChildren; - } - } catch (error) { - if (!(error instanceof Error)) { - throw error; - } - } - - const sessionIdResponse = await client.session.children({ - sessionID: sessionId, - directory, - }); + const sessionIdResponse = await client.session.children( + { sessionID: sessionId, directory }, + { signal }, + ); return readOpenCodeChildSessionInfosFromResponse(sessionIdResponse) ?? []; } @@ -3131,19 +3208,19 @@ class OpenCodeAgentSession implements AgentSession { private sessionTotalCostUsd: number | undefined; private mcpConfigured = false; private mcpSetupPromise: Promise | null = null; - /** Tracks the role of each message by ID to distinguish user from assistant messages */ private messageRoles = new Map(); private pendingUserMessageText: string | null = null; private pendingClientMessageId: string | null = null; private emittedUserMessageIds = new Set(); - /** Tracks streamed textual part IDs to suppress final full-text echoes from OpenCode. */ - private streamedPartKeys = new Set(); - /** Tracks assistant messages already emitted from structured payloads. */ + private materializedParts = new Map< + string, + { messageId: string; emittedText: string; closed: boolean } + >(); + private activeDispatchMessageId: string | null = null; private emittedStructuredMessageIds = new Set(); private compactionSummaryMessageIds = new Set(); private emittedCompactionPartIds = new Set(); private suppressAssistantMessagesUntilIdle = { active: false }; - /** Tracks the type of each part by ID, learned from message.part.updated events. */ private partTypes = new Map(); private availableModesCache: AgentMode[] | null = null; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); @@ -3167,15 +3244,20 @@ class OpenCodeAgentSession implements AgentSession { >(); private readonly childTranslationStates = new Map(); private readonly childSessionCwds = new Map(); + private readonly childStatuses = new Map< + string, + Extract["event"], { type: "upsert" }> + >(); private readonly pendingPermissionDirectories = new Map(); private childHydrationPromise: Promise | null = null; private childHydrationCompleted = false; private readonly unrelatedSessionIds = new Set(); private selectedModelContextWindowMaxTokens: number | undefined; private releaseServer: (() => Promise) | null; - private eventStreamAbortController: AbortController | null = null; - private eventStreamReady: Deferred | null = null; - private eventStreamTask: Promise | null = null; + private ingress = Promise.resolve(); + private gapRepairRevision = 0; + private recoveryAbortController = new AbortController(); + private unsubscribeEvents: (() => void) | null = null; private closed = false; private readonly persistSession: boolean; private deletedFromProvider = false; @@ -3185,6 +3267,7 @@ class OpenCodeAgentSession implements AgentSession { sessionId: string, logger: Logger, modelContextWindowsByModelKey: ReadonlyMap = new Map(), + private readonly events: OpenCodeEventSource = EMPTY_OPENCODE_EVENT_SOURCE, releaseServer?: () => Promise, persistSession = true, private readonly agentId?: string, @@ -3203,7 +3286,18 @@ class OpenCodeAgentSession implements AgentSession { this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( config.model, ); - this.startEventStream(); + this.unsubscribeEvents = this.events.subscribe((input) => { + if ("type" in input && input.type === "server-exited") + this.recoveryAbortController.abort(input.error); + this.ingress = this.ingress + .then(() => this.consumeEventSourceInput(input)) + .catch((error) => { + this.logger.warn( + { err: error, sessionId: this.sessionId }, + "OpenCode event ingress failed", + ); + }); + }); } get id(): string | null { @@ -3334,9 +3428,7 @@ class OpenCodeAgentSession implements AgentSession { } private async waitUntilProviderIdle(): Promise { - if (this.turnState.status !== "stopping") { - return; - } + if (this.turnState.status !== "stopping") return; await withTimeout( this.observeProviderStopBoundary(this.turnState.stop), OPENCODE_PENDING_ABORT_START_TIMEOUT_MS, @@ -3345,34 +3437,21 @@ class OpenCodeAgentSession implements AgentSession { } private async observeProviderStopBoundary(stop: OpenCodeStop): Promise { + let delayMs = 100; while (this.isStopping(stop)) { - const eventStreamTask = this.eventStreamTask; - const boundary = await (eventStreamTask - ? Promise.race([ - stop.terminal.promise.then(() => "terminal" as const), - eventStreamTask.then( - () => "stream_ended" as const, - () => "stream_ended" as const, - ), - ]) - : Promise.resolve("stream_ended" as const)); - if (boundary === "terminal") { - return; - } + const boundary = await Promise.race([ + stop.terminal.promise.then(() => "terminal" as const), + waitForOpenCodeStopProbe(delayMs, this.recoveryAbortController.signal).then( + () => "probe" as const, + ), + ]); + if (boundary === "terminal") return; await this.reconcileStopWithProviderStatus(stop); + delayMs = Math.min(delayMs * 2, OPENCODE_STOP_STATUS_MAX_DELAY_MS); } } - /** - * The event stream dropped mid-stop, so the canceled run's terminal may have - * gone with it. `session.status` is the provider's authority on whether the - * runner is still going, and an idle answer ends the stop through the same - * transition an SSE terminal takes. A failed probe is not evidence either way, - * so the stop keeps waiting on the replacement stream instead of rejecting a - * caller that a later terminal would have released. - */ private async reconcileStopWithProviderStatus(stop: OpenCodeStop): Promise { - await this.ensureEventStreamReady(); try { if ((await this.readProviderRunnerStatus()) === "idle") { this.finishStoppingTurn(stop); @@ -3386,7 +3465,10 @@ class OpenCodeAgentSession implements AgentSession { } private async readProviderRunnerStatus(): Promise { - const response = await this.client.session.status({ directory: this.config.cwd }); + const response = await this.client.session.status( + { directory: this.config.cwd }, + { signal: this.recoveryAbortController.signal }, + ); if (response.error) { throw new Error( `Failed to confirm OpenCode session status: ${toDiagnosticErrorMessage(response.error)}`, @@ -3412,10 +3494,20 @@ class OpenCodeAgentSession implements AgentSession { prompt: AgentPromptInput, options?: AgentRunOptions, ): Promise<{ turnId: string }> { + if (this.closed) { + throw new Error("OpenCode session is closed"); + } if (this.turnState.status === "running") { throw new Error("A foreground turn is already active"); } - await this.awaitRunnerQuiescence(); + try { + await this.awaitRunnerQuiescence(); + } catch (error) { + this.rethrowRunnerWaitError(error); + } + if (this.closed) { + throw new Error("OpenCode session is closed"); + } if (this.turnState.status !== "idle") { throw new Error("OpenCode is still stopping the previous turn"); } @@ -3439,7 +3531,7 @@ class OpenCodeAgentSession implements AgentSession { const effectiveMode = resolveOpenCodeRuntimeAgentId(this.currentMode); try { - await this.ensureEventStreamReady(); + await withTimeout(this.events.ready(), 10_000, "OpenCode event stream first record"); } catch (error) { if (this.abortController === turnAbortController) { this.abortController = null; @@ -3448,12 +3540,14 @@ class OpenCodeAgentSession implements AgentSession { } const turnId = this.createTurnId(); + this.materializedParts.clear(); this.turnState = { status: "running", turnId }; this.notifySubscribers({ type: "turn_started", provider: "opencode" }, turnId); const slashCommand = await this.resolveSlashCommandInvocation(prompt); if (slashCommand) { if (slashCommand.commandName === "compact" || slashCommand.commandName === "summarize") { + this.activeDispatchMessageId = null; this.suppressAssistantMessagesUntilIdle.active = true; void this.client.session .summarize({ @@ -3491,12 +3585,14 @@ class OpenCodeAgentSession implements AgentSession { // command() is only dispatch acknowledgement. OpenCode session events are // the source of truth for when the command turn becomes idle or fails. + this.activeDispatchMessageId = createOpenCodeMessageId(); void this.client.session .command({ sessionID: this.sessionId, directory: this.config.cwd, command: slashCommand.commandName, arguments: slashCommand.args ?? "", + messageID: this.activeDispatchMessageId, ...(this.config.model ? { model: this.config.model } : {}), ...(effectiveMode ? { agent: effectiveMode } : {}), ...(effectiveVariant ? { variant: effectiveVariant } : {}), @@ -3540,6 +3636,8 @@ class OpenCodeAgentSession implements AgentSession { ); }); } else { + const dispatchMessageId = createOpenCodeMessageId(); + this.activeDispatchMessageId = dispatchMessageId; // Wrap in an async IIFE so a synchronous throw from promptAsync (e.g. // SDK input validation) is caught alongside async rejections. A plain // `.then().catch()` chain would let a sync throw escape unhandled. @@ -3564,6 +3662,7 @@ class OpenCodeAgentSession implements AgentSession { const promptResponse = await this.client.session.promptAsync({ sessionID: this.sessionId, directory: this.config.cwd, + messageID: dispatchMessageId, parts, ...(options?.outputSchema ? { @@ -3617,6 +3716,11 @@ class OpenCodeAgentSession implements AgentSession { return { turnId }; } + + private rethrowRunnerWaitError(error: unknown): never { + if (this.closed) throw new Error("OpenCode session is closed", { cause: error }); + throw error; + } subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); this.startExternalStatusReconciliation(); @@ -3640,7 +3744,7 @@ class OpenCodeAgentSession implements AgentSession { } private async reconcileExternalRunnerStatus(): Promise { - await this.ensureEventStreamReady(); + await this.events.ready(); const observedRevision = this.runnerStatusRevision; const runnerStatus = await this.readProviderRunnerStatus(); if ( @@ -3676,54 +3780,136 @@ class OpenCodeAgentSession implements AgentSession { }); } - private async hydrateChildSessions(): Promise { - const queue = [this.sessionId]; - const visited = new Set(); - while (queue.length > 0 && visited.size < OPENCODE_CHILD_SESSION_HYDRATION_LIMIT) { - const parentSessionId = queue.shift(); - if (!parentSessionId || visited.has(parentSessionId)) { - continue; - } - visited.add(parentSessionId); - const children = await listOpenCodeChildSessions( - this.client, - parentSessionId, - this.config.cwd, - ); - if (this.closed) return; - for (const child of children) { - const detectionEvents: AgentStreamEvent[] = []; - appendOpenCodeChildSessionDetected( - child, - this.createTranslationState(), - detectionEvents, - "completed", - ); - for (const event of detectionEvents) { - this.recordProviderInternalEvent(event); - this.notifySubscribers(event, null); - } - try { - await this.hydrateChildSessionTimeline(child); - } catch (error) { - this.logger.warn( - { err: error, sessionId: child.id }, - "OpenCode child timeline hydration failed", - ); - } - if (visited.size + queue.length < OPENCODE_CHILD_SESSION_HYDRATION_LIMIT) { - queue.push(child.id); - } - } + private async hydrateChildSessions(recovered = false): Promise { + const discovered = await this.discoverChildSessions(); + const statusesByDirectory = await this.readChildStatuses(discovered); + for (const child of discovered) { + await this.hydrateDiscoveredChild(child, statusesByDirectory, recovered); } } - private async hydrateChildSessionTimeline(child: OpenCodeChildSessionInfo): Promise { - const messages = await readOpenCodeSessionMessagesFromSdk(this.client, { - id: child.id, - directory: child.directory ?? this.config.cwd, - ...(child.revert ? { revert: child.revert } : {}), - } as OpenCodePersistedSession); + private async discoverChildSessions(): Promise { + const queue = [{ id: this.sessionId, directory: this.config.cwd }]; + const visited = new Set(); + const discovered: OpenCodeChildSessionInfo[] = []; + while (queue.length > 0 && visited.size < OPENCODE_CHILD_SESSION_HYDRATION_LIMIT) { + const parent = queue.shift(); + if (!parent || visited.has(parent.id)) { + continue; + } + visited.add(parent.id); + const children = await listOpenCodeChildSessions( + this.client, + parent.id, + parent.directory, + this.recoveryAbortController.signal, + ); + if (this.closed) return discovered; + for (const child of children) { + discovered.push(child); + if (visited.size + queue.length < OPENCODE_CHILD_SESSION_HYDRATION_LIMIT) { + queue.push({ id: child.id, directory: child.directory ?? parent.directory }); + } + } + } + return discovered; + } + + private async readChildStatuses( + children: OpenCodeChildSessionInfo[], + ): Promise | null>> { + const statusesByDirectory = new Map | null>(); + for (const directory of new Set(children.map((child) => child.directory ?? this.config.cwd))) { + const response = await this.client.session + .status({ directory }, { signal: this.recoveryAbortController.signal }) + .catch(() => null); + const record = readOpenCodeRecord(response); + statusesByDirectory.set( + directory, + record && !record.error ? readOpenCodeRecord(record.data) : null, + ); + } + return statusesByDirectory; + } + + private async hydrateDiscoveredChild( + child: OpenCodeChildSessionInfo, + statusesByDirectory: Map | null>, + recovered: boolean, + ): Promise { + const directory = child.directory ?? this.config.cwd; + const snapshot = statusesByDirectory.get(directory) ?? null; + const status = readOpenCodeRecord(snapshot?.[child.id]); + const active = snapshot === null || status?.type === "busy" || status?.type === "retry"; + const detectedStatus = active ? "running" : "completed"; + const detectionEvents: AgentStreamEvent[] = []; + appendOpenCodeChildSessionDetected( + child, + this.createTranslationState(), + detectionEvents, + recovered ? null : detectedStatus, + ); + for (const event of detectionEvents) { + this.recordProviderInternalEvent(event); + this.notifySubscribers(event, null); + } + let messages: OpenCodeSessionMessage[] | null = null; + try { + messages = await this.hydrateChildSessionTimeline(child, recovered); + } catch (error) { + this.logger.warn( + { err: error, sessionId: child.id }, + "OpenCode child timeline hydration failed", + ); + } + if (!recovered) return; + const latestAssistant = messages?.findLast((message) => message.info.role === "assistant"); + const recoveredStatus = recoverChildStatus(snapshot !== null, status?.type, latestAssistant); + if (!recoveredStatus) return; + const statusEvent: AgentStreamEvent = { + type: "provider_subagent", + provider: "opencode", + event: { type: "upsert", id: child.id, status: recoveredStatus }, + }; + this.recordProviderInternalEvent(statusEvent); + this.notifySubscribers(statusEvent, null); + } + + private async hydrateChildSessionTimeline( + child: OpenCodeChildSessionInfo, + recovered = false, + ): Promise { + const messages = await readOpenCodeSessionMessagesFromSdk( + this.client, + { + id: child.id, + directory: child.directory ?? this.config.cwd, + ...(child.revert ? { revert: child.revert } : {}), + } as OpenCodePersistedSession, + this.recoveryAbortController.signal, + ); + if (recovered) { + for (const message of messages) { + await this.consumeOpenCodeStreamEvent({ + rawEvent: { + directory: child.directory ?? this.config.cwd, + payload: { type: "message.updated", properties: { info: message.info } }, + }, + eventCount: 0, + }); + for (const part of message.parts) { + await this.consumeOpenCodeStreamEvent({ + rawEvent: { + directory: child.directory ?? this.config.cwd, + payload: { type: "message.part.updated", properties: { part } }, + }, + eventCount: 0, + }); + } + } + this.emitHydratedChildPresentation(child, messages); + return messages; + } const translationState = this.getChildTranslationState(child.id); let latestReplayedMessage: OpenCodeSessionMessage | null = null; for (const message of messages) { @@ -3756,6 +3942,7 @@ class OpenCodeAgentSession implements AgentSession { } } this.emitHydratedChildPresentation(child, messages); + return messages; } /** @@ -3817,126 +4004,199 @@ class OpenCodeAgentSession implements AgentSession { unregisterOpenCodeChildSessionServerUrl(event.event.id); this.childTranslationStates.delete(event.event.id); this.childSessionCwds.delete(event.event.id); + this.childStatuses.delete(event.event.id); this.subagentPresentationByChildId.delete(event.event.id); } } - private startEventStream(): void { - void this.ensureEventStreamReady().catch((error) => { - this.logger.warn({ err: error, sessionId: this.sessionId }, "OpenCode event stream failed"); - }); + private async consumeEventSourceInput(input: OpenCodeEventSourceInput): Promise { + if ("type" in input && input.type === "reconnected") { + await this.reconcileAfterGap(++this.gapRepairRevision); + return; + } + if ("type" in input && input.type === "server-exited") { + if (this.turnState.status === "stopping") return this.finishStoppingTurn(this.turnState.stop); + const turnId = this.activeForegroundTurnId; + if (turnId) { + this.finishForegroundTurn( + { type: "turn_failed", provider: "opencode", error: input.error.message }, + turnId, + ); + } + return; + } + await this.consumeOpenCodeStreamEvent({ rawEvent: input, eventCount: 0 }); } - private ensureEventStreamReady(): Promise { - if (this.closed) { - return Promise.reject(new Error("OpenCode session is closed")); + private async reconcileAfterGap(revision: number, refresh = true, delay = 100): Promise { + if (revision !== this.gapRepairRevision) return; + const turnId = this.activeForegroundTurnId; + const dispatchMessageId = this.activeDispatchMessageId; + await this.refreshGapScope(refresh); + if (revision !== this.gapRepairRevision || !turnId) return; + const runnerStatus = await this.readProviderRunnerStatus().catch(() => null); + if (revision !== this.gapRepairRevision) return; + if (runnerStatus === null) { + this.scheduleGapRepair(revision, delay); + return; } - if (this.eventStreamReady) { - return this.eventStreamReady.promise; + if (dispatchMessageId === null) { + if (runnerStatus === "idle") { + this.suppressAssistantMessagesUntilIdle.active = false; + this.finishForegroundTurn({ type: "turn_completed", provider: "opencode" }, turnId); + } + return; } - - const eventStreamAbortController = new AbortController(); - const eventStreamReady = createDeferred(); - this.eventStreamAbortController = eventStreamAbortController; - this.eventStreamReady = eventStreamReady; - const eventStreamTask = this.consumeEventStream( - eventStreamAbortController, - eventStreamReady, - ).finally(() => { - if (this.eventStreamAbortController === eventStreamAbortController) { - this.eventStreamAbortController = null; - this.eventStreamReady = null; + const messages = await readOpenCodeSessionMessagesFromSdk( + this.client, + { id: this.sessionId, directory: this.config.cwd } as OpenCodePersistedSession, + this.recoveryAbortController.signal, + ).catch(() => null); + if (revision !== this.gapRepairRevision) return; + if (messages === null) { + this.scheduleGapRepair(revision, delay); + return; + } + const boundary = messages.findIndex((message) => message.info.id === dispatchMessageId); + if (boundary < 0) { + if (runnerStatus === "idle") { + this.finishForegroundTurn( + { type: "turn_failed", provider: "opencode", error: "Active dispatch not found" }, + turnId, + ); } - if (this.eventStreamTask === eventStreamTask) { - this.eventStreamTask = null; + return; + } + for (const message of messages.slice(boundary)) { + if (revision !== this.gapRepairRevision) return; + await this.consumeOpenCodeStreamEvent({ + rawEvent: { + directory: this.config.cwd, + payload: { type: "message.updated", properties: { info: message.info } }, + }, + eventCount: 0, + }); + for (const part of message.parts) { + if (revision !== this.gapRepairRevision) return; + await this.consumeOpenCodeStreamEvent({ + rawEvent: { + directory: this.config.cwd, + payload: { type: "message.part.updated", properties: { part } }, + }, + eventCount: 0, + }); } - }); - this.eventStreamTask = eventStreamTask; - void eventStreamTask.catch((error) => { - this.logger.warn( - { err: error, sessionId: this.sessionId }, - "OpenCode event stream task failed", - ); - }); - - return eventStreamReady.promise; + } + if (runnerStatus === "idle" && this.activeForegroundTurnId === turnId) { + await this.settleRecoveredGap(messages, boundary); + } } - private async consumeEventStream( - eventStreamAbortController: AbortController, - eventStreamReady: Deferred, + private async refreshGapScope(refresh: boolean): Promise { + if (!refresh) return; + await this.hydrateChildSessions(true).catch(() => undefined); + await this.reconcileBlockingRequests(); + } + + private async settleRecoveredGap( + messages: OpenCodeSessionMessage[], + boundary: number, ): Promise { - this.traceOpenCode("provider.opencode.subscribe.start", { - sessionId: this.sessionId, - cwd: this.config.cwd, - }); - let eventStreamReadyResolved = false; - try { - const result = await this.client.global.event({ - signal: eventStreamAbortController.signal, - sseMaxRetryAttempts: 0, - }); - eventStreamReadyResolved = true; - this.traceOpenCode("provider.opencode.subscribe.ready", { - sessionId: this.sessionId, - }); - eventStreamReady.resolve(); - - let eventCount = 0; - for await (const rawEvent of result.stream) { - eventCount += 1; - await this.consumeOpenCodeStreamEvent({ rawEvent, eventCount }); - } - - this.traceOpenCode("provider.opencode.stream.eof", { - eventCount, - aborted: eventStreamAbortController.signal.aborted, - activeTurnId: this.activeForegroundTurnId, - }); - - if (!eventStreamAbortController.signal.aborted) { - if (!eventStreamReadyResolved) { - eventStreamReady.reject(new Error("OpenCode event stream ended before it became ready")); + const latestAssistant = messages + .slice(boundary) + .findLast((message) => message.info.role === "assistant"); + const persistedError = + latestAssistant && "error" in latestAssistant.info ? latestAssistant.info.error : undefined; + const payload = persistedError + ? { + type: "session.error" as const, + properties: { sessionID: this.sessionId, error: persistedError }, } - const activeTurnId = this.activeForegroundTurnId; - if (activeTurnId) { - this.traceOpenCode("provider.opencode.turn.fail_eof", { - turnId: activeTurnId, - eventCount, - }); - this.finishForegroundTurn( - { - type: "turn_failed", - provider: "opencode", - error: "OpenCode event stream ended before the turn reached a terminal state", + : { + type: "session.status" as const, + properties: { sessionID: this.sessionId, status: { type: "idle" as const } }, + }; + await this.consumeOpenCodeStreamEvent({ + rawEvent: { directory: this.config.cwd, payload }, + eventCount: 0, + }); + } + + private scheduleGapRepair(revision: number, delayMs: number): void { + setTimeout(() => { + if (revision !== this.gapRepairRevision || this.recoveryAbortController.signal.aborted) + return; + const nextDelay = Math.min(delayMs * 2, OPENCODE_STOP_STATUS_MAX_DELAY_MS); + this.ingress = this.ingress + .then(() => this.reconcileAfterGap(revision, false, nextDelay)) + .catch(() => undefined); + }, delayMs).unref(); + } + + private async reconcileBlockingRequests(): Promise { + const directories = new Set([this.config.cwd, ...this.childSessionCwds.values()]); + for (const directory of directories) { + const [permissions, questions] = await Promise.all([ + this.client.permission + .list({ directory }, { signal: this.recoveryAbortController.signal }) + .catch(() => null), + this.client.question + .list({ directory }, { signal: this.recoveryAbortController.signal }) + .catch(() => null), + ]); + for (const [kind, response] of [ + ["tool", permissions], + ["question", questions], + ] as const) { + const record = readOpenCodeRecord(response); + if (!record || record.error || !Array.isArray(record.data)) continue; + const liveIds = new Set(); + for (const properties of record.data) { + const propertiesRecord = readOpenCodeRecord(properties); + const id = readNonEmptyString(propertiesRecord?.id); + const sessionId = readNonEmptyString(propertiesRecord?.sessionID); + if (!id || !sessionId || !this.isOwnedSessionId(sessionId)) continue; + liveIds.add(id); + await this.consumeOpenCodeStreamEvent({ + rawEvent: { + directory, + payload: { + type: kind === "question" ? "question.asked" : "permission.asked", + properties, + }, }, - activeTurnId, + eventCount: 0, + }); + } + for (const [id, pendingDirectory] of this.pendingPermissionDirectories) { + const pending = this.pendingPermissions.get(id); + if ( + pendingDirectory !== directory || + liveIds.has(id) || + (kind === "question") !== (pending?.kind === "question") + ) { + continue; + } + this.pendingPermissionDirectories.delete(id); + this.pendingPermissions.delete(id); + this.notifySubscribers( + { + type: "permission_resolved", + provider: "opencode", + requestId: id, + resolution: { behavior: "allow" }, + }, + null, ); } } - } catch (error) { - this.traceOpenCode("provider.opencode.subscribe.error", { - turnId: this.activeForegroundTurnId ?? undefined, - error: - error instanceof Error ? { name: error.name, message: error.message } : String(error), - }); - if (!eventStreamReadyResolved) { - eventStreamReady.reject(error); - } - const activeTurnId = this.activeForegroundTurnId; - if (!eventStreamAbortController.signal.aborted && activeTurnId) { - this.finishForegroundTurn( - { - type: "turn_failed", - provider: "opencode", - error: toDiagnosticErrorMessage(error), - }, - activeTurnId, - ); - } } } + private isOwnedSessionId(sessionId: string): boolean { + return sessionId === this.sessionId || this.knownChildSessionIds.has(sessionId); + } + private async consumeOpenCodeStreamEvent(params: { rawEvent: unknown; eventCount: number; @@ -4057,12 +4317,14 @@ class OpenCodeAgentSession implements AgentSession { private startAutonomousTurn(): string { const turnId = this.createTurnId(); + this.materializedParts.clear(); this.turnState = { status: "running", turnId }; this.runningToolCalls.clear(); this.subAgentsByCallId.clear(); this.subAgentCallIdByChildSessionId.clear(); this.pendingUserMessageText = null; this.pendingClientMessageId = null; + this.activeDispatchMessageId = null; this.abortController = null; this.notifySubscribers({ type: "turn_started", provider: "opencode" }, turnId); return turnId; @@ -4098,7 +4360,6 @@ class OpenCodeAgentSession implements AgentSession { return this.turnState.status === "stopping" && this.turnState.stop === stop; } - /** Stops the session runner and returns the abort this Stop issued. */ private issueStop(turnId: string | null): Promise { if (this.turnState.status === "stopping") { // Stop pressed again during a stop retries that same stop. There is one @@ -4124,7 +4385,6 @@ class OpenCodeAgentSession implements AgentSession { return abort; } - /** Issues one more session-scoped abort for this stop and returns it. */ private issueOwnedAbort(stop: OpenCodeStop): Promise { const abort = this.runOwnedAbort(stop.pendingCancellationTurnId); // Abort is session-scoped, so its settlement is too: an older request lands @@ -4226,6 +4486,10 @@ class OpenCodeAgentSession implements AgentSession { if (this.closed) { return; } + if (event.type === "provider_subagent" && event.event.type === "upsert" && event.event.status) { + if (isDeepStrictEqual(this.childStatuses.get(event.event.id), event.event)) return; + this.childStatuses.set(event.event.id, structuredClone(event.event)); + } const turnId = turnIdOverride === null ? null : (turnIdOverride ?? this.activeForegroundTurnId); const tagged = turnId ? { ...event, turnId } : event; this.traceOpenCode("provider.opencode.event_emit", { @@ -4235,13 +4499,12 @@ class OpenCodeAgentSession implements AgentSession { for (const callback of this.subscribers) { try { callback(tagged); - } catch { - // Subscriber callback error isolation - } + } catch {} } } private createTurnId(): string { + this.gapRepairRevision += 1; return `opencode-turn-${this.nextTurnOrdinal++}`; } @@ -4405,25 +4668,12 @@ class OpenCodeAgentSession implements AgentSession { async close(): Promise { try { - // Flip closed before clearing subscribers so any event the SDK delivers - // after the abort (between here and subscribers.clear) is swallowed by - // notifySubscribers instead of bubbling through provider-runner as an - // unhandled rejection in whichever test the daemon hops to next. this.closed = true; this.abortController?.abort(); - const eventStreamTask = this.eventStreamTask; - this.eventStreamAbortController?.abort(); - if (eventStreamTask) { - await eventStreamTask.catch((error) => { - this.logger.debug( - { err: error, sessionId: this.sessionId }, - "OpenCode event stream failed during close", - ); - }); - } - this.eventStreamAbortController = null; - this.eventStreamReady = null; - this.eventStreamTask = null; + this.recoveryAbortController.abort(); + this.unsubscribeEvents?.(); + this.unsubscribeEvents = null; + await this.ingress.catch(() => undefined); this.subscribers.clear(); await abortOpenCodeSession({ client: this.client, @@ -4582,7 +4832,7 @@ class OpenCodeAgentSession implements AgentSession { emittedUserMessageIds: this.emittedUserMessageIds, accumulatedUsage: this.accumulatedUsage, sessionTotalCostUsd: this.sessionTotalCostUsd, - streamedPartKeys: this.streamedPartKeys, + materializedParts: this.materializedParts, emittedStructuredMessageIds: this.emittedStructuredMessageIds, compactionSummaryMessageIds: this.compactionSummaryMessageIds, emittedCompactionPartIds: this.emittedCompactionPartIds, @@ -4593,6 +4843,12 @@ class OpenCodeAgentSession implements AgentSession { knownChildSessionIds: this.knownChildSessionIds, subagentPresentationByChildId: this.subagentPresentationByChildId, modelContextWindowsByModelKey: this.modelContextWindowsByModelKey, + onMaterializationMismatch: (diagnostic) => { + this.logger.warn( + { ...diagnostic, sessionId: this.sessionId }, + "OpenCode final part snapshot replaced streamed content", + ); + }, onAssistantModelContextWindowResolved: (contextWindowMaxTokens) => { this.accumulatedUsage.contextWindowMaxTokens = contextWindowMaxTokens; if (!this.config.model) { @@ -4613,7 +4869,7 @@ class OpenCodeAgentSession implements AgentSession { messageRoles: new Map(), emittedUserMessageIds: new Set(), accumulatedUsage: {}, - streamedPartKeys: new Set(), + materializedParts: new Map(), emittedStructuredMessageIds: new Set(), compactionSummaryMessageIds: new Set(), emittedCompactionPartIds: new Set(), @@ -4626,6 +4882,12 @@ class OpenCodeAgentSession implements AgentSession { knownChildSessionIds: new Set(), subagentPresentationByChildId: this.subagentPresentationByChildId, modelContextWindowsByModelKey: this.modelContextWindowsByModelKey, + onMaterializationMismatch: (diagnostic) => { + this.logger.warn( + { ...diagnostic, sessionId }, + "OpenCode final part snapshot replaced streamed content", + ); + }, }; this.childTranslationStates.set(sessionId, state); return state; diff --git a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts index d6a1d5578..200632cc6 100644 --- a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts +++ b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts @@ -1,6 +1,7 @@ import type { ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; import { mkdtempSync, rmSync } from "node:fs"; +import { createServer, type ServerResponse } from "node:http"; import net from "node:net"; import os from "node:os"; import path from "node:path"; @@ -27,6 +28,38 @@ afterEach(() => { }); describe("OpenCodeServerManager generations", () => { + test("shares one real SDK event stream across acquisitions until generation shutdown", async () => { + const responses: ServerResponse[] = []; + let requestCount = 0; + const upstream = createServer((_request, response) => { + requestCount += 1; + response.writeHead(200, { "content-type": "text/event-stream" }); + response.flushHeaders(); + responses.push(response); + }); + await new Promise((resolve) => upstream.listen(0, "127.0.0.1", resolve)); + const address = upstream.address(); + if (!address || typeof address === "string") throw new Error("Missing upstream address"); + const { manager } = createTestManager([address.port]); + + const first = await manager.acquireCurrent(); + const second = await manager.acquireCurrent(); + expect(first.events).toBe(second.events); + await vi.waitFor(() => expect(responses).toHaveLength(1)); + responses[0]?.write( + `data: ${JSON.stringify({ directory: "/workspace", payload: { type: "server.connected", properties: {} } })}\n\n`, + ); + await first.events.ready(); + expect(requestCount).toBe(1); + + await first.release(); + expect(requestCount).toBe(1); + await second.release(); + expect(requestCount).toBe(1); + await manager.shutdown(); + await new Promise((resolve) => upstream.close(() => resolve())); + }); + test("uses an explicit base environment for the server process", async () => { const baseEnv = { HOME: "/isolated/home", PATH: "/isolated/bin" }; const { manager, runtime } = createTestManager([4091], { baseEnv }); diff --git a/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts b/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts new file mode 100644 index 000000000..a6e572a7e --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts @@ -0,0 +1,405 @@ +import { createServer, type ServerResponse } from "node:http"; +import { createOpencodeClient, type OpencodeClient } from "@opencode-ai/sdk/v2/client"; +import { afterEach, describe, expect, test } from "vitest"; + +import { + OpenCodeEventConsumer, + type OpenCodeEventConsumerTiming, + type OpenCodeEventSourceInput, +} from "./event-consumer.js"; + +describe("OpenCodeEventConsumer", () => { + const cleanups: Array<() => Promise> = []; + afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())); + }); + + test("settles when close wins immediately before an injected backoff wait", async () => { + const upstream = await createSseUpstream(); + upstream.failNext(1); + let consumer!: OpenCodeEventConsumer; + let closePromise: Promise | null = null; + let observedAlreadyAborted = false; + const timing: OpenCodeEventConsumerTiming = { + arm: () => () => undefined, + get wait() { + closePromise = consumer.close(); + return async (_delayMs, signal) => { + observedAlreadyAborted = signal.aborted; + if (signal.aborted) throw signal.reason; + }; + }, + }; + consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(upstream.close); + + await eventually(() => expect(observedAlreadyAborted).toBe(true)); + await expect(closePromise).resolves.toBeUndefined(); + }); + + test("becomes ready on the first record and reconnects once after EOF", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const processExit = deferred(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: processExit.promise, + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + + await upstream.connected(1); + expect(await promiseState(consumer.ready())).toBe("pending"); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + expect(inputs).toEqual([connectedRecord("/one")]); + + upstream.end(0); + await timing.waiting(); + expect(timing.waitDelays).toEqual([100]); + timing.advanceWait(); + await upstream.connected(2); + upstream.send(1, connectedRecord("/two")); + await eventually(() => expect(inputs).toHaveLength(3)); + expect(inputs).toEqual([ + connectedRecord("/one"), + { type: "reconnected" }, + connectedRecord("/two"), + ]); + expect(upstream.requests).toHaveLength(2); + expect(upstream.requests.map((request) => request.url)).toEqual([ + "/global/event", + "/global/event", + ]); + }); + + test("reconnects a stalled response without publishing a terminal", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + await upstream.connected(1); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + + timing.expireWatchdog(); + expect(timing.watchdogDelays).toEqual([30_000, 30_000]); + await timing.waiting(); + timing.advanceWait(); + await upstream.connected(2); + upstream.send(1, connectedRecord("/two")); + await eventually(() => expect(inputs).toHaveLength(3)); + expect(inputs[1]).toEqual({ type: "reconnected" }); + }); + + test("publishes one terminal and stops reconnecting on process exit", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const processExit = deferred(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: processExit.promise, + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + await upstream.connected(1); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + + processExit.resolve(new Error("process exited")); + await eventually(() => expect(inputs).toHaveLength(2)); + expect(inputs[1]).toMatchObject({ type: "server-exited" }); + expect(upstream.requests).toHaveLength(1); + }); + + test("backs off repeated zero-record EOFs exponentially up to the cap", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + + for (let connection = 0; connection < 8; connection += 1) { + await upstream.connected(connection + 1); + upstream.end(connection); + await timing.waiting(); + timing.advanceWait(); + } + + expect(timing.waitDelays).toEqual([100, 200, 400, 800, 1_600, 3_200, 5_000, 5_000]); + }); + + test("isolates a throwing subscriber from the shared transport", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe(() => { + throw new Error("listener failed"); + }); + consumer.subscribe((input) => inputs.push(input)); + + await upstream.connected(1); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + expect(inputs).toEqual([connectedRecord("/one")]); + }); + + test("reconnects after a socket error with a delivered-record backoff reset", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + await upstream.connected(1); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + upstream.destroy(0); + await timing.waiting(); + expect(timing.waitDelays).toEqual([100]); + }); + + test("routes initial HTTP failures through consumer backoff before readiness", async () => { + const upstream = await createSseUpstream(); + upstream.failNext(2); + const timing = new ControlledTiming(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + + await timing.waiting(); + expect(upstream.requests).toHaveLength(1); + expect(await promiseState(consumer.ready())).toBe("pending"); + timing.advanceWait(); + await timing.waiting(); + expect(upstream.requests).toHaveLength(2); + timing.advanceWait(); + await upstream.connected(1); + upstream.send(0, connectedRecord("/ready")); + await consumer.ready(); + expect(timing.waitDelays).toEqual([100, 200]); + }); + + test("disables SDK SSE retries at the real request seam", async () => { + const upstream = await createSseUpstream(); + const realClient = createOpencodeClient({ baseUrl: upstream.url }); + let requestOptions: unknown; + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + createClient: () => + ({ + ...realClient, + global: { + ...realClient.global, + event: (options: unknown) => { + requestOptions = options; + return realClient.global.event(options as never); + }, + }, + }) as OpencodeClient, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + await upstream.connected(1); + + expect(requestOptions).toMatchObject({ sseMaxRetryAttempts: 0 }); + }); + + test("rejects readiness and publishes terminal when the process exits before a record", async () => { + const upstream = await createSseUpstream(); + const processExit = deferred(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: processExit.promise, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + await upstream.connected(1); + + processExit.resolve(new Error("process exited before ready")); + + await expect(consumer.ready()).rejects.toThrow("process exited before ready"); + expect(inputs).toEqual([expect.objectContaining({ type: "server-exited" })]); + }); + + test("intentional close does not publish a false terminal", async () => { + const upstream = await createSseUpstream(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + }); + cleanups.push(upstream.close); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + await upstream.connected(1); + upstream.send(0, connectedRecord("/one")); + await consumer.ready(); + + await consumer.close(); + + expect(inputs).toEqual([connectedRecord("/one")]); + }); +}); + +class ControlledTiming implements OpenCodeEventConsumerTiming { + private waits: Array<() => void> = []; + private watchdog: (() => void) | null = null; + readonly waitDelays: number[] = []; + readonly watchdogDelays: number[] = []; + arm(delayMs: number, callback: () => void): () => void { + this.watchdogDelays.push(delayMs); + this.watchdog = callback; + return () => { + if (this.watchdog === callback) this.watchdog = null; + }; + } + wait(delayMs: number, signal: AbortSignal): Promise { + this.waitDelays.push(delayMs); + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + this.waits.push(resolve); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + } + async waiting(): Promise { + await eventually(() => expect(this.waits).toHaveLength(1)); + } + advanceWait(): void { + this.waits.shift()?.(); + } + expireWatchdog(): void { + const watchdog = this.watchdog; + this.watchdog = null; + if (!watchdog) throw new Error("No watchdog is armed"); + watchdog(); + } +} + +function connectedRecord(directory: string) { + return { directory, payload: { type: "server.connected", properties: {} } }; +} + +async function createSseUpstream() { + const responses: ServerResponse[] = []; + const requests: Array<{ url: string | undefined }> = []; + let failuresRemaining = 0; + const server = createServer((request, response) => { + requests.push({ url: request.url }); + if (failuresRemaining > 0) { + failuresRemaining -= 1; + response.writeHead(503).end(); + return; + } + response.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-cache", + connection: "keep-alive", + }); + response.flushHeaders(); + responses.push(response); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test server address"); + return { + url: `http://127.0.0.1:${address.port}`, + requests, + connected: async (count: number) => eventually(() => expect(responses).toHaveLength(count)), + send(index: number, value: unknown) { + responses[index]?.write(`data: ${JSON.stringify(value)}\n\n`); + }, + end(index: number) { + responses[index]?.end(); + }, + destroy(index: number) { + responses[index]?.destroy(new Error("socket failed")); + }, + failNext(count: number) { + failuresRemaining = count; + }, + close: async () => new Promise((resolve) => server.close(() => resolve())), + }; +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((next) => { + resolve = next; + }); + return { promise, resolve }; +} + +async function promiseState(promise: Promise): Promise<"pending" | "settled"> { + return await Promise.race([ + promise.then(() => "settled" as const), + new Promise<"pending">((resolve) => setImmediate(() => resolve("pending"))), + ]); +} + +async function eventually(assertion: () => void): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + try { + assertion(); + return; + } catch { + await new Promise((resolve) => setImmediate(resolve)); + } + } + assertion(); +} diff --git a/packages/server/src/server/agent/providers/opencode/event-consumer.ts b/packages/server/src/server/agent/providers/opencode/event-consumer.ts new file mode 100644 index 000000000..cbce3d7a2 --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.ts @@ -0,0 +1,169 @@ +import { + createOpencodeClient, + type GlobalEvent, + type OpencodeClient, +} from "@opencode-ai/sdk/v2/client"; + +export type OpenCodeEventSourceInput = + | GlobalEvent + | { type: "reconnected" } + | { type: "server-exited"; error: Error }; + +export interface OpenCodeEventSource { + ready(): Promise; + subscribe(listener: (input: OpenCodeEventSourceInput) => void): () => void; +} + +export interface OpenCodeEventConsumerTiming { + arm(delayMs: number, callback: () => void): () => void; + wait(delayMs: number, signal: AbortSignal): Promise; +} + +export interface OpenCodeEventConsumerOptions { + serverUrl: string; + processExit: Promise; + createClient?: (baseUrl: string) => OpencodeClient; + timing?: OpenCodeEventConsumerTiming; +} + +const WATCHDOG_MS = 30_000; +const MAX_BACKOFF_MS = 5_000; + +const systemTiming: OpenCodeEventConsumerTiming = { + arm(delayMs, callback) { + const handle = setTimeout(callback, delayMs); + return () => clearTimeout(handle); + }, + wait(delayMs, signal) { + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const handle = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, delayMs); + const onAbort = () => { + clearTimeout(handle); + reject(signal.reason); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); + }, +}; + +export class OpenCodeEventConsumer implements OpenCodeEventSource { + private readonly listeners = new Set<(input: OpenCodeEventSourceInput) => void>(); + private readonly client: OpencodeClient; + private readonly timing: OpenCodeEventConsumerTiming; + private readonly readyPromise: Promise; + private resolveReady!: () => void; + private rejectReady!: (error: Error) => void; + private connectionAbort = new AbortController(); + private connectionTask: Promise; + private connected = false; + private closed = false; + + constructor(options: OpenCodeEventConsumerOptions) { + this.client = + options.createClient?.(options.serverUrl) ?? + createOpencodeClient({ baseUrl: options.serverUrl }); + this.timing = options.timing ?? systemTiming; + this.readyPromise = new Promise((resolve, reject) => { + this.resolveReady = resolve; + this.rejectReady = reject; + }); + void this.readyPromise.catch(() => undefined); + this.connectionTask = this.consume(options.processExit); + void this.connectionTask.catch(() => undefined); + } + + ready(): Promise { + return this.readyPromise; + } + + subscribe(listener: (input: OpenCodeEventSourceInput) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + async close(): Promise { + if (this.closed) return; + this.closed = true; + this.listeners.clear(); + const error = new Error("OpenCode event source closed"); + this.rejectReady(error); + this.connectionAbort.abort(error); + await this.connectionTask.catch(() => undefined); + } + + private async consume(processExit: Promise): Promise { + void processExit.then((error) => this.exit(error)); + let reconnectAttempt = 0; + while (!this.closed) { + const delivered = await this.consumeConnection(this.connectionAbort.signal).catch( + () => false, + ); + if (this.closed) return; + reconnectAttempt = delivered ? 0 : reconnectAttempt + 1; + const delayMs = Math.min(100 * 2 ** Math.max(0, reconnectAttempt - 1), MAX_BACKOFF_MS); + await this.timing.wait(delayMs, this.connectionAbort.signal).catch(() => undefined); + } + } + + private async consumeConnection(signal: AbortSignal): Promise { + const requestAbort = new AbortController(); + const abortRequest = () => requestAbort.abort(signal.reason); + signal.addEventListener("abort", abortRequest, { once: true }); + let cancelWatchdog: () => void = () => undefined; + let delivered = false; + try { + const result = await this.client.global.event({ + signal: requestAbort.signal, + sseMaxRetryAttempts: 0, + }); + const armWatchdog = () => { + cancelWatchdog(); + cancelWatchdog = this.timing.arm(WATCHDOG_MS, () => requestAbort.abort()); + }; + armWatchdog(); + for await (const event of result.stream) { + if (this.closed) return delivered; + armWatchdog(); + if (!delivered) { + delivered = true; + if (this.connected) this.publish({ type: "reconnected" }); + this.connected = true; + this.resolveReady(); + } + this.publish(event); + } + return delivered; + } finally { + cancelWatchdog(); + signal.removeEventListener("abort", abortRequest); + requestAbort.abort(); + } + } + + private exit(error: Error): void { + if (this.closed) return; + this.closed = true; + this.connectionAbort.abort(error); + if (!this.connected) this.rejectReady(error); + this.publish({ type: "server-exited", error }); + this.listeners.clear(); + } + + private publish(input: OpenCodeEventSourceInput): void { + for (const listener of this.listeners) { + try { + listener(input); + } catch { + // A session callback cannot tear down the generation-owned transport. + } + } + } +} + +export type OpenCodeEventConsumerFactory = ( + options: Pick, +) => OpenCodeEventConsumer; diff --git a/packages/server/src/server/agent/providers/opencode/event-translator.test.ts b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts index 2c5d6792c..50a3a9edf 100644 --- a/packages/server/src/server/agent/providers/opencode/event-translator.test.ts +++ b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts @@ -29,7 +29,7 @@ function createState(sessionId = "session-1"): OpenCodeEventTranslationState { sessionId, messageRoles: new Map(), accumulatedUsage: {}, - streamedPartKeys: new Set(), + materializedParts: new Map(), emittedStructuredMessageIds: new Set(), compactionSummaryMessageIds: new Set(), emittedCompactionPartIds: new Set(), @@ -38,6 +38,130 @@ function createState(sessionId = "session-1"): OpenCodeEventTranslationState { } describe("translateOpenCodeEvent", () => { + it("emits only the missing suffix from a final full text part", () => { + const state = createState(); + translateOpenCodeEvent( + { + type: "message.updated", + properties: { + info: { id: "message-1", sessionID: "session-1", role: "assistant" }, + }, + }, + state, + ); + const first = translateOpenCodeEvent( + { + type: "message.part.delta", + properties: { + sessionID: "session-1", + messageID: "message-1", + partID: "part-1", + field: "text", + delta: "Hello", + }, + }, + state, + ); + const incomplete = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "part-1", + sessionID: "session-1", + messageID: "message-1", + type: "text", + text: "", + time: { start: 1 }, + }, + }, + }, + state, + ); + const second = translateOpenCodeEvent( + { + type: "message.part.delta", + properties: { + sessionID: "session-1", + messageID: "message-1", + partID: "part-1", + field: "text", + delta: " world", + }, + }, + state, + ); + const final = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "part-1", + sessionID: "session-1", + messageID: "message-1", + type: "text", + text: "Hello world!", + time: { start: 1, end: 2 }, + }, + }, + }, + state, + ); + + expect(incomplete).toEqual([]); + expect([first, second, final].flatMap((events) => events)).toMatchObject([ + { item: { text: "Hello" } }, + { item: { text: " world" } }, + { item: { text: "!" } }, + ]); + expect( + translateOpenCodeEvent( + { + type: "message.part.delta", + properties: { + sessionID: "session-1", + messageID: "message-1", + partID: "part-1", + field: "text", + delta: " delayed", + }, + }, + state, + ), + ).toEqual([]); + }); + + it("diagnoses a non-prefix final text snapshot", () => { + const diagnostics: unknown[] = []; + const state = createState(); + state.onMaterializationMismatch = (diagnostic) => diagnostics.push(diagnostic); + state.materializedParts.set("part-1", { + messageId: "message-1", + emittedText: "streamed text", + closed: false, + }); + + expect( + translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "part-1", + sessionID: "session-1", + messageID: "message-1", + type: "text", + text: "mutated text", + time: { start: 1, end: 2 }, + }, + }, + }, + state, + ), + ).toEqual([]); + expect(diagnostics).toEqual([{ partId: "part-1", messageId: "message-1", kind: "text" }]); + }); + it("resolves context window max tokens from assistant message.updated model metadata", () => { const resolvedContextWindowMaxTokens: number[] = []; const state = createState(); @@ -1185,7 +1309,11 @@ describe("translateOpenCodeEvent", () => { it("emits turn_completed from session.status idle", () => { const state = createState(); - state.streamedPartKeys.add("text:part-1"); + state.materializedParts.set("part-1", { + messageId: "message-1", + emittedText: "partial", + closed: false, + }); state.partTypes.set("part-1", "text"); const result = translateOpenCodeEvent( @@ -1206,13 +1334,17 @@ describe("translateOpenCodeEvent", () => { usage: undefined, }, ]); - expect(state.streamedPartKeys.size).toBe(0); + expect(state.materializedParts.size).toBe(1); expect(state.partTypes.size).toBe(0); }); it("forwards session.status retry as a non-terminal timeline error item", () => { const state = createState(); - state.streamedPartKeys.add("text:part-1"); + state.materializedParts.set("part-1", { + messageId: "message-1", + emittedText: "partial", + closed: false, + }); state.partTypes.set("part-1", "text"); const result = translateOpenCodeEvent( @@ -1240,7 +1372,7 @@ describe("translateOpenCodeEvent", () => { ]); // Streaming state must NOT be reset — the turn is still alive, opencode // will eventually either succeed or emit session.idle / session.error. - expect(state.streamedPartKeys.size).toBe(1); + expect(state.materializedParts.size).toBe(1); expect(state.partTypes.size).toBe(1); }); diff --git a/packages/server/src/server/agent/providers/opencode/server-manager.ts b/packages/server/src/server/agent/providers/opencode/server-manager.ts index 0c850514d..f19ba54d5 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -15,12 +15,18 @@ import { type ProviderRuntimeSettings, } from "../../provider-launch-config.js"; import { resolveOpenCodeHomeDir } from "./paths.js"; +import { + OpenCodeEventConsumer, + type OpenCodeEventConsumerFactory, + type OpenCodeEventSource, +} from "./event-consumer.js"; const OPENCODE_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 5_000; const OPENCODE_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; export interface OpenCodeServerAcquisition { server: { port: number; url: string }; + events: OpenCodeEventSource; release: () => Promise; } @@ -39,6 +45,7 @@ export interface OpenCodeServerGeneration { refCount: number; retired: boolean; ready: Promise; + events: OpenCodeEventConsumer; managedProcessId?: string; managedProcessRecord?: Promise<{ id: string } | null>; } @@ -61,6 +68,7 @@ export interface OpenCodeServerManagerOptions { resolveCommandPrefix?: OpenCodeCommandPrefixResolver; resolveHomeDir?: () => string; spawnServerProcess?: OpenCodeServerProcessSpawner; + createEventSource?: OpenCodeEventConsumerFactory; } export class OpenCodeServerManager implements OpenCodeServerManagerLike { @@ -80,6 +88,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { private readonly resolveCommandPrefix: OpenCodeCommandPrefixResolver; private readonly resolveHomeDir: () => string; private readonly spawnServerProcess: OpenCodeServerProcessSpawner; + private readonly createEventSource: OpenCodeEventConsumerFactory; constructor(options: OpenCodeServerManagerOptions) { this.logger = options.logger; @@ -94,6 +103,8 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { (() => resolveProviderCommandPrefix(this.runtimeSettings?.command, resolveOpenCodeBinary)); this.resolveHomeDir = options.resolveHomeDir ?? resolveOpenCodeHomeDir; this.spawnServerProcess = options.spawnServerProcess ?? spawnProcess; + this.createEventSource = + options.createEventSource ?? ((input) => new OpenCodeEventConsumer(input)); } static getInstance( @@ -191,6 +202,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { let releasePromise: Promise | null = null; return { server: { port: server.port, url: server.url }, + events: server.events, release: async () => { if (releasePromise) { return releasePromise; @@ -316,6 +328,10 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { args: serverArgs, port, }); + let resolveProcessExit!: (error: Error) => void; + const processExit = new Promise((resolve) => { + resolveProcessExit = resolve; + }); const server: OpenCodeServerGeneration = { process: serverProcess, port, @@ -323,6 +339,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { refCount: 0, retired: false, ready: Promise.resolve(), + events: this.createEventSource({ serverUrl: url, processExit }), managedProcessRecord, }; void managedProcessRecord.then((record) => { @@ -396,6 +413,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { }); serverProcess.on("exit", (code) => { + resolveProcessExit(new Error(`OpenCode server exited with code ${code}`)); this.removeManagedServerRecord(server); if (!started) { failStartup( @@ -447,6 +465,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } private async killServer(server: OpenCodeServerGeneration): Promise { + await server.events.close(); if ( (server.process.exitCode !== null && server.process.exitCode !== undefined) || (server.process.signalCode !== null && server.process.signalCode !== undefined) diff --git a/packages/server/src/server/agent/providers/opencode/test-server-manager.ts b/packages/server/src/server/agent/providers/opencode/test-server-manager.ts index e4cc8a65e..fc1748dd2 100644 --- a/packages/server/src/server/agent/providers/opencode/test-server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/test-server-manager.ts @@ -1,5 +1,10 @@ import type { OpenCodeServerAcquisition, OpenCodeServerManagerLike } from "./server-manager.js"; +const events = { + ready: async () => undefined, + subscribe: () => () => undefined, +}; + export interface TestOpenCodeServerAcquisition { kind: "current" | "new" | "dedicated" | "existing"; env?: Record; @@ -41,6 +46,7 @@ export class TestOpenCodeServerManager implements OpenCodeServerManagerLike { this.acquisitions.push(acquisition); return { server: this.server, + events, release: async () => { acquisition.released = true; }, diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts index f17ba9da6..1ded611f7 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts @@ -1,5 +1,6 @@ import type { OpencodeClient } from "@opencode-ai/sdk/v2/client"; +import type { OpenCodeEventSourceInput } from "../event-consumer.js"; import type { OpenCodeServerAcquisition, OpenCodeServerManagerLike } from "../server-manager.js"; interface OpenCodeResponse { @@ -16,10 +17,21 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike { }> = []; readonly clientCreations: Array<{ baseUrl: string; directory: string }> = []; private readonly clients: TestOpenCodeClient[] = []; + private readonly eventListeners = new Set<(input: OpenCodeEventSourceInput) => void>(); + readonly events = { + ready: async () => undefined, + subscribe: (listener: (input: OpenCodeEventSourceInput) => void) => { + this.eventListeners.add(listener); + return () => this.eventListeners.delete(listener); + }, + }; server = { port: 1234, url: "http://127.0.0.1:1234" }; enqueueClient(client: TestOpenCodeClient): void { + client.observeEvents((event) => { + for (const listener of this.eventListeners) listener(event as OpenCodeEventSourceInput); + }); this.clients.push(client); } @@ -53,6 +65,7 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike { this.acquisitions.push(acquisition); return { server: this.server, + events: this.events, release: async () => { acquisition.releaseCount += 1; }, @@ -79,9 +92,11 @@ export class TestOpenCodeClient { mcpAdd: [] as unknown[], mcpConnect: [] as unknown[], permissionReply: [] as unknown[], + permissionList: [] as unknown[], providerList: [] as unknown[], providerListOptions: [] as unknown[], questionReject: [] as unknown[], + questionList: [] as unknown[], questionReply: [] as unknown[], sessionAbort: [] as unknown[], sessionCommand: [] as unknown[], @@ -106,6 +121,10 @@ export class TestOpenCodeClient { mcpAddResponse: OpenCodeResponse = {}; mcpConnectResponse: OpenCodeResponse = {}; permissionReplyResponse: OpenCodeResponse = {}; + permissionListResponse: OpenCodeResponse = { data: [] }; + permissionListImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } }; providerListImplementation: | ((parameters: unknown, options: unknown) => Promise) @@ -114,6 +133,10 @@ export class TestOpenCodeClient { | ((options: unknown) => Promise<{ stream: AsyncIterable }>) | null = null; questionRejectResponse: OpenCodeResponse = {}; + questionListResponse: OpenCodeResponse = { data: [] }; + questionListImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; questionReplyResponse: OpenCodeResponse = {}; sessionAbortResponse: OpenCodeResponse = {}; sessionAbortImplementation: ((parameters: unknown) => Promise) | null = null; @@ -123,19 +146,29 @@ export class TestOpenCodeClient { sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } }; sessionDeleteResponse: OpenCodeResponse = {}; sessionChildrenResponses: OpenCodeResponse[] = []; - sessionChildrenImplementation: ((parameters: unknown) => Promise) | null = null; + sessionChildrenImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; sessionGetResponse: OpenCodeResponse = { data: { id: "session-1", directory: "/workspace/repo", title: null }, }; sessionMessagesResponse: OpenCodeResponse = { data: [] }; + sessionMessagesImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; + sessionPromptAsyncImplementation: ((parameters: unknown) => Promise) | null = + null; sessionPromptAsyncEvents: unknown[] = [idleEvent()]; sessionPromptAsyncResponse: OpenCodeResponse = {}; sessionStatusResponse: OpenCodeResponse = { data: {} }; - sessionStatusImplementation: ((parameters: unknown) => Promise) | null = null; + sessionStatusImplementation: + | ((parameters: unknown, options: unknown) => Promise) + | null = null; sessionSummarizeEvents: unknown[] = [idleEvent()]; sessionSummarizeResponse: OpenCodeResponse = { data: {} }; sessionUpdateResponse: OpenCodeResponse = {}; private readonly queuedEventStream = createQueuedEventStream(); + private eventObserver: ((event: unknown) => void) | null = null; constructor() { this.eventStream = this.queuedEventStream.stream; @@ -143,6 +176,11 @@ export class TestOpenCodeClient { emitEvent(event: unknown): void { this.queuedEventStream.emit(event); + this.eventObserver?.(event); + } + + observeEvents(observer: (event: unknown) => void): void { + this.eventObserver = observer; } asSdkClient(): OpencodeClient { @@ -200,6 +238,12 @@ export class TestOpenCodeClient { }, }, permission: { + list: async (parameters: unknown, options: unknown) => { + this.calls.permissionList.push(parameters); + return this.permissionListImplementation + ? await this.permissionListImplementation(parameters, options) + : this.permissionListResponse; + }, reply: async (parameters: unknown) => { this.calls.permissionReply.push(parameters); return this.permissionReplyResponse; @@ -215,6 +259,12 @@ export class TestOpenCodeClient { }, }, question: { + list: async (parameters: unknown, options: unknown) => { + this.calls.questionList.push(parameters); + return this.questionListImplementation + ? await this.questionListImplementation(parameters, options) + : this.questionListResponse; + }, reject: async (parameters: unknown) => { this.calls.questionReject.push(parameters); return this.questionRejectResponse; @@ -249,10 +299,10 @@ export class TestOpenCodeClient { this.calls.sessionDelete.push(parameters); return this.sessionDeleteResponse; }, - children: async (parameters: unknown) => { + children: async (parameters: unknown, options: unknown) => { this.calls.sessionChildren.push(parameters); if (this.sessionChildrenImplementation) { - return await this.sessionChildrenImplementation(parameters); + return await this.sessionChildrenImplementation(parameters, options); } return this.sessionChildrenResponses.shift() ?? { data: [] }; }, @@ -260,21 +310,26 @@ export class TestOpenCodeClient { this.calls.sessionGet.push(parameters); return this.sessionGetResponse; }, - messages: async (parameters: unknown) => { + messages: async (parameters: unknown, options: unknown) => { this.calls.sessionMessages.push(parameters); - return this.sessionMessagesResponse; + return this.sessionMessagesImplementation + ? await this.sessionMessagesImplementation(parameters, options) + : this.sessionMessagesResponse; }, promptAsync: async (parameters: unknown) => { this.calls.sessionPromptAsync.push(parameters); + if (this.sessionPromptAsyncImplementation) { + return await this.sessionPromptAsyncImplementation(parameters); + } for (const event of this.sessionPromptAsyncEvents) { this.emitEvent(event); } return this.sessionPromptAsyncResponse; }, - status: async (parameters: unknown) => { + status: async (parameters: unknown, options: unknown) => { this.calls.sessionStatus.push(parameters); return this.sessionStatusImplementation - ? await this.sessionStatusImplementation(parameters) + ? await this.sessionStatusImplementation(parameters, options) : this.sessionStatusResponse; }, summarize: async (parameters: unknown) => {