feat: ipc socket

packages/orchestrator/src/config.ts

@@ -0,0 +1,27 @@
+import { homedir } from "node:os";
+import { join } from "node:path";
+
+const CONFIG_DIR_NAME = ".pi";
+const ENV_ORCHESTRATOR_DIR = "PI_ORCHESTRATOR_DIR";
+
+export function getOrchestratorDir(): string {
+	const envDir = process.env[ENV_ORCHESTRATOR_DIR];
+	if (envDir) {
+		return envDir;
+	}
+
+	const piDir = process.env.PI_CONFIG_DIR || join(homedir(), CONFIG_DIR_NAME);
+	return join(piDir, "orchestrator");
+}
+
+export function getMachinePath(): string {
+	return join(getOrchestratorDir(), "machine.json");
+}
+
+export function getInstancesPath(): string {
+	return join(getOrchestratorDir(), "instances.json");
+}
+
+export function getSocketPath(): string {
+	return join(getOrchestratorDir(), "orchestrator.sock");
+}

packages/orchestrator/src/index.ts

@@ -0,0 +1,3 @@
+export * from "./config.ts";
+export * from "./ipc/client.ts";
+export * from "./ipc/protocol.ts";

packages/orchestrator/src/ipc/client.ts

@@ -0,0 +1,63 @@
+import { createConnection } from "node:net";
+import { getSocketPath } from "../config.ts";
+import { encodeMessage, type OrchestratorRequest, type OrchestratorResponse, parseResponseLine } from "./protocol.ts";
+
+export async function sendIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse> {
+	const socketPath = getSocketPath();
+
+	return new Promise<OrchestratorResponse>((resolve, reject) => {
+		const socket = createConnection(socketPath);
+		let buffer = "";
+		let settled = false;
+
+		const cleanup = () => {
+			socket.removeAllListeners();
+			socket.end();
+		};
+
+		socket.on("connect", () => {
+			socket.write(encodeMessage(request));
+		});
+
+		socket.on("data", (chunk: Buffer | string) => {
+			buffer += chunk.toString();
+			const newlineIndex = buffer.indexOf("\n");
+			if (newlineIndex === -1) {
+				return;
+			}
+
+			const line = buffer.slice(0, newlineIndex).trim();
+			if (!line) {
+				return;
+			}
+
+			try {
+				settled = true;
+				resolve(parseResponseLine(line));
+				cleanup();
+			} catch (error) {
+				settled = true;
+				reject(error);
+				cleanup();
+			}
+		});
+
+		socket.on("error", (error) => {
+			if (settled) {
+				return;
+			}
+			settled = true;
+			reject(error);
+			cleanup();
+		});
+
+		socket.on("end", () => {
+			if (settled) {
+				return;
+			}
+			settled = true;
+			reject(new Error(`Orchestrator socket closed before a response was received: ${socketPath}`));
+			cleanup();
+		});
+	});
+}

packages/orchestrator/src/ipc/protocol.ts

@@ -0,0 +1,73 @@
+export interface SpawnRequest {
+	type: "spawn";
+	cwd: string;
+	label?: string;
+	provider?: string;
+	model?: string;
+}
+
+export interface ListRequest {
+	type: "list";
+}
+
+export interface StopRequest {
+	type: "stop";
+	instanceId: string;
+}
+
+export interface StatusRequest {
+	type: "status";
+	instanceId: string;
+}
+
+export type OrchestratorRequest = SpawnRequest | ListRequest | StopRequest | StatusRequest;
+
+export interface InstanceSummary {
+	id: string;
+	status: "starting" | "online" | "stopping" | "stopped" | "error";
+	cwd: string;
+	label?: string;
+	sessionId?: string;
+}
+
+export interface ResponseBase {
+	type: string;
+	ok: boolean;
+	error?: string;
+}
+
+export interface SpawnResponse extends ResponseBase {
+	type: "spawn_result";
+	instance?: InstanceSummary;
+}
+
+export interface ListResponse extends ResponseBase {
+	type: "list_result";
+	instances?: InstanceSummary[];
+}
+
+export interface StopResponse extends ResponseBase {
+	type: "stop_result";
+	instanceId?: string;
+}
+
+export interface StatusResponse extends ResponseBase {
+	type: "status_result";
+	instance?: InstanceSummary;
+}
+
+export type OrchestratorResponse = SpawnResponse | ListResponse | StopResponse | StatusResponse;
+
+export function encodeMessage(message: OrchestratorRequest | OrchestratorResponse): string {
+	return `${JSON.stringify(message)}\n`;
+}
+
+export function parseRequestLine(line: string): OrchestratorRequest {
+	const value = JSON.parse(line) as OrchestratorRequest;
+	return value;
+}
+
+export function parseResponseLine(line: string): OrchestratorResponse {
+	const value = JSON.parse(line) as OrchestratorResponse;
+	return value;
+}