diff --git a/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts b/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts index 5ce99a5d8b..fd3ad3c25e 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts @@ -7,6 +7,60 @@ import { OpenCodeV2AgentClient } from "./agent.js"; import { V2Harness } from "../test-utils/v2-harness.js"; describe("OpenCode v2 session lifecycle", () => { + test("reconnects after a helper exits and restores session configuration on the next turn", async () => { + const first = new V2Harness(); + const second = new V2Harness(); + let exit!: (error: Error) => void; + const exited = new Promise((resolve) => { + exit = resolve; + }); + let acquisitions = 0; + const client = new OpenCodeV2AgentClient({ + logger: createTestLogger(), + runtime: { + acquire: async () => { + acquisitions += 1; + return acquisitions === 1 ? { ...first.connection, exited } : second.connection; + }, + shutdown: async () => undefined, + }, + }); + first.wait = () => new Promise(() => undefined); + second.prompt = async (input) => { + second.history.push({ + id: "answer", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "model" }, + time: { created: 2 }, + content: [{ type: "text", text: "recovered" }], + }); + second.prompts.push(input.text); + }; + const session = await client.createSession( + { + provider: "opencode", + cwd: "/tmp/project", + mcpServers: { tools: { type: "stdio", command: "tools" } }, + }, + { env: { TOKEN: "test" } }, + ); + const running = session.run("before exit"); + await expect.poll(() => first.prompts).toEqual(["before exit"]); + exit(new Error("helper exited")); + await expect(running).rejects.toThrow("helper exited"); + expect(acquisitions).toBe(1); + try { + expect((await session.run("after exit")).finalText).toBe("recovered"); + expect(second.prompts).toEqual(["after exit"]); + expect(second.mcpAdds).toEqual(["tools"]); + expect(second.environments).toEqual([{ sessionID: "session", variables: { TOKEN: "test" } }]); + expect(first.releases).toBe(1); + } finally { + await session.close(); + } + await expect(session.startTurn("after close")).rejects.toThrow("OpenCode session is closed"); + }); test("fails initialization when the event stream ends before connecting", async () => { const harness = new V2Harness(); harness.api.event.subscribe = async function* () {}; diff --git a/packages/server/src/server/agent/providers/opencode/v2/agent.ts b/packages/server/src/server/agent/providers/opencode/v2/agent.ts index ef268b3578..34ed21bd6c 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/agent.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/agent.ts @@ -155,6 +155,11 @@ export class OpenCodeV2AgentClient implements AgentClient { launch: AgentLaunchContext | undefined, persist: boolean, ) { + const acquire = () => + this.runtime.acquire( + requiresDedicatedV2Server(config, launch) ? { env: launch?.env, dedicated: true } : {}, + ); + let ownedConnection = connection; const unbind = this.options.bridge?.bindSession({ sessionId: info.id, env: launch?.env ?? {}, @@ -162,7 +167,7 @@ export class OpenCodeV2AgentClient implements AgentClient { }); const bound = new Map void>(); const bindChild = (childId: string) => { - this.connections.set(childId, connection); + this.connections.set(childId, ownedConnection); if (bound.has(childId)) return; const childUnbind = this.options.bridge?.bindSession({ sessionId: childId, @@ -176,7 +181,12 @@ export class OpenCodeV2AgentClient implements AgentClient { unbind?.(); for (const cleanup of bound.values()) cleanup(); for (const [id, owner] of this.connections) - if (owner === connection) this.connections.delete(id); + if (owner === ownedConnection) this.connections.delete(id); + }; + const moved = (next: V2Connection) => { + for (const [id, owner] of this.connections) + if (owner === ownedConnection) this.connections.set(id, next); + ownedConnection = next; }; const session = new OpenCodeV2Session( connection, @@ -187,6 +197,8 @@ export class OpenCodeV2AgentClient implements AgentClient { Boolean(this.options.bridge), releaseBindings, bindChild, + acquire, + moved, ); try { if (this.options.bridge) { diff --git a/packages/server/src/server/agent/providers/opencode/v2/children.ts b/packages/server/src/server/agent/providers/opencode/v2/children.ts index bbb5fe7ed8..95853b71c7 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/children.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/children.ts @@ -10,7 +10,7 @@ import { messages } from "./history.js"; import type { V2Api } from "./api.js"; interface ChildrenOptions { - client: V2Api; + client(): V2Api; id: string; emit(event: AgentStreamEvent): void; reconcilePermissions(id: string): Promise; @@ -34,14 +34,14 @@ export class SessionChildren { this.children.has(event.data.sessionID) ) { await this.reconcileChild( - await this.options.client.session.get({ sessionID: event.data.sessionID }), + await this.options.client().session.get({ sessionID: event.data.sessionID }), ); } } async reconcile(parentID: string) { let cursor: string | undefined; do { - const page = await this.options.client.session.list({ + const page = await this.options.client().session.list({ ...(cursor ? { cursor } : { parentID }), limit: 100, }); @@ -60,7 +60,7 @@ export class SessionChildren { this.children.set(info.id, timeline); this.options.bindChild?.(info.id); } - const active = await this.options.client.session.active(); + const active = await this.options.client().session.active(); let status: "running" | "failed" | "canceled" | "completed" = "completed"; if (active[info.id]) status = "running"; else if (info.outcome === "failed") status = "failed"; @@ -79,7 +79,7 @@ export class SessionChildren { this.childStates.set(info.id, signature); this.options.emit({ type: "provider_subagent", provider: "opencode", event: presentation }); } - for (const event of timeline.messages(await messages(this.options.client, info.id))) { + for (const event of timeline.messages(await messages(this.options.client(), info.id))) { if (event.type === "timeline") this.options.emit({ type: "provider_subagent", diff --git a/packages/server/src/server/agent/providers/opencode/v2/permissions.ts b/packages/server/src/server/agent/providers/opencode/v2/permissions.ts index 12965a6d7e..fcd0213282 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/permissions.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/permissions.ts @@ -13,7 +13,7 @@ export class SessionPermissions { private readonly permissionOwners = new Map(); private readonly forms = new Map(); constructor( - private readonly client: V2Api, + private readonly getClient: () => V2Api, private readonly id: string, private readonly config: AgentSessionConfig, private readonly emit: (event: AgentStreamEvent) => void, @@ -37,7 +37,7 @@ export class SessionPermissions { const form = this.forms.get(requestId); if (form) { if (response.behavior === "deny") - await this.client.session.form.cancel({ sessionID: form.sessionID, formID: form.id }); + await this.getClient().session.form.cancel({ sessionID: form.sessionID, formID: form.id }); else { const raw = response.updatedInput?.answers; const answer: Record = {}; @@ -49,7 +49,7 @@ export class SessionPermissions { const normalized = formAnswer(field, value); if (normalized !== undefined) answer[field.key] = normalized; } - await this.client.session.form.reply({ + await this.getClient().session.form.reply({ sessionID: form.sessionID, formID: form.id, answer, @@ -57,7 +57,7 @@ export class SessionPermissions { } this.forms.delete(requestId); } else { - await this.client.permission.reply({ + await this.getClient().permission.reply({ sessionID: this.permissionOwners.get(requestId) ?? this.id, requestID: requestId, decision: permissionReply(response), @@ -67,8 +67,8 @@ export class SessionPermissions { } async reconcile(sessionID: string) { const [permissions, forms] = await Promise.all([ - this.client.permission.list({ sessionID }), - this.client.session.form.list({ sessionID }), + this.getClient().permission.list({ sessionID }), + this.getClient().session.form.list({ sessionID }), ]); for (const request of permissions) { if (this.pending.has(request.id)) continue; diff --git a/packages/server/src/server/agent/providers/opencode/v2/runtime.ts b/packages/server/src/server/agent/providers/opencode/v2/runtime.ts index 6f659cb44a..cab5544582 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/runtime.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/runtime.ts @@ -130,6 +130,14 @@ export class V2Runtime { private startTracked(env?: Record): Promise { const pending = this.start(env); this.starts.add(pending); + void pending.then( + (generation) => + generation.exited.then(() => { + if (this.current === pending) this.current = null; + return undefined; + }), + () => undefined, + ); void pending.then( () => this.starts.delete(pending), () => this.starts.delete(pending), diff --git a/packages/server/src/server/agent/providers/opencode/v2/session.ts b/packages/server/src/server/agent/providers/opencode/v2/session.ts index 4df28f3875..27ed2974c4 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/session.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/session.ts @@ -46,6 +46,10 @@ export class OpenCodeV2Session implements AgentSession { private readonly turns: SessionTurns; private readonly children: SessionChildren; private stream: Promise | null = null; + private streamAbort = new AbortController(); + private exited = false; + private reconnecting: Promise | null = null; + private launchEnv: Record | undefined; private sync: Promise = Promise.resolve(); private dirty = false; private refreshTimer: ReturnType | null = null; @@ -53,7 +57,7 @@ export class OpenCodeV2Session implements AgentSession { private history: SessionMessageInfo[] = []; private modes: AgentMode[] = []; constructor( - private readonly connection: V2Connection, + private connection: V2Connection, private info: SessionInfo, private readonly config: AgentSessionConfig, private readonly logger: Logger, @@ -61,19 +65,24 @@ export class OpenCodeV2Session implements AgentSession { private readonly requiresPaseoPlugin: boolean, private readonly unbind?: () => void, bindChild?: (id: string) => void, + private readonly acquire?: () => Promise, + private readonly moved?: (connection: V2Connection) => void, ) { - this.permissions = new SessionPermissions(connection.client, info.id, config, (event) => - this.emit(event), + this.permissions = new SessionPermissions( + () => this.client, + info.id, + config, + (event) => this.emit(event), ); this.children = new SessionChildren({ - client: connection.client, + client: () => this.client, id: info.id, emit: (event) => this.emit(event), reconcilePermissions: (id) => this.permissions.reconcile(id), bindChild, }); this.turns = new SessionTurns({ - client: connection.client, + client: () => this.client, id: info.id, cwd: config.cwd, signal: this.abort.signal, @@ -98,18 +107,31 @@ export class OpenCodeV2Session implements AgentSession { return this.connection.client; } async initialize(launch?: AgentLaunchContext) { - void this.connection.exited.then((error) => { - if (this.closed) return; - this.abort.abort(error); + this.launchEnv = launch?.env; + this.watchExit(this.connection); + await this.configureConnection(); + const location = { directory: this.config.cwd }; + this.modes = modesFromV2((await this.client.agent.list({ location })).data); + this.timeline.messages(await messages(this.client, this.id)); + await this.startStream(); + await this.children.reconcile(this.id); + } + private watchExit(connection: V2Connection) { + void connection.exited.then((error) => { + if (this.closed || this.connection !== connection) return undefined; + this.exited = true; + this.streamAbort.abort(error); this.turns.fail(error); return undefined; }); + } + private async configureConnection() { const location = { directory: this.config.cwd }; await waitForLocationReady({ client: this.client, location, signal: this.abort.signal }); if (this.requiresPaseoPlugin) await awaitPaseoPlugin({ client: this.client, location, signal: this.abort.signal }); - if (launch?.env) - await this.client.session.environment({ sessionID: this.id, variables: launch.env }); + if (this.launchEnv) + await this.client.session.environment({ sessionID: this.id, variables: this.launchEnv }); for (const [server, config] of Object.entries(this.config.mcpServers ?? {})) { await this.client.mcp.add({ server, @@ -142,8 +164,8 @@ export class OpenCodeV2Session implements AgentSession { key: "paseo", value: system, }); - this.modes = modesFromV2((await this.client.agent.list({ location })).data); - this.timeline.messages(await messages(this.client, this.id)); + } + private async startStream() { let ready!: () => void; let fail!: (error: unknown) => void; const first = new Promise((resolve, reject) => { @@ -155,7 +177,41 @@ export class OpenCodeV2Session implements AgentSession { AbortSignal.any([this.abort.signal, AbortSignal.timeout(30_000)]), first, ); - await this.children.reconcile(this.id); + } + private async reconnectIfExited() { + if (this.closed) throw new Error("OpenCode session is closed"); + if (!this.exited) return; + this.reconnecting ??= this.reconnect().finally(() => { + this.reconnecting = null; + }); + await this.reconnecting; + } + private async reconnect() { + if (!this.acquire) throw new Error("OpenCode helper server exited"); + await this.stream; + await this.sync.catch(() => undefined); + const next = await this.acquire(); + if (this.closed) { + await next.release(); + throw new Error("OpenCode session is closed"); + } + const old = this.connection; + this.connection = next; + this.streamAbort = new AbortController(); + this.exited = false; + this.watchExit(next); + try { + await this.configureConnection(); + await this.startStream(); + } catch (error) { + this.connection = old; + this.exited = true; + this.streamAbort.abort(); + await next.release(); + throw error; + } + this.moved?.(next); + await old.release(); } subscribe(callback: (event: AgentStreamEvent) => void) { this.listeners.add(callback); @@ -186,10 +242,12 @@ export class OpenCodeV2Session implements AgentSession { this.emit({ ...event, ...(this.turns.id ? { turnId: this.turns.id } : {}) }); } async *streamHistory() { + await this.reconnectIfExited(); const history = new V2Timeline(false); yield* history.messages(await messages(this.client, this.id)); } async getRuntimeInfo() { + await this.reconnectIfExited(); this.info = await this.client.session.get({ sessionID: this.id }); return { provider: "opencode", @@ -206,6 +264,7 @@ export class OpenCodeV2Session implements AgentSession { return this.info.agent ?? null; } async setMode(modeId: string) { + await this.reconnectIfExited(); await this.client.session.switchAgent({ sessionID: this.id, agent: modeId }); this.info.agent = modeId; this.config.modeId = modeId; @@ -217,6 +276,7 @@ export class OpenCodeV2Session implements AgentSession { }); } async setModel(model: string | null) { + await this.reconnectIfExited(); const selected = model ? modelRef(model, this.config.thinkingOptionId) : (await this.client.model.default({ location: { directory: this.config.cwd } })).data; @@ -237,6 +297,7 @@ export class OpenCodeV2Session implements AgentSession { }); } async setThinkingOption(variant: string | null) { + await this.reconnectIfExited(); const model = this.info.model; if (!model) throw new Error("Select an OpenCode model before changing its variant"); await this.client.session.switchModel({ @@ -257,6 +318,7 @@ export class OpenCodeV2Session implements AgentSession { await this.respondToPermission(request.id, { behavior: "allow" }); } async listCommands() { + await this.reconnectIfExited(); return commands(this.client, this.config.cwd); } describePersistence(): AgentPersistenceHandle { @@ -278,7 +340,8 @@ export class OpenCodeV2Session implements AgentSession { item.type === "assistant_message" ? current + item.text : current, }); } - startTurn(prompt: AgentPromptInput, options?: AgentRunOptions) { + async startTurn(prompt: AgentPromptInput, options?: AgentRunOptions) { + await this.reconnectIfExited(); return this.turns.startTurn(prompt, options); } steerActiveTurn(prompt: AgentPromptInput, options: SteerActiveTurnOptions): Promise { @@ -289,6 +352,7 @@ export class OpenCodeV2Session implements AgentSession { } async revertBoth(input: { messageId: string }) { await this.interrupt(); + await this.reconnectIfExited(); await this.client.session.revert.stage({ sessionID: this.id, messageID: input.messageId, @@ -300,6 +364,7 @@ export class OpenCodeV2Session implements AgentSession { return [...this.permissions.list()]; } async respondToPermission(requestId: string, response: AgentPermissionResponse) { + await this.reconnectIfExited(); await this.permissions.respondToPermission(requestId, response); } private async reconcileSnapshot() { @@ -377,11 +442,12 @@ export class OpenCodeV2Session implements AgentSession { this.turns.observeActiveTurn(); } private async consume(ready: () => void, fail: (error: unknown) => void) { + const signal = AbortSignal.any([this.abort.signal, this.streamAbort.signal]); let connected = false; - while (!this.closed && !this.abort.signal.aborted) { + while (!this.closed && !signal.aborted) { try { - for await (const event of this.client.event.subscribe({ signal: this.abort.signal })) { - if (this.closed || this.abort.signal.aborted) return; + for await (const event of this.client.event.subscribe({ signal })) { + if (this.closed || signal.aborted) return; if (event.type === "server.connected") { this.timeline.resetStreams(); await this.reconcileConnection(); @@ -393,7 +459,7 @@ export class OpenCodeV2Session implements AgentSession { } if (!connected) throw new Error("OpenCode event stream ended before connecting"); } catch (error) { - if (this.closed || this.abort.signal.aborted) return; + if (this.closed || signal.aborted) return; if (!connected) { fail(error); return; @@ -403,15 +469,17 @@ export class OpenCodeV2Session implements AgentSession { "OpenCode event stream interrupted; reconciling on reconnect", ); } - await delay(100, undefined, { signal: this.abort.signal }).catch(() => undefined); + await delay(100, undefined, { signal }).catch(() => undefined); } } async close() { if (this.closed) return; this.closed = true; this.abort.abort(); + this.streamAbort.abort(); if (this.refreshTimer) clearTimeout(this.refreshTimer); await this.stream; + await this.reconnecting?.catch(() => undefined); await this.sync.catch(() => undefined); this.unbind?.(); try { diff --git a/packages/server/src/server/agent/providers/opencode/v2/turns.ts b/packages/server/src/server/agent/providers/opencode/v2/turns.ts index cc5872e10d..5b4611b8f8 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/turns.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/turns.ts @@ -25,7 +25,7 @@ interface TurnSnapshot { history: SessionMessageInfo[]; } interface TurnOptions { - client: V2Api; + client(): V2Api; id: string; cwd: string; signal: AbortSignal; @@ -115,16 +115,16 @@ export class SessionTurns { ) { const command = input.text.match(/^\/([^\s]+)(?:\s+([\s\S]*))?$/); if (command?.[1] === "compact" || command?.[1] === "summarize") { - await this.options.client.session.compact({ sessionID: this.options.id }); + await this.options.client().session.compact({ sessionID: this.options.id }); return; } const selected = command - ? (await commands(this.options.client, this.options.cwd)).find( + ? (await commands(this.options.client(), this.options.cwd)).find( (item) => item.name === command[1], ) : undefined; if (selected && selected.kind !== "skill") { - await this.options.client.session.command({ + await this.options.client().session.command({ sessionID: this.options.id, name: selected.name, text: command?.[2] ?? "", @@ -132,7 +132,7 @@ export class SessionTurns { }); return; } - await this.options.client.session.prompt({ + await this.options.client().session.prompt({ sessionID: this.options.id, ...input, ...(selected?.kind === "skill" @@ -148,10 +148,9 @@ export class SessionTurns { }); } private async finish(id: string) { - await this.options.client.session.wait( - { sessionID: this.options.id }, - { signal: this.options.signal }, - ); + await this.options + .client() + .session.wait({ sessionID: this.options.id }, { signal: this.options.signal }); const { info, history } = await this.options.reconcile(); if (this.turn?.id !== id) return; if (info.outcome !== "failed" && info.outcome !== "interrupted") @@ -184,10 +183,12 @@ export class SessionTurns { } private async readExecutionError(): Promise { let message = "OpenCode execution failed"; - for await (const event of this.options.client.session.log( - { sessionID: this.options.id, follow: false }, - { signal: this.options.signal }, - )) { + for await (const event of this.options + .client() + .session.log( + { sessionID: this.options.id, follow: false }, + { signal: this.options.signal }, + )) { if (event.type === "session.execution.failed") message = event.data.error.message; } return message; @@ -209,7 +210,7 @@ export class SessionTurns { options: SteerActiveTurnOptions, ): Promise { if (this.turn?.id !== options.expectedTurnId || this.stopping) return { status: "unavailable" }; - await this.options.client.session.prompt({ + await this.options.client().session.prompt({ sessionID: this.options.id, ...this.promptInput(prompt), delivery: "steer", @@ -227,7 +228,7 @@ export class SessionTurns { const stop = (async () => { // A stop sent before the queued prompt is accepted would interrupt an idle session. await turn?.submitted; - await this.options.client.session.interrupt({ sessionID: this.options.id }); + await this.options.client().session.interrupt({ sessionID: this.options.id }); await turn?.completion; })(); this.stopping = stop;