feat: rpc bridge

packages/orchestrator/src/cli.ts

@@ -14,7 +14,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 --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 --help\n  orchestrator --version`,
 	);
 }
 
@@ -80,6 +80,23 @@ async function main(): Promise<void> {
 		return;
 	}
 
+	if (args[0] === "rpc") {
+		const instanceId = args[1];
+		const commandJson = args[2];
+		if (!instanceId || !commandJson) {
+			console.error("Usage: orchestrator rpc <instance-id> <json-command>");
+			process.exit(1);
+		}
+		printResponse(
+			await sendIpcRequest({
+				type: "rpc",
+				instanceId,
+				command: JSON.parse(commandJson),
+			}),
+		);
+		return;
+	}
+
 	console.error(`Unknown command: ${args[0]}`);
 	printHelp();
 	process.exit(1);

packages/orchestrator/src/handler.ts

@@ -5,6 +5,8 @@ import type {
 	ListResponse,
 	OrchestratorRequest,
 	OrchestratorResponse,
+	RpcBridgeResponse,
+	RpcRequest,
 	SpawnRequest,
 	SpawnResponse,
 	StatusRequest,
@@ -38,6 +40,7 @@ export async function handleIpcRequest(request: SpawnRequest): Promise<SpawnResp
 export async function handleIpcRequest(request: ListRequest): Promise<ListResponse | ErrorResponse>;
 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: OrchestratorRequest): Promise<OrchestratorResponse>;
 export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse> {
 	switch (request.type) {
@@ -86,5 +89,18 @@ export async function handleIpcRequest(request: OrchestratorRequest): Promise<Or
 				instanceId: request.instanceId,
 			};
 		}
+
+		case "rpc": {
+			const response = await supervisor.handleRpc(request.instanceId, request.command);
+			if (!response) {
+				return unknownInstanceError(request.instanceId);
+			}
+
+			return {
+				type: "rpc_result",
+				ok: true,
+				response,
+			};
+		}
 	}
 }

packages/orchestrator/src/ipc/protocol.ts

@@ -1,3 +1,4 @@
+import type { RpcCommand, RpcResponse } from "@earendil-works/pi-coding-agent";
 import type { InstanceStatus } from "../types.ts";
 
 export interface SpawnRequest {
@@ -22,11 +23,18 @@ export interface StatusRequest {
 	instanceId: string;
 }
 
+export interface RpcRequest {
+	type: "rpc";
+	instanceId: string;
+	command: RpcCommand;
+}
+
 export interface RequestMap {
 	spawn: SpawnRequest;
 	list: ListRequest;
 	stop: StopRequest;
 	status: StatusRequest;
+	rpc: RpcRequest;
 }
 
 export type OrchestratorRequest = RequestMap[keyof RequestMap];
@@ -64,6 +72,11 @@ export interface StatusResponse extends ResponseBase {
 	instance?: InstanceSummary;
 }
 
+export interface RpcBridgeResponse extends ResponseBase {
+	type: "rpc_result";
+	response: RpcResponse;
+}
+
 export interface ErrorResponse extends ResponseBase {
 	type: "error";
 	ok: false;
@@ -75,6 +88,7 @@ export interface ResponseMap {
 	list: ListResponse;
 	stop: StopResponse;
 	status: StatusResponse;
+	rpc: RpcBridgeResponse;
 }
 
 export type OrchestratorResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;

packages/orchestrator/src/ipc/server.ts

@@ -9,6 +9,8 @@ import {
 	type OrchestratorRequest,
 	type OrchestratorResponse,
 	parseRequestLine,
+	type RpcBridgeResponse,
+	type RpcRequest,
 	type SpawnRequest,
 	type SpawnResponse,
 	type StatusRequest,
@@ -22,6 +24,7 @@ export interface IpcRequestHandler {
 	(request: ListRequest): Promise<ListResponse | ErrorResponse> | ListResponse | ErrorResponse;
 	(request: StopRequest): Promise<StopResponse | ErrorResponse> | StopResponse | ErrorResponse;
 	(request: StatusRequest): Promise<StatusResponse | ErrorResponse> | StatusResponse | ErrorResponse;
+	(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse> | RpcBridgeResponse | ErrorResponse;
 	(request: OrchestratorRequest): Promise<OrchestratorResponse> | OrchestratorResponse;
 }
 

packages/orchestrator/src/rpc-bridge.ts

@@ -0,0 +1,63 @@
+import type { AgentSessionRuntime, RpcCommand, RpcResponse, RpcSessionState } from "@earendil-works/pi-coding-agent";
+
+function success<T extends RpcCommand["type"]>(id: string | undefined, command: T, data?: object | null): RpcResponse {
+	if (data === undefined) {
+		return { id, type: "response", command, success: true } as RpcResponse;
+	}
+	return { id, type: "response", command, success: true, data } as RpcResponse;
+}
+
+function error(id: string | undefined, command: string, message: string): RpcResponse {
+	return { id, type: "response", command, success: false, error: message };
+}
+
+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 success(id, "prompt");
+		}
+
+		case "abort": {
+			await session.abort();
+			return success(id, "abort");
+		}
+
+		case "get_state": {
+			const state: RpcSessionState = {
+				model: session.model,
+				thinkingLevel: session.thinkingLevel,
+				isStreaming: session.isStreaming,
+				isCompacting: session.isCompacting,
+				steeringMode: session.steeringMode,
+				followUpMode: session.followUpMode,
+				sessionFile: session.sessionFile,
+				sessionId: session.sessionId,
+				sessionName: session.sessionName,
+				autoCompactionEnabled: session.autoCompactionEnabled,
+				messageCount: session.messages.length,
+				pendingMessageCount: session.pendingMessageCount,
+			};
+			return success(id, "get_state", state);
+		}
+
+		case "get_last_assistant_text": {
+			const text = session.getLastAssistantText() ?? null;
+			return success(id, "get_last_assistant_text", { text });
+		}
+
+		case "get_messages": {
+			return success(id, "get_messages", { messages: session.messages });
+		}
+
+		default:
+			return error(id, command.type, `Unsupported RPC command: ${command.type}`);
+	}
+}

packages/orchestrator/src/supervisor.ts

@@ -6,8 +6,11 @@ import {
 	createAgentSessionRuntime,
 	createAgentSessionServices,
 	getAgentDir,
+	type RpcCommand,
+	type RpcResponse,
 	SessionManager,
 } from "@earendil-works/pi-coding-agent";
+import { handleRpcCommand } from "./rpc-bridge.ts";
 import { getInstance, loadInstances, removeInstance, upsertInstance } from "./storage.ts";
 import type { InstanceRecord } from "./types.ts";
 
@@ -94,6 +97,15 @@ export class OrchestratorSupervisor {
 		removeInstance(instanceId);
 		return cloneInstance(live.record);
 	}
+
+	async handleRpc(instanceId: string, command: RpcCommand): Promise<RpcResponse | undefined> {
+		const live = this.liveInstances.get(instanceId);
+		if (!live) {
+			return undefined;
+		}
+
+		return handleRpcCommand(live.runtime, command);
+	}
 }
 
 export const supervisor = new OrchestratorSupervisor();