From 2f853bbce0ac112a0a69e9188ce22ea3dbdf1dfd Mon Sep 17 00:00:00 2001 From: Cristina Poncela Cubeiro <140309543+cristinaponcela@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:47:09 +0200 Subject: [PATCH] fix: class RpcProcessInstance as state machine --- packages/orchestrator/src/rpc-process.ts | 271 +++++++++++------------ packages/orchestrator/src/supervisor.ts | 208 ++++++++++++----- 2 files changed, 289 insertions(+), 190 deletions(-) diff --git a/packages/orchestrator/src/rpc-process.ts b/packages/orchestrator/src/rpc-process.ts index 3983a03c..b770862c 100644 --- a/packages/orchestrator/src/rpc-process.ts +++ b/packages/orchestrator/src/rpc-process.ts @@ -16,187 +16,186 @@ interface PendingRequest { reject(error: Error): void; } -export interface RpcProcessInstance { - process: ChildProcess; - send(command: RpcCommand): Promise; - handleUiResponse(response: RpcExtensionUIResponse): void; - setUiRequestHandler(handler?: (request: RpcExtensionUIRequest) => void): void; - onEvent(listener: (event: AgentSessionEvent) => void): () => void; - onExit(listener: (error?: Error) => void): () => void; - dispose(): Promise; -} - const require = createRequire(import.meta.url); -function getRpcSpawnCommand(): { command: string; args: string[] } { - if (isBunBinary) { - return { - command: join(dirname(process.execPath), process.platform === "win32" ? "pi.exe" : "pi"), - args: ["--mode", "rpc"], - }; - } - return { - command: process.execPath, - args: [require.resolve("@earendil-works/pi-coding-agent/rpc-entry")], - }; -} - function toError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); } -export function createRpcProcessInstance(options: { cwd: string }): RpcProcessInstance { - const rpcCommand = getRpcSpawnCommand(); - const child = spawn(rpcCommand.command, rpcCommand.args, { - cwd: options.cwd, - env: process.env, - stdio: ["pipe", "pipe", "pipe"], - }); - if (!child.stdin || !child.stdout) { - throw new Error("Failed to create RPC process stdio"); +export class RpcProcessInstance { + readonly process: ChildProcess; + + private exited = false; + private nextRequestId = 0; + private stdoutBuffer = ""; + private stderrBuffer = ""; + private readonly pendingRequests = new Map(); + private readonly eventListeners = new Set<(event: AgentSessionEvent) => void>(); + private readonly exitListeners = new Set<(error?: Error) => void>(); + private uiRequestHandler: ((request: RpcExtensionUIRequest) => void) | undefined; + + constructor(options: { cwd: string }) { + const rpcCommand = this.getSpawnCommand(); + this.process = spawn(rpcCommand.command, rpcCommand.args, { + cwd: options.cwd, + env: process.env, + stdio: ["pipe", "pipe", "pipe"], + }); + if (!this.process.stdin || !this.process.stdout) { + throw new Error("Failed to create RPC process stdio"); + } + this.attachListeners(); } - let exited = false; - let nextRequestId = 0; - let stdoutBuffer = ""; - let stderrBuffer = ""; - const pendingRequests = new Map(); - const eventListeners = new Set<(event: AgentSessionEvent) => void>(); - const exitListeners = new Set<(error?: Error) => void>(); - - const rejectAllPending = (error: Error) => { - for (const [id, pending] of pendingRequests) { - pendingRequests.delete(id); - pending.reject(error); + private getSpawnCommand(): { command: string; args: string[] } { + if (isBunBinary) { + return { + command: join(dirname(process.execPath), process.platform === "win32" ? "pi.exe" : "pi"), + args: ["--mode", "rpc"], + }; } - }; + return { + command: process.execPath, + args: [require.resolve("@earendil-works/pi-coding-agent/rpc-entry")], + }; + } - const notifyExit = (error?: Error) => { - for (const listener of exitListeners) { - listener(error); - } - }; + private attachListeners(): void { + this.process.stdout?.setEncoding("utf8"); + this.process.stdout?.on("data", (chunk: string) => { + this.stdoutBuffer += chunk; + while (true) { + const newlineIndex = this.stdoutBuffer.indexOf("\n"); + if (newlineIndex === -1) { + break; + } + const line = this.stdoutBuffer.slice(0, newlineIndex).trim(); + this.stdoutBuffer = this.stdoutBuffer.slice(newlineIndex + 1); + if (!line) { + continue; + } + this.handleLine(line); + } + }); - let uiRequestHandler: ((request: RpcExtensionUIRequest) => void) | undefined; + this.process.stderr?.setEncoding("utf8"); + this.process.stderr?.on("data", (chunk: string) => { + this.stderrBuffer += chunk; + }); - const handleLine = (line: string) => { + this.process.once("error", (error) => { + this.exited = true; + const wrapped = new Error(`RPC process error: ${error.message}. Stderr: ${this.stderrBuffer}`); + this.rejectAllPending(wrapped); + this.notifyExit(wrapped); + }); + + this.process.once("exit", (code, signal) => { + this.exited = true; + const error = new Error(`RPC process exited (code=${code} signal=${signal}). Stderr: ${this.stderrBuffer}`); + this.rejectAllPending(error); + this.notifyExit(error); + }); + } + + private handleLine(line: string): void { const parsed = JSON.parse(line) as { type?: string; id?: string }; - switch (parsed.type) { case "response": { if (!parsed.id) { return; } - const pending = pendingRequests.get(parsed.id); + const pending = this.pendingRequests.get(parsed.id); if (!pending) { return; } - pendingRequests.delete(parsed.id); + this.pendingRequests.delete(parsed.id); pending.resolve(parsed as RpcResponse); return; } case "extension_ui_request": { - uiRequestHandler?.(parsed as RpcExtensionUIRequest); + this.uiRequestHandler?.(parsed as RpcExtensionUIRequest); return; } default: { - for (const listener of eventListeners) { + for (const listener of this.eventListeners) { listener(parsed as AgentSessionEvent); } } } - }; + } - child.stdout.setEncoding("utf8"); - child.stdout.on("data", (chunk: string) => { - stdoutBuffer += chunk; - while (true) { - const newlineIndex = stdoutBuffer.indexOf("\n"); - if (newlineIndex === -1) { - break; - } - const line = stdoutBuffer.slice(0, newlineIndex).trim(); - stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1); - if (!line) { - continue; - } - handleLine(line); + private rejectAllPending(error: Error): void { + for (const [id, pending] of this.pendingRequests) { + this.pendingRequests.delete(id); + pending.reject(error); } - }); + } - child.stderr?.setEncoding("utf8"); - child.stderr?.on("data", (chunk: string) => { - stderrBuffer += chunk; - }); - - child.once("error", (error) => { - exited = true; - const wrapped = new Error(`RPC process error: ${error.message}. Stderr: ${stderrBuffer}`); - rejectAllPending(wrapped); - notifyExit(wrapped); - }); - - child.once("exit", (code, signal) => { - exited = true; - const error = new Error(`RPC process exited (code=${code} signal=${signal}). Stderr: ${stderrBuffer}`); - rejectAllPending(error); - notifyExit(error); - }); - - const send = async (command: RpcCommand): Promise => { - if (exited) { - throw new Error(`RPC process is not running. Stderr: ${stderrBuffer}`); + private notifyExit(error?: Error): void { + for (const listener of this.exitListeners) { + listener(error); } - const id = command.id ?? `orchestrator_${++nextRequestId}_${randomUUID()}`; + } + + send(command: RpcCommand): Promise { + if (this.exited) { + throw new Error(`RPC process is not running. Stderr: ${this.stderrBuffer}`); + } + const id = command.id ?? `orchestrator_${++this.nextRequestId}_${randomUUID()}`; const fullCommand = { ...command, id }; return new Promise((resolve, reject) => { - pendingRequests.set(id, { resolve, reject }); - child.stdin.write(`${JSON.stringify(fullCommand)}\n`, (error) => { + this.pendingRequests.set(id, { resolve, reject }); + this.process.stdin?.write(`${JSON.stringify(fullCommand)}\n`, (error) => { if (!error) { return; } - pendingRequests.delete(id); + this.pendingRequests.delete(id); reject(toError(error)); }); }); - }; + } - return { - process: child, - send, - handleUiResponse(response) { - if (exited) { - return; - } - child.stdin.write(`${JSON.stringify(response)}\n`); - }, - setUiRequestHandler(handler) { - uiRequestHandler = handler; - }, - onEvent(listener) { - eventListeners.add(listener); - return () => { - eventListeners.delete(listener); - }; - }, - onExit(listener) { - exitListeners.add(listener); - return () => { - exitListeners.delete(listener); - }; - }, - async dispose() { - uiRequestHandler = undefined; - rejectAllPending(new Error("RPC process disposed")); - if (exited) { - return; - } - child.kill("SIGTERM"); - await new Promise((resolve) => { - child.once("exit", () => resolve()); - }); - }, - }; + handleUiResponse(response: RpcExtensionUIResponse): void { + if (this.exited) { + return; + } + this.process.stdin?.write(`${JSON.stringify(response)}\n`); + } + + setUiRequestHandler(handler?: (request: RpcExtensionUIRequest) => void): void { + this.uiRequestHandler = handler; + } + + onEvent(listener: (event: AgentSessionEvent) => void): () => void { + this.eventListeners.add(listener); + return () => { + this.eventListeners.delete(listener); + }; + } + + onExit(listener: (error?: Error) => void): () => void { + this.exitListeners.add(listener); + return () => { + this.exitListeners.delete(listener); + }; + } + + async dispose(): Promise { + this.uiRequestHandler = undefined; + this.rejectAllPending(new Error("RPC process disposed")); + if (this.exited) { + return; + } + this.process.kill("SIGTERM"); + await new Promise((resolve) => { + this.process.once("exit", () => resolve()); + }); + } +} + +export function createRpcProcessInstance(options: { cwd: string }): RpcProcessInstance { + return new RpcProcessInstance(options); } diff --git a/packages/orchestrator/src/supervisor.ts b/packages/orchestrator/src/supervisor.ts index a1007814..1e375189 100644 --- a/packages/orchestrator/src/supervisor.ts +++ b/packages/orchestrator/src/supervisor.ts @@ -10,11 +10,17 @@ import type { import { radiusPresence } from "./radius.ts"; import { createRpcProcessInstance, type RpcProcessInstance } from "./rpc-process.ts"; import { getInstance, loadInstances, removeInstance, saveInstances, upsertInstance } from "./storage.ts"; -import type { InstanceRecord } from "./types.ts"; +import type { InstanceRecord, InstanceStatus } from "./types.ts"; + +interface LiveInstanceResources { + rpcProcess?: RpcProcessInstance; + radiusPiId?: string; + sessionId?: string; +} interface LiveInstance { - rpc: RpcProcessInstance; record: InstanceRecord; + resources: LiveInstanceResources; subscribers: Set; onUiRequest?: (request: RpcExtensionUIRequest) => void; unsubscribeEvents?: () => void; @@ -57,51 +63,133 @@ function isGetStateSuccess( export class OrchestratorSupervisor { private readonly liveInstances = new Map(); - private async syncInstanceRecord(live: LiveInstance): Promise { - const response = await live.rpc.send({ type: "get_state" }); - if (!isGetStateSuccess(response)) { - live.record = { - ...live.record, - lastSeenAt: new Date().toISOString(), - }; - upsertInstance(live.record); - return; - } + private setStatus(live: LiveInstance, status: InstanceStatus): void { live.record = { ...live.record, - sessionId: response.data.sessionId, - sessionFile: response.data.sessionFile, + status, lastSeenAt: new Date().toISOString(), }; upsertInstance(live.record); } - private bindLiveInstance(live: LiveInstance): void { + private updateRecord(live: LiveInstance, updates: Partial): void { + live.record = { + ...live.record, + ...updates, + lastSeenAt: new Date().toISOString(), + }; + if (updates.radiusPiId !== undefined) { + live.resources.radiusPiId = updates.radiusPiId; + } + if (updates.sessionId !== undefined) { + live.resources.sessionId = updates.sessionId; + } + upsertInstance(live.record); + } + + private clearBindings(live: LiveInstance): void { live.unsubscribeEvents?.(); live.unsubscribeExit?.(); - live.unsubscribeEvents = live.rpc.onEvent((event) => { + live.unsubscribeEvents = undefined; + live.unsubscribeExit = undefined; + live.onUiRequest = undefined; + live.resources.rpcProcess?.setUiRequestHandler(undefined); + } + + private bindRpcProcess(live: LiveInstance, rpcProcess: RpcProcessInstance): void { + this.clearBindings(live); + live.resources.rpcProcess = rpcProcess; + live.unsubscribeEvents = rpcProcess.onEvent((event) => { for (const subscriber of live.subscribers) { subscriber(event); } }); - live.unsubscribeExit = live.rpc.onExit(() => { - live.record = { - ...live.record, - status: "stopped", - lastSeenAt: new Date().toISOString(), - }; - upsertInstance(live.record); - this.liveInstances.delete(live.record.id); + live.unsubscribeExit = rpcProcess.onExit((error) => { + void this.handleUnexpectedRpcExit(live, error); }); - live.rpc.setUiRequestHandler((request) => { + rpcProcess.setUiRequestHandler((request) => { live.onUiRequest?.(request); }); } + private async handleUnexpectedRpcExit(live: LiveInstance, _error?: Error): Promise { + if (this.liveInstances.get(live.record.id) !== live) { + return; + } + if (live.record.status === "stopping" || live.record.status === "stopped") { + return; + } + this.setStatus(live, "error"); + this.clearBindings(live); + live.resources.rpcProcess = undefined; + if (live.resources.radiusPiId) { + try { + await radiusPresence.disconnectPi(live.record); + this.updateRecord(live, { radiusPiId: undefined }); + } catch (error) { + console.error(`Failed to disconnect Radius Pi ${live.record.id}: ${String(error)}`); + } + } + this.liveInstances.delete(live.record.id); + } + + private getRpcProcess(live: LiveInstance): RpcProcessInstance | undefined { + return live.resources.rpcProcess; + } + + private async syncInstanceRecord(live: LiveInstance): Promise { + const rpcProcess = this.getRpcProcess(live); + if (!rpcProcess) { + this.updateRecord(live, {}); + return; + } + const response = await rpcProcess.send({ type: "get_state" }); + if (!isGetStateSuccess(response)) { + this.updateRecord(live, {}); + return; + } + this.updateRecord(live, { + sessionId: response.data.sessionId, + sessionFile: response.data.sessionFile, + }); + } + + private async cleanupAcquiredResources(live: LiveInstance): Promise { + const rpcProcess = live.resources.rpcProcess; + this.clearBindings(live); + if (live.resources.radiusPiId) { + await radiusPresence.disconnectPi(live.record); + live.resources.radiusPiId = undefined; + live.record = { + ...live.record, + radiusPiId: undefined, + lastSeenAt: new Date().toISOString(), + }; + } + live.resources.sessionId = undefined; + if (rpcProcess) { + live.resources.rpcProcess = undefined; + await rpcProcess.dispose(); + } + } + + private async failSpawn(live: LiveInstance, error: unknown): Promise { + this.setStatus(live, "error"); + try { + await this.cleanupAcquiredResources(live); + } finally { + this.setStatus(live, "stopped"); + this.liveInstances.delete(live.record.id); + } + throw error; + } + updateInstance(instance: InstanceRecord): void { const live = this.liveInstances.get(instance.id); if (live) { live.record = instance; + live.resources.radiusPiId = instance.radiusPiId; + live.resources.sessionId = instance.sessionId; } upsertInstance(instance); } @@ -118,21 +206,22 @@ export class OrchestratorSupervisor { } | undefined { const live = this.liveInstances.get(instanceId); - if (!live) { + const rpcProcess = live ? this.getRpcProcess(live) : undefined; + if (!live || !rpcProcess) { return undefined; } live.subscribers.add(onEvent); live.onUiRequest = onUiRequest; return { handleRpc: async (command) => { - const response = await live.rpc.send(command); + const response = await rpcProcess.send(command); if (shouldRefreshSessionMetadata(command)) { await this.syncInstanceRecord(live); } return response; }, handleUiResponse: (response) => { - live.rpc.handleUiResponse(response); + rpcProcess.handleUiResponse(response); }, close: () => { if (live.onUiRequest === onUiRequest) { @@ -179,28 +268,33 @@ export class OrchestratorSupervisor { } async spawnInstance(options: { cwd: string; label?: string }): Promise { - const rpc = createRpcProcessInstance({ cwd: options.cwd }); const now = new Date().toISOString(); - const record: InstanceRecord = { - id: randomUUID(), - status: "online", - cwd: options.cwd, - createdAt: now, - lastSeenAt: now, - label: options.label, - }; - - const registeredRecord = await radiusPresence.registerPi(record); const live: LiveInstance = { - rpc, - record: registeredRecord, + record: { + id: randomUUID(), + status: "starting", + cwd: options.cwd, + createdAt: now, + lastSeenAt: now, + label: options.label, + }, + resources: {}, subscribers: new Set(), }; - this.bindLiveInstance(live); - this.liveInstances.set(registeredRecord.id, live); - await this.syncInstanceRecord(live); + this.liveInstances.set(live.record.id, live); upsertInstance(live.record); - return cloneInstance(live.record); + + try { + const rpcProcess = createRpcProcessInstance({ cwd: options.cwd }); + this.bindRpcProcess(live, rpcProcess); + await this.syncInstanceRecord(live); + const registeredRecord = await radiusPresence.registerPi(live.record); + this.updateRecord(live, { radiusPiId: registeredRecord.radiusPiId }); + this.setStatus(live, "online"); + return cloneInstance(live.record); + } catch (error) { + return await this.failSpawn(live, error); + } } async stopInstance(instanceId: string): Promise { @@ -209,23 +303,29 @@ export class OrchestratorSupervisor { return undefined; } - await radiusPresence.disconnectPi(live.record); - live.unsubscribeEvents?.(); - live.unsubscribeExit?.(); - live.onUiRequest = undefined; - await live.rpc.dispose(); - this.liveInstances.delete(instanceId); - removeInstance(instanceId); + this.setStatus(live, "stopping"); + try { + await this.cleanupAcquiredResources(live); + } finally { + live.record = { + ...live.record, + status: "stopped", + lastSeenAt: new Date().toISOString(), + }; + this.liveInstances.delete(instanceId); + removeInstance(instanceId); + } return cloneInstance(live.record); } async handleRpc(instanceId: string, command: RpcCommand): Promise { const live = this.liveInstances.get(instanceId); - if (!live) { + const rpcProcess = live ? this.getRpcProcess(live) : undefined; + if (!live || !rpcProcess) { return undefined; } - const response = await live.rpc.send(command); + const response = await rpcProcess.send(command); if (shouldRefreshSessionMetadata(command)) { await this.syncInstanceRecord(live); }