From 9a0c53d79056b2ce143cf6746df00f7331bf4afc Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 16 Aug 2026 11:05:07 +0200 Subject: [PATCH] fix(opencode): recover turns after event stream drops (#3395) * fix(opencode): recover turns after event stream drops Own the global event transport for the lifetime of the OpenCode server generation instead of coupling it to individual turns. Reconcile scoped provider state after reconnect so transient EOF, errors, and stalls do not falsely fail healthy foreground or delegated work. * fix(opencode): settle recovery lifecycle edges Retry only failed post-gap snapshots, preserve terminal child evidence when status is unavailable, and treat provider process exit as an authoritative stop boundary. * fix(opencode): keep gap repair snapshot-scoped Refresh child and blocking-request state once per reconnect, retry only rejected root snapshot reads, and settle a successful snapshot without the active dispatch instead of polling transcripts. * fix(opencode): keep live ingress ahead of repair retries Schedule failed snapshot retries back onto the serialized session ingress so newly arrived live terminal events run first. Keep retry backoff scoped to root reads and invalidate stale retries across later reconnects. * fix(opencode): invalidate stale gap repairs between turns A delayed recovery retry could inspect a replacement turn after the original turn completed. Turn creation now invalidates older repair work, and the slash-command timeout regression drives the generation-owned event source used by sessions. * fix(opencode): bind gap repair to its turn A queued or in-flight repair could resume after its original turn ended and read a replacement turn's mutable dispatch state. Recovery now captures its boundary and revalidates the repair revision after provider awaits and while materializing snapshots. * fix(opencode): preserve child status on failed recovery --- .../providers/opencode-agent.real.e2e.test.ts | 515 +++++++++ .../providers/opencode-agent.recovery.test.ts | 457 ++++++++ ...encode-agent.slash-command-timeout.test.ts | 35 +- .../agent/providers/opencode-agent.test.ts | 1005 +++++++++++++++-- .../server/agent/providers/opencode-agent.ts | 786 ++++++++----- .../providers/opencode-server-manager.test.ts | 33 + .../providers/opencode/event-consumer.test.ts | 405 +++++++ .../providers/opencode/event-consumer.ts | 169 +++ .../opencode/event-translator.test.ts | 142 ++- .../providers/opencode/server-manager.ts | 19 + .../providers/opencode/test-server-manager.ts | 6 + .../test-utils/test-opencode-harness.ts | 71 +- 12 files changed, 3275 insertions(+), 368 deletions(-) create mode 100644 packages/server/src/server/agent/providers/opencode-agent.real.e2e.test.ts create mode 100644 packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts create mode 100644 packages/server/src/server/agent/providers/opencode/event-consumer.test.ts create mode 100644 packages/server/src/server/agent/providers/opencode/event-consumer.ts 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) => {