From 555ab2e5482b1617b244c2dd2cf6eeb25d846e99 Mon Sep 17 00:00:00 2001 From: Cristina Poncela Cubeiro <140309543+cristinaponcela@users.noreply.github.com> Date: Mon, 22 Jun 2026 14:56:41 +0200 Subject: [PATCH] cleanup: small optimizations --- packages/orchestrator/src/handler.ts | 4 +-- packages/orchestrator/src/ipc/server.ts | 10 +++---- packages/orchestrator/src/rpc-process.ts | 37 ++++++++++++++---------- packages/orchestrator/src/serve.ts | 4 +-- packages/orchestrator/src/supervisor.ts | 30 +++++++++++++++++-- 5 files changed, 58 insertions(+), 27 deletions(-) diff --git a/packages/orchestrator/src/handler.ts b/packages/orchestrator/src/handler.ts index 89fcb154..fbffa971 100644 --- a/packages/orchestrator/src/handler.ts +++ b/packages/orchestrator/src/handler.ts @@ -129,7 +129,7 @@ export async function handleIpcRequest(request: OrchestratorRequest): Promise void, onSessionEvent: (event: AgentSessionEvent) => void, @@ -140,7 +140,7 @@ export function attachIpcInstance( close(): void; } | undefined { - const handle = supervisor.attachInstance(instanceId, onSessionEvent, onUiRequest); + const handle = supervisor.openRpcStream(instanceId, onSessionEvent, onUiRequest); if (!handle) { return undefined; } diff --git a/packages/orchestrator/src/ipc/server.ts b/packages/orchestrator/src/ipc/server.ts index 1c45a6db..18ed7d49 100644 --- a/packages/orchestrator/src/ipc/server.ts +++ b/packages/orchestrator/src/ipc/server.ts @@ -30,7 +30,7 @@ export interface IpcRequestHandler { (request: RpcRequest): Promise | RpcBridgeResponse | ErrorResponse; (request: RpcStreamRequest): Promise | RpcReadyResponse | ErrorResponse; (request: OrchestratorRequest): Promise | OrchestratorResponse; - attach( + openRpcStream( instanceId: string, onResponse: (response: import("@earendil-works/pi-coding-agent").RpcResponse) => void, onSessionEvent: (event: import("@earendil-works/pi-coding-agent").AgentSessionEvent) => void, @@ -72,7 +72,7 @@ export async function startIpcServer(handler: IpcRequestHandler): Promise { socket.write(encodeMessage(response)); @@ -84,7 +84,7 @@ export async function startIpcServer(handler: IpcRequestHandler): Promise { try { const rpcRequest = JSON.parse(rpcLine) as RpcClientMessage; - await attachment.handleRequest(rpcRequest); + await rpcStream.handleRequest(rpcRequest); } catch (rpcError) { socket.write( encodeMessage({ @@ -121,7 +121,7 @@ export async function startIpcServer(handler: IpcRequestHandler): Promise attachment.close()); + socket.once("close", () => rpcStream.close()); return; } diff --git a/packages/orchestrator/src/rpc-process.ts b/packages/orchestrator/src/rpc-process.ts index 99da6ab0..221fe2eb 100644 --- a/packages/orchestrator/src/rpc-process.ts +++ b/packages/orchestrator/src/rpc-process.ts @@ -69,24 +69,31 @@ export function createRpcProcessInstance(options: { cwd: string }): RpcProcessIn const handleLine = (line: string) => { const parsed = JSON.parse(line) as { type?: string; id?: string }; - if (parsed.type === "response") { - if (parsed.id) { + + switch (parsed.type) { + case "response": { + if (!parsed.id) { + return; + } const pending = pendingRequests.get(parsed.id); - if (pending) { - pendingRequests.delete(parsed.id); - pending.resolve(parsed as RpcResponse); + if (!pending) { + return; + } + pendingRequests.delete(parsed.id); + pending.resolve(parsed as RpcResponse); + return; + } + + case "extension_ui_request": { + uiRequestHandler?.(parsed as RpcExtensionUIRequest); + return; + } + + default: { + for (const listener of eventListeners) { + listener(parsed as AgentSessionEvent); } } - return; - } - - if (parsed.type === "extension_ui_request") { - uiRequestHandler?.(parsed as RpcExtensionUIRequest); - return; - } - - for (const listener of eventListeners) { - listener(parsed as AgentSessionEvent); } }; diff --git a/packages/orchestrator/src/serve.ts b/packages/orchestrator/src/serve.ts index acc4e9fc..ffa444ab 100644 --- a/packages/orchestrator/src/serve.ts +++ b/packages/orchestrator/src/serve.ts @@ -1,7 +1,7 @@ import { existsSync, mkdirSync, unlinkSync } from "node:fs"; import { dirname } from "node:path"; import { getSocketPath } from "./config.ts"; -import { attachIpcInstance, handleIpcRequest } from "./handler.ts"; +import { handleIpcRequest, openRpcStream } from "./handler.ts"; import { startIpcServer } from "./ipc/server.ts"; import { getRadiusOrchestratorBaseUrl, isRadiusEnabled, radiusPresence } from "./radius.ts"; import { supervisor } from "./supervisor.ts"; @@ -21,7 +21,7 @@ export async function serve(): Promise { } const server = await startIpcServer( Object.assign(handleIpcRequest, { - attach: attachIpcInstance, + openRpcStream, }), ); console.log(`orchestrator listening on ${socketPath}`); diff --git a/packages/orchestrator/src/supervisor.ts b/packages/orchestrator/src/supervisor.ts index bb484f49..a1007814 100644 --- a/packages/orchestrator/src/supervisor.ts +++ b/packages/orchestrator/src/supervisor.ts @@ -25,6 +25,26 @@ function cloneInstance(record: InstanceRecord): InstanceRecord { return { ...record }; } +// Only refresh persisted session metadata after commands that can plausibly change +// the instance identity/details we store in instances.json. Most RPCs mutate transient +// runtime state only, so forcing a follow-up get_state after every command is wasted IO. +// +// - new_session / switch_session / fork / clone can change sessionId/sessionFile +// - set_session_name changes a persisted session detail we may want reflected externally +// - prompt can materialize or advance persisted session state after the child processes it +const SESSION_METADATA_COMMANDS: ReadonlySet = new Set([ + "new_session", + "switch_session", + "fork", + "clone", + "set_session_name", + "prompt", +]); + +function shouldRefreshSessionMetadata(command: RpcCommand): boolean { + return SESSION_METADATA_COMMANDS.has(command.type); +} + function isGetStateSuccess( response: RpcResponse, ): response is Extract< @@ -86,7 +106,7 @@ export class OrchestratorSupervisor { upsertInstance(instance); } - attachInstance( + openRpcStream( instanceId: string, onEvent: (event: AgentSessionEvent) => void, onUiRequest: (request: RpcExtensionUIRequest) => void, @@ -106,7 +126,9 @@ export class OrchestratorSupervisor { return { handleRpc: async (command) => { const response = await live.rpc.send(command); - await this.syncInstanceRecord(live); + if (shouldRefreshSessionMetadata(command)) { + await this.syncInstanceRecord(live); + } return response; }, handleUiResponse: (response) => { @@ -204,7 +226,9 @@ export class OrchestratorSupervisor { } const response = await live.rpc.send(command); - await this.syncInstanceRecord(live); + if (shouldRefreshSessionMetadata(command)) { + await this.syncInstanceRecord(live); + } return response; }