fix: bridge missing rpc commands

packages/orchestrator/src/cli.ts

@@ -1,9 +1,12 @@
 #!/usr/bin/env node
 import { readFileSync } from "node:fs";
+import { createConnection } from "node:net";
 import { dirname, join } from "node:path";
 import { cwd } from "node:process";
 import { fileURLToPath } from "node:url";
+import { getSocketPath } from "./config.ts";
 import { sendIpcRequest } from "./ipc/client.ts";
+import { encodeMessage } from "./ipc/protocol.ts";
 import { serve } from "./serve.ts";
 
 const __filename = fileURLToPath(import.meta.url);
@@ -14,7 +17,7 @@ const packageJson = JSON.parse(readFileSync(join(__dirname, "../package.json"),
 
 function printHelp(): void {
 	console.log(
-		`orchestrator v${packageJson.version}\n\nUsage:\n  orchestrator serve\n  orchestrator list\n  orchestrator spawn [--cwd <path>] [--label <label>]\n  orchestrator status <instance-id>\n  orchestrator stop <instance-id>\n  orchestrator rpc <instance-id> <json-command>\n  orchestrator --help\n  orchestrator --version`,
+		`orchestrator v${packageJson.version}\n\nUsage:\n  orchestrator serve\n  orchestrator list\n  orchestrator spawn [--cwd <path>] [--label <label>]\n  orchestrator status <instance-id>\n  orchestrator stop <instance-id>\n  orchestrator rpc <instance-id> <json-command>\n  orchestrator attach <instance-id>\n  orchestrator --help\n  orchestrator --version`,
 	);
 }
 
@@ -30,6 +33,39 @@ function getFlagValue(args: string[], flag: string): string | undefined {
 	return args[index + 1];
 }
 
+async function attach(instanceId: string): Promise<void> {
+	const socket = createConnection(getSocketPath());
+	process.stdin.setEncoding("utf8");
+
+	await new Promise<void>((resolve, reject) => {
+		socket.once("connect", () => {
+			socket.write(encodeMessage({ type: "attach", instanceId }));
+			resolve();
+		});
+		socket.once("error", reject);
+	});
+
+	socket.on("data", (chunk: Buffer | string) => {
+		process.stdout.write(chunk.toString());
+	});
+	socket.on("error", (error) => {
+		console.error(error instanceof Error ? error.message : String(error));
+		process.exit(1);
+	});
+	socket.on("end", () => {
+		process.exit(0);
+	});
+	process.stdin.on("data", (chunk: string) => {
+		const lines = chunk
+			.split("\n")
+			.map((line) => line.trim())
+			.filter((line) => line.length > 0);
+		for (const line of lines) {
+			socket.write(encodeMessage({ type: "attach_rpc", command: JSON.parse(line) }));
+		}
+	});
+}
+
 async function main(): Promise<void> {
 	const args = process.argv.slice(2);
 
@@ -97,6 +133,16 @@ async function main(): Promise<void> {
 		return;
 	}
 
+	if (args[0] === "attach") {
+		const instanceId = args[1];
+		if (!instanceId) {
+			console.error("Usage: orchestrator attach <instance-id>");
+			process.exit(1);
+		}
+		await attach(instanceId);
+		return;
+	}
+
 	console.error(`Unknown command: ${args[0]}`);
 	printHelp();
 	process.exit(1);

packages/orchestrator/src/handler.ts

@@ -1,4 +1,8 @@
+import type { AgentSessionEvent, RpcCommand } from "@earendil-works/pi-coding-agent";
 import type {
+	AttachReadyResponse,
+	AttachRequest,
+	AttachRpcResponse,
 	ErrorResponse,
 	InstanceSummary,
 	ListRequest,
@@ -43,6 +47,7 @@ export async function handleIpcRequest(request: ListRequest): Promise<ListRespon
 export async function handleIpcRequest(request: StopRequest): Promise<StopResponse | ErrorResponse>;
 export async function handleIpcRequest(request: StatusRequest): Promise<StatusResponse | ErrorResponse>;
 export async function handleIpcRequest(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse>;
+export async function handleIpcRequest(request: AttachRequest): Promise<AttachReadyResponse | ErrorResponse>;
 export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse>;
 export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse> {
 	switch (request.type) {
@@ -104,5 +109,38 @@ export async function handleIpcRequest(request: OrchestratorRequest): Promise<Or
 				response,
 			};
 		}
+
+		case "attach": {
+			const instance = supervisor.getInstance(request.instanceId);
+			if (!instance) {
+				return unknownInstanceError(request.instanceId);
+			}
+			return {
+				type: "attach_ready",
+				ok: true,
+				instance: toInstanceSummary(instance),
+			};
+		}
 	}
 }
+
+export function attachIpcInstance(
+	instanceId: string,
+	onResponse: (response: AttachRpcResponse) => void,
+	onSessionEvent: (event: AgentSessionEvent) => void,
+): { handleRequest(request: { type: "attach_rpc"; command: RpcCommand }): Promise<void>; close(): void } | undefined {
+	const handle = supervisor.attachInstance(instanceId, onSessionEvent);
+	if (!handle) {
+		return undefined;
+	}
+
+	return {
+		async handleRequest(request): Promise<void> {
+			const response = await handle.handleRpc(request.command);
+			onResponse({ type: "attach_rpc_result", response });
+		},
+		close(): void {
+			handle.close();
+		},
+	};
+}

packages/orchestrator/src/ipc/protocol.ts

@@ -1,4 +1,4 @@
-import type { RpcCommand, RpcResponse } from "@earendil-works/pi-coding-agent";
+import type { AgentSessionEvent, RpcCommand, RpcResponse } from "@earendil-works/pi-coding-agent";
 import type { InstanceStatus } from "../types.ts";
 
 export interface SpawnRequest {
@@ -29,12 +29,23 @@ export interface RpcRequest {
 	command: RpcCommand;
 }
 
+export interface AttachRequest {
+	type: "attach";
+	instanceId: string;
+}
+
+export interface AttachRpcRequest {
+	type: "attach_rpc";
+	command: RpcCommand;
+}
+
 export interface RequestMap {
 	spawn: SpawnRequest;
 	list: ListRequest;
 	stop: StopRequest;
 	status: StatusRequest;
 	rpc: RpcRequest;
+	attach: AttachRequest;
 }
 
 export type OrchestratorRequest = RequestMap[keyof RequestMap];
@@ -79,6 +90,21 @@ export interface RpcBridgeResponse extends ResponseBase {
 	response: RpcResponse;
 }
 
+export interface AttachReadyResponse extends ResponseBase {
+	type: "attach_ready";
+	instance?: InstanceSummary;
+}
+
+export interface AttachEventResponse {
+	type: "attach_event";
+	event: AgentSessionEvent;
+}
+
+export interface AttachRpcResponse {
+	type: "attach_rpc_result";
+	response: RpcResponse;
+}
+
 export interface ErrorResponse extends ResponseBase {
 	type: "error";
 	ok: false;
@@ -91,9 +117,12 @@ export interface ResponseMap {
 	stop: StopResponse;
 	status: StatusResponse;
 	rpc: RpcBridgeResponse;
+	attach: AttachReadyResponse;
 }
 
 export type OrchestratorResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;
+export type AttachClientRequest = AttachRpcRequest;
+export type AttachServerResponse = AttachReadyResponse | AttachEventResponse | AttachRpcResponse | ErrorResponse;
 
 export type ResponseFor<T extends OrchestratorRequest> = T extends { type: infer K }
 	? K extends keyof ResponseMap
@@ -101,7 +130,7 @@ export type ResponseFor<T extends OrchestratorRequest> = T extends { type: infer
 		: ErrorResponse
 	: ErrorResponse;
 
-export function encodeMessage(message: OrchestratorRequest | OrchestratorResponse): string {
+export function encodeMessage(message: unknown): string {
 	return `${JSON.stringify(message)}\n`;
 }
 

packages/orchestrator/src/ipc/server.ts

@@ -2,6 +2,10 @@ import { existsSync, unlinkSync } from "node:fs";
 import { createConnection, createServer, type Server } from "node:net";
 import { getSocketPath } from "../config.ts";
 import {
+	type AttachClientRequest,
+	type AttachReadyResponse,
+	type AttachRequest,
+	type AttachRpcResponse,
 	type ErrorResponse,
 	encodeMessage,
 	type ListRequest,
@@ -25,7 +29,18 @@ export interface IpcRequestHandler {
 	(request: StopRequest): Promise<StopResponse | ErrorResponse> | StopResponse | ErrorResponse;
 	(request: StatusRequest): Promise<StatusResponse | ErrorResponse> | StatusResponse | ErrorResponse;
 	(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse> | RpcBridgeResponse | ErrorResponse;
+	(request: AttachRequest): Promise<AttachReadyResponse | ErrorResponse> | AttachReadyResponse | ErrorResponse;
 	(request: OrchestratorRequest): Promise<OrchestratorResponse> | OrchestratorResponse;
+	attach(
+		instanceId: string,
+		onEvent: (response: AttachRpcResponse) => void,
+		onSessionEvent: (event: import("@earendil-works/pi-coding-agent").AgentSessionEvent) => void,
+	):
+		| {
+				handleRequest(request: AttachClientRequest): Promise<void>;
+				close(): void;
+		  }
+		| undefined;
 }
 
 export async function startIpcServer(handler: IpcRequestHandler): Promise<Server> {
@@ -50,6 +65,63 @@ export async function startIpcServer(handler: IpcRequestHandler): Promise<Server
 
 			try {
 				const request = parseRequestLine(line);
+				if (request.type === "attach") {
+					const response = await handler(request);
+					if (!response.ok || !response.instance) {
+						socket.end(encodeMessage(response));
+						return;
+					}
+
+					const attachment = handler.attach(
+						request.instanceId,
+						(response) => {
+							socket.write(encodeMessage(response));
+						},
+						(event) => {
+							socket.write(encodeMessage({ type: "attach_event", event }));
+						},
+					);
+					if (!attachment) {
+						socket.end(
+							encodeMessage({ type: "error", ok: false, error: `Unknown instance: ${request.instanceId}` }),
+						);
+						return;
+					}
+
+					socket.write(encodeMessage(response));
+					socket.removeAllListeners("data");
+					socket.on("data", (attachChunk: Buffer | string) => {
+						buffer += attachChunk.toString();
+						for (;;) {
+							const attachNewlineIndex = buffer.indexOf("\n");
+							if (attachNewlineIndex === -1) {
+								break;
+							}
+							const attachLine = buffer.slice(0, attachNewlineIndex).trim();
+							buffer = buffer.slice(attachNewlineIndex + 1);
+							if (!attachLine) {
+								continue;
+							}
+							void (async () => {
+								try {
+									const attachRequest = JSON.parse(attachLine) as AttachClientRequest;
+									await attachment.handleRequest(attachRequest);
+								} catch (attachError) {
+									socket.write(
+										encodeMessage({
+											type: "error",
+											ok: false,
+											error: attachError instanceof Error ? attachError.message : String(attachError),
+										}),
+									);
+								}
+							})();
+						}
+					});
+					socket.once("close", () => attachment.close());
+					return;
+				}
+
 				const response = await handler(request);
 				socket.end(encodeMessage(response));
 			} catch (error) {

packages/orchestrator/src/rpc-bridge.ts

@@ -11,22 +11,33 @@ function error(id: string | undefined, command: string, message: string): RpcRes
 	return { id, type: "response", command, success: false, error: message };
 }
 
-function unsupported(id: string | undefined, command: RpcCommand["type"]): RpcResponse {
-	return error(id, command, `Unsupported RPC command in orchestrator bridge: ${command}`);
-}
-
 export async function handleRpcCommand(runtime: AgentSessionRuntime, command: RpcCommand): Promise<RpcResponse> {
 	const session = runtime.session;
 	const id = command.id;
 
 	switch (command.type) {
 		case "prompt": {
-			await session.prompt(command.message, {
-				images: command.images,
-				streamingBehavior: command.streamingBehavior,
-				source: "rpc",
+			return await new Promise<RpcResponse>((resolve) => {
+				let settled = false;
+				void session
+					.prompt(command.message, {
+						images: command.images,
+						streamingBehavior: command.streamingBehavior,
+						source: "rpc",
+						preflightResult: (didSucceed) => {
+							if (!settled && didSucceed) {
+								settled = true;
+								resolve(success(id, "prompt"));
+							}
+						},
+					})
+					.catch((rpcError: unknown) => {
+						if (!settled) {
+							settled = true;
+							resolve(error(id, "prompt", rpcError instanceof Error ? rpcError.message : String(rpcError)));
+						}
+					});
 			});
-			return success(id, "prompt");
 		}
 
 		case "steer": {
@@ -44,11 +55,30 @@ export async function handleRpcCommand(runtime: AgentSessionRuntime, command: Rp
 			return success(id, "abort");
 		}
 
-		case "new_session":
-		case "switch_session":
-		case "fork":
-		case "clone":
-			return unsupported(id, command.type);
+		case "new_session": {
+			const options = command.parentSession ? { parentSession: command.parentSession } : undefined;
+			const result = await runtime.newSession(options);
+			return success(id, "new_session", result);
+		}
+
+		case "switch_session": {
+			const result = await runtime.switchSession(command.sessionPath);
+			return success(id, "switch_session", result);
+		}
+
+		case "fork": {
+			const result = await runtime.fork(command.entryId);
+			return success(id, "fork", { text: result.selectedText ?? "", cancelled: result.cancelled });
+		}
+
+		case "clone": {
+			const leafId = session.sessionManager.getLeafId();
+			if (!leafId) {
+				return error(id, "clone", "Cannot clone session: no current entry selected");
+			}
+			const result = await runtime.fork(leafId, { position: "at" });
+			return success(id, "clone", { cancelled: result.cancelled });
+		}
 
 		case "get_state": {
 			const state: RpcSessionState = {

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 { handleIpcRequest } from "./handler.ts";
+import { attachIpcInstance, handleIpcRequest } from "./handler.ts";
 import { startIpcServer } from "./ipc/server.ts";
 import { getRadiusOrchestratorBaseUrl, isRadiusEnabled, radiusPresence } from "./radius.ts";
 import { supervisor } from "./supervisor.ts";
@@ -19,7 +19,11 @@ export async function serve(): Promise<void> {
 	} else {
 		console.log("radius integration disabled: set PI_RADIUS_API_KEY to enable");
 	}
-	const server = await startIpcServer(handleIpcRequest);
+	const server = await startIpcServer(
+		Object.assign(handleIpcRequest, {
+			attach: attachIpcInstance,
+		}),
+	);
 	console.log(`orchestrator listening on ${socketPath}`);
 
 	let shutdownPromise: Promise<void> | undefined;

packages/orchestrator/src/supervisor.ts

@@ -1,5 +1,7 @@
 import { randomUUID } from "node:crypto";
 import {
+	type AgentSessionEvent,
+	type AgentSessionEventListener,
 	type AgentSessionRuntime,
 	type CreateAgentSessionRuntimeFactory,
 	createAgentSessionFromServices,
@@ -18,6 +20,8 @@ import type { InstanceRecord } from "./types.ts";
 interface LiveInstance {
 	runtime: AgentSessionRuntime;
 	record: InstanceRecord;
+	subscribers: Set<AgentSessionEventListener>;
+	unsubscribeSession?: () => void;
 }
 
 function cloneInstance(record: InstanceRecord): InstanceRecord {
@@ -56,6 +60,29 @@ async function createRuntime(cwd: string): Promise<AgentSessionRuntime> {
 export class OrchestratorSupervisor {
 	private readonly liveInstances = new Map<string, LiveInstance>();
 
+	private syncInstanceRecord(live: LiveInstance): void {
+		live.record = {
+			...live.record,
+			sessionId: live.runtime.session.sessionId,
+			sessionFile: live.runtime.session.sessionFile,
+			lastSeenAt: new Date().toISOString(),
+		};
+		upsertInstance(live.record);
+	}
+
+	private bindLiveInstance(live: LiveInstance): void {
+		live.unsubscribeSession?.();
+		live.unsubscribeSession = live.runtime.session.subscribe((event) => {
+			for (const subscriber of live.subscribers) {
+				subscriber(event);
+			}
+		});
+		live.runtime.setRebindSession(async () => {
+			this.syncInstanceRecord(live);
+			this.bindLiveInstance(live);
+		});
+	}
+
 	updateInstance(instance: InstanceRecord): void {
 		const live = this.liveInstances.get(instance.id);
 		if (live) {
@@ -64,6 +91,27 @@ export class OrchestratorSupervisor {
 		upsertInstance(instance);
 	}
 
+	attachInstance(
+		instanceId: string,
+		onEvent: (event: AgentSessionEvent) => void,
+	): { handleRpc(command: RpcCommand): Promise<RpcResponse>; close(): void } | undefined {
+		const live = this.liveInstances.get(instanceId);
+		if (!live) {
+			return undefined;
+		}
+		live.subscribers.add(onEvent);
+		return {
+			handleRpc: async (command) => {
+				const response = await handleRpcCommand(live.runtime, command);
+				this.syncInstanceRecord(live);
+				return response;
+			},
+			close: () => {
+				live.subscribers.delete(onEvent);
+			},
+		};
+	}
+
 	getLiveInstance(instanceId: string): InstanceRecord | undefined {
 		const live = this.liveInstances.get(instanceId);
 		return live ? cloneInstance(live.record) : undefined;
@@ -114,7 +162,9 @@ export class OrchestratorSupervisor {
 		};
 
 		const registeredRecord = await radiusPresence.registerPi(record);
-		this.liveInstances.set(registeredRecord.id, { runtime, record: registeredRecord });
+		const live: LiveInstance = { runtime, record: registeredRecord, subscribers: new Set() };
+		this.bindLiveInstance(live);
+		this.liveInstances.set(registeredRecord.id, live);
 		upsertInstance(registeredRecord);
 		return cloneInstance(registeredRecord);
 	}
@@ -126,6 +176,8 @@ export class OrchestratorSupervisor {
 		}
 
 		await radiusPresence.disconnectPi(live.record);
+		live.unsubscribeSession?.();
+		live.runtime.setRebindSession(undefined);
 		await live.runtime.dispose();
 		this.liveInstances.delete(instanceId);
 		removeInstance(instanceId);
@@ -138,7 +190,9 @@ export class OrchestratorSupervisor {
 			return undefined;
 		}
 
-		return handleRpcCommand(live.runtime, command);
+		const response = await handleRpcCommand(live.runtime, command);
+		this.syncInstanceRecord(live);
+		return response;
 	}
 
 	async shutdown(): Promise<void> {