diff --git a/packages/orchestrator/src/config.ts b/packages/orchestrator/src/config.ts new file mode 100644 index 00000000..64fb1008 --- /dev/null +++ b/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"); +} diff --git a/packages/orchestrator/src/index.ts b/packages/orchestrator/src/index.ts index e69de29b..67ecbe7a 100644 --- a/packages/orchestrator/src/index.ts +++ b/packages/orchestrator/src/index.ts @@ -0,0 +1,3 @@ +export * from "./config.ts"; +export * from "./ipc/client.ts"; +export * from "./ipc/protocol.ts"; diff --git a/packages/orchestrator/src/ipc/client.ts b/packages/orchestrator/src/ipc/client.ts new file mode 100644 index 00000000..8029654a --- /dev/null +++ b/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 { + const socketPath = getSocketPath(); + + return new Promise((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(); + }); + }); +} diff --git a/packages/orchestrator/src/ipc/protocol.ts b/packages/orchestrator/src/ipc/protocol.ts new file mode 100644 index 00000000..7c275279 --- /dev/null +++ b/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; +}