refactor(daemon): share a single opencode runtime across instances (#29)

Author: Aleexc12

packages/daemon/src/__tests__/daemon.test.ts

@@ -451,6 +451,85 @@ describe("Daemon streaming", () => {
 		expect(job.outputText).toBe("reply:plain");
 	});
 
+	test("two instances stream independently without cross-leak", async () => {
+		registry.streamingDeltas = ["x", "y", "z"];
+		registry.deltaIntervalMs = 20;
+
+		const alpha = await daemon.handleRequest(
+			req({
+				id: "instance-alpha",
+				method: "instance.create",
+				params: { name: "Alpha", directory: "/tmp/alpha" },
+			}),
+		);
+		const beta = await daemon.handleRequest(
+			req({
+				id: "instance-beta",
+				method: "instance.create",
+				params: { name: "Beta", directory: "/tmp/beta" },
+			}),
+		);
+		const alphaInstance = expectSuccess(alpha, "instance.create").instance;
+		const betaInstance = expectSuccess(beta, "instance.create").instance;
+
+		const submitAlpha = await daemon.handleRequest(
+			req({
+				id: "submit-alpha",
+				method: "job.submit",
+				params: {
+					instanceId: alphaInstance.id,
+					session: { type: "new" },
+					task: { type: "prompt", prompt: "alpha-prompt" },
+				},
+			}),
+		);
+		const submitBeta = await daemon.handleRequest(
+			req({
+				id: "submit-beta",
+				method: "job.submit",
+				params: {
+					instanceId: betaInstance.id,
+					session: { type: "new" },
+					task: { type: "prompt", prompt: "beta-prompt" },
+				},
+			}),
+		);
+		const alphaJobId = expectSuccess(submitAlpha, "job.submit").job.id;
+		const betaJobId = expectSuccess(submitBeta, "job.submit").job.id;
+
+		const alphaEvents: Array<{ type: string }> = [];
+		const betaEvents: Array<{ type: string }> = [];
+		const unsubA = daemon.subscribeJob(alphaJobId, (e) => alphaEvents.push(e));
+		const unsubB = daemon.subscribeJob(betaJobId, (e) => betaEvents.push(e));
+
+		await Bun.sleep(200);
+		unsubA();
+		unsubB();
+
+		const alphaJob = expectSuccess(
+			await daemon.handleRequest(
+				req({ id: "g-a", method: "job.get", params: { jobId: alphaJobId } }),
+			),
+			"job.get",
+		).job;
+		const betaJob = expectSuccess(
+			await daemon.handleRequest(
+				req({ id: "g-b", method: "job.get", params: { jobId: betaJobId } }),
+			),
+			"job.get",
+		).job;
+
+		expect(alphaJob.outputText).toBe("xyz");
+		expect(betaJob.outputText).toBe("xyz");
+		expect(alphaEvents.some((e) => e.type === "delta")).toBe(true);
+		expect(betaEvents.some((e) => e.type === "delta")).toBe(true);
+		expect(alphaEvents[alphaEvents.length - 1]?.type).toBe("done");
+		expect(betaEvents[betaEvents.length - 1]?.type).toBe("done");
+
+		expect(registry.directoriesStarted).toContain("/tmp/alpha");
+		expect(registry.directoriesStarted).toContain("/tmp/beta");
+	});
+
 	test("late subscriber gets snapshot of accumulated text and continues without duplicates", async () => {
 		registry.streamingDeltas = ["alpha ", "beta ", "gamma ", "delta"];
 		registry.deltaIntervalMs = 25;

packages/daemon/src/__tests__/helpers.ts

@@ -61,6 +61,8 @@ export class FakeOpencodeRegistry implements OpencodeRuntimeManager {
 		prompt: string;
 	}> = [];
 	readonly abortCalls: Array<{ instanceId: string; sessionId: string }> = [];
+	readonly directoriesStarted: string[] = [];
+	readonly disposeCalls: Array<{ directory?: string }> = [];
 	globalMaxConcurrent = 0;
 	/** Configurable per-test: an array of text deltas the fake will emit
 	 * via the wired onEvent handler before returning the final response
@@ -84,7 +86,11 @@ export class FakeOpencodeRegistry implements OpencodeRuntimeManager {
 		this.onEvent?.(instanceId, event);
 	}
 
-	async ensureStarted(instanceId: string): Promise<ManagedOpencodeRuntime> {
+	async ensureStarted(
+		instanceId: string,
+		directory: string,
+	): Promise<ManagedOpencodeRuntime> {
+		this.directoriesStarted.push(directory);
 		const existing = this.runtimes.get(instanceId);
 		if (existing) {
 			return existing;
@@ -93,7 +99,10 @@ export class FakeOpencodeRegistry implements OpencodeRuntimeManager {
 		const runtime: ManagedOpencodeRuntime = {
 			client: {
 				instance: {
-					dispose: async () => undefined,
+					dispose: async (parameters) => {
+						this.disposeCalls.push({ directory: parameters?.directory });
+						return undefined;
+					},
 				},
 				session: {
 					create: async () => {
@@ -204,7 +213,16 @@ export class FakeOpencodeRegistry implements OpencodeRuntimeManager {
 		this.runtimes.clear();
 	}
 
-	async queryProviders(_directory?: string, _refresh?: boolean) {
+	async queryProviders(
+		directories: string[],
+		_directory?: string,
+		refresh?: boolean,
+	) {
+		if (refresh) {
+			for (const dir of directories) {
+				this.disposeCalls.push({ directory: dir });
+			}
+		}
 		return { providers: [], connected: [] };
 	}
 }

packages/daemon/src/opencode.ts

@@ -50,7 +50,7 @@ export interface ProviderListResult {
 export interface OpencodeRuntimeClient {
 	session: OpencodeSessionClient;
 	instance: {
-		dispose(): Promise<unknown>;
+		dispose(parameters?: { directory?: string }): Promise<unknown>;
 	};
 	provider: {
 		list(parameters?: { directory?: string }): Promise<ProviderListResult>;
@@ -69,33 +69,42 @@ export interface ManagedOpencodeRuntime {
 export type OpencodeRuntimeEvent = OpencodeEvent;
 
 export interface OpencodeRuntimeManager {
-	ensureStarted(instanceId: string): Promise<ManagedOpencodeRuntime>;
+	ensureStarted(
+		instanceId: string,
+		directory: string,
+	): Promise<ManagedOpencodeRuntime>;
 	get(instanceId: string): ManagedOpencodeRuntime | undefined;
 	isRunning(instanceId: string): boolean;
 	stop(instanceId: string): Promise<void>;
 	stopAll(): Promise<void>;
-	/** Register the handler that receives every event surfaced by managed
-	 * runtimes. Called by the Daemon during construction so that wiring is
+	/** Register the handler that receives every event surfaced by the shared
+	 * runtime. Called by the Daemon during construction so that wiring is
 	 * uniform regardless of whether the registry was injected or default. */
 	setOnEvent(
 		handler: (instanceId: string, event: OpencodeRuntimeEvent) => void,
 	): void;
 	queryProviders(
+		directories: string[],
 		directory?: string,
 		refresh?: boolean,
 	): Promise<ProviderListResult>;
 }
 
-interface RuntimeEntry {
-	runtime?: ManagedOpencodeRuntime;
-	starting?: Promise<ManagedOpencodeRuntime>;
+interface SharedRuntime extends ManagedOpencodeRuntime {
+	rawEventSubscribe(parameters: {
+		directory: string;
+	}): Promise<{ stream: AsyncIterable<OpencodeRuntimeEvent> }>;
 }
 
-const SYSTEM_INSTANCE_ID = "__system__";
+interface InstanceSubscription {
+	directory: string;
+	cancel(): void;
+}
 
 export class OpencodeRegistry implements OpencodeRuntimeManager {
-	private readonly runtimes = new Map<string, RuntimeEntry>();
-	private readonly eventSubscriptions = new Map<string, { cancel(): void }>();
+	private shared?: SharedRuntime;
+	private sharedStarting?: Promise<SharedRuntime>;
+	private readonly subscriptions = new Map<string, InstanceSubscription>();
 	private onEvent?: (instanceId: string, event: OpencodeRuntimeEvent) => void;
 
 	setOnEvent(
@@ -104,25 +113,38 @@ export class OpencodeRegistry implements OpencodeRuntimeManager {
 		this.onEvent = handler;
 	}
 
-	async ensureStarted(instanceId: string): Promise<ManagedOpencodeRuntime> {
-		const entry = this.runtimes.get(instanceId);
-		if (entry?.runtime) {
-			return entry.runtime;
-		}
+	async ensureStarted(
+		instanceId: string,
+		directory: string,
+	): Promise<ManagedOpencodeRuntime> {
+		const runtime = await this.ensureSharedRuntime();
 
-		if (entry?.starting) {
-			return entry.starting;
+		const existing = this.subscriptions.get(instanceId);
+		if (!existing) {
+			const events = await runtime.rawEventSubscribe({ directory });
+			const subscription = this.consumeEvents(instanceId, events);
+			this.subscriptions.set(instanceId, {
+				directory,
+				cancel: subscription.cancel,
+			});
 		}
 
-		const starting = createOpencode().then(async ({ client, server }) => {
-			const events = await client.event.subscribe();
-			const subscription = this.consumeEvents(instanceId, events);
-			this.eventSubscriptions.set(instanceId, subscription);
+		return runtime;
+	}
+
+	private async ensureSharedRuntime(): Promise<SharedRuntime> {
+		if (this.shared) {
+			return this.shared;
+		}
+		if (this.sharedStarting) {
+			return this.sharedStarting;
+		}
 
-			const runtime: ManagedOpencodeRuntime = {
+		const starting = createOpencode({ port: 0 }).then(({ client, server }) => {
+			const runtime: SharedRuntime = {
 				client: {
 					instance: {
-						dispose: () => client.instance.dispose(),
+						dispose: (parameters) => client.instance.dispose(parameters),
 					},
 					session: {
 						create: async (parameters) => {
@@ -184,17 +206,19 @@ export class OpencodeRegistry implements OpencodeRuntimeManager {
 					},
 				},
 				server,
+				rawEventSubscribe: async (parameters) => {
+					return client.event.subscribe(parameters);
+				},
 			};
-			this.runtimes.set(instanceId, { runtime });
+			this.shared = runtime;
 			return runtime;
 		});
 
-		this.runtimes.set(instanceId, { starting });
+		this.sharedStarting = starting;
 		try {
 			return await starting;
-		} catch (error) {
-			this.runtimes.delete(instanceId);
-			throw error;
+		} finally {
+			this.sharedStarting = undefined;
 		}
 	}
 
@@ -223,84 +247,52 @@ export class OpencodeRegistry implements OpencodeRuntimeManager {
 	}
 
 	get(instanceId: string): ManagedOpencodeRuntime | undefined {
-		return this.runtimes.get(instanceId)?.runtime;
+		return this.subscriptions.has(instanceId) ? this.shared : undefined;
 	}
 
 	isRunning(instanceId: string): boolean {
-		return this.runtimes.has(instanceId) && Boolean(this.get(instanceId));
+		return this.subscriptions.has(instanceId);
 	}
 
 	async stop(instanceId: string): Promise<void> {
-		this.eventSubscriptions.get(instanceId)?.cancel();
-		this.eventSubscriptions.delete(instanceId);
-
-		const entry = this.runtimes.get(instanceId);
-		if (!entry) {
+		const subscription = this.subscriptions.get(instanceId);
+		if (!subscription) {
 			return;
 		}
-
-		try {
-			const runtime =
-				entry.runtime ?? (entry.starting ? await entry.starting : undefined);
-			await runtime?.client.instance.dispose();
-			runtime?.server.close();
-		} finally {
-			this.runtimes.delete(instanceId);
-		}
+		subscription.cancel();
+		this.subscriptions.delete(instanceId);
 	}
 
 	async stopAll(): Promise<void> {
-		await Promise.allSettled(
-			[...this.runtimes.keys()].map((instanceId) => this.stop(instanceId)),
-		);
-	}
-
-	/**
-	 * Get or create a long-lived system runtime for provider queries and other
-	 * lightweight operations that don't belong to a user-created instance.
-	 */
-	private ensureSystemRuntime(): Promise<ManagedOpencodeRuntime> {
-		return this.ensureStarted(SYSTEM_INSTANCE_ID);
-	}
+		for (const subscription of this.subscriptions.values()) {
+			subscription.cancel();
+		}
+		this.subscriptions.clear();
 
-	private async healthCheck(runtime: ManagedOpencodeRuntime): Promise<boolean> {
-		return runtime.client.ping();
+		const runtime = this.shared;
+		this.shared = undefined;
+		if (runtime) {
+			try {
+				runtime.server.close();
+			} catch {
+				// best-effort shutdown
+			}
+		}
 	}
 
 	async queryProviders(
+		directories: string[],
 		directory?: string,
 		refresh?: boolean,
 	): Promise<ProviderListResult> {
-		// When refresh is requested, dispose the runtime's internal instance so
-		// OpenCode re-reads auth.json and rebuilds its provider cache.
+		const runtime = await this.ensureSharedRuntime();
 		if (refresh) {
-			const runtime = await this.ensureSystemRuntime();
-			try {
-				await runtime.client.instance.dispose();
-			} catch {
-				// dispose may fail if the instance was already gone — ignore
-			}
-			return runtime.client.provider.list({ directory });
-		}
-
-		// Prefer an existing user runtime if one is available
-		for (const [id, entry] of this.runtimes.entries()) {
-			if (id !== SYSTEM_INSTANCE_ID && entry.runtime) {
-				try {
-					return await entry.runtime.client.provider.list({ directory });
-				} catch {}
-			}
+			await Promise.allSettled(
+				directories.map((dir) =>
+					runtime.client.instance.dispose({ directory: dir }),
+				),
+			);
 		}
-
-		// Fall back to the long-lived system runtime
-		let runtime = await this.ensureSystemRuntime();
-
-		// Health check — restart if the system instance died
-		if (!(await this.healthCheck(runtime))) {
-			this.runtimes.delete(SYSTEM_INSTANCE_ID);
-			runtime = await this.ensureSystemRuntime();
-		}
-
 		return runtime.client.provider.list({ directory });
 	}
 }