chore: rename orchestrator to server (#6898)
This commit is contained in:
Cristina Poncela Cubeiro
committed by
GitHub
GitHub
parent
9e7582aa03
commit
8495f9d0d6
@@ -0,0 +1,63 @@
|
||||
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();
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -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 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;
|
||||
}
|
||||
@@ -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,
|
||||
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);
|
||||
});
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user