mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat: remove legacy server implementation (#7614)
This commit is contained in:
Christian Klotz
GitHub
parent
4b85cd9786
commit
05bf9df651
18 changed files with 8 additions and 2012 deletions
4
package-lock.json
generated
4
package-lock.json
generated
|
|
@ -5715,12 +5715,8 @@
|
|||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-ai": "^0.83.0",
|
||||
"@earendil-works/pi-coding-agent": "^0.83.0",
|
||||
"@earendil-works/pi-protocol": "^0.83.0"
|
||||
},
|
||||
"bin": {
|
||||
"server": "dist/legacy/cli.js"
|
||||
},
|
||||
"devDependencies": {
|
||||
"shx": "0.4.0",
|
||||
"vitest": "4.1.9"
|
||||
|
|
|
|||
|
|
@ -13,8 +13,8 @@
|
|||
],
|
||||
"scripts": {
|
||||
"clean": "npm run clean --workspaces",
|
||||
"build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build",
|
||||
"build:offline": "cd packages/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build",
|
||||
"build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
|
||||
"build:offline": "cd packages/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build",
|
||||
"check": "biome check --write --error-on-warnings . && npm run check:pinned-deps && npm run check:ts-imports && npm run check:shrinkwrap && npm run check:install-lock:coding-agent && 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",
|
||||
|
|
|
|||
|
|
@ -1,18 +1,12 @@
|
|||
# @earendil-works/pi-server
|
||||
|
||||
Experimental. This package is under active development and may change or be removed without notice. Its CLI, APIs, and behavior are not yet stable.
|
||||
Experimental. This package is under active development and may change or be removed without notice. Its APIs and behavior are not yet stable.
|
||||
|
||||
Server package for pi.
|
||||
|
||||
## CLI
|
||||
|
||||
```bash
|
||||
server --help
|
||||
```
|
||||
|
||||
## Session server core
|
||||
|
||||
The package also exports the new `PiServer` session server. This API is additive while the legacy child-process supervisor and `server` CLI are migrated.
|
||||
The package exports the `PiServer` session server.
|
||||
|
||||
```ts
|
||||
import type { PiSessionBackend } from "@earendil-works/pi-server";
|
||||
|
|
@ -39,7 +33,9 @@ const server = createUnixServer(backend, {
|
|||
await server.start();
|
||||
```
|
||||
|
||||
`PiServer` composes transport listeners through the `PiServerListener` interface. Each listener must complete any transport-specific authentication and authorization before passing a connection to `PiServer`. For example, a WebSocket listener can validate credentials during the HTTP upgrade, while the Unix listener relies on socket filesystem permissions. The Unix submodule exports the `createUnixListener()` building block and `createUnixServer()` preset, keeping the common case concise without coupling the primary server to Unix sockets. The listener uses length-prefixed CBOR messages from `@earendil-works/pi-protocol`. It does not yet replace the legacy JSONL IPC control plane, child-process supervisor, standalone `server` CLI, or Radius presence integration.
|
||||
`PiServer` composes transport listeners through the `PiServerListener` interface. Each listener must complete any transport-specific authentication and authorization before passing a connection to `PiServer`. For example, a WebSocket listener can validate credentials during the HTTP upgrade, while the Unix listener relies on socket filesystem permissions. The Unix submodule exports the `createUnixListener()` building block and `createUnixServer()` preset, keeping the common case concise without coupling the primary server to Unix sockets. The listener uses length-prefixed CBOR messages from `@earendil-works/pi-protocol`.
|
||||
|
||||
This package does not provide a standalone CLI or coding-agent backend. Applications supply the `PiSessionBackend` implementation.
|
||||
|
||||
## Transport testing
|
||||
|
||||
|
|
@ -50,7 +46,3 @@ Custom transports can use `@earendil-works/pi-server/testing` for deterministic
|
|||
`@earendil-works/pi-ai` domain objects and `@earendil-works/pi-protocol` wire DTOs remain independent. This package owns their boundary and exports `toProtocolModelMetadata()`, `toProtocolAssistantMessage()`, `toProtocolUserMessage()`, and `toProtocolToolResultMessage()`.
|
||||
|
||||
The adapters reject invalid tool inputs, identifiers, timestamps, and mismatched tool results; `toProtocolToolResultMessage()` requires the original `ToolCall` so it can verify the association and convert its arguments itself. Diagnostic details are explicitly sanitized. Closed `pi-ai` unions are mapped exhaustively, and compile-time field manifests enumerate current `pi-ai` properties so additions require an explicit review. The protocol mirrors `pi-ai` vocabulary such as `toolCall` and `toolUse` where the semantics are identical. Protocol schemas enforce consistent lifecycle states, and tests encode adapter output through the runtime schemas so incompatible changes fail in the bridging package.
|
||||
|
||||
## Legacy server migration
|
||||
|
||||
The existing IPC, supervisor, process management, persistence, and Radius modules remain available during migration. The new Unix session protocol supersedes the legacy socket framing and RPC proxy only after the coding-agent backend and CLI replacement have landed. Radius is presence and registration infrastructure, not a transport, and requires a separate integration with the new server lifecycle before the legacy supervisor can be removed.
|
||||
|
|
|
|||
|
|
@ -17,15 +17,8 @@
|
|||
"./unix": {
|
||||
"types": "./dist/transports/unix/index.d.ts",
|
||||
"import": "./dist/transports/unix/index.js"
|
||||
},
|
||||
"./legacy": {
|
||||
"types": "./dist/legacy/index.d.ts",
|
||||
"import": "./dist/legacy/index.js"
|
||||
}
|
||||
},
|
||||
"bin": {
|
||||
"server": "./dist/legacy/cli.js"
|
||||
},
|
||||
"files": [
|
||||
"dist",
|
||||
"README.md",
|
||||
|
|
@ -34,7 +27,7 @@
|
|||
"scripts": {
|
||||
"clean": "shx rm -rf dist",
|
||||
"dev": "tsgo -p tsconfig.build.json --watch --preserveWatchOutput",
|
||||
"build": "tsgo -p tsconfig.build.json && shx chmod +x dist/legacy/cli.js",
|
||||
"build": "tsgo -p tsconfig.build.json",
|
||||
"test": "vitest --run",
|
||||
"typecheck": "tsgo -p tsconfig.test.json",
|
||||
"prepublishOnly": "npm run clean && npm run build"
|
||||
|
|
@ -55,7 +48,6 @@
|
|||
},
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-ai": "^0.83.0",
|
||||
"@earendil-works/pi-coding-agent": "^0.83.0",
|
||||
"@earendil-works/pi-protocol": "^0.83.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
export * from "./errors.ts";
|
||||
export * from "./legacy/index.ts";
|
||||
export * from "./listener.ts";
|
||||
export * from "./protocol.ts";
|
||||
export * from "./server.ts";
|
||||
|
|
|
|||
|
|
@ -1,161 +0,0 @@
|
|||
#!/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(
|
||||
`server v${packageJson.version}\n\nUsage:\n server serve\n server list\n server spawn [--cwd <path>] [--label <label>]\n server status <instance-id>\n server stop <instance-id>\n server rpc <instance-id> <json-command>\n server rpc-stream <instance-id>\n server --help\n server --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: server 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: server 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: server 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: server rpc-stream <instance-id>");
|
||||
process.exit(1);
|
||||
}
|
||||
await rpcStream(instanceId);
|
||||
return;
|
||||
}
|
||||
|
||||
console.error(`Unknown command: ${args[0]}`);
|
||||
printHelp();
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
await main();
|
||||
|
|
@ -1,69 +0,0 @@
|
|||
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_SERVER_DIR = "PI_SERVER_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 getServerDir(): string {
|
||||
const envDir = process.env[ENV_SERVER_DIR];
|
||||
if (envDir) {
|
||||
return envDir;
|
||||
}
|
||||
|
||||
const piDir = process.env.PI_CONFIG_DIR || join(homedir(), CONFIG_DIR_NAME);
|
||||
return join(piDir, "server");
|
||||
}
|
||||
|
||||
export function getAuthPath(): string {
|
||||
return join(getServerDir(), "auth.json");
|
||||
}
|
||||
|
||||
export function getMachinePath(): string {
|
||||
return join(getServerDir(), "machine.json");
|
||||
}
|
||||
|
||||
export function getInstancesPath(): string {
|
||||
return join(getServerDir(), "instances.json");
|
||||
}
|
||||
|
||||
export function getSocketPath(): string {
|
||||
return join(getServerDir(), "server.sock");
|
||||
}
|
||||
|
|
@ -1,161 +0,0 @@
|
|||
import type {
|
||||
AgentSessionEvent,
|
||||
RpcCommand,
|
||||
RpcExtensionUIRequest,
|
||||
RpcExtensionUIResponse,
|
||||
RpcResponse,
|
||||
} from "@earendil-works/pi-coding-agent";
|
||||
import type {
|
||||
ErrorResponse,
|
||||
InstanceSummary,
|
||||
ListRequest,
|
||||
ListResponse,
|
||||
RpcBridgeResponse,
|
||||
RpcReadyResponse,
|
||||
RpcRequest,
|
||||
RpcStreamRequest,
|
||||
ServerRequest,
|
||||
ServerResponse,
|
||||
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: ServerRequest): Promise<ServerResponse>;
|
||||
export async function handleIpcRequest(request: ServerRequest): Promise<ServerResponse> {
|
||||
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();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -1,11 +0,0 @@
|
|||
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 "./radius.ts";
|
||||
export * from "./rpc-process.ts";
|
||||
export * from "./serve.ts";
|
||||
export * from "./storage.ts";
|
||||
export * from "./supervisor.ts";
|
||||
export * from "./types.ts";
|
||||
|
|
@ -1,63 +0,0 @@
|
|||
import { createConnection } from "node:net";
|
||||
import { getSocketPath } from "../config.ts";
|
||||
import { encodeMessage, parseResponseLine, type ServerRequest, type ServerResponse } from "./protocol.ts";
|
||||
|
||||
export async function sendIpcRequest(request: ServerRequest): Promise<ServerResponse> {
|
||||
const socketPath = getSocketPath();
|
||||
|
||||
return new Promise<ServerResponse>((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(`Server socket closed before a response was received: ${socketPath}`));
|
||||
cleanup();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
@ -1,142 +0,0 @@
|
|||
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 ServerRequest = 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 ServerResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;
|
||||
export type RpcClientMessage = RpcCommand | RpcExtensionUIResponse;
|
||||
export type RpcServerMessage =
|
||||
| RpcReadyResponse
|
||||
| RpcResponse
|
||||
| AgentSessionEvent
|
||||
| RpcExtensionUIRequest
|
||||
| ErrorResponse;
|
||||
export type ProtocolMessage = ServerRequest | ServerResponse | RpcClientMessage | RpcServerMessage;
|
||||
|
||||
export type ResponseFor<T extends ServerRequest> = 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): ServerRequest {
|
||||
const value = JSON.parse(line) as ServerRequest;
|
||||
return value;
|
||||
}
|
||||
|
||||
export function parseResponseLine(line: string): ServerResponse {
|
||||
const value = JSON.parse(line) as ServerResponse;
|
||||
return value;
|
||||
}
|
||||
|
|
@ -1,209 +0,0 @@
|
|||
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,
|
||||
parseRequestLine,
|
||||
type RpcBridgeResponse,
|
||||
type RpcReadyResponse,
|
||||
type RpcRequest,
|
||||
type RpcStreamRequest,
|
||||
type ServerRequest,
|
||||
type ServerResponse,
|
||||
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: ServerRequest): Promise<ServerResponse> | ServerResponse;
|
||||
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(`server 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);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
@ -1,440 +0,0 @@
|
|||
import { hostname, platform } from "node:os";
|
||||
import type { OAuthCredential } from "@earendil-works/pi-ai";
|
||||
import { readStoredCredential } from "@earendil-works/pi-coding-agent";
|
||||
import { getServerDir, 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_SERVER_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, getRadiusServerBaseUrl()), {
|
||||
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, getRadiusServerBaseUrl()), {
|
||||
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 getRadiusServerBaseUrl(): string {
|
||||
const explicitUrl = process.env.PI_RADIUS_SERVER_URL;
|
||||
if (explicitUrl) {
|
||||
return explicitUrl;
|
||||
}
|
||||
|
||||
return new URL(DEFAULT_SERVER_BASE_PATH, getRadiusUrl()).toString();
|
||||
}
|
||||
|
||||
function getStoredRadiusCredential(): OAuthCredential | undefined {
|
||||
const credential = readStoredCredential(RADIUS_PROVIDER);
|
||||
return credential?.type === "oauth" ? credential : undefined;
|
||||
}
|
||||
|
||||
export function getRadiusAccessToken(): string {
|
||||
const storedCredential = getStoredRadiusCredential();
|
||||
if (typeof storedCredential?.access === "string" && storedCredential.access) {
|
||||
return storedCredential.access;
|
||||
}
|
||||
|
||||
const apiKey = process.env.RADIUS_API_KEY;
|
||||
if (apiKey) {
|
||||
return apiKey;
|
||||
}
|
||||
|
||||
throw new Error("Radius credentials are required in ~/.pi/agent/auth.json or RADIUS_API_KEY");
|
||||
}
|
||||
|
||||
export function isRadiusEnabled(): boolean {
|
||||
return !!getStoredRadiusCredential()?.access || !!process.env.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: getServerDir(),
|
||||
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();
|
||||
|
|
@ -1,201 +0,0 @@
|
|||
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 ?? `server_${++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);
|
||||
}
|
||||
|
|
@ -1,77 +0,0 @@
|
|||
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 { getRadiusServerBaseUrl, 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} -> ${getRadiusServerBaseUrl()}`);
|
||||
if (machine) {
|
||||
console.log(`radius machine id: ${machine.id}`);
|
||||
}
|
||||
} else {
|
||||
console.log("radius integration disabled: login radius in ~/.pi/agent/auth.json or set RADIUS_API_KEY");
|
||||
}
|
||||
} catch (error) {
|
||||
server.close();
|
||||
if (existsSync(socketPath)) {
|
||||
unlinkSync(socketPath);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
console.log(`server 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.
|
||||
});
|
||||
}
|
||||
|
|
@ -1,70 +0,0 @@
|
|||
import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { getInstancesPath, getMachinePath, getServerDir } from "./config.ts";
|
||||
import type { InstanceRecord, MachineRecord } from "./types.ts";
|
||||
|
||||
function ensureServerDir(): void {
|
||||
const serverDir = getServerDir();
|
||||
if (!existsSync(serverDir)) {
|
||||
mkdirSync(serverDir, { 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 {
|
||||
ensureServerDir();
|
||||
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 {
|
||||
ensureServerDir();
|
||||
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);
|
||||
}
|
||||
|
|
@ -1,354 +0,0 @@
|
|||
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 ServerSupervisor {
|
||||
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 ServerSupervisor();
|
||||
|
||||
radiusPresence.setCoordinator({
|
||||
getLiveInstance(instanceId) {
|
||||
return supervisor.getLiveInstance(instanceId);
|
||||
},
|
||||
listLiveInstances() {
|
||||
return supervisor.listLiveInstances();
|
||||
},
|
||||
updateInstance(instance) {
|
||||
supervisor.updateInstance(instance);
|
||||
},
|
||||
});
|
||||
|
|
@ -1,25 +0,0 @@
|
|||
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;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue