feat(experimental): pi orchestrator
This commit is contained in:
Cristina Poncela Cubeiro
Generated
+18
@@ -793,6 +793,10 @@
|
||||
"resolved": "packages/coding-agent",
|
||||
"link": true
|
||||
},
|
||||
"node_modules/@earendil-works/pi-orchestrator": {
|
||||
"resolved": "packages/orchestrator",
|
||||
"link": true
|
||||
},
|
||||
"node_modules/@earendil-works/pi-tui": {
|
||||
"resolved": "packages/tui",
|
||||
"link": true
|
||||
@@ -6134,6 +6138,20 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"packages/orchestrator": {
|
||||
"name": "@earendil-works/pi-orchestrator",
|
||||
"version": "0.80.2",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-coding-agent": "0.80.2"
|
||||
},
|
||||
"devDependencies": {
|
||||
"shx": "0.4.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=22.19.0"
|
||||
}
|
||||
},
|
||||
"packages/tui": {
|
||||
"name": "@earendil-works/pi-tui",
|
||||
"version": "0.80.2",
|
||||
|
||||
+1
-1
@@ -12,7 +12,7 @@
|
||||
],
|
||||
"scripts": {
|
||||
"clean": "npm run clean --workspaces",
|
||||
"build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build",
|
||||
"build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build && cd ../orchestrator && npm run build",
|
||||
"check": "biome check --write --error-on-warnings . && npm run check:pinned-deps && npm run check:ts-imports && npm run check:shrinkwrap && tsgo --noEmit && npm run check:browser-smoke",
|
||||
"check:browser-smoke": "node scripts/check-browser-smoke.mjs",
|
||||
"check:pinned-deps": "node scripts/check-pinned-deps.mjs",
|
||||
|
||||
@@ -15,6 +15,9 @@
|
||||
".": {
|
||||
"types": "./dist/index.d.ts",
|
||||
"import": "./dist/index.js"
|
||||
},
|
||||
"./rpc-entry": {
|
||||
"import": "./dist/rpc-entry.js"
|
||||
}
|
||||
},
|
||||
"files": [
|
||||
@@ -27,7 +30,7 @@
|
||||
],
|
||||
"scripts": {
|
||||
"clean": "shx rm -rf dist",
|
||||
"build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js && npm run copy-assets",
|
||||
"build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js dist/rpc-entry.js && npm run copy-assets",
|
||||
"build:binary": "npm --prefix ../tui run build && npm --prefix ../ai run build && npm --prefix ../agent run build && npm run build && bun build --compile ./dist/bun/cli.js ./src/utils/image-resize-worker.ts --outfile dist/pi && npm run copy-binary-assets",
|
||||
"copy-assets": "shx mkdir -p dist/modes/interactive/theme && shx cp src/modes/interactive/theme/*.json dist/modes/interactive/theme/ && shx mkdir -p dist/modes/interactive/assets && shx cp src/modes/interactive/assets/*.png dist/modes/interactive/assets/ && shx mkdir -p dist/core/export-html/vendor && shx cp src/core/export-html/template.html src/core/export-html/template.css src/core/export-html/template.js dist/core/export-html/ && shx cp src/core/export-html/vendor/*.js dist/core/export-html/vendor/",
|
||||
"copy-binary-assets": "shx cp package.json dist/ && shx cp README.md dist/ && shx cp CHANGELOG.md dist/ && shx mkdir -p dist/theme && shx cp src/modes/interactive/theme/*.json dist/theme/ && shx mkdir -p dist/assets && shx cp src/modes/interactive/assets/*.png dist/assets/ && shx mkdir -p dist/export-html/vendor && shx cp src/core/export-html/template.html dist/export-html/ && shx cp src/core/export-html/vendor/*.js dist/export-html/vendor/ && shx cp -r docs dist/ && shx cp -r examples dist/ && shx cp ../../node_modules/@silvia-odwyer/photon-node/photon_rs_bg.wasm dist/",
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
#!/usr/bin/env node
|
||||
import { APP_NAME } from "./config.ts";
|
||||
import { configureHttpDispatcher } from "./core/http-dispatcher.ts";
|
||||
import { main } from "./main.ts";
|
||||
|
||||
process.title = `${APP_NAME}-rpc`;
|
||||
process.env.PI_CODING_AGENT = "true";
|
||||
process.emitWarning = (() => {}) as typeof process.emitWarning;
|
||||
|
||||
configureHttpDispatcher();
|
||||
|
||||
main(["--mode", "rpc", ...process.argv.slice(2)]);
|
||||
@@ -0,0 +1,3 @@
|
||||
# Changelog
|
||||
|
||||
## [Unreleased]
|
||||
@@ -0,0 +1,11 @@
|
||||
# @earendil-works/pi-orchestrator
|
||||
|
||||
Experimental. This package is under active development and may change or be removed without notice. Its CLI, APIs, and behavior are not yet stable.
|
||||
|
||||
Orchestrator package for pi.
|
||||
|
||||
## CLI
|
||||
|
||||
```bash
|
||||
orchestrator --help
|
||||
```
|
||||
@@ -0,0 +1,45 @@
|
||||
{
|
||||
"name": "@earendil-works/pi-orchestrator",
|
||||
"version": "0.80.2",
|
||||
"description": "experimental orchestrator package for pi",
|
||||
"type": "module",
|
||||
"main": "./dist/index.js",
|
||||
"types": "./dist/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./dist/index.d.ts",
|
||||
"import": "./dist/index.js"
|
||||
}
|
||||
},
|
||||
"files": [
|
||||
"dist",
|
||||
"README.md",
|
||||
"CHANGELOG.md"
|
||||
],
|
||||
"scripts": {
|
||||
"clean": "shx rm -rf dist",
|
||||
"dev": "tsgo -p tsconfig.build.json --watch --preserveWatchOutput",
|
||||
"build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js",
|
||||
"prepublishOnly": "npm run clean && npm run build"
|
||||
},
|
||||
"keywords": [
|
||||
"pi",
|
||||
"orchestrator"
|
||||
],
|
||||
"author": "Earendil Works",
|
||||
"license": "MIT",
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "git+https://github.com/earendil-works/pi.git",
|
||||
"directory": "packages/orchestrator"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=22.19.0"
|
||||
},
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-coding-agent": "0.80.2"
|
||||
},
|
||||
"devDependencies": {
|
||||
"shx": "0.4.0"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
#!/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 type { RpcCommand, RpcExtensionUIResponse } from "@earendil-works/pi-coding-agent";
|
||||
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);
|
||||
const __dirname = dirname(__filename);
|
||||
const packageJson = JSON.parse(readFileSync(join(__dirname, "../package.json"), "utf-8")) as {
|
||||
version: string;
|
||||
};
|
||||
|
||||
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 rpc-stream <instance-id>\n orchestrator --help\n orchestrator --version\n\nRPC stream stdin expects JSONL RpcCommand or extension_ui_response messages.`,
|
||||
);
|
||||
}
|
||||
|
||||
function printResponse(response: unknown): void {
|
||||
console.log(JSON.stringify(response, null, 2));
|
||||
}
|
||||
|
||||
function getFlagValue(args: string[], flag: string): string | undefined {
|
||||
const index = args.indexOf(flag);
|
||||
if (index === -1 || index + 1 >= args.length) {
|
||||
return undefined;
|
||||
}
|
||||
return args[index + 1];
|
||||
}
|
||||
|
||||
async function rpcStream(instanceId: string): Promise<void> {
|
||||
const socket = createConnection(getSocketPath());
|
||||
let stdinBuffer = "";
|
||||
process.stdin.setEncoding("utf8");
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once("connect", () => {
|
||||
socket.write(encodeMessage({ type: "rpc_stream", instanceId }));
|
||||
resolve();
|
||||
});
|
||||
socket.once("error", reject);
|
||||
});
|
||||
|
||||
socket.on("data", (chunk: Buffer | string) => {
|
||||
process.stdout.write(chunk.toString());
|
||||
});
|
||||
console.error(`connected to rpc stream ${instanceId}; send JSONL RpcCommand or extension_ui_response on stdin`);
|
||||
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) => {
|
||||
stdinBuffer += chunk;
|
||||
while (true) {
|
||||
const newlineIndex = stdinBuffer.indexOf("\n");
|
||||
if (newlineIndex === -1) {
|
||||
return;
|
||||
}
|
||||
const line = stdinBuffer.slice(0, newlineIndex).trim();
|
||||
stdinBuffer = stdinBuffer.slice(newlineIndex + 1);
|
||||
if (!line) {
|
||||
continue;
|
||||
}
|
||||
const parsed = JSON.parse(line) as RpcCommand | RpcExtensionUIResponse;
|
||||
socket.write(encodeMessage(parsed));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const args = process.argv.slice(2);
|
||||
|
||||
if (args.length === 0 || args[0] === "--help" || args[0] === "-h") {
|
||||
printHelp();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
if (args[0] === "--version" || args[0] === "-v") {
|
||||
console.log(packageJson.version);
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
if (args[0] === "serve") {
|
||||
await serve();
|
||||
return;
|
||||
}
|
||||
|
||||
if (args[0] === "list") {
|
||||
printResponse(await sendIpcRequest({ type: "list" }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (args[0] === "spawn") {
|
||||
const spawnCwd = getFlagValue(args, "--cwd") ?? cwd();
|
||||
const label = getFlagValue(args, "--label");
|
||||
printResponse(await sendIpcRequest({ type: "spawn", cwd: spawnCwd, label }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (args[0] === "status") {
|
||||
const instanceId = args[1];
|
||||
if (!instanceId) {
|
||||
console.error("Usage: orchestrator status <instance-id>");
|
||||
process.exit(1);
|
||||
}
|
||||
printResponse(await sendIpcRequest({ type: "status", instanceId }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (args[0] === "stop") {
|
||||
const instanceId = args[1];
|
||||
if (!instanceId) {
|
||||
console.error("Usage: orchestrator stop <instance-id>");
|
||||
process.exit(1);
|
||||
}
|
||||
printResponse(await sendIpcRequest({ type: "stop", instanceId }));
|
||||
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;
|
||||
}
|
||||
|
||||
if (args[0] === "rpc-stream") {
|
||||
const instanceId = args[1];
|
||||
if (!instanceId) {
|
||||
console.error("Usage: orchestrator rpc-stream <instance-id>");
|
||||
process.exit(1);
|
||||
}
|
||||
await rpcStream(instanceId);
|
||||
return;
|
||||
}
|
||||
|
||||
console.error(`Unknown command: ${args[0]}`);
|
||||
printHelp();
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
await main();
|
||||
@@ -0,0 +1,69 @@
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
import { homedir } from "node:os";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
const CONFIG_DIR_NAME = ".pi";
|
||||
const ENV_ORCHESTRATOR_DIR = "PI_ORCHESTRATOR_DIR";
|
||||
|
||||
const __filename = fileURLToPath(import.meta.url);
|
||||
const __dirname = dirname(__filename);
|
||||
|
||||
/**
|
||||
* Detect if we're running as a Bun compiled binary.
|
||||
* Bun binaries have import.meta.url containing "$bunfs", "~BUN", or "%7EBUN" (Bun's virtual filesystem path)
|
||||
*/
|
||||
export const isBunBinary =
|
||||
import.meta.url.includes("$bunfs") || import.meta.url.includes("~BUN") || import.meta.url.includes("%7EBUN");
|
||||
|
||||
interface PackageJson {
|
||||
version?: string;
|
||||
}
|
||||
|
||||
function getPackageJsonPath(): string {
|
||||
let dir = __dirname;
|
||||
while (dir !== dirname(dir)) {
|
||||
const packageJsonPath = join(dir, "package.json");
|
||||
if (existsSync(packageJsonPath)) {
|
||||
return packageJsonPath;
|
||||
}
|
||||
dir = dirname(dir);
|
||||
}
|
||||
return join(__dirname, "package.json");
|
||||
}
|
||||
|
||||
let pkg: PackageJson = {};
|
||||
try {
|
||||
pkg = JSON.parse(readFileSync(getPackageJsonPath(), "utf-8")) as PackageJson;
|
||||
} catch (e: unknown) {
|
||||
const err = e as NodeJS.ErrnoException;
|
||||
if (err.code !== "ENOENT") throw e;
|
||||
}
|
||||
|
||||
export const VERSION: string = pkg.version || "0.0.0";
|
||||
|
||||
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 getAuthPath(): string {
|
||||
return join(getOrchestratorDir(), "auth.json");
|
||||
}
|
||||
|
||||
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");
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
import type {
|
||||
AgentSessionEvent,
|
||||
RpcCommand,
|
||||
RpcExtensionUIRequest,
|
||||
RpcExtensionUIResponse,
|
||||
RpcResponse,
|
||||
} from "@earendil-works/pi-coding-agent";
|
||||
import type {
|
||||
ErrorResponse,
|
||||
InstanceSummary,
|
||||
ListRequest,
|
||||
ListResponse,
|
||||
OrchestratorRequest,
|
||||
OrchestratorResponse,
|
||||
RpcBridgeResponse,
|
||||
RpcReadyResponse,
|
||||
RpcRequest,
|
||||
RpcStreamRequest,
|
||||
SpawnRequest,
|
||||
SpawnResponse,
|
||||
StatusRequest,
|
||||
StatusResponse,
|
||||
StopRequest,
|
||||
StopResponse,
|
||||
} from "./ipc/protocol.ts";
|
||||
import { supervisor } from "./supervisor.ts";
|
||||
import type { InstanceRecord } from "./types.ts";
|
||||
|
||||
function toInstanceSummary(instance: InstanceRecord): InstanceSummary {
|
||||
return {
|
||||
id: instance.id,
|
||||
status: instance.status,
|
||||
cwd: instance.cwd,
|
||||
label: instance.label,
|
||||
sessionId: instance.sessionId,
|
||||
sessionFile: instance.sessionFile,
|
||||
radiusPiId: instance.radiusPiId,
|
||||
};
|
||||
}
|
||||
|
||||
function unknownInstanceError(instanceId: string): ErrorResponse {
|
||||
return {
|
||||
type: "error",
|
||||
ok: false,
|
||||
error: `Unknown instance: ${instanceId}`,
|
||||
};
|
||||
}
|
||||
|
||||
// Overhead types
|
||||
export async function handleIpcRequest(request: SpawnRequest): Promise<SpawnResponse | ErrorResponse>;
|
||||
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: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse>;
|
||||
export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse>;
|
||||
export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse> {
|
||||
switch (request.type) {
|
||||
case "spawn": {
|
||||
const instance = await supervisor.spawnInstance({
|
||||
cwd: request.cwd,
|
||||
label: request.label,
|
||||
});
|
||||
return {
|
||||
type: "spawn_result",
|
||||
ok: true,
|
||||
instance: toInstanceSummary(instance),
|
||||
};
|
||||
}
|
||||
|
||||
case "list": {
|
||||
return {
|
||||
type: "list_result",
|
||||
ok: true,
|
||||
instances: supervisor.listInstances().map(toInstanceSummary),
|
||||
};
|
||||
}
|
||||
|
||||
case "status": {
|
||||
const instance = supervisor.getInstance(request.instanceId);
|
||||
if (!instance) {
|
||||
return unknownInstanceError(request.instanceId);
|
||||
}
|
||||
|
||||
return {
|
||||
type: "status_result",
|
||||
ok: true,
|
||||
instance: toInstanceSummary(instance),
|
||||
};
|
||||
}
|
||||
|
||||
case "stop": {
|
||||
const instance = await supervisor.stopInstance(request.instanceId);
|
||||
if (!instance) {
|
||||
return unknownInstanceError(request.instanceId);
|
||||
}
|
||||
|
||||
return {
|
||||
type: "stop_result",
|
||||
ok: true,
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
case "rpc_stream": {
|
||||
const instance = supervisor.getInstance(request.instanceId);
|
||||
if (!instance) {
|
||||
return unknownInstanceError(request.instanceId);
|
||||
}
|
||||
return {
|
||||
type: "rpc_ready",
|
||||
ok: true,
|
||||
instance: toInstanceSummary(instance),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function openRpcStream(
|
||||
instanceId: string,
|
||||
onResponse: (response: RpcResponse) => void,
|
||||
onSessionEvent: (event: AgentSessionEvent) => void,
|
||||
onUiRequest: (request: RpcExtensionUIRequest) => void,
|
||||
):
|
||||
| {
|
||||
handleRequest(request: RpcCommand | RpcExtensionUIResponse): Promise<void>;
|
||||
close(): void;
|
||||
}
|
||||
| undefined {
|
||||
const handle = supervisor.openRpcStream(instanceId, onSessionEvent, onUiRequest);
|
||||
if (!handle) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return {
|
||||
async handleRequest(request): Promise<void> {
|
||||
if (request.type === "extension_ui_response") {
|
||||
handle.handleUiResponse(request);
|
||||
return;
|
||||
}
|
||||
const response = await handle.handleRpc(request);
|
||||
onResponse(response);
|
||||
},
|
||||
close(): void {
|
||||
handle.close();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
export * from "./config.ts";
|
||||
export * from "./handler.ts";
|
||||
export * from "./ipc/client.ts";
|
||||
export * from "./ipc/protocol.ts";
|
||||
export * from "./ipc/server.ts";
|
||||
export * from "./rpc-process.ts";
|
||||
export * from "./serve.ts";
|
||||
export * from "./storage.ts";
|
||||
export * from "./supervisor.ts";
|
||||
export * from "./types.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();
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
import type {
|
||||
AgentSessionEvent,
|
||||
RpcCommand,
|
||||
RpcExtensionUIRequest,
|
||||
RpcExtensionUIResponse,
|
||||
RpcResponse,
|
||||
} from "@earendil-works/pi-coding-agent";
|
||||
import type { InstanceStatus } from "../types.ts";
|
||||
|
||||
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 interface RpcRequest {
|
||||
type: "rpc";
|
||||
instanceId: string;
|
||||
command: RpcCommand;
|
||||
}
|
||||
|
||||
export interface RpcStreamRequest {
|
||||
type: "rpc_stream";
|
||||
instanceId: string;
|
||||
}
|
||||
|
||||
export interface RequestMap {
|
||||
spawn: SpawnRequest;
|
||||
list: ListRequest;
|
||||
stop: StopRequest;
|
||||
status: StatusRequest;
|
||||
rpc: RpcRequest;
|
||||
rpc_stream: RpcStreamRequest;
|
||||
}
|
||||
|
||||
export type OrchestratorRequest = RequestMap[keyof RequestMap];
|
||||
|
||||
export interface InstanceSummary {
|
||||
id: string;
|
||||
status: InstanceStatus;
|
||||
cwd: string;
|
||||
label?: string;
|
||||
sessionId?: string;
|
||||
sessionFile?: string;
|
||||
radiusPiId?: string;
|
||||
}
|
||||
|
||||
export interface ResponseBase {
|
||||
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 interface RpcBridgeResponse extends ResponseBase {
|
||||
type: "rpc_result";
|
||||
response: RpcResponse;
|
||||
}
|
||||
|
||||
export interface RpcReadyResponse extends ResponseBase {
|
||||
type: "rpc_ready";
|
||||
instance?: InstanceSummary;
|
||||
}
|
||||
|
||||
export interface ErrorResponse extends ResponseBase {
|
||||
type: "error";
|
||||
ok: false;
|
||||
error: string;
|
||||
}
|
||||
|
||||
export interface ResponseMap {
|
||||
spawn: SpawnResponse;
|
||||
list: ListResponse;
|
||||
stop: StopResponse;
|
||||
status: StatusResponse;
|
||||
rpc: RpcBridgeResponse;
|
||||
rpc_stream: RpcReadyResponse;
|
||||
}
|
||||
|
||||
export type OrchestratorResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;
|
||||
export type RpcClientMessage = RpcCommand | RpcExtensionUIResponse;
|
||||
export type RpcServerMessage =
|
||||
| RpcReadyResponse
|
||||
| RpcResponse
|
||||
| AgentSessionEvent
|
||||
| RpcExtensionUIRequest
|
||||
| ErrorResponse;
|
||||
export type ProtocolMessage = OrchestratorRequest | OrchestratorResponse | RpcClientMessage | RpcServerMessage;
|
||||
|
||||
export type ResponseFor<T extends OrchestratorRequest> = T extends { type: infer K }
|
||||
? K extends keyof ResponseMap
|
||||
? ResponseMap[K] | ErrorResponse
|
||||
: ErrorResponse
|
||||
: ErrorResponse;
|
||||
|
||||
export function encodeMessage(message: ProtocolMessage): 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;
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
import { existsSync, unlinkSync } from "node:fs";
|
||||
import { createConnection, createServer, type Server } from "node:net";
|
||||
import type { AgentSessionEvent, RpcExtensionUIRequest, RpcResponse } from "@earendil-works/pi-coding-agent";
|
||||
import { getSocketPath } from "../config.ts";
|
||||
import {
|
||||
type ErrorResponse,
|
||||
encodeMessage,
|
||||
type ListRequest,
|
||||
type ListResponse,
|
||||
type OrchestratorRequest,
|
||||
type OrchestratorResponse,
|
||||
parseRequestLine,
|
||||
type RpcBridgeResponse,
|
||||
type RpcReadyResponse,
|
||||
type RpcRequest,
|
||||
type RpcStreamRequest,
|
||||
type SpawnRequest,
|
||||
type SpawnResponse,
|
||||
type StatusRequest,
|
||||
type StatusResponse,
|
||||
type StopRequest,
|
||||
type StopResponse,
|
||||
} from "./protocol.ts";
|
||||
|
||||
export interface IpcRequestHandler {
|
||||
(request: SpawnRequest): Promise<SpawnResponse | ErrorResponse> | SpawnResponse | ErrorResponse;
|
||||
(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: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse> | RpcReadyResponse | ErrorResponse;
|
||||
(request: OrchestratorRequest): Promise<OrchestratorResponse> | OrchestratorResponse;
|
||||
openRpcStream(
|
||||
instanceId: string,
|
||||
onResponse: (response: RpcResponse) => void,
|
||||
onSessionEvent: (event: AgentSessionEvent) => void,
|
||||
onUiRequest: (request: RpcExtensionUIRequest) => void,
|
||||
):
|
||||
| {
|
||||
handleRequest(request: RpcRequest["command"] | { type: "extension_ui_response" }): Promise<void>;
|
||||
close(): void;
|
||||
}
|
||||
| undefined;
|
||||
}
|
||||
|
||||
export async function startIpcServer(handler: IpcRequestHandler): Promise<Server> {
|
||||
const socketPath = getSocketPath();
|
||||
await removeStaleSocketIfNeeded(socketPath);
|
||||
|
||||
const server = createServer((socket) => {
|
||||
let buffer = "";
|
||||
|
||||
socket.on("data", async (chunk: Buffer | string) => {
|
||||
buffer += chunk.toString();
|
||||
const newlineIndex = buffer.indexOf("\n");
|
||||
if (newlineIndex === -1) {
|
||||
return;
|
||||
}
|
||||
|
||||
const line = buffer.slice(0, newlineIndex).trim();
|
||||
buffer = buffer.slice(newlineIndex + 1);
|
||||
if (!line) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const request = parseRequestLine(line);
|
||||
if (request.type === "rpc_stream") {
|
||||
const response = await handler(request);
|
||||
if (!response.ok || response.type !== "rpc_ready" || !response.instance) {
|
||||
socket.end(encodeMessage(response));
|
||||
return;
|
||||
}
|
||||
|
||||
socket.removeAllListeners("data");
|
||||
const rpcStream = handler.openRpcStream(
|
||||
request.instanceId,
|
||||
(response) => {
|
||||
socket.write(encodeMessage(response));
|
||||
},
|
||||
(event) => {
|
||||
socket.write(encodeMessage(event));
|
||||
},
|
||||
(request) => {
|
||||
socket.write(encodeMessage(request));
|
||||
},
|
||||
);
|
||||
if (!rpcStream) {
|
||||
socket.end(
|
||||
encodeMessage({ type: "error", ok: false, error: `Unknown instance: ${request.instanceId}` }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
socket.write(encodeMessage(response));
|
||||
let rpcRequestQueue = Promise.resolve();
|
||||
socket.on("data", (rpcChunk: Buffer | string) => {
|
||||
buffer += rpcChunk.toString();
|
||||
for (;;) {
|
||||
const rpcNewlineIndex = buffer.indexOf("\n");
|
||||
if (rpcNewlineIndex === -1) {
|
||||
break;
|
||||
}
|
||||
const rpcLine = buffer.slice(0, rpcNewlineIndex).trim();
|
||||
buffer = buffer.slice(rpcNewlineIndex + 1);
|
||||
if (!rpcLine) {
|
||||
continue;
|
||||
}
|
||||
rpcRequestQueue = rpcRequestQueue
|
||||
.then(async () => {
|
||||
try {
|
||||
await rpcStream.handleRequest(JSON.parse(rpcLine));
|
||||
} catch (rpcError: unknown) {
|
||||
socket.write(
|
||||
encodeMessage({
|
||||
type: "error",
|
||||
ok: false,
|
||||
error: rpcError instanceof Error ? rpcError.message : String(rpcError),
|
||||
}),
|
||||
);
|
||||
}
|
||||
})
|
||||
.catch((rpcError: Error) => {
|
||||
socket.write(
|
||||
encodeMessage({
|
||||
type: "error",
|
||||
ok: false,
|
||||
error: rpcError.message,
|
||||
}),
|
||||
);
|
||||
});
|
||||
}
|
||||
});
|
||||
socket.once("close", () => rpcStream.close());
|
||||
return;
|
||||
}
|
||||
|
||||
const response = await handler(request);
|
||||
socket.end(encodeMessage(response));
|
||||
} catch (error: unknown) {
|
||||
const response: ErrorResponse = {
|
||||
type: "error",
|
||||
ok: false,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
socket.end(encodeMessage(response));
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(socketPath, () => {
|
||||
server.off("error", reject);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
|
||||
return server;
|
||||
}
|
||||
|
||||
async function removeStaleSocketIfNeeded(socketPath: string): Promise<void> {
|
||||
if (!existsSync(socketPath)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const isLive = await isSocketLive(socketPath);
|
||||
if (isLive) {
|
||||
throw new Error(`orchestrator is already running: ${socketPath}`);
|
||||
}
|
||||
|
||||
unlinkSync(socketPath);
|
||||
}
|
||||
|
||||
async function isSocketLive(socketPath: string): Promise<boolean> {
|
||||
return new Promise<boolean>((resolve, reject) => {
|
||||
const socket = createConnection(socketPath);
|
||||
let settled = false;
|
||||
|
||||
const finish = (result: boolean) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
socket.removeAllListeners();
|
||||
socket.destroy();
|
||||
resolve(result);
|
||||
};
|
||||
|
||||
socket.on("connect", () => finish(true));
|
||||
socket.on("error", (error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ECONNREFUSED" || error.code === "ENOENT") {
|
||||
finish(false);
|
||||
return;
|
||||
}
|
||||
if (error.code === "EPIPE" || error.code === "ECONNRESET") {
|
||||
finish(false);
|
||||
return;
|
||||
}
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
socket.removeAllListeners();
|
||||
socket.destroy();
|
||||
reject(error);
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,445 @@
|
||||
import { hostname, platform } from "node:os";
|
||||
import { AuthStorage, type OAuthCredential } from "@earendil-works/pi-coding-agent";
|
||||
import { getOrchestratorDir, getSocketPath, VERSION } from "./config.ts";
|
||||
import { loadMachine, saveMachine } from "./storage.ts";
|
||||
import type { InstanceRecord, MachineRecord, RadiusRegistration } from "./types.ts";
|
||||
|
||||
const DEFAULT_RADIUS_URL = "https://radius.pi.dev/";
|
||||
const DEFAULT_ORCHESTRATOR_BASE_PATH = "/v1/";
|
||||
const NOT_FOUND_RETRY_THRESHOLD = 3;
|
||||
const HEARTBEAT_BACKOFF_BASE_MS = 1_000;
|
||||
const HEARTBEAT_BACKOFF_MAX_MS = 30_000;
|
||||
const RADIUS_PROVIDER = "radius";
|
||||
|
||||
interface RegisterMachineResponse extends RadiusRegistration {
|
||||
id: string;
|
||||
}
|
||||
|
||||
interface RegisterPiResponse extends RadiusRegistration {
|
||||
id: string;
|
||||
}
|
||||
|
||||
interface RadiusPresenceCoordinator {
|
||||
getLiveInstance(instanceId: string): InstanceRecord | undefined;
|
||||
listLiveInstances(): InstanceRecord[];
|
||||
updateInstance(instance: InstanceRecord): void;
|
||||
}
|
||||
|
||||
interface PiHeartbeatState {
|
||||
timer?: NodeJS.Timeout;
|
||||
intervalMs: number;
|
||||
radiusPiId: string;
|
||||
consecutiveNotFoundCount: number;
|
||||
transientFailureCount: number;
|
||||
}
|
||||
|
||||
class RadiusHttpError extends Error {
|
||||
readonly status: number;
|
||||
|
||||
constructor(status: number, message: string) {
|
||||
super(message);
|
||||
this.name = "RadiusHttpError";
|
||||
this.status = status;
|
||||
}
|
||||
}
|
||||
|
||||
async function post<T>(path: string, body: unknown): Promise<T> {
|
||||
const response = await fetch(new URL(path, getRadiusOrchestratorBaseUrl()), {
|
||||
method: "POST",
|
||||
headers: {
|
||||
Authorization: `Bearer ${getRadiusAccessToken()}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new RadiusHttpError(response.status, `Radius request failed: ${response.status} ${await response.text()}`);
|
||||
}
|
||||
|
||||
return (await response.json()) as T;
|
||||
}
|
||||
|
||||
async function maybePost(path: string, body: unknown): Promise<void> {
|
||||
const response = await fetch(new URL(path, getRadiusOrchestratorBaseUrl()), {
|
||||
method: "POST",
|
||||
headers: {
|
||||
Authorization: `Bearer ${getRadiusAccessToken()}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new RadiusHttpError(response.status, `Radius request failed: ${response.status} ${await response.text()}`);
|
||||
}
|
||||
}
|
||||
|
||||
function isNotFoundError(error: unknown): error is RadiusHttpError {
|
||||
return error instanceof RadiusHttpError && error.status === 404;
|
||||
}
|
||||
|
||||
function computeBackoffDelayMs(failureCount: number): number {
|
||||
const exponentialDelay = Math.min(
|
||||
HEARTBEAT_BACKOFF_MAX_MS,
|
||||
HEARTBEAT_BACKOFF_BASE_MS * 2 ** Math.max(0, failureCount - 1),
|
||||
);
|
||||
const jitterMs = Math.floor(Math.random() * Math.max(250, exponentialDelay / 4));
|
||||
return Math.min(HEARTBEAT_BACKOFF_MAX_MS, exponentialDelay + jitterMs);
|
||||
}
|
||||
|
||||
function formatRadiusError(error: unknown): string {
|
||||
if (error instanceof RadiusHttpError) {
|
||||
return `HTTP ${error.status}: ${error.message}`;
|
||||
}
|
||||
if (error instanceof Error) {
|
||||
return error.message;
|
||||
}
|
||||
return String(error);
|
||||
}
|
||||
|
||||
function logRadiusRetry(scope: string, action: string, delayMs: number, failureCount: number, error: unknown): void {
|
||||
console.error(
|
||||
`${scope} ${action} failed (attempt ${failureCount}); retrying in ${delayMs}ms: ${formatRadiusError(error)}`,
|
||||
);
|
||||
}
|
||||
|
||||
export function getRadiusUrl(): string {
|
||||
return process.env.PI_RADIUS_URL || DEFAULT_RADIUS_URL;
|
||||
}
|
||||
|
||||
export function getRadiusOrchestratorBaseUrl(): string {
|
||||
const explicitUrl = process.env.PI_RADIUS_ORCHESTRATOR_URL;
|
||||
if (explicitUrl) {
|
||||
return explicitUrl;
|
||||
}
|
||||
|
||||
return new URL(DEFAULT_ORCHESTRATOR_BASE_PATH, getRadiusUrl()).toString();
|
||||
}
|
||||
|
||||
const radiusAuthStorage = AuthStorage.create();
|
||||
|
||||
function getStoredRadiusCredential(): OAuthCredential | undefined {
|
||||
radiusAuthStorage.reload();
|
||||
const credential = radiusAuthStorage.get(RADIUS_PROVIDER);
|
||||
if (!credential || credential.type !== "oauth") {
|
||||
return undefined;
|
||||
}
|
||||
return credential;
|
||||
}
|
||||
|
||||
export function getRadiusAccessToken(): string {
|
||||
const storedCredential = getStoredRadiusCredential();
|
||||
if (typeof storedCredential?.access === "string" && storedCredential.access) {
|
||||
return storedCredential.access;
|
||||
}
|
||||
|
||||
const apiKey = process.env.PI_RADIUS_API_KEY;
|
||||
if (apiKey) {
|
||||
return apiKey;
|
||||
}
|
||||
|
||||
throw new Error("Radius credentials are required in ~/.pi/agent/auth.json or PI_RADIUS_API_KEY");
|
||||
}
|
||||
|
||||
export function isRadiusEnabled(): boolean {
|
||||
return !!getStoredRadiusCredential()?.access || !!process.env.PI_RADIUS_API_KEY;
|
||||
}
|
||||
|
||||
export class RadiusPresence {
|
||||
private machineHeartbeatTimer?: NodeJS.Timeout;
|
||||
private machineHeartbeatIntervalMs = 0;
|
||||
private machineConsecutiveNotFoundCount = 0;
|
||||
private machineTransientFailureCount = 0;
|
||||
private readonly piHeartbeatStates = new Map<string, PiHeartbeatState>();
|
||||
private machine?: MachineRecord;
|
||||
private coordinator?: RadiusPresenceCoordinator;
|
||||
|
||||
setCoordinator(coordinator: RadiusPresenceCoordinator): void {
|
||||
this.coordinator = coordinator;
|
||||
}
|
||||
|
||||
async start(label?: string): Promise<MachineRecord | undefined> {
|
||||
if (!isRadiusEnabled()) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const registered = await this.registerMachine(label);
|
||||
this.startMachineHeartbeat(registered.heartbeatIntervalMs);
|
||||
return this.machine;
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
if (this.machineHeartbeatTimer) {
|
||||
clearTimeout(this.machineHeartbeatTimer);
|
||||
this.machineHeartbeatTimer = undefined;
|
||||
}
|
||||
for (const [instanceId, state] of this.piHeartbeatStates) {
|
||||
if (state.timer) {
|
||||
clearTimeout(state.timer);
|
||||
}
|
||||
this.piHeartbeatStates.delete(instanceId);
|
||||
}
|
||||
if (!this.machine || !isRadiusEnabled()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await maybePost(`machines/${this.machine.id}/disconnect`, {});
|
||||
} catch (error) {
|
||||
if (!isNotFoundError(error)) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async registerPi(instance: InstanceRecord): Promise<InstanceRecord> {
|
||||
if (!isRadiusEnabled()) {
|
||||
return instance;
|
||||
}
|
||||
const machine = this.machine ?? loadMachine();
|
||||
if (!machine) {
|
||||
throw new Error("No registered machine available for Pi registration");
|
||||
}
|
||||
const registered = await post<RegisterPiResponse>("pis/register", {
|
||||
machineId: machine.id,
|
||||
label: instance.label,
|
||||
cwd: instance.cwd,
|
||||
hostname: hostname(),
|
||||
pid: process.pid,
|
||||
transport: "local-rpc",
|
||||
capabilities: { rpc: true, relay: false, iroh: false },
|
||||
sessionId: instance.sessionId,
|
||||
});
|
||||
const registeredInstance = { ...instance, radiusPiId: registered.id };
|
||||
this.startPiHeartbeat(instance.id, registered.heartbeatIntervalMs, registered.id);
|
||||
return registeredInstance;
|
||||
}
|
||||
|
||||
async disconnectPi(instance: InstanceRecord): Promise<void> {
|
||||
const state = this.piHeartbeatStates.get(instance.id);
|
||||
if (state) {
|
||||
if (state.timer) {
|
||||
clearTimeout(state.timer);
|
||||
}
|
||||
this.piHeartbeatStates.delete(instance.id);
|
||||
}
|
||||
if (!isRadiusEnabled() || !instance.radiusPiId) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await maybePost(`pis/${instance.radiusPiId}/disconnect`, {});
|
||||
} catch (error) {
|
||||
if (!isNotFoundError(error)) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async registerMachine(label?: string): Promise<RegisterMachineResponse> {
|
||||
const existingMachine = this.machine ?? loadMachine();
|
||||
const registered = await post<RegisterMachineResponse>("machines/register", {
|
||||
machineId: existingMachine?.id,
|
||||
label,
|
||||
hostname: hostname(),
|
||||
platform: platform(),
|
||||
arch: process.arch,
|
||||
version: VERSION,
|
||||
capabilities: { spawn: true, relay: false, iroh: false },
|
||||
});
|
||||
|
||||
const timestamp = new Date().toISOString();
|
||||
this.machine = {
|
||||
id: registered.id,
|
||||
createdAt: existingMachine?.createdAt ?? timestamp,
|
||||
lastSeenAt: timestamp,
|
||||
label,
|
||||
};
|
||||
saveMachine(this.machine);
|
||||
this.machineConsecutiveNotFoundCount = 0;
|
||||
this.machineTransientFailureCount = 0;
|
||||
return registered;
|
||||
}
|
||||
|
||||
private startMachineHeartbeat(intervalMs: number): void {
|
||||
this.machineHeartbeatIntervalMs = intervalMs;
|
||||
this.scheduleMachineHeartbeat(intervalMs);
|
||||
}
|
||||
|
||||
private scheduleMachineHeartbeat(delayMs: number): void {
|
||||
if (this.machineHeartbeatTimer) {
|
||||
clearTimeout(this.machineHeartbeatTimer);
|
||||
}
|
||||
this.machineHeartbeatTimer = setTimeout(() => {
|
||||
void this.heartbeatMachine();
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
private startPiHeartbeat(instanceId: string, intervalMs: number, radiusPiId: string): void {
|
||||
const existingState = this.piHeartbeatStates.get(instanceId);
|
||||
if (existingState?.timer) {
|
||||
clearTimeout(existingState.timer);
|
||||
}
|
||||
const state: PiHeartbeatState = existingState ?? {
|
||||
intervalMs,
|
||||
radiusPiId,
|
||||
consecutiveNotFoundCount: 0,
|
||||
transientFailureCount: 0,
|
||||
};
|
||||
state.intervalMs = intervalMs;
|
||||
state.radiusPiId = radiusPiId;
|
||||
state.consecutiveNotFoundCount = 0;
|
||||
state.transientFailureCount = 0;
|
||||
this.piHeartbeatStates.set(instanceId, state);
|
||||
this.schedulePiHeartbeat(instanceId, intervalMs);
|
||||
}
|
||||
|
||||
private schedulePiHeartbeat(instanceId: string, delayMs: number): void {
|
||||
const state = this.piHeartbeatStates.get(instanceId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
if (state.timer) {
|
||||
clearTimeout(state.timer);
|
||||
}
|
||||
state.timer = setTimeout(() => {
|
||||
void this.heartbeatPi(instanceId);
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
private async heartbeatMachine(): Promise<void> {
|
||||
if (!this.machine || !isRadiusEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await maybePost(`machines/${this.machine.id}/heartbeat`, {
|
||||
cwd: getOrchestratorDir(),
|
||||
socketPath: getSocketPath(),
|
||||
});
|
||||
this.machineConsecutiveNotFoundCount = 0;
|
||||
this.machineTransientFailureCount = 0;
|
||||
this.scheduleMachineHeartbeat(this.machineHeartbeatIntervalMs);
|
||||
} catch (error) {
|
||||
if (!isNotFoundError(error)) {
|
||||
this.machineTransientFailureCount += 1;
|
||||
const delayMs = computeBackoffDelayMs(this.machineTransientFailureCount);
|
||||
logRadiusRetry("Radius machine", "heartbeat", delayMs, this.machineTransientFailureCount, error);
|
||||
this.scheduleMachineHeartbeat(delayMs);
|
||||
return;
|
||||
}
|
||||
|
||||
this.machineTransientFailureCount = 0;
|
||||
this.machineConsecutiveNotFoundCount += 1;
|
||||
if (this.machineConsecutiveNotFoundCount < NOT_FOUND_RETRY_THRESHOLD) {
|
||||
this.scheduleMachineHeartbeat(this.machineHeartbeatIntervalMs);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.reRegisterMachineAndPis();
|
||||
} catch (recoveryError) {
|
||||
this.machineTransientFailureCount += 1;
|
||||
const delayMs = computeBackoffDelayMs(this.machineTransientFailureCount);
|
||||
logRadiusRetry(
|
||||
"Radius machine",
|
||||
"re-registration",
|
||||
delayMs,
|
||||
this.machineTransientFailureCount,
|
||||
recoveryError,
|
||||
);
|
||||
this.scheduleMachineHeartbeat(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async heartbeatPi(instanceId: string): Promise<void> {
|
||||
if (!isRadiusEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const state = this.piHeartbeatStates.get(instanceId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await maybePost(`pis/${state.radiusPiId}/heartbeat`, {});
|
||||
state.consecutiveNotFoundCount = 0;
|
||||
state.transientFailureCount = 0;
|
||||
this.schedulePiHeartbeat(instanceId, state.intervalMs);
|
||||
} catch (error) {
|
||||
if (!isNotFoundError(error)) {
|
||||
state.transientFailureCount += 1;
|
||||
const delayMs = computeBackoffDelayMs(state.transientFailureCount);
|
||||
logRadiusRetry(`Radius Pi ${instanceId}`, "heartbeat", delayMs, state.transientFailureCount, error);
|
||||
this.schedulePiHeartbeat(instanceId, delayMs);
|
||||
return;
|
||||
}
|
||||
|
||||
state.transientFailureCount = 0;
|
||||
state.consecutiveNotFoundCount += 1;
|
||||
if (state.consecutiveNotFoundCount < NOT_FOUND_RETRY_THRESHOLD) {
|
||||
this.schedulePiHeartbeat(instanceId, state.intervalMs);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const recovered = await this.reRegisterPi(instanceId);
|
||||
if (!recovered) {
|
||||
const delayMs = computeBackoffDelayMs(1);
|
||||
console.error(`Radius Pi ${instanceId} re-registration skipped; retrying in ${delayMs}ms`);
|
||||
this.schedulePiHeartbeat(instanceId, delayMs);
|
||||
}
|
||||
} catch (recoveryError) {
|
||||
state.transientFailureCount += 1;
|
||||
const delayMs = computeBackoffDelayMs(state.transientFailureCount);
|
||||
logRadiusRetry(
|
||||
`Radius Pi ${instanceId}`,
|
||||
"re-registration",
|
||||
delayMs,
|
||||
state.transientFailureCount,
|
||||
recoveryError,
|
||||
);
|
||||
this.schedulePiHeartbeat(instanceId, delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async reRegisterMachineAndPis(): Promise<void> {
|
||||
const registered = await this.registerMachine(this.machine?.label);
|
||||
this.startMachineHeartbeat(registered.heartbeatIntervalMs);
|
||||
|
||||
const instances = this.coordinator?.listLiveInstances() ?? [];
|
||||
for (const instance of instances) {
|
||||
try {
|
||||
await this.reRegisterPi(instance.id);
|
||||
} catch (error) {
|
||||
console.error(`Radius Pi ${instance.id} re-registration failed: ${formatRadiusError(error)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async reRegisterPi(instanceId: string): Promise<boolean> {
|
||||
const instance = this.coordinator?.getLiveInstance(instanceId);
|
||||
if (!instance) {
|
||||
const state = this.piHeartbeatStates.get(instanceId);
|
||||
if (state) {
|
||||
if (state.timer) {
|
||||
clearTimeout(state.timer);
|
||||
}
|
||||
this.piHeartbeatStates.delete(instanceId);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!this.machine) {
|
||||
await this.reRegisterMachineAndPis();
|
||||
return true;
|
||||
}
|
||||
|
||||
const registeredInstance = await this.registerPi(instance);
|
||||
this.coordinator?.updateInstance(registeredInstance);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
export const radiusPresence = new RadiusPresence();
|
||||
@@ -0,0 +1,201 @@
|
||||
import { type ChildProcess, spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createRequire } from "node:module";
|
||||
import { dirname, join } from "node:path";
|
||||
import type {
|
||||
AgentSessionEvent,
|
||||
RpcCommand,
|
||||
RpcExtensionUIRequest,
|
||||
RpcExtensionUIResponse,
|
||||
RpcResponse,
|
||||
} from "@earendil-works/pi-coding-agent";
|
||||
import { isBunBinary } from "./config.ts";
|
||||
|
||||
interface PendingRequest {
|
||||
resolve(response: RpcResponse): void;
|
||||
reject(error: Error): void;
|
||||
}
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
|
||||
function toError(error: unknown): Error {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
|
||||
export class RpcProcessInstance {
|
||||
readonly process: ChildProcess;
|
||||
|
||||
private exited = false;
|
||||
private nextRequestId = 0;
|
||||
private stdoutBuffer = "";
|
||||
private stderrBuffer = "";
|
||||
private readonly pendingRequests = new Map<string, PendingRequest>();
|
||||
private readonly eventListeners = new Set<(event: AgentSessionEvent) => void>();
|
||||
private readonly exitListeners = new Set<(error?: Error) => void>();
|
||||
private uiRequestHandler: ((request: RpcExtensionUIRequest) => void) | undefined;
|
||||
|
||||
constructor(options: { cwd: string }) {
|
||||
const rpcCommand = this.getSpawnCommand();
|
||||
this.process = spawn(rpcCommand.command, rpcCommand.args, {
|
||||
cwd: options.cwd,
|
||||
env: process.env,
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
});
|
||||
if (!this.process.stdin || !this.process.stdout) {
|
||||
throw new Error("Failed to create RPC process stdio");
|
||||
}
|
||||
this.attachListeners();
|
||||
}
|
||||
|
||||
private getSpawnCommand(): { command: string; args: string[] } {
|
||||
if (isBunBinary) {
|
||||
return {
|
||||
command: join(dirname(process.execPath), process.platform === "win32" ? "pi.exe" : "pi"),
|
||||
args: ["--mode", "rpc"],
|
||||
};
|
||||
}
|
||||
return {
|
||||
command: process.execPath,
|
||||
args: [require.resolve("@earendil-works/pi-coding-agent/rpc-entry")],
|
||||
};
|
||||
}
|
||||
|
||||
private attachListeners(): void {
|
||||
this.process.stdout?.setEncoding("utf8");
|
||||
this.process.stdout?.on("data", (chunk: string) => {
|
||||
this.stdoutBuffer += chunk;
|
||||
while (true) {
|
||||
const newlineIndex = this.stdoutBuffer.indexOf("\n");
|
||||
if (newlineIndex === -1) {
|
||||
break;
|
||||
}
|
||||
const line = this.stdoutBuffer.slice(0, newlineIndex).trim();
|
||||
this.stdoutBuffer = this.stdoutBuffer.slice(newlineIndex + 1);
|
||||
if (!line) {
|
||||
continue;
|
||||
}
|
||||
this.handleLine(line);
|
||||
}
|
||||
});
|
||||
|
||||
this.process.stderr?.setEncoding("utf8");
|
||||
this.process.stderr?.on("data", (chunk: string) => {
|
||||
this.stderrBuffer += chunk;
|
||||
});
|
||||
|
||||
this.process.once("error", (error) => {
|
||||
this.exited = true;
|
||||
const wrapped = new Error(`RPC process error: ${error.message}. Stderr: ${this.stderrBuffer}`);
|
||||
this.rejectAllPending(wrapped);
|
||||
this.notifyExit(wrapped);
|
||||
});
|
||||
|
||||
this.process.once("exit", (code, signal) => {
|
||||
this.exited = true;
|
||||
const error = new Error(`RPC process exited (code=${code} signal=${signal}). Stderr: ${this.stderrBuffer}`);
|
||||
this.rejectAllPending(error);
|
||||
this.notifyExit(error);
|
||||
});
|
||||
}
|
||||
|
||||
private handleLine(line: string): void {
|
||||
const parsed = JSON.parse(line) as { type?: string; id?: string };
|
||||
switch (parsed.type) {
|
||||
case "response": {
|
||||
if (!parsed.id) {
|
||||
return;
|
||||
}
|
||||
const pending = this.pendingRequests.get(parsed.id);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
this.pendingRequests.delete(parsed.id);
|
||||
pending.resolve(parsed as RpcResponse);
|
||||
return;
|
||||
}
|
||||
|
||||
case "extension_ui_request": {
|
||||
this.uiRequestHandler?.(parsed as RpcExtensionUIRequest);
|
||||
return;
|
||||
}
|
||||
|
||||
default: {
|
||||
for (const listener of this.eventListeners) {
|
||||
listener(parsed as AgentSessionEvent);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private rejectAllPending(error: Error): void {
|
||||
for (const [id, pending] of this.pendingRequests) {
|
||||
this.pendingRequests.delete(id);
|
||||
pending.reject(error);
|
||||
}
|
||||
}
|
||||
|
||||
private notifyExit(error?: Error): void {
|
||||
for (const listener of this.exitListeners) {
|
||||
listener(error);
|
||||
}
|
||||
}
|
||||
|
||||
send(command: RpcCommand): Promise<RpcResponse> {
|
||||
if (this.exited) {
|
||||
throw new Error(`RPC process is not running. Stderr: ${this.stderrBuffer}`);
|
||||
}
|
||||
const id = command.id ?? `orchestrator_${++this.nextRequestId}_${randomUUID()}`;
|
||||
const fullCommand = { ...command, id };
|
||||
return new Promise<RpcResponse>((resolve, reject) => {
|
||||
this.pendingRequests.set(id, { resolve, reject });
|
||||
this.process.stdin?.write(`${JSON.stringify(fullCommand)}\n`, (error) => {
|
||||
if (!error) {
|
||||
return;
|
||||
}
|
||||
this.pendingRequests.delete(id);
|
||||
reject(toError(error));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
handleUiResponse(response: RpcExtensionUIResponse): void {
|
||||
if (this.exited) {
|
||||
return;
|
||||
}
|
||||
this.process.stdin?.write(`${JSON.stringify(response)}\n`);
|
||||
}
|
||||
|
||||
setUiRequestHandler(handler?: (request: RpcExtensionUIRequest) => void): void {
|
||||
this.uiRequestHandler = handler;
|
||||
}
|
||||
|
||||
onEvent(listener: (event: AgentSessionEvent) => void): () => void {
|
||||
this.eventListeners.add(listener);
|
||||
return () => {
|
||||
this.eventListeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
onExit(listener: (error?: Error) => void): () => void {
|
||||
this.exitListeners.add(listener);
|
||||
return () => {
|
||||
this.exitListeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
async dispose(): Promise<void> {
|
||||
this.uiRequestHandler = undefined;
|
||||
this.rejectAllPending(new Error("RPC process disposed"));
|
||||
if (this.exited) {
|
||||
return;
|
||||
}
|
||||
this.process.kill("SIGTERM");
|
||||
await new Promise<void>((resolve) => {
|
||||
this.process.once("exit", () => resolve());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function createRpcProcessInstance(options: { cwd: string }): RpcProcessInstance {
|
||||
return new RpcProcessInstance(options);
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
import { existsSync, mkdirSync, unlinkSync } from "node:fs";
|
||||
import { dirname } from "node:path";
|
||||
import { getSocketPath } from "./config.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";
|
||||
|
||||
export async function serve(): Promise<void> {
|
||||
const socketPath = getSocketPath();
|
||||
mkdirSync(dirname(socketPath), { recursive: true });
|
||||
const server = await startIpcServer(
|
||||
Object.assign(handleIpcRequest, {
|
||||
openRpcStream,
|
||||
}),
|
||||
);
|
||||
|
||||
try {
|
||||
await supervisor.recoverAfterRestart();
|
||||
if (isRadiusEnabled()) {
|
||||
const machine = await radiusPresence.start();
|
||||
console.log(`radius integration enabled: ${socketPath} -> ${getRadiusOrchestratorBaseUrl()}`);
|
||||
if (machine) {
|
||||
console.log(`radius machine id: ${machine.id}`);
|
||||
}
|
||||
} else {
|
||||
console.log("radius integration disabled: login radius in ~/.pi/agent/auth.json or set PI_RADIUS_API_KEY");
|
||||
}
|
||||
} catch (error) {
|
||||
server.close();
|
||||
if (existsSync(socketPath)) {
|
||||
unlinkSync(socketPath);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
console.log(`orchestrator listening on ${socketPath}`);
|
||||
|
||||
let shutdownPromise: Promise<void> | undefined;
|
||||
const shutdown = async (exitCode: number) => {
|
||||
if (shutdownPromise) {
|
||||
await shutdownPromise;
|
||||
process.exit(exitCode);
|
||||
}
|
||||
|
||||
shutdownPromise = (async () => {
|
||||
server.close();
|
||||
await supervisor.shutdown();
|
||||
await radiusPresence.stop();
|
||||
if (existsSync(socketPath)) {
|
||||
unlinkSync(socketPath);
|
||||
}
|
||||
})();
|
||||
|
||||
await shutdownPromise;
|
||||
process.exit(exitCode);
|
||||
};
|
||||
|
||||
process.on("SIGINT", () => {
|
||||
void shutdown(0);
|
||||
});
|
||||
process.on("SIGTERM", () => {
|
||||
void shutdown(0);
|
||||
});
|
||||
process.on("uncaughtException", (error) => {
|
||||
console.error(error);
|
||||
void shutdown(1);
|
||||
});
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
console.error(reason);
|
||||
void shutdown(1);
|
||||
});
|
||||
|
||||
await new Promise<void>(() => {
|
||||
// Keep the process alive until a signal or fatal error triggers shutdown.
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { getInstancesPath, getMachinePath, getOrchestratorDir } from "./config.ts";
|
||||
import type { InstanceRecord, MachineRecord } from "./types.ts";
|
||||
|
||||
function ensureOrchestratorDir(): void {
|
||||
const orchestratorDir = getOrchestratorDir();
|
||||
if (!existsSync(orchestratorDir)) {
|
||||
mkdirSync(orchestratorDir, { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
export function loadMachine(): MachineRecord | undefined {
|
||||
const machinePath = getMachinePath();
|
||||
if (!existsSync(machinePath)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const data = readFileSync(machinePath, "utf-8");
|
||||
return JSON.parse(data) as MachineRecord;
|
||||
}
|
||||
|
||||
export function saveMachine(machine: MachineRecord): void {
|
||||
ensureOrchestratorDir();
|
||||
writeFileSync(getMachinePath(), JSON.stringify(machine, null, 2));
|
||||
}
|
||||
|
||||
export function deleteMachine(): void {
|
||||
const machinePath = getMachinePath();
|
||||
if (!existsSync(machinePath)) {
|
||||
return;
|
||||
}
|
||||
rmSync(machinePath);
|
||||
}
|
||||
|
||||
export function loadInstances(): InstanceRecord[] {
|
||||
const instancesPath = getInstancesPath();
|
||||
if (!existsSync(instancesPath)) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const data = readFileSync(instancesPath, "utf-8");
|
||||
return JSON.parse(data) as InstanceRecord[];
|
||||
}
|
||||
|
||||
export function saveInstances(instances: InstanceRecord[]): void {
|
||||
ensureOrchestratorDir();
|
||||
writeFileSync(getInstancesPath(), JSON.stringify(instances, null, 2));
|
||||
}
|
||||
|
||||
export function getInstance(instanceId: string): InstanceRecord | undefined {
|
||||
return loadInstances().find((instance) => instance.id === instanceId);
|
||||
}
|
||||
|
||||
export function upsertInstance(instance: InstanceRecord): void {
|
||||
const instances = loadInstances();
|
||||
const index = instances.findIndex((existing) => existing.id === instance.id);
|
||||
if (index === -1) {
|
||||
instances.push(instance);
|
||||
saveInstances(instances);
|
||||
return;
|
||||
}
|
||||
|
||||
instances[index] = instance;
|
||||
saveInstances(instances);
|
||||
}
|
||||
|
||||
export function removeInstance(instanceId: string): void {
|
||||
const instances = loadInstances().filter((instance) => instance.id !== instanceId);
|
||||
saveInstances(instances);
|
||||
}
|
||||
@@ -0,0 +1,354 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type {
|
||||
AgentSessionEvent,
|
||||
AgentSessionEventListener,
|
||||
RpcCommand,
|
||||
RpcExtensionUIRequest,
|
||||
RpcExtensionUIResponse,
|
||||
RpcResponse,
|
||||
} from "@earendil-works/pi-coding-agent";
|
||||
import { radiusPresence } from "./radius.ts";
|
||||
import { createRpcProcessInstance, type RpcProcessInstance } from "./rpc-process.ts";
|
||||
import { getInstance, loadInstances, removeInstance, saveInstances, upsertInstance } from "./storage.ts";
|
||||
import type { InstanceRecord, InstanceStatus } from "./types.ts";
|
||||
|
||||
interface LiveInstanceResources {
|
||||
rpcProcess?: RpcProcessInstance;
|
||||
radiusPiId?: string;
|
||||
sessionId?: string;
|
||||
}
|
||||
|
||||
interface LiveInstance {
|
||||
record: InstanceRecord;
|
||||
resources: LiveInstanceResources;
|
||||
subscribers: Set<AgentSessionEventListener>;
|
||||
onUiRequest?: (request: RpcExtensionUIRequest) => void;
|
||||
unsubscribeEvents?: () => void;
|
||||
unsubscribeExit?: () => void;
|
||||
}
|
||||
|
||||
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<RpcCommand["type"]> = 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<
|
||||
RpcResponse,
|
||||
{ success: true; command: "get_state"; data: { sessionId: string; sessionFile?: string } }
|
||||
> {
|
||||
return response.success === true && response.command === "get_state" && "data" in response;
|
||||
}
|
||||
|
||||
export class OrchestratorSupervisor {
|
||||
private readonly liveInstances = new Map<string, LiveInstance>();
|
||||
|
||||
private setStatus(live: LiveInstance, status: InstanceStatus): void {
|
||||
live.record = {
|
||||
...live.record,
|
||||
status,
|
||||
lastSeenAt: new Date().toISOString(),
|
||||
};
|
||||
upsertInstance(live.record);
|
||||
}
|
||||
|
||||
private updateRecord(live: LiveInstance, updates: Partial<InstanceRecord>): void {
|
||||
live.record = {
|
||||
...live.record,
|
||||
...updates,
|
||||
lastSeenAt: new Date().toISOString(),
|
||||
};
|
||||
if (updates.radiusPiId !== undefined) {
|
||||
live.resources.radiusPiId = updates.radiusPiId;
|
||||
}
|
||||
if (updates.sessionId !== undefined) {
|
||||
live.resources.sessionId = updates.sessionId;
|
||||
}
|
||||
upsertInstance(live.record);
|
||||
}
|
||||
|
||||
private clearBindings(live: LiveInstance): void {
|
||||
live.unsubscribeEvents?.();
|
||||
live.unsubscribeExit?.();
|
||||
live.unsubscribeEvents = undefined;
|
||||
live.unsubscribeExit = undefined;
|
||||
live.onUiRequest = undefined;
|
||||
live.resources.rpcProcess?.setUiRequestHandler(undefined);
|
||||
}
|
||||
|
||||
private bindRpcProcess(live: LiveInstance, rpcProcess: RpcProcessInstance): void {
|
||||
this.clearBindings(live);
|
||||
live.resources.rpcProcess = rpcProcess;
|
||||
live.unsubscribeEvents = rpcProcess.onEvent((event) => {
|
||||
for (const subscriber of live.subscribers) {
|
||||
subscriber(event);
|
||||
}
|
||||
});
|
||||
live.unsubscribeExit = rpcProcess.onExit((error) => {
|
||||
void this.handleUnexpectedRpcExit(live, error);
|
||||
});
|
||||
rpcProcess.setUiRequestHandler((request) => {
|
||||
live.onUiRequest?.(request);
|
||||
});
|
||||
}
|
||||
|
||||
private async handleUnexpectedRpcExit(live: LiveInstance, _error?: Error): Promise<void> {
|
||||
if (this.liveInstances.get(live.record.id) !== live) {
|
||||
return;
|
||||
}
|
||||
if (live.record.status === "stopping" || live.record.status === "stopped") {
|
||||
return;
|
||||
}
|
||||
this.setStatus(live, "error");
|
||||
this.clearBindings(live);
|
||||
live.resources.rpcProcess = undefined;
|
||||
if (live.resources.radiusPiId) {
|
||||
try {
|
||||
await radiusPresence.disconnectPi(live.record);
|
||||
this.updateRecord(live, { radiusPiId: undefined });
|
||||
} catch (error) {
|
||||
console.error(`Failed to disconnect Radius Pi ${live.record.id}: ${String(error)}`);
|
||||
}
|
||||
}
|
||||
this.liveInstances.delete(live.record.id);
|
||||
}
|
||||
|
||||
private getRpcProcess(live: LiveInstance): RpcProcessInstance | undefined {
|
||||
return live.resources.rpcProcess;
|
||||
}
|
||||
|
||||
private async syncInstanceRecord(live: LiveInstance): Promise<void> {
|
||||
const rpcProcess = this.getRpcProcess(live);
|
||||
if (!rpcProcess) {
|
||||
this.updateRecord(live, {});
|
||||
return;
|
||||
}
|
||||
const response = await rpcProcess.send({ type: "get_state" });
|
||||
if (!isGetStateSuccess(response)) {
|
||||
this.updateRecord(live, {});
|
||||
return;
|
||||
}
|
||||
this.updateRecord(live, {
|
||||
sessionId: response.data.sessionId,
|
||||
sessionFile: response.data.sessionFile,
|
||||
});
|
||||
}
|
||||
|
||||
private async cleanupAcquiredResources(live: LiveInstance): Promise<void> {
|
||||
const rpcProcess = live.resources.rpcProcess;
|
||||
this.clearBindings(live);
|
||||
if (live.resources.radiusPiId) {
|
||||
await radiusPresence.disconnectPi(live.record);
|
||||
live.resources.radiusPiId = undefined;
|
||||
live.record = {
|
||||
...live.record,
|
||||
radiusPiId: undefined,
|
||||
lastSeenAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
live.resources.sessionId = undefined;
|
||||
if (rpcProcess) {
|
||||
live.resources.rpcProcess = undefined;
|
||||
await rpcProcess.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
private async failSpawn(live: LiveInstance, error: unknown): Promise<never> {
|
||||
this.setStatus(live, "error");
|
||||
try {
|
||||
await this.cleanupAcquiredResources(live);
|
||||
} finally {
|
||||
this.setStatus(live, "stopped");
|
||||
this.liveInstances.delete(live.record.id);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
updateInstance(instance: InstanceRecord): void {
|
||||
const live = this.liveInstances.get(instance.id);
|
||||
if (live) {
|
||||
live.record = instance;
|
||||
live.resources.radiusPiId = instance.radiusPiId;
|
||||
live.resources.sessionId = instance.sessionId;
|
||||
}
|
||||
upsertInstance(instance);
|
||||
}
|
||||
|
||||
openRpcStream(
|
||||
instanceId: string,
|
||||
onEvent: (event: AgentSessionEvent) => void,
|
||||
onUiRequest: (request: RpcExtensionUIRequest) => void,
|
||||
):
|
||||
| {
|
||||
handleRpc(command: RpcCommand): Promise<RpcResponse>;
|
||||
handleUiResponse(response: RpcExtensionUIResponse): void;
|
||||
close(): void;
|
||||
}
|
||||
| undefined {
|
||||
const live = this.liveInstances.get(instanceId);
|
||||
const rpcProcess = live ? this.getRpcProcess(live) : undefined;
|
||||
if (!live || !rpcProcess) {
|
||||
return undefined;
|
||||
}
|
||||
live.subscribers.add(onEvent);
|
||||
live.onUiRequest = onUiRequest;
|
||||
return {
|
||||
handleRpc: async (command) => {
|
||||
const response = await rpcProcess.send(command);
|
||||
if (shouldRefreshSessionMetadata(command)) {
|
||||
await this.syncInstanceRecord(live);
|
||||
}
|
||||
return response;
|
||||
},
|
||||
handleUiResponse: (response) => {
|
||||
rpcProcess.handleUiResponse(response);
|
||||
},
|
||||
close: () => {
|
||||
if (live.onUiRequest === onUiRequest) {
|
||||
live.onUiRequest = undefined;
|
||||
}
|
||||
live.subscribers.delete(onEvent);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
getLiveInstance(instanceId: string): InstanceRecord | undefined {
|
||||
const live = this.liveInstances.get(instanceId);
|
||||
return live ? cloneInstance(live.record) : undefined;
|
||||
}
|
||||
|
||||
listLiveInstances(): InstanceRecord[] {
|
||||
return [...this.liveInstances.values()].map((live) => cloneInstance(live.record));
|
||||
}
|
||||
|
||||
async recoverAfterRestart(): Promise<void> {
|
||||
const recoveredAt = new Date().toISOString();
|
||||
const instances = loadInstances().map((instance) => ({
|
||||
...instance,
|
||||
status: instance.status === "online" || instance.status === "starting" ? "stopped" : instance.status,
|
||||
lastSeenAt: recoveredAt,
|
||||
}));
|
||||
for (const instance of instances) {
|
||||
await radiusPresence.disconnectPi(instance);
|
||||
}
|
||||
saveInstances(instances);
|
||||
}
|
||||
|
||||
listInstances(): InstanceRecord[] {
|
||||
return loadInstances().map(cloneInstance);
|
||||
}
|
||||
|
||||
getInstance(instanceId: string): InstanceRecord | undefined {
|
||||
const live = this.liveInstances.get(instanceId);
|
||||
if (live) {
|
||||
return cloneInstance(live.record);
|
||||
}
|
||||
const stored = getInstance(instanceId);
|
||||
return stored ? cloneInstance(stored) : undefined;
|
||||
}
|
||||
|
||||
async spawnInstance(options: { cwd: string; label?: string }): Promise<InstanceRecord> {
|
||||
const now = new Date().toISOString();
|
||||
const live: LiveInstance = {
|
||||
record: {
|
||||
id: randomUUID(),
|
||||
status: "starting",
|
||||
cwd: options.cwd,
|
||||
createdAt: now,
|
||||
lastSeenAt: now,
|
||||
label: options.label,
|
||||
},
|
||||
resources: {},
|
||||
subscribers: new Set(),
|
||||
};
|
||||
this.liveInstances.set(live.record.id, live);
|
||||
upsertInstance(live.record);
|
||||
|
||||
try {
|
||||
const rpcProcess = createRpcProcessInstance({ cwd: options.cwd });
|
||||
this.bindRpcProcess(live, rpcProcess);
|
||||
await this.syncInstanceRecord(live);
|
||||
const registeredRecord = await radiusPresence.registerPi(live.record);
|
||||
this.updateRecord(live, { radiusPiId: registeredRecord.radiusPiId });
|
||||
this.setStatus(live, "online");
|
||||
return cloneInstance(live.record);
|
||||
} catch (error) {
|
||||
return await this.failSpawn(live, error);
|
||||
}
|
||||
}
|
||||
|
||||
async stopInstance(instanceId: string): Promise<InstanceRecord | undefined> {
|
||||
const live = this.liveInstances.get(instanceId);
|
||||
if (!live) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
this.setStatus(live, "stopping");
|
||||
try {
|
||||
await this.cleanupAcquiredResources(live);
|
||||
} finally {
|
||||
live.record = {
|
||||
...live.record,
|
||||
status: "stopped",
|
||||
lastSeenAt: new Date().toISOString(),
|
||||
};
|
||||
this.liveInstances.delete(instanceId);
|
||||
removeInstance(instanceId);
|
||||
}
|
||||
return cloneInstance(live.record);
|
||||
}
|
||||
|
||||
async handleRpc(instanceId: string, command: RpcCommand): Promise<RpcResponse | undefined> {
|
||||
const live = this.liveInstances.get(instanceId);
|
||||
const rpcProcess = live ? this.getRpcProcess(live) : undefined;
|
||||
if (!live || !rpcProcess) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const response = await rpcProcess.send(command);
|
||||
if (shouldRefreshSessionMetadata(command)) {
|
||||
await this.syncInstanceRecord(live);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
for (const instanceId of [...this.liveInstances.keys()]) {
|
||||
await this.stopInstance(instanceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const supervisor = new OrchestratorSupervisor();
|
||||
|
||||
radiusPresence.setCoordinator({
|
||||
getLiveInstance(instanceId) {
|
||||
return supervisor.getLiveInstance(instanceId);
|
||||
},
|
||||
listLiveInstances() {
|
||||
return supervisor.listLiveInstances();
|
||||
},
|
||||
updateInstance(instance) {
|
||||
supervisor.updateInstance(instance);
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
export type InstanceStatus = "starting" | "online" | "stopping" | "stopped" | "error";
|
||||
|
||||
export interface MachineRecord {
|
||||
id: string;
|
||||
createdAt: string;
|
||||
lastSeenAt?: string;
|
||||
label?: string;
|
||||
}
|
||||
|
||||
export interface RadiusRegistration {
|
||||
heartbeatIntervalMs: number;
|
||||
expiresInMs: number;
|
||||
}
|
||||
|
||||
export interface InstanceRecord {
|
||||
id: string;
|
||||
status: InstanceStatus;
|
||||
cwd: string;
|
||||
createdAt: string;
|
||||
lastSeenAt?: string;
|
||||
label?: string;
|
||||
sessionId?: string;
|
||||
sessionFile?: string;
|
||||
radiusPiId?: string;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"extends": "../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"outDir": "./dist",
|
||||
"rootDir": "./src"
|
||||
},
|
||||
"include": ["src/**/*.ts"],
|
||||
"exclude": ["node_modules", "dist", "**/*.d.ts", "src/**/*.d.ts"]
|
||||
}
|
||||
@@ -13,6 +13,8 @@
|
||||
"@earendil-works/pi-coding-agent": ["./packages/coding-agent/src/index.ts"],
|
||||
"@earendil-works/pi-coding-agent/hooks": ["./packages/coding-agent/src/core/hooks/index.ts"],
|
||||
"@earendil-works/pi-coding-agent/*": ["./packages/coding-agent/src/*"],
|
||||
"@earendil-works/pi-orchestrator": ["./packages/orchestrator/src/index.ts"],
|
||||
"@earendil-works/pi-orchestrator/*": ["./packages/orchestrator/src/*"],
|
||||
"typebox": ["./node_modules/typebox"],
|
||||
"@earendil-works/pi-tui": ["./packages/tui/src/index.ts"],
|
||||
"@earendil-works/pi-tui/*": ["./packages/tui/src/*"],
|
||||
|
||||
Reference in New Issue
Block a user