diff --git a/packages/client/README.md b/packages/client/README.md index 447ccccd3..7d04b979e 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -27,18 +27,7 @@ The client verifies that the physical endpoint reports the expected logical `ser `attachSession()` currently returns only `{ sessionId }`. Remote Session and Harness methods will be added directly from the new shared interfaces in a later slice. The client does not reconnect or replay requests automatically. After disconnection, call `reconnect()` and explicitly repeat safe control-plane actions. -Server lifecycle control is not part of `PiClient`. Launchers use the separate `@earendil-works/pi-client/control` entry point to request a drain and wait for that server generation to close. - -```ts -import { requestServerDrain } from "@earendil-works/pi-client/control"; - -await requestServerDrain({ serverId, transportFactory }); -// Drain was acknowledged. Verifying endpoint release and starting a successor are launcher policy. -``` - -`requestServerDrain()` applies a 15-second total timeout across connection, handshake, and drain acknowledgement. Pass `timeoutMs` to override it. A timeout is ambiguous and is never replayed automatically. The transport-neutral helper does not prove process termination or endpoint release; a Unix launcher separately waits for its stable socket path to disappear. - -The control API is administrative, not an authorization boundary. Only expose it over a transport that authenticates launchers as administrators. The experimental local launcher relies on a mode-`0600` Unix socket and treats every same-user peer as trusted. +The experimental local coordinator only provides a stable endpoint and relays traffic. Replaceable server processes own session and worker lifecycle outside the public client protocol. Call transport handlers as follows: diff --git a/packages/client/package.json b/packages/client/package.json index f094d812a..45d5a103f 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -10,10 +10,6 @@ "types": "./dist/index.d.ts", "import": "./dist/index.js" }, - "./control": { - "types": "./dist/control.d.ts", - "import": "./dist/control.js" - }, "./unix": { "types": "./dist/unix.d.ts", "import": "./dist/unix.js" diff --git a/packages/client/src/control.ts b/packages/client/src/control.ts deleted file mode 100644 index 69431ef86..000000000 --- a/packages/client/src/control.ts +++ /dev/null @@ -1,125 +0,0 @@ -import { - createRpcClient, - encodeClientMessage, - isServerId, - ProtocolValidationError, - ServerControlRpc, - type ServerControlRpcCall, -} from "@earendil-works/pi-protocol"; -import { Connection } from "./connection.ts"; -import { PiDisconnectedError, PiServerError, toError } from "./errors.ts"; -import { createPromiseResolvers } from "./promise.ts"; -import type { ByteTransportFactory } from "./transport.ts"; - -const DEFAULT_SERVER_CONTROL_TIMEOUT_MS = 15_000; -const MAX_TIMER_DELAY_MS = 2_147_483_647; - -export interface ServerControlOptions { - readonly transportFactory: ByteTransportFactory; - /** Logical server identity expected at the physical endpoint. */ - readonly serverId: string; - readonly maxFrameLength?: number; - /** Total time allowed for handshake and drain acknowledgement. Defaults to 15 seconds. */ - readonly timeoutMs?: number; -} - -export class ServerControlTimeoutError extends Error { - readonly timeoutMs: number; - - constructor(timeoutMs: number) { - super(`Server control timed out after ${timeoutMs} ms`); - this.name = "ServerControlTimeoutError"; - this.timeoutMs = timeoutMs; - } -} - -interface PendingControlRequest { - readonly id: string; - resolve(result: unknown): void; - reject(error: Error): void; -} - -/** - * Ask one server generation to drain and wait for its acknowledgement. - * Physical endpoint shutdown is transport-specific launcher policy. - */ -export async function requestServerDrain(options: ServerControlOptions): Promise { - if (!isServerId(options.serverId)) { - throw new TypeError("Server control serverId must be a canonical lowercase UUIDv4"); - } - const timeoutMs = options.timeoutMs ?? DEFAULT_SERVER_CONTROL_TIMEOUT_MS; - if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0 || timeoutMs > MAX_TIMER_DELAY_MS) { - throw new TypeError(`Server control timeoutMs must be an integer between 1 and ${MAX_TIMER_DELAY_MS}`); - } - - let pending: PendingControlRequest | undefined; - const connection = new Connection({ - transportFactory: options.transportFactory, - serverId: options.serverId, - maxFrameLength: options.maxFrameLength, - onHandshake: () => {}, - onMessage: (message) => { - if (!pending || message.id !== pending.id) { - connection.fail(new ProtocolValidationError("Control response has no matching request")); - return; - } - const request = pending; - pending = undefined; - if (message.ok) request.resolve(message.result); - else request.reject(new PiServerError(message.error)); - }, - onStateChange: (change) => { - if (change.state !== "disconnected") return; - const request = pending; - pending = undefined; - request?.reject(change.error ?? new PiDisconnectedError()); - }, - }); - const control = createRpcClient( - ServerControlRpc, - (call) => - sendControlRequest(connection, options.serverId, call, (request) => { - pending = request; - }), - (message) => new ProtocolValidationError(message), - ); - - const timeout = setTimeout(() => { - connection.disconnect(new ServerControlTimeoutError(timeoutMs)); - }, timeoutMs); - timeout.unref(); - - try { - await connection.connect(); - await control.drain(); - } finally { - clearTimeout(timeout); - const request = pending; - pending = undefined; - request?.reject(new PiDisconnectedError("Server control client closed")); - connection.disconnect("Server control client closed"); - } -} - -function sendControlRequest( - connection: Connection, - serverId: string, - call: ServerControlRpcCall, - setPending: (request: PendingControlRequest) => void, -): Promise { - const id = "control-1"; - const { promise, resolve, reject } = createPromiseResolvers(); - setPending({ id, resolve, reject }); - let frame: Uint8Array; - try { - frame = encodeClientMessage( - { type: "request", id, serverId, call }, - { maxFrameLength: connection.maxFrameLength }, - ); - } catch (error) { - reject(toError(error)); - return promise; - } - connection.send(frame); - return promise; -} diff --git a/packages/client/test/control.test.ts b/packages/client/test/control.test.ts deleted file mode 100644 index 725961616..000000000 --- a/packages/client/test/control.test.ts +++ /dev/null @@ -1,63 +0,0 @@ -import { expect, test, vi } from "vitest"; -import { requestServerDrain, ServerControlTimeoutError } from "../src/control.ts"; -import { MemoryByteServer } from "./support.ts"; - -test("launcher control resolves after the drain acknowledgement", async () => { - const server = new MemoryByteServer(); - const draining = requestServerDrain({ - serverId: "00000000-0000-4000-8000-000000000001", - transportFactory: (handlers) => server.connect(handlers), - }); - await server.waitForMessages(2); - expect(server.messages[1]).toEqual({ - type: "request", - id: "control-1", - serverId: "00000000-0000-4000-8000-000000000001", - call: { method: "drain", args: [] }, - }); - - server.send({ - type: "response", - id: "control-1", - ok: true, - result: {}, - }); - await expect(draining).resolves.toBeUndefined(); - expect(server.clientCloseCount).toBe(1); -}); - -test("fails a control request that does not complete within its total timeout", async () => { - vi.useFakeTimers(); - try { - const server = new MemoryByteServer(); - const draining = requestServerDrain({ - serverId: "00000000-0000-4000-8000-000000000001", - transportFactory: (handlers) => server.connect(handlers), - timeoutMs: 25, - }); - await server.waitForMessages(2); - const observed = draining.then( - () => undefined, - (error: unknown) => error, - ); - - await vi.advanceTimersByTimeAsync(25); - const thrown = await observed; - expect(thrown).toBeInstanceOf(ServerControlTimeoutError); - expect(thrown).toMatchObject({ timeoutMs: 25 }); - expect(server.clientCloseCount).toBe(1); - } finally { - vi.useRealTimers(); - } -}); - -test("validates the control timeout", async () => { - const server = new MemoryByteServer(); - await expect( - requestServerDrain({ - serverId: "00000000-0000-4000-8000-000000000001", - transportFactory: (handlers) => server.connect(handlers), - timeoutMs: 0, - }), - ).rejects.toThrow(/timeoutMs/); -}); diff --git a/packages/coding-agent/src/cli/experimental/coordinated-session-workers.ts b/packages/coding-agent/src/cli/experimental/coordinated-session-workers.ts new file mode 100644 index 000000000..c5eafd4ae --- /dev/null +++ b/packages/coding-agent/src/cli/experimental/coordinated-session-workers.ts @@ -0,0 +1,353 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { isAbsolute } from "node:path"; +import type { JsonlSessionMetadata } from "@earendil-works/pi-agent-core"; +import type { HostedHarnessHandle } from "@earendil-works/pi-server"; +import Type, { type Static } from "typebox"; +import { Check } from "typebox/value"; +import type { CoordinatorServer, CoordinatorServerEvent } from "./coordinator-client.ts"; +import { + createExperimentalSessionWorkerLaunchSpec, + SESSION_WORKER_CONTROL_ADDRESS_ENV, + SESSION_WORKER_CONTROL_TOKEN_ENV, + SESSION_WORKER_COORDINATED_ENV, + SESSION_WORKER_PEER_ID_ENV, + SESSION_WORKER_SESSION_KEY_ENV, +} from "./session-worker.ts"; + +const WORKER_STARTUP_TIMEOUT_MS = 15_000; +const WORKER_SHUTDOWN_TIMEOUT_MS = 10_000; +const WORKER_DISCOVERY_TIMEOUT_MS = 5_000; + +const JsonlSessionMetadataSchema = Type.Object({ + id: Type.String({ minLength: 1 }), + createdAt: Type.Integer(), + storageVersion: Type.Integer(), + cwd: Type.String(), + path: Type.String(), + modifiedAt: Type.Number(), + parentSessionId: Type.Optional(Type.String()), + legacyParentSessionPath: Type.Optional(Type.String()), +}); + +const WorkerMessageSchema = Type.Union([ + Type.Object({ + type: Type.Literal("worker_ready"), + token: Type.String(), + sessionKey: Type.String(), + sessionId: Type.String(), + pid: Type.Integer({ minimum: 1 }), + metadata: JsonlSessionMetadataSchema, + }), + Type.Object({ + type: Type.Literal("worker_failed"), + token: Type.String(), + sessionKey: Type.String(), + message: Type.String(), + }), +]); +type WorkerMessage = Static; + +interface WorkerRecord { + readonly peerId: string; + readonly metadata: JsonlSessionMetadata; + readonly pid: number; + readonly terminated: Promise; + resolveTerminated(error: Error | undefined): void; + expectedStop: boolean; +} + +interface PendingLaunch { + readonly sessionKey: string; + readonly peerId: string; + readonly token: string; + readonly child: ChildProcess; + readonly timer: NodeJS.Timeout; + readonly promise: Promise; + resolve(worker: WorkerRecord): void; + reject(error: Error): void; +} + +/** Session and process bookkeeping owned by one replaceable Pi server process. */ +export class CoordinatedSessionWorkers { + readonly workerPids = new Map(); + readonly #coordinator: CoordinatorServer; + readonly #sessionDir: string; + readonly #workersBySession = new Map(); + readonly #workersByPeer = new Map(); + readonly #pending = new Map(); + readonly #removeListener: () => void; + #discoveryPeers?: Set; + #resolveDiscovery?: () => void; + #detached = false; + #shuttingDown = false; + + constructor(coordinator: CoordinatorServer, sessionDir: string) { + this.#coordinator = coordinator; + this.#sessionDir = sessionDir; + this.#removeListener = coordinator.onEvent((event) => this.#handleCoordinatorEvent(event)); + } + + get trackedSessions(): readonly JsonlSessionMetadata[] { + return [...this.#workersBySession.values()].map((worker) => worker.metadata); + } + + async discover(peerIds: ReadonlySet): Promise { + if (this.#detached) return; + const undiscovered = new Set( + [...peerIds].filter((peerId) => !this.#workersByPeer.has(peerId) && !this.#pendingPeer(peerId)), + ); + if (undiscovered.size === 0) return; + this.#discoveryPeers = undiscovered; + const discovered = new Promise((resolve) => { + this.#resolveDiscovery = resolve; + }); + await this.#coordinator.broadcast({ type: "discover_workers" }); + let timer: NodeJS.Timeout | undefined; + await Promise.race([ + discovered, + new Promise((resolve) => { + timer = setTimeout(resolve, WORKER_DISCOVERY_TIMEOUT_MS); + timer.unref(); + }), + ]); + if (timer) clearTimeout(timer); + this.#discoveryPeers = undefined; + this.#resolveDiscovery = undefined; + } + + async createHarness(metadata: JsonlSessionMetadata): Promise { + if (this.#detached || this.#shuttingDown) throw new Error("Experimental server is shutting down"); + const existing = this.#workersBySession.get(metadata.path); + if (existing) return { terminated: existing.terminated, close: async () => {} }; + const pending = this.#pending.get(metadata.path); + if (pending) { + const worker = await pending.promise; + return { terminated: worker.terminated, close: async () => {} }; + } + const worker = await this.#launch(metadata); + return { terminated: worker.terminated, close: async () => {} }; + } + + async shutdown(): Promise { + if (this.#detached || this.#shuttingDown) return; + this.#shuttingDown = true; + for (const pending of this.#pending.values()) { + void this.#coordinator.send(pending.peerId, { type: "shutdown" }).catch(() => {}); + } + const workers = [...this.#workersBySession.values()]; + for (const worker of workers) { + worker.expectedStop = true; + void this.#coordinator.send(worker.peerId, { type: "shutdown" }).catch(() => {}); + } + const finished = Promise.all([ + ...workers.map((worker) => worker.terminated), + ...[...this.#pending.values()].map((pending) => { + if (pending.child.exitCode !== null || pending.child.signalCode !== null) return Promise.resolve(); + return new Promise((resolve) => pending.child.once("exit", () => resolve())); + }), + ]).then(() => undefined); + let timer: NodeJS.Timeout | undefined; + const timedOut = new Promise((resolve) => { + timer = setTimeout(() => resolve(true), WORKER_SHUTDOWN_TIMEOUT_MS); + timer.unref(); + }); + if ((await Promise.race([finished.then(() => false), timedOut])) === true) { + for (const worker of workers) { + try { + process.kill(worker.pid, "SIGKILL"); + } catch (error) { + if (!(error instanceof Error && "code" in error && error.code === "ESRCH")) throw error; + } + } + for (const pending of this.#pending.values()) pending.child.kill("SIGKILL"); + await finished; + } + if (timer) clearTimeout(timer); + this.#detachState(); + } + + /** Forget workers without stopping them when this server is replaced. */ + detach(): void { + if (this.#detached) return; + this.#detached = true; + for (const pending of this.#pending.values()) { + clearTimeout(pending.timer); + pending.reject(new Error("Experimental server was replaced")); + } + this.#detachState(); + } + + #launch(metadata: JsonlSessionMetadata): Promise { + const sessionKey = metadata.path; + const peerId = `worker-${randomUUID()}`; + const token = randomUUID(); + const launch = createExperimentalSessionWorkerLaunchSpec(metadata, { sessionDir: this.#sessionDir }); + let child: ChildProcess; + try { + child = spawn(launch.command, launch.args, { + cwd: launch.cwd, + detached: true, + env: { + ...launch.env, + [SESSION_WORKER_CONTROL_ADDRESS_ENV]: this.#coordinator.controlPath, + [SESSION_WORKER_CONTROL_TOKEN_ENV]: token, + [SESSION_WORKER_SESSION_KEY_ENV]: Buffer.from(sessionKey).toString("base64url"), + [SESSION_WORKER_COORDINATED_ENV]: "1", + [SESSION_WORKER_PEER_ID_ENV]: peerId, + }, + stdio: "ignore", + windowsHide: true, + }); + child.unref(); + } catch (error) { + return Promise.reject(error); + } + let resolve!: (worker: WorkerRecord) => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + const timer = setTimeout( + () => this.#failPending(sessionKey, new Error("Session worker startup timed out")), + WORKER_STARTUP_TIMEOUT_MS, + ); + timer.unref(); + const pending = { sessionKey, peerId, token, child, timer, promise, resolve, reject }; + this.#pending.set(sessionKey, pending); + child.once("error", (error) => this.#failPending(sessionKey, error)); + child.once("exit", (code, signal) => this.#childExited(pending, code, signal)); + return promise; + } + + #handleCoordinatorEvent(event: CoordinatorServerEvent): void { + if (this.#detached) return; + if (event.type === "peer_disconnected") { + this.#markDiscovered(event.peerId); + const worker = this.#workersByPeer.get(event.peerId); + if (worker) { + this.#removeWorker( + worker, + worker.expectedStop + ? undefined + : new Error(`Session worker ${worker.metadata.id} disconnected unexpectedly`), + ); + } + const pending = this.#pendingPeer(event.peerId); + if (pending) this.#failPending(pending.sessionKey, new Error("Session worker disconnected during startup")); + return; + } + if (event.type !== "message" || !Check(WorkerMessageSchema, event.payload)) return; + const message: WorkerMessage = event.payload; + if (message.type === "worker_failed") { + const pending = this.#pending.get(message.sessionKey); + if (pending?.peerId === event.from && pending.token === message.token) { + this.#failPending(message.sessionKey, new Error(`Session worker failed: ${message.message}`)); + } + return; + } + this.#recordReadyWorker(event.from, message); + } + + #recordReadyWorker(peerId: string, message: Extract): void { + if ( + message.sessionKey !== message.metadata.path || + message.sessionId !== message.metadata.id || + !isAbsolute(message.metadata.cwd) || + !isAbsolute(message.metadata.path) + ) { + return; + } + this.#markDiscovered(peerId); + const existing = this.#workersBySession.get(message.sessionKey); + if (existing && existing.peerId !== peerId) { + void this.#coordinator.send(peerId, { type: "shutdown" }).catch(() => {}); + return; + } + const pending = this.#pending.get(message.sessionKey); + if ( + pending && + (pending.peerId !== peerId || pending.token !== message.token || pending.child.pid !== message.pid) + ) { + return; + } + if (existing) return; + let resolveTerminated!: (error: Error | undefined) => void; + const terminated = new Promise((resolve) => { + resolveTerminated = resolve; + }); + const worker: WorkerRecord = { + peerId, + metadata: message.metadata, + pid: message.pid, + terminated, + resolveTerminated, + expectedStop: false, + }; + this.#workersBySession.set(message.sessionKey, worker); + this.#workersByPeer.set(peerId, worker); + this.workerPids.set(message.sessionId, message.pid); + if (pending) { + this.#pending.delete(message.sessionKey); + clearTimeout(pending.timer); + pending.resolve(worker); + } + } + + #childExited(pending: PendingLaunch, code: number | null, signal: NodeJS.Signals | null): void { + if (this.#pending.get(pending.sessionKey) === pending) { + this.#failPending( + pending.sessionKey, + new Error(`Session worker exited before readiness (${signal ?? code ?? "unknown"})`), + ); + return; + } + const worker = this.#workersByPeer.get(pending.peerId); + if (worker) { + this.#removeWorker( + worker, + worker.expectedStop + ? undefined + : new Error(`Session worker ${worker.metadata.id} exited unexpectedly (${signal ?? code ?? "unknown"})`), + ); + } + } + + #failPending(sessionKey: string, error: Error): void { + const pending = this.#pending.get(sessionKey); + if (!pending) return; + this.#pending.delete(sessionKey); + clearTimeout(pending.timer); + if (pending.child.exitCode === null && pending.child.signalCode === null) pending.child.kill("SIGKILL"); + pending.reject(error); + } + + #removeWorker(worker: WorkerRecord, error: Error | undefined): void { + if (this.#workersByPeer.get(worker.peerId) !== worker) return; + this.#workersByPeer.delete(worker.peerId); + this.#workersBySession.delete(worker.metadata.path); + if (this.workerPids.get(worker.metadata.id) === worker.pid) this.workerPids.delete(worker.metadata.id); + worker.resolveTerminated(error); + } + + #pendingPeer(peerId: string): PendingLaunch | undefined { + return [...this.#pending.values()].find((pending) => pending.peerId === peerId); + } + + #markDiscovered(peerId: string): void { + if (!this.#discoveryPeers?.delete(peerId) || this.#discoveryPeers.size !== 0) return; + this.#resolveDiscovery?.(); + } + + #detachState(): void { + this.#removeListener(); + this.#pending.clear(); + this.#workersByPeer.clear(); + this.#workersBySession.clear(); + this.workerPids.clear(); + this.#discoveryPeers = undefined; + this.#resolveDiscovery?.(); + this.#resolveDiscovery = undefined; + } +} diff --git a/packages/coding-agent/src/cli/experimental/coordinator-client.ts b/packages/coding-agent/src/cli/experimental/coordinator-client.ts new file mode 100644 index 000000000..177048a1f --- /dev/null +++ b/packages/coding-agent/src/cli/experimental/coordinator-client.ts @@ -0,0 +1,282 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { createConnection, type Socket } from "node:net"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import Type, { type Static } from "typebox"; +import { Check } from "typebox/value"; + +const COORDINATOR_PROTOCOL_VERSION = 1; +const COORDINATOR_START_TIMEOUT_MS = 10_000; +const COORDINATOR_RETRY_MS = 10; +const MAX_CONTROL_LINE_BYTES = 1024 * 1024; + +const CoordinatorMessageSchema = Type.Union([ + Type.Object({ + type: Type.Literal("server_registered"), + serverConnectionId: Type.String(), + peers: Type.Array(Type.String()), + }), + Type.Object({ type: Type.Literal("server_replaced") }), + Type.Object({ type: Type.Literal("peer_connected"), peerId: Type.String() }), + Type.Object({ type: Type.Literal("peer_disconnected"), peerId: Type.String() }), + Type.Object({ type: Type.Literal("message"), from: Type.String(), payload: Type.Unknown() }), +]); +type CoordinatorMessage = Static; + +export type CoordinatorServerEvent = + | { readonly type: "peer_connected"; readonly peerId: string } + | { readonly type: "peer_disconnected"; readonly peerId: string } + | { readonly type: "message"; readonly from: string; readonly payload: unknown }; + +export interface CoordinatorServerOptions { + readonly controlPath: string; + readonly endpoint: string; + readonly serverConnectionId?: string; +} + +/** The server-side endpoint of the coordinator's intentionally opaque message router. */ +export class CoordinatorServer { + readonly serverConnectionId: string; + readonly replaced: Promise; + readonly peerIds = new Set(); + readonly #controlPath: string; + readonly #endpoint: string; + readonly #listeners = new Set<(event: CoordinatorServerEvent) => void>(); + #socket?: Socket; + #registered = false; + #closed = false; + #replacedValue = false; + #resolveRegistered?: () => void; + #rejectRegistered?: (error: Error) => void; + #resolveReplaced!: () => void; + + constructor(options: CoordinatorServerOptions) { + this.serverConnectionId = options.serverConnectionId ?? randomUUID(); + this.#controlPath = options.controlPath; + this.#endpoint = options.endpoint; + this.replaced = new Promise((resolve) => { + this.#resolveReplaced = resolve; + }); + } + + get controlPath(): string { + return this.#controlPath; + } + + get wasReplaced(): boolean { + return this.#replacedValue; + } + + onEvent(listener: (event: CoordinatorServerEvent) => void): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + async connect(): Promise { + if (this.#socket) throw new Error("Coordinator server is already connected"); + const socket = await connectSocket(this.#controlPath); + this.#socket = socket; + attachJsonLineReader(socket, (message) => this.#handleMessage(message)); + const registered = new Promise((resolve, reject) => { + this.#resolveRegistered = resolve; + this.#rejectRegistered = reject; + }); + socket.once("close", () => this.#disconnected(new Error("Coordinator connection closed"))); + socket.once("error", (error) => this.#disconnected(error)); + await writeJsonLine(socket, { + type: "register_server", + protocol: COORDINATOR_PROTOCOL_VERSION, + serverConnectionId: this.serverConnectionId, + endpoint: this.#endpoint, + }); + await registered; + } + + send(peerId: string, payload: unknown): Promise { + return this.#write({ type: "send", to: peerId, payload }); + } + + broadcast(payload: unknown): Promise { + return this.#write({ type: "broadcast", payload }); + } + + close(): void { + if (this.#closed) return; + this.#closed = true; + this.#socket?.destroy(); + this.#socket = undefined; + this.peerIds.clear(); + this.#rejectRegistered?.(new Error("Coordinator server closed")); + this.#resolveRegistered = undefined; + this.#rejectRegistered = undefined; + } + + async #write(message: unknown): Promise { + if (!this.#registered || !this.#socket || this.#closed) throw new Error("Coordinator server is not connected"); + await writeJsonLine(this.#socket, message); + } + + #handleMessage(value: unknown): void { + if (!Check(CoordinatorMessageSchema, value)) { + this.#socket?.destroy(new Error("Coordinator sent an invalid message")); + return; + } + const message: CoordinatorMessage = value; + if (message.type === "server_registered") { + if (message.serverConnectionId !== this.serverConnectionId) { + this.#socket?.destroy(new Error("Coordinator returned an invalid server registration")); + return; + } + for (const peerId of message.peers) this.peerIds.add(peerId); + this.#registered = true; + this.#resolveRegistered?.(); + this.#resolveRegistered = undefined; + this.#rejectRegistered = undefined; + return; + } + if (message.type === "server_replaced") { + this.#markReplaced(); + return; + } + if (message.type === "peer_connected") { + this.peerIds.add(message.peerId); + this.#emit({ type: "peer_connected", peerId: message.peerId }); + return; + } + if (message.type === "peer_disconnected") { + this.peerIds.delete(message.peerId); + this.#emit({ type: "peer_disconnected", peerId: message.peerId }); + return; + } + if (message.type === "message") { + this.#emit({ type: "message", from: message.from, payload: message.payload }); + return; + } + this.#socket?.destroy(new Error("Coordinator sent an unsupported message")); + } + + #emit(event: CoordinatorServerEvent): void { + for (const listener of this.#listeners) listener(event); + } + + #disconnected(error: Error): void { + this.#socket = undefined; + if (this.#closed) return; + this.#rejectRegistered?.(error); + this.#resolveRegistered = undefined; + this.#rejectRegistered = undefined; + this.#markReplaced(); + } + + #markReplaced(): void { + if (this.#replacedValue) return; + this.#replacedValue = true; + this.#resolveReplaced(); + } +} + +export function getExperimentalCoordinatorControlPath(serverId: string, socketDirectory: string): string { + return join(socketDirectory, `.c-${serverId.slice(0, 12)}.sock`); +} + +export function getExperimentalServerPath(socketDirectory: string): string { + return join(socketDirectory, `.s-${randomUUID().slice(0, 12)}.sock`); +} + +export interface CoordinatorStartupLease { + close(): void; +} + +export async function ensureExperimentalCoordinator( + publicPath: string, + controlPath: string, + coordinatorUrl = new URL( + import.meta.url.endsWith(".js") ? "coordinator-process.js" : "coordinator-process.ts", + import.meta.url, + ), +): Promise { + const existing = await tryConnect(controlPath); + if (existing) return { close: () => existing.destroy() }; + const args = [ + ...(coordinatorUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : []), + fileURLToPath(coordinatorUrl), + publicPath, + controlPath, + ]; + const child = spawn(process.execPath, args, { + cwd: process.cwd(), + detached: true, + env: process.env, + stdio: "ignore", + windowsHide: true, + }); + child.unref(); + const deadline = Date.now() + COORDINATOR_START_TIMEOUT_MS; + while (true) { + const socket = await tryConnect(controlPath); + if (socket) return { close: () => socket.destroy() }; + if (child.exitCode !== null || child.signalCode !== null) throw new Error("Coordinator exited during startup"); + if (Date.now() >= deadline) throw new Error("Timed out waiting for coordinator startup"); + await new Promise((resolve) => setTimeout(resolve, COORDINATOR_RETRY_MS)); + } +} + +function connectSocket(path: string): Promise { + return new Promise((resolve, reject) => { + const socket = createConnection(path); + const onConnect = (): void => { + socket.off("error", onError); + resolve(socket); + }; + const onError = (error: Error): void => { + socket.off("connect", onConnect); + reject(error); + }; + socket.once("connect", onConnect); + socket.once("error", onError); + }); +} + +async function tryConnect(path: string): Promise { + try { + return await connectSocket(path); + } catch (error) { + const code = error instanceof Error && "code" in error ? error.code : undefined; + if (code === "ENOENT" || code === "ECONNREFUSED") return undefined; + throw error; + } +} + +function attachJsonLineReader(socket: Socket, onMessage: (message: unknown) => void): void { + let buffered = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffered += chunk; + if (Buffer.byteLength(buffered) > MAX_CONTROL_LINE_BYTES) { + socket.destroy(new Error("Coordinator message is too large")); + return; + } + while (true) { + const newline = buffered.indexOf("\n"); + if (newline === -1) return; + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + try { + onMessage(JSON.parse(line)); + } catch { + socket.destroy(new Error("Coordinator sent invalid JSON")); + return; + } + } + }); +} + +function writeJsonLine(socket: Socket, message: unknown): Promise { + return new Promise((resolve, reject) => { + socket.write(`${JSON.stringify(message)}\n`, (error) => { + if (error) reject(error); + else resolve(); + }); + }); +} diff --git a/packages/coding-agent/src/cli/experimental/coordinator-process.ts b/packages/coding-agent/src/cli/experimental/coordinator-process.ts new file mode 100644 index 000000000..c30ea8351 --- /dev/null +++ b/packages/coding-agent/src/cli/experimental/coordinator-process.ts @@ -0,0 +1,329 @@ +import { chmod, lstat, unlink } from "node:fs/promises"; +import { createConnection, createServer, type Server, type Socket } from "node:net"; + +// This process is intentionally a transport shim. It depends only on Node +// built-ins and never interprets Pi, session, worker, or lifecycle payloads. + +const COORDINATOR_PROTOCOL_VERSION = 1; +const MAX_CONTROL_LINE_BYTES = 1024 * 1024; +const EMPTY_STARTUP_GRACE_MS = 30_000; +const EMPTY_SHUTDOWN_GRACE_MS = 250; + +interface ServerPeer { + readonly serverConnectionId: string; + readonly endpoint: string; + readonly socket: Socket; +} + +interface RoutedPeer { + readonly peerId: string; + readonly socket: Socket; +} + +interface ControlMessage { + readonly type: string; + readonly protocol?: unknown; + readonly serverConnectionId?: unknown; + readonly endpoint?: unknown; + readonly peerId?: unknown; + readonly to?: unknown; + readonly payload?: unknown; +} + +const publicPath = process.argv[2]; +const controlPath = process.argv[3]; +if (!publicPath || !controlPath) throw new Error("Coordinator requires public and control socket paths"); + +const peers = new Map(); +const controlConnections = new Set(); +const publicConnections = new Map(); +let currentServer: ServerPeer | undefined; +let shuttingDown = false; +let emptyTimer: NodeJS.Timeout | undefined; + +const publicServer = createServer((socket) => acceptPublicConnection(socket)); +const controlServer = createServer((socket) => acceptControlConnection(socket)); + +async function main(): Promise { + await removeStaleSocket(controlPath); + await removeStaleSocket(publicPath); + try { + await listen(controlServer, controlPath); + await restrictSocket(controlPath); + await listen(publicServer, publicPath); + await restrictSocket(publicPath); + } catch (error) { + await Promise.all([cleanupSocket(controlPath), cleanupSocket(publicPath)]); + throw error; + } + scheduleEmptyShutdown(EMPTY_STARTUP_GRACE_MS); +} + +function acceptControlConnection(socket: Socket): void { + if (shuttingDown) { + socket.destroy(); + return; + } + controlConnections.add(socket); + let server: ServerPeer | undefined; + let peer: RoutedPeer | undefined; + attachJsonLineReader(socket, (message) => { + if (!server && !peer) { + if (message.type === "register_server") { + server = registerServer(socket, message); + return; + } + if (message.type === "register_peer") { + peer = registerPeer(socket, message); + return; + } + throw new Error("Coordinator connection did not register a role"); + } + if (server) { + if (server === currentServer) handleRoutedMessage("server", message); + return; + } + handleRoutedMessage(peer!.peerId, message); + }); + const disconnect = (): void => { + controlConnections.delete(socket); + if (server && currentServer === server) currentServer = undefined; + if (peer && peers.get(peer.peerId) === peer) { + peers.delete(peer.peerId); + if (currentServer) writeJsonLine(currentServer.socket, { type: "peer_disconnected", peerId: peer.peerId }); + } + checkEmpty(); + }; + socket.once("close", disconnect); + socket.once("error", () => socket.destroy()); +} + +function registerServer(socket: Socket, message: ControlMessage): ServerPeer { + if (message.protocol !== COORDINATOR_PROTOCOL_VERSION) throw new Error("Unsupported coordinator protocol"); + if (typeof message.serverConnectionId !== "string" || message.serverConnectionId.length === 0) { + throw new Error("Coordinator serverConnectionId must be a string"); + } + if (typeof message.endpoint !== "string" || message.endpoint.length === 0) { + throw new Error("Coordinator endpoint must be a string"); + } + const { serverConnectionId, endpoint } = message; + const server = { serverConnectionId, endpoint, socket }; + const previous = currentServer; + currentServer = server; + cancelEmptyShutdown(); + writeJsonLine(socket, { + type: "server_registered", + serverConnectionId, + peers: [...peers.keys()], + }); + if (previous && previous !== server) { + closePublicConnections(); + writeJsonLine(previous.socket, { type: "server_replaced" }); + } + return server; +} + +function registerPeer(socket: Socket, message: ControlMessage): RoutedPeer { + if (message.protocol !== COORDINATOR_PROTOCOL_VERSION) throw new Error("Unsupported coordinator protocol"); + if (typeof message.peerId !== "string" || message.peerId.length === 0) { + throw new Error("Coordinator peerId must be a string"); + } + const { peerId } = message; + if (peerId === "server" || peers.has(peerId)) throw new Error(`Coordinator peer is already connected: ${peerId}`); + const peer = { peerId, socket }; + peers.set(peerId, peer); + cancelEmptyShutdown(); + writeJsonLine(socket, { type: "peer_registered", peerId }); + if (currentServer) { + writeJsonLine(currentServer.socket, { type: "peer_connected", peerId }); + } + return peer; +} + +function handleRoutedMessage(from: string, message: ControlMessage): void { + if (message.type === "send") { + if (typeof message.to !== "string" || message.to.length === 0) { + throw new Error("Coordinator message target must be a string"); + } + const { to } = message; + const target = to === "server" ? currentServer?.socket : peers.get(to)?.socket; + if (target) writeJsonLine(target, { type: "message", from, payload: message.payload }); + return; + } + if (message.type === "broadcast") { + if (from !== "server") throw new Error("Only the current server may broadcast"); + for (const peer of peers.values()) { + writeJsonLine(peer.socket, { type: "message", from, payload: message.payload }); + } + return; + } + throw new Error(`Unknown coordinator routing message: ${message.type}`); +} + +function acceptPublicConnection(client: Socket): void { + if (shuttingDown || !currentServer) { + client.destroy(); + return; + } + cancelEmptyShutdown(); + const upstream = createConnection(currentServer.endpoint); + publicConnections.set(client, upstream); + let finalized = false; + const finalize = (): void => { + if (finalized) return; + finalized = true; + publicConnections.delete(client); + checkEmpty(); + }; + upstream.once("connect", () => { + client.pipe(upstream); + upstream.pipe(client); + }); + upstream.once("error", () => client.destroy()); + upstream.once("close", () => { + client.destroy(); + finalize(); + }); + client.once("error", () => upstream.destroy()); + client.once("close", () => { + upstream.destroy(); + finalize(); + }); +} + +function closePublicConnections(): void { + for (const [client, upstream] of publicConnections) { + client.destroy(); + upstream.destroy(); + } + publicConnections.clear(); +} + +function checkEmpty(): void { + if (shuttingDown || currentServer || peers.size > 0 || publicConnections.size > 0 || controlConnections.size > 0) { + cancelEmptyShutdown(); + return; + } + scheduleEmptyShutdown(EMPTY_SHUTDOWN_GRACE_MS); +} + +function scheduleEmptyShutdown(delayMs: number): void { + if (emptyTimer || shuttingDown) return; + emptyTimer = setTimeout(() => { + emptyTimer = undefined; + if (!currentServer && peers.size === 0 && publicConnections.size === 0 && controlConnections.size === 0) { + void shutdownCoordinator(); + } + }, delayMs); + emptyTimer.unref(); +} + +function cancelEmptyShutdown(): void { + if (!emptyTimer) return; + clearTimeout(emptyTimer); + emptyTimer = undefined; +} + +async function shutdownCoordinator(): Promise { + if (shuttingDown) return; + shuttingDown = true; + cancelEmptyShutdown(); + closePublicConnections(); + for (const connection of controlConnections) connection.destroy(); + await Promise.all([closeServer(publicServer), closeServer(controlServer)]); + await Promise.all([cleanupSocket(publicPath), cleanupSocket(controlPath)]); + process.exit(0); +} + +function writeJsonLine(socket: Socket, message: unknown): void { + if (!socket.destroyed) socket.write(`${JSON.stringify(message)}\n`); +} + +function attachJsonLineReader(socket: Socket, onMessage: (message: ControlMessage) => void): void { + let buffered = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffered += chunk; + if (Buffer.byteLength(buffered) > MAX_CONTROL_LINE_BYTES) { + socket.destroy(new Error("Coordinator message is too large")); + return; + } + while (true) { + const newline = buffered.indexOf("\n"); + if (newline === -1) return; + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + try { + const message = JSON.parse(line) as ControlMessage | null; + if (typeof message?.type !== "string") throw new Error("Coordinator message must have a type"); + onMessage(message); + } catch (error) { + socket.destroy(error instanceof Error ? error : new Error(String(error))); + return; + } + } + }); +} + +function listen(server: Server, path: string): Promise { + return new Promise((resolve, reject) => { + const onError = (error: Error): void => { + server.off("listening", onListening); + reject(error); + }; + const onListening = (): void => { + server.off("error", onError); + resolve(); + }; + server.once("error", onError); + server.once("listening", onListening); + server.listen(path); + }); +} + +function closeServer(server: Server): Promise { + if (!server.listening) return Promise.resolve(); + return new Promise((resolve) => server.close(() => resolve())); +} + +async function restrictSocket(path: string): Promise { + if (process.platform !== "win32") await chmod(path, 0o600); +} + +async function removeStaleSocket(path: string): Promise { + try { + const stats = await lstat(path); + if (!stats.isSocket()) throw new Error(`Coordinator path is not a socket: ${path}`); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") return; + throw error; + } + const live = await new Promise((resolve, reject) => { + const socket = createConnection(path); + socket.once("connect", () => { + socket.destroy(); + resolve(true); + }); + socket.once("error", (error: NodeJS.ErrnoException) => { + socket.destroy(); + if (error.code === "ECONNREFUSED" || error.code === "ENOENT") resolve(false); + else reject(error); + }); + }); + if (live) throw new Error(`Coordinator socket is already active: ${path}`); + await cleanupSocket(path); +} + +async function cleanupSocket(path: string): Promise { + if (process.platform === "win32") return; + await unlink(path).catch((error: unknown) => { + if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) throw error; + }); +} + +process.once("SIGINT", () => void shutdownCoordinator()); +process.once("SIGTERM", () => void shutdownCoordinator()); +void main().catch((error: unknown) => { + console.error(error); + process.exit(1); +}); diff --git a/packages/coding-agent/src/cli/experimental/runtime.ts b/packages/coding-agent/src/cli/experimental/runtime.ts index 297a50487..8e2c0a43a 100644 --- a/packages/coding-agent/src/cli/experimental/runtime.ts +++ b/packages/coding-agent/src/cli/experimental/runtime.ts @@ -2,23 +2,27 @@ import { randomUUID } from "node:crypto"; import { chmod, lstat, mkdir } from "node:fs/promises"; import { homedir } from "node:os"; import { basename, join } from "node:path"; -import { setTimeout as delay } from "node:timers/promises"; import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-agent-core"; import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; import { PiClient } from "@earendil-works/pi-client"; -import { requestServerDrain } from "@earendil-works/pi-client/control"; import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix"; import { isServerId } from "@earendil-works/pi-protocol"; -import type { PiServer, PiServerHost } from "@earendil-works/pi-server"; +import type { HostedHarnessHandle, PiServer, PiServerHost } from "@earendil-works/pi-server"; import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix"; import { getAgentDir } from "../../config.ts"; import { resolvePath } from "../../utils/paths.ts"; import type { ClientCommand } from "./commands/client.ts"; +import { CoordinatedSessionWorkers } from "./coordinated-session-workers.ts"; +import { + CoordinatorServer, + type CoordinatorStartupLease, + ensureExperimentalCoordinator, + getExperimentalCoordinatorControlPath, + getExperimentalServerPath, +} from "./coordinator-client.ts"; import { acquireExperimentalServerProfile } from "./server-profile.ts"; import { startExperimentalSessionWorker } from "./session-worker.ts"; -const SOCKET_RELEASE_TIMEOUT_MS = 10_000; -const SOCKET_RELEASE_POLL_MS = 10; const EXPERIMENTAL_SOCKET_ROOT = "/tmp"; export interface ExperimentalServer { @@ -47,7 +51,7 @@ export interface StartExperimentalServerOptions { readonly sessionDir?: string; } -export interface StartExperimentalServerGenerationOptions { +export interface StartExperimentalCoordinatedServerOptions { /** Persistent profile directory. Defaults to ~/.pi/server. */ readonly directory?: string; /** Physical socket directory. Defaults to a short, private per-user runtime directory. */ @@ -65,18 +69,21 @@ export function resolveExperimentalSessionDirectory(sessionDir?: string): string return resolvePath(sessionDir ?? join(getAgentDir(), "experimental", "sessions")); } +interface ExperimentalWorkerController { + readonly workerPids: ReadonlyMap; + readonly trackedSessions: readonly JsonlSessionMetadata[]; + createHarness(metadata: JsonlSessionMetadata): Promise; +} + /** Start the experimental durable list-and-attach server composition. */ export async function startExperimentalServer( options: StartExperimentalServerOptions = {}, ): Promise { - const serverId = options.serverId ?? randomUUID(); const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir); - const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() }); - const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir }); - const workerPids = new Map(); - const host: PiServerHost = { - sessions: repo, + const controller: ExperimentalWorkerController = { + workerPids, + trackedSessions: [], createHarness: async (metadata) => { const sessionId = metadata.id; const worker = await startExperimentalSessionWorker(metadata, { sessionDir }); @@ -96,6 +103,27 @@ export async function startExperimentalServer( }; }, }; + return startExperimentalServerBackend(options, controller); +} + +async function startExperimentalServerBackend( + options: StartExperimentalServerOptions, + workers: ExperimentalWorkerController, +): Promise { + const serverId = options.serverId ?? randomUUID(); + const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir); + const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() }); + const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir }); + const host: PiServerHost = { + sessions: { + list: async () => { + const sessions = new Map((await repo.list()).map((metadata) => [metadata.path, metadata])); + for (const metadata of workers.trackedSessions) sessions.set(metadata.path, metadata); + return [...sessions.values()]; + }, + }, + createHarness: (metadata) => workers.createHarness(metadata), + }; let socketPath = options.path; if (socketPath === undefined) { const socketDirectory = options.directory ?? getExperimentalSocketDirectory(); @@ -137,72 +165,100 @@ export async function startExperimentalServer( sessionDir, socketPath, server, - workerPids, + workerPids: workers.workerPids, closed, close() { - if (closePromise === undefined) - closePromise = server.close().then( - () => closed, - () => closed, - ); + closePromise ??= server.close().then( + () => closed, + () => closed, + ); return closePromise; }, }; } -/** Start the experimental local server, replacing an existing generation for the same profile. */ -export async function startExperimentalServerGeneration( - options: StartExperimentalServerGenerationOptions = {}, +/** Start a replaceable experimental server behind the stable coordinator endpoint. */ +export async function startExperimentalCoordinatedServer( + options: StartExperimentalCoordinatedServerOptions = {}, ): Promise { const directory = options.directory ?? join(homedir(), ".pi", "server"); const socketDirectory = options.socketDirectory ?? getExperimentalSocketDirectory(); const { serverId, release } = await acquireExperimentalServerProfile(directory); - let runtime: ExperimentalServer; + let backend: ExperimentalServer | undefined; + let coordinator: CoordinatorServer | undefined; + let startupLease: CoordinatorStartupLease | undefined; + let workers: CoordinatedSessionWorkers | undefined; + let released = false; try { await ensurePrivateSocketDirectory(socketDirectory); const socketPath = getUnixSocketPath(serverId, socketDirectory); - await drainExistingGeneration(serverId, socketPath); - runtime = await startExperimentalServer({ - directory, - path: socketPath, - serverId, - sessionDir: options.sessionDir, - }); - } catch (error) { - try { - await release(); - } catch (releaseError) { - throw new AggregateError([error, releaseError], "Server generation failed and launcher lock release failed"); - } - throw error; - } + const controlPath = getExperimentalCoordinatorControlPath(serverId, socketDirectory); + const serverPath = getExperimentalServerPath(socketDirectory); + startupLease = await ensureExperimentalCoordinator(socketPath, controlPath); + coordinator = new CoordinatorServer({ controlPath, endpoint: serverPath }); + const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir); + workers = new CoordinatedSessionWorkers(coordinator, sessionDir); + backend = await startExperimentalServerBackend( + { path: serverPath, serverId, sessionDir: options.sessionDir }, + workers, + ); + await coordinator.connect(); + startupLease.close(); + startupLease = undefined; + await workers.discover(coordinator.peerIds); - try { + const activeBackend = backend; + const activeCoordinator = coordinator; + const activeWorkers = workers; + void activeCoordinator.replaced + .then(async () => { + activeWorkers.detach(); + await activeBackend.close(); + }) + .finally(() => activeCoordinator.close()) + .catch(() => {}); + let closePromise: Promise | undefined; + const runtime: ExperimentalServer = { + serverId, + sessionDir: activeBackend.sessionDir, + socketPath, + server: activeBackend.server, + workerPids: activeWorkers.workerPids, + closed: activeBackend.closed, + close() { + closePromise ??= (async () => { + try { + await activeBackend.close(); + } finally { + try { + if (activeCoordinator.wasReplaced) activeWorkers.detach(); + else await activeWorkers.shutdown(); + } finally { + activeCoordinator.close(); + } + } + })(); + return closePromise; + }, + }; + released = true; await release(); + return runtime; } catch (error) { - try { - await runtime.close(); - } catch (closeError) { - throw new AggregateError([error, closeError], "Launcher lock release and server cleanup failed"); + startupLease?.close(); + if (coordinator?.wasReplaced) workers?.detach(); + const cleanup = await Promise.allSettled([ + backend?.close(), + coordinator?.wasReplaced ? undefined : workers?.shutdown(), + Promise.resolve(coordinator?.close()), + released ? undefined : release(), + ]); + const cleanupErrors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); + if (cleanupErrors.length > 0) { + throw new AggregateError([error, ...cleanupErrors], "Coordinated server startup and cleanup failed"); } throw error; } - return runtime; -} - -async function drainExistingGeneration(serverId: string, socketPath: string): Promise { - try { - await requestServerDrain({ - serverId, - transportFactory: createUnixTransportFactory({ path: socketPath }), - }); - await waitForSocketRelease(socketPath); - } catch (error) { - // At this launcher boundary, a missing or refused stable socket proves that - // no generation accepted the drain request. All ambiguous failures propagate. - if (hasErrorCode(error, "ENOENT") || hasErrorCode(error, "ECONNREFUSED")) return; - throw error; - } } /** Discover local servers, then list sessions or attach to one selected session. */ @@ -269,20 +325,6 @@ function routeFromExplicitPath(path: string): UnixServerRoute { return { serverId, path }; } -async function waitForSocketRelease(path: string): Promise { - const deadline = Date.now() + SOCKET_RELEASE_TIMEOUT_MS; - while (true) { - try { - await lstat(path); - } catch (error) { - if (hasErrorCode(error, "ENOENT")) return; - throw error; - } - if (Date.now() >= deadline) throw new Error(`Timed out waiting for server socket to close: ${path}`); - await delay(SOCKET_RELEASE_POLL_MS); - } -} - function getExperimentalSocketDirectory(): string { if (process.platform === "win32" || typeof process.getuid !== "function") { throw new Error("Experimental Unix server transport requires a POSIX user ID"); @@ -299,14 +341,3 @@ async function ensurePrivateSocketDirectory(directory: string): Promise { throw new Error(`Unix socket directory is not owned by the current user: ${directory}`); await chmod(directory, 0o700); } - -function hasErrorCode(error: unknown, code: string): boolean { - let current = error; - const seen = new Set(); - while (current instanceof Error && !seen.has(current)) { - seen.add(current); - if ("code" in current && current.code === code) return true; - current = current.cause; - } - return false; -} diff --git a/packages/coding-agent/src/cli/experimental/session-worker-process.ts b/packages/coding-agent/src/cli/experimental/session-worker-process.ts index e5fba0d14..e8e608504 100644 --- a/packages/coding-agent/src/cli/experimental/session-worker-process.ts +++ b/packages/coding-agent/src/cli/experimental/session-worker-process.ts @@ -1,3 +1,4 @@ +import { createConnection, type Socket } from "node:net"; import { isAbsolute } from "node:path"; import { AgentHarness, @@ -7,22 +8,163 @@ import { type Session, } from "@earendil-works/pi-agent-core"; import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; +import lockfile from "proper-lockfile"; +import Type, { type Static } from "typebox"; +import { Check } from "typebox/value"; import { ModelRuntime } from "../../core/model-runtime.ts"; +import { + SESSION_WORKER_CONTROL_ADDRESS_ENV, + SESSION_WORKER_CONTROL_TOKEN_ENV, + SESSION_WORKER_COORDINATED_ENV, + SESSION_WORKER_PEER_ID_ENV, + SESSION_WORKER_SESSION_KEY_ENV, +} from "./session-worker.ts"; -// Prototype-only process control. Agent operations will use the shared -// transport-independent protocol rather than additional ad hoc IPC messages. +// Process lifecycle control only. Agent operations use the shared +// transport-independent protocol rather than this channel. export type SessionWorkerCommand = { type: "shutdown" }; export type SessionWorkerEvent = | { type: "ready"; sessionId: string; pid: number } | { type: "failed"; message: string }; -function send(event: SessionWorkerEvent): Promise { - return new Promise((resolve, reject) => { - if (!process.send) { - reject(new Error("Session worker requires an IPC channel")); +const WorkerMetadataSchema = Type.Object({ + id: Type.String({ minLength: 1 }), + createdAt: Type.Integer(), + storageVersion: Type.Integer(), + cwd: Type.String(), + path: Type.String(), + modifiedAt: Type.Number(), + parentSessionId: Type.Optional(Type.String()), + legacyParentSessionPath: Type.Optional(Type.String()), +}); +const CoordinatorInputSchema = Type.Union([ + Type.Object({ type: Type.Literal("peer_registered"), peerId: Type.String() }), + Type.Object({ type: Type.Literal("message"), from: Type.Literal("server"), payload: Type.Unknown() }), +]); +const WorkerCommandSchema = Type.Union([ + Type.Object({ type: Type.Literal("shutdown") }), + Type.Object({ type: Type.Literal("discover_workers") }), +]); +type CoordinatorInput = Static; +type WorkerCommand = Static; + +type SessionWorkerControlEvent = + | { + type: "register_worker"; + protocol: 1; + token: string; + sessionKey: string; + sessionId: string; + pid: number; + } + | { + type: "worker_ready"; + token: string; + sessionKey: string; + sessionId: string; + pid: number; + metadata: JsonlSessionMetadata; + } + | { type: "worker_failed"; token: string; sessionKey: string; message: string }; + +interface WorkerControl { + readonly coordinated: boolean; + readonly messages: AsyncIterable; + readonly socket: Socket; + send(event: SessionWorkerControlEvent): Promise; +} + +let failureControl: WorkerControl | undefined; + +async function connectControl(): Promise { + const address = process.env[SESSION_WORKER_CONTROL_ADDRESS_ENV]; + const token = process.env[SESSION_WORKER_CONTROL_TOKEN_ENV]; + const encodedSessionKey = process.env[SESSION_WORKER_SESSION_KEY_ENV]; + if (!address || !token || !encodedSessionKey) throw new Error("Session worker requires a control address"); + const socket = createConnection(address); + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + const messages = createJsonLineMessages(socket); + const coordinated = process.env[SESSION_WORKER_COORDINATED_ENV] === "1"; + if (coordinated) { + const peerId = process.env[SESSION_WORKER_PEER_ID_ENV]; + if (!peerId) throw new Error("Coordinated session worker requires a peer ID"); + await writeJsonLine(socket, { type: "register_peer", protocol: 1, peerId }); + const registered = await messages[Symbol.asyncIterator]().next(); + if ( + registered.done || + !Check(CoordinatorInputSchema, registered.value) || + registered.value.type !== "peer_registered" + ) { + throw new Error("Coordinator rejected the session worker registration"); + } + } + return { + coordinated, + messages, + socket, + send: (event) => writeJsonLine(socket, coordinated ? { type: "send", to: "server", payload: event } : event), + }; +} + +async function readCommands(control: WorkerControl, onShutdown: () => void, onDiscovery: () => void): Promise { + for await (const value of control.messages) { + if (!control.coordinated) { + if (Check(WorkerCommandSchema, value) && value.type === "shutdown") onShutdown(); + continue; + } + if (!Check(CoordinatorInputSchema, value)) { + control.socket.destroy(new Error("Coordinator sent an invalid worker message")); return; } - process.send(event, (error) => { + const message: CoordinatorInput = value; + if (message.type !== "message" || !Check(WorkerCommandSchema, message.payload)) continue; + const command: WorkerCommand = message.payload; + if (command.type === "shutdown") onShutdown(); + else onDiscovery(); + } +} + +function createJsonLineMessages(socket: Socket): AsyncIterable { + const queued: unknown[] = []; + const waiters: ((value: unknown) => void)[] = []; + let buffered = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffered += chunk; + while (true) { + const newline = buffered.indexOf("\n"); + if (newline === -1) return; + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + try { + const value: unknown = JSON.parse(line); + const waiter = waiters.shift(); + if (waiter) waiter(value); + else queued.push(value); + } catch { + socket.destroy(new Error("Session worker received invalid control JSON")); + return; + } + } + }); + return { + [Symbol.asyncIterator]() { + return { + next: async () => { + const value = queued.shift() ?? (await new Promise((resolve) => waiters.push(resolve))); + return { done: false as const, value }; + }, + }; + }, + }; +} + +function writeJsonLine(socket: Socket, message: unknown): Promise { + return new Promise((resolve, reject) => { + socket.write(`${JSON.stringify(message)}\n`, (error) => { if (error) reject(error); else resolve(); }); @@ -37,27 +179,10 @@ function parseMetadata(value: string | undefined): JsonlSessionMetadata { } catch (error) { throw new Error("Session worker received invalid session metadata", { cause: error }); } - if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + if (!Check(WorkerMetadataSchema, parsed) || !isAbsolute(parsed.cwd) || !isAbsolute(parsed.path)) { throw new Error("Session worker received invalid session metadata"); } - const candidate = parsed as Record; - if ( - typeof candidate.id !== "string" || - candidate.id.length === 0 || - !Number.isSafeInteger(candidate.createdAt) || - !Number.isSafeInteger(candidate.storageVersion) || - typeof candidate.cwd !== "string" || - !isAbsolute(candidate.cwd) || - typeof candidate.path !== "string" || - !isAbsolute(candidate.path) || - typeof candidate.modifiedAt !== "number" || - !Number.isFinite(candidate.modifiedAt) || - (candidate.parentSessionId !== undefined && typeof candidate.parentSessionId !== "string") || - (candidate.legacyParentSessionPath !== undefined && typeof candidate.legacyParentSessionPath !== "string") - ) { - throw new Error("Session worker received invalid session metadata"); - } - return candidate as unknown as JsonlSessionMetadata; + return parsed; } async function closeResources(resources: { @@ -65,6 +190,7 @@ async function closeResources(resources: { session?: Session; repo: JsonlSessionRepo; executionEnv: NodeExecutionEnv; + releaseOwnership: () => Promise; }): Promise { const errors: unknown[] = []; try { @@ -83,6 +209,11 @@ async function closeResources(resources: { } catch (error) { errors.push(error); } + try { + await resources.releaseOwnership(); + } catch (error) { + errors.push(error); + } if (errors.length === 1) throw errors[0]; if (errors.length > 1) throw new AggregateError(errors, "Session worker cleanup failed"); } @@ -92,7 +223,27 @@ async function run(): Promise { if (!sessionDir || !isAbsolute(sessionDir)) throw new Error("Session worker requires an absolute session directory"); const metadata = parseMetadata(process.argv[3]); const sessionId = metadata.id; + const control = await connectControl(); + const token = process.env[SESSION_WORKER_CONTROL_TOKEN_ENV]!; + const sessionKey = Buffer.from(process.env[SESSION_WORKER_SESSION_KEY_ENV]!, "base64url").toString(); + failureControl = control; + if (!control.coordinated) { + await control.send({ + type: "register_worker", + protocol: 1, + token, + sessionKey, + sessionId, + pid: process.pid, + }); + } + const releaseOwnership = await lockfile.lock(metadata.path, { + realpath: true, + stale: 2_000, + update: 1_000, + retries: { retries: 320, factor: 1, minTimeout: 25, maxTimeout: 25, maxRetryTime: 8_000 }, + }); const executionEnv = new NodeExecutionEnv({ cwd: metadata.cwd }); const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir }); let session: Session | undefined; @@ -105,7 +256,7 @@ async function run(): Promise { ({ harness } = await AgentHarness.create({ session, models: modelRuntime, model, tools: [], resources: {} })); } catch (error) { try { - await closeResources({ harness, session, repo, executionEnv }); + await closeResources({ harness, session, repo, executionEnv, releaseOwnership }); } catch (cleanupError) { throw new AggregateError([error, cleanupError], "Session worker startup and cleanup failed"); } @@ -114,7 +265,7 @@ async function run(): Promise { let closing: Promise | undefined; const close = (): Promise => { - closing ??= closeResources({ harness, repo, executionEnv }); + closing ??= closeResources({ harness, repo, executionEnv, releaseOwnership }); return closing; }; const closeAndExit = (): void => { @@ -127,15 +278,29 @@ async function run(): Promise { ); }; - process.on("message", (message: SessionWorkerCommand) => { - if (message?.type === "shutdown") closeAndExit(); - }); - process.once("disconnect", closeAndExit); + let ready = false; + const announce = (): void => { + if (!ready) return; + void control + .send({ + type: "worker_ready", + token, + sessionKey, + sessionId, + pid: process.pid, + metadata, + }) + .catch(() => closeAndExit()); + }; + void readCommands(control, closeAndExit, announce); + control.socket.once("close", closeAndExit); + control.socket.once("error", () => closeAndExit()); process.once("SIGTERM", closeAndExit); process.once("SIGINT", closeAndExit); try { - await send({ type: "ready", sessionId, pid: process.pid }); + ready = true; + await control.send({ type: "worker_ready", token, sessionKey, sessionId, pid: process.pid, metadata }); } catch (error) { try { await close(); @@ -148,6 +313,11 @@ async function run(): Promise { void run().catch(async (error: unknown) => { const message = error instanceof Error ? error.message : String(error); - await send({ type: "failed", message }).catch(() => {}); + const token = process.env[SESSION_WORKER_CONTROL_TOKEN_ENV]; + const encodedSessionKey = process.env[SESSION_WORKER_SESSION_KEY_ENV]; + if (token && encodedSessionKey) { + const sessionKey = Buffer.from(encodedSessionKey, "base64url").toString(); + await failureControl?.send({ type: "worker_failed", token, sessionKey, message }).catch(() => {}); + } process.exit(1); }); diff --git a/packages/coding-agent/src/cli/experimental/session-worker.ts b/packages/coding-agent/src/cli/experimental/session-worker.ts index ceee791b9..f6e6cfe25 100644 --- a/packages/coding-agent/src/cli/experimental/session-worker.ts +++ b/packages/coding-agent/src/cli/experimental/session-worker.ts @@ -1,11 +1,35 @@ -import { type ChildProcess, fork } from "node:child_process"; -import { isAbsolute } from "node:path"; +import { type ChildProcess, spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { chmod, unlink } from "node:fs/promises"; +import { createServer, type Server, type Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { isAbsolute, join } from "node:path"; +import { fileURLToPath } from "node:url"; import type { JsonlSessionMetadata } from "@earendil-works/pi-agent-core"; +import Type, { type Static } from "typebox"; +import { Check } from "typebox/value"; import type { SessionWorkerCommand, SessionWorkerEvent } from "./session-worker-process.ts"; -const DEFAULT_STARTUP_TIMEOUT_MS = 5_000; +const DEFAULT_STARTUP_TIMEOUT_MS = 10_000; const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000; const MAX_TIMER_DELAY_MS = 2_147_483_647; +const WorkerControlInputSchema = Type.Union([ + Type.Object({ type: Type.Literal("worker_failed"), message: Type.String() }), + Type.Object({ type: Type.Literal("worker_ready"), sessionId: Type.String(), pid: Type.Integer({ minimum: 1 }) }), + Type.Object({ + type: Type.Literal("register_worker"), + protocol: Type.Literal(1), + token: Type.String(), + sessionId: Type.String(), + pid: Type.Integer({ minimum: 1 }), + }), +]); +type WorkerControlInput = Static; +export const SESSION_WORKER_CONTROL_ADDRESS_ENV = "PI_SESSION_WORKER_CONTROL_ADDRESS"; +export const SESSION_WORKER_CONTROL_TOKEN_ENV = "PI_SESSION_WORKER_CONTROL_TOKEN"; +export const SESSION_WORKER_SESSION_KEY_ENV = "PI_SESSION_WORKER_SESSION_KEY_BASE64"; +export const SESSION_WORKER_COORDINATED_ENV = "PI_SESSION_WORKER_COORDINATED"; +export const SESSION_WORKER_PEER_ID_ENV = "PI_SESSION_WORKER_PEER_ID"; export interface ExperimentalSessionWorker { readonly sessionId: string; @@ -14,6 +38,13 @@ export interface ExperimentalSessionWorker { close(): Promise; } +export interface SessionWorkerLaunchSpec { + readonly command: string; + readonly args: readonly string[]; + readonly cwd: string; + readonly env: Readonly>; +} + export interface StartExperimentalSessionWorkerOptions { readonly sessionDir: string; readonly startupTimeoutMs?: number; @@ -21,29 +52,74 @@ export interface StartExperimentalSessionWorkerOptions { readonly workerUrl?: URL; } +export function createExperimentalSessionWorkerLaunchSpec( + metadata: JsonlSessionMetadata, + options: Pick, +): SessionWorkerLaunchSpec { + if (!isAbsolute(options.sessionDir)) throw new TypeError("Session worker sessionDir must be absolute"); + const workerUrl = + options.workerUrl ?? + new URL( + import.meta.url.endsWith(".js") ? "session-worker-process.js" : "session-worker-process.ts", + import.meta.url, + ); + return { + command: process.execPath, + args: [ + ...(workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : []), + fileURLToPath(workerUrl), + options.sessionDir, + JSON.stringify(metadata), + ], + cwd: process.cwd(), + env: Object.fromEntries( + Object.entries(process.env).flatMap(([name, value]) => (value === undefined ? [] : [[name, value]])), + ), + }; +} + export async function startExperimentalSessionWorker( metadata: JsonlSessionMetadata, options: StartExperimentalSessionWorkerOptions, ): Promise { - if (!isAbsolute(options.sessionDir)) throw new TypeError("Session worker sessionDir must be absolute"); - const sessionId = metadata.id; const startupTimeoutMs = validateTimeout(options.startupTimeoutMs ?? DEFAULT_STARTUP_TIMEOUT_MS, "startupTimeoutMs"); const shutdownTimeoutMs = validateTimeout( options.shutdownTimeoutMs ?? DEFAULT_SHUTDOWN_TIMEOUT_MS, "shutdownTimeoutMs", ); - const workerUrl = options.workerUrl ?? defaultWorkerUrl(); - const child = fork(workerUrl, [options.sessionDir, JSON.stringify(metadata)], { - execArgv: workerExecArgv(workerUrl), - stdio: ["ignore", "ignore", "inherit", "ipc"], - }); + const launch = createExperimentalSessionWorkerLaunchSpec(metadata, options); + const token = randomUUID(); + const controlId = randomUUID(); + const controlAddress = + process.platform === "win32" ? `\\\\.\\pipe\\pi-worker-${controlId}` : join(tmpdir(), `pi-w-${controlId}.sock`); + const control = await createWorkerControlServer(controlAddress, token, metadata.id); + let child: ChildProcess; + try { + child = spawn(launch.command, launch.args, { + cwd: launch.cwd, + detached: true, + env: { + ...launch.env, + [SESSION_WORKER_CONTROL_ADDRESS_ENV]: controlAddress, + [SESSION_WORKER_CONTROL_TOKEN_ENV]: token, + [SESSION_WORKER_SESSION_KEY_ENV]: Buffer.from(metadata.path).toString("base64url"), + }, + stdio: "ignore", + windowsHide: true, + }); + child.unref(); + } catch (error) { + await control.close(); + throw error; + } try { - const ready = await waitForReady(child, sessionId, startupTimeoutMs); - return new ChildSessionWorker(child, ready.sessionId, ready.pid, shutdownTimeoutMs); + const ready = await waitForReady(child, control, metadata.id, startupTimeoutMs); + return new SpawnedSessionWorker(child, control, ready.sessionId, ready.pid, shutdownTimeoutMs); } catch (error) { if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); await waitForExit(child); + await control.close(); throw error; } } @@ -55,38 +131,163 @@ function validateTimeout(value: number, name: string): number { return value; } -function workerExecArgv(workerUrl: URL): string[] { - return workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : process.execArgv; +type WorkerControllerEvent = SessionWorkerEvent | { type: "hello"; token: string; sessionId: string; pid: number }; + +interface WorkerControlServer { + readonly connected: Promise; + readonly events: AsyncIterable; + send(message: SessionWorkerCommand): Promise; + close(): Promise; } -function defaultWorkerUrl(): URL { - return new URL( - import.meta.url.endsWith(".js") ? "session-worker-process.js" : "session-worker-process.ts", - import.meta.url, - ); +async function createWorkerControlServer( + address: string, + token: string, + expectedSessionId: string, +): Promise { + let accepted: Socket | undefined; + let resolveConnected!: (socket: Socket) => void; + const connected = new Promise((resolve) => { + resolveConnected = resolve; + }); + const messages: WorkerControllerEvent[] = []; + const waiters: ((message: WorkerControllerEvent) => void)[] = []; + const server = createServer((socket) => { + if (accepted) { + socket.destroy(); + return; + } + accepted = socket; + resolveConnected(socket); + attachJsonLineReader(socket, (value) => { + if (!Check(WorkerControlInputSchema, value)) { + socket.destroy(new Error("Session worker sent an invalid control message")); + return; + } + const message: WorkerControlInput = value; + let event: WorkerControllerEvent; + if (message.type === "worker_failed") { + event = { type: "failed", message: message.message }; + } else if (message.type === "worker_ready") { + event = { type: "ready", sessionId: message.sessionId, pid: message.pid }; + } else { + event = { + type: "hello", + token: message.token, + sessionId: message.sessionId, + pid: message.pid, + }; + } + if (event.type === "hello" && (event.token !== token || event.sessionId !== expectedSessionId)) { + socket.destroy(new Error("Session worker sent invalid control credentials")); + return; + } + const waiter = waiters.shift(); + if (waiter) waiter(event); + else messages.push(event); + }); + }); + await listen(server, address); + if (process.platform !== "win32") await chmod(address, 0o600); + + let closePromise: Promise | undefined; + return { + connected, + events: { + [Symbol.asyncIterator]() { + return { + next: async () => { + const message = + messages.shift() ?? + (await new Promise<(typeof messages)[number]>((resolve) => waiters.push(resolve))); + return { done: false as const, value: message }; + }, + }; + }, + }, + async send(message) { + const socket = await connected; + await new Promise((resolve, reject) => { + socket.write(`${JSON.stringify(message)}\n`, (error) => { + if (error) reject(error); + else resolve(); + }); + }); + }, + close() { + closePromise ??= closeWorkerControl(server, accepted, address); + return closePromise; + }, + }; } -class ChildSessionWorker implements ExperimentalSessionWorker { +async function waitForReady( + child: ChildProcess, + control: WorkerControlServer, + expectedSessionId: string, + startupTimeoutMs: number, +): Promise> { + const events = control.events[Symbol.asyncIterator](); + const startup = (async () => { + const hello = (await events.next()).value; + if (hello.type !== "hello" || hello.sessionId !== expectedSessionId || hello.pid !== child.pid) { + throw new Error("Session worker reported an invalid identity"); + } + const event = (await events.next()).value; + if (event.type === "failed") throw new Error(`Session worker failed: ${event.message}`); + if (event.type !== "ready" || event.sessionId !== expectedSessionId || event.pid !== child.pid) { + throw new Error("Session worker reported an invalid identity"); + } + return event; + })(); + const exited = waitForExit(child).then(({ code, signal }) => { + throw new Error(`Session worker exited before readiness (${signal ?? code ?? "unknown"})`); + }); + let timeout: NodeJS.Timeout | undefined; + const timedOut = new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error(`Session worker startup timed out after ${startupTimeoutMs}ms`)), + startupTimeoutMs, + ); + timeout.unref(); + }); + try { + return await Promise.race([startup, exited, timedOut]); + } finally { + if (timeout) clearTimeout(timeout); + } +} + +class SpawnedSessionWorker implements ExperimentalSessionWorker { readonly terminated: Promise; readonly sessionId: string; readonly pid: number; readonly #child: ChildProcess; + readonly #control: WorkerControlServer; readonly #exit: Promise<{ code: number | null; signal: NodeJS.Signals | null }>; readonly #shutdownTimeoutMs: number; #closePromise?: Promise; #closeRequested = false; - constructor(child: ChildProcess, sessionId: string, pid: number, shutdownTimeoutMs: number) { + constructor( + child: ChildProcess, + control: WorkerControlServer, + sessionId: string, + pid: number, + shutdownTimeoutMs: number, + ) { this.#child = child; + this.#control = control; this.sessionId = sessionId; this.pid = pid; this.#shutdownTimeoutMs = shutdownTimeoutMs; this.#exit = waitForExit(child); - this.terminated = this.#exit.then(({ code, signal }) => - this.#closeRequested + this.terminated = this.#exit.then(async ({ code, signal }) => { + await this.#control.close(); + return this.#closeRequested ? undefined - : new Error(`Session worker ${sessionId} exited unexpectedly (${signal ?? code ?? "unknown"})`), - ); + : new Error(`Session worker ${sessionId} exited unexpectedly (${signal ?? code ?? "unknown"})`); + }); } close(): Promise { @@ -96,69 +297,62 @@ class ChildSessionWorker implements ExperimentalSessionWorker { } async #close(): Promise { - if (this.#child.exitCode !== null || this.#child.signalCode !== null) return; - if (!this.#child.connected) { - this.#child.kill("SIGKILL"); - } else { - this.#child.send({ type: "shutdown" } satisfies SessionWorkerCommand, (error) => { - if (error) this.#child.kill("SIGKILL"); - }); - } - const timeout = setTimeout(() => this.#child.kill("SIGKILL"), this.#shutdownTimeoutMs); - timeout.unref(); - try { - await this.#exit; - } finally { - clearTimeout(timeout); + if (this.#child.exitCode === null && this.#child.signalCode === null) { + await this.#control.send({ type: "shutdown" }).catch(() => this.#child.kill("SIGKILL")); + const timeout = setTimeout(() => this.#child.kill("SIGKILL"), this.#shutdownTimeoutMs); + timeout.unref(); + try { + await this.#exit; + } finally { + clearTimeout(timeout); + } } + await this.#control.close(); } } -function waitForReady( - child: ChildProcess, - expectedSessionId: string, - startupTimeoutMs: number, -): Promise> { +function attachJsonLineReader(socket: Socket, onMessage: (message: unknown) => void): void { + let buffered = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffered += chunk; + while (true) { + const newline = buffered.indexOf("\n"); + if (newline === -1) return; + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + try { + onMessage(JSON.parse(line)); + } catch { + socket.destroy(new Error("Session worker sent invalid JSON")); + return; + } + } + }); +} + +function listen(server: Server, address: string): Promise { return new Promise((resolve, reject) => { - let timeout: NodeJS.Timeout | undefined; - const cleanup = (): void => { - if (timeout) clearTimeout(timeout); - child.off("message", onMessage); - child.off("error", onError); - child.off("exit", onExit); - }; - const fail = (error: Error): void => { - cleanup(); + const onError = (error: Error): void => { + server.off("listening", onListening); reject(error); }; - const onMessage = (message: SessionWorkerEvent): void => { - if (message?.type === "failed") { - fail(new Error(`Session worker failed: ${message.message}`)); - return; - } - if (message?.type !== "ready") return; - if (message.sessionId !== expectedSessionId || message.pid !== child.pid) { - fail(new Error("Session worker reported an invalid identity")); - return; - } - cleanup(); - resolve(message); + const onListening = (): void => { + server.off("error", onError); + resolve(); }; - const onError = (error: Error): void => fail(error); - const onExit = (code: number | null, signal: NodeJS.Signals | null): void => { - fail(new Error(`Session worker exited before readiness (${signal ?? code ?? "unknown"})`)); - }; - timeout = setTimeout( - () => fail(new Error(`Session worker startup timed out after ${startupTimeoutMs}ms`)), - startupTimeoutMs, - ); - timeout.unref(); - child.on("message", onMessage); - child.once("error", onError); - child.once("exit", onExit); + server.once("error", onError); + server.once("listening", onListening); + server.listen(address); }); } +async function closeWorkerControl(server: Server, socket: Socket | undefined, address: string): Promise { + socket?.destroy(); + if (server.listening) await new Promise((resolve) => server.close(() => resolve())); + if (process.platform !== "win32") await unlink(address).catch(() => undefined); +} + function waitForExit(child: ChildProcess): Promise<{ code: number | null; signal: NodeJS.Signals | null }> { if (child.exitCode !== null || child.signalCode !== null) { return Promise.resolve({ code: child.exitCode, signal: child.signalCode }); diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index f5d8ed0c1..d2bf234fd 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -29,7 +29,7 @@ import { resolveCredentialForPrint } from "./cli/credential-print.ts"; import { experimentalCli } from "./cli/experimental/cli.ts"; import type { ClientCommand } from "./cli/experimental/commands/client.ts"; import type { ServerCommand } from "./cli/experimental/commands/server.ts"; -import { runExperimentalClient, startExperimentalServerGeneration } from "./cli/experimental/runtime.ts"; +import { runExperimentalClient, startExperimentalCoordinatedServer } from "./cli/experimental/runtime.ts"; import { processFileArguments } from "./cli/file-processor.ts"; import { buildInitialMessage } from "./cli/initial-message.ts"; import { listModels } from "./cli/list-models.ts"; @@ -590,7 +590,7 @@ async function waitForTermination(serverClosed: Promise): Promise { async function runExperimentalServerCommand(command: ServerCommand): Promise { if (command.auth !== undefined) throw new Error("Authentication is not supported by the local demo server"); if (command.listen !== undefined) throw new Error("The local demo server uses its server-addressed Unix socket"); - const runtime = await startExperimentalServerGeneration({ sessionDir: command.sessionDir }); + const runtime = await startExperimentalCoordinatedServer({ sessionDir: command.sessionDir }); console.log(`Server: ${runtime.serverId}`); console.log(`Socket: ${runtime.socketPath}`); try { diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index edd6ac501..cb5614b34 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -1,11 +1,13 @@ import { chmod, lstat, mkdtemp, rm } from "node:fs/promises"; import { join, resolve } from "node:path"; +import { PiClient } from "@earendil-works/pi-client"; +import { createUnixTransportFactory } from "@earendil-works/pi-client/unix"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { type ExperimentalServer, runExperimentalClient, + startExperimentalCoordinatedServer, startExperimentalServer, - startExperimentalServerGeneration, } from "../src/cli/experimental/runtime.ts"; import { configureExperimentalWorkerModel, createExperimentalSessions } from "./experimental-session-support.ts"; @@ -130,54 +132,101 @@ describe("experimental durable server composition", () => { for (const pid of pids) expect(processExists(pid)).toBe(false); }); - test("serializes concurrent launchers so only one generation remains active", async () => { + test("coordinated server replaces an exited worker on the next attach", async () => { + const directory = await mkdtemp(join("/tmp", "pew-")); + directories.add(directory); + const runtime = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + servers.add(runtime); + await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); + const firstPid = runtime.workerPids.get("demo-1"); + expect(firstPid).toEqual(expect.any(Number)); + + process.kill(firstPid!, "SIGKILL"); + await expect.poll(() => runtime.workerPids.has("demo-1")).toBe(false); + await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); + const replacementPid = runtime.workerPids.get("demo-1"); + expect(replacementPid).toEqual(expect.any(Number)); + expect(replacementPid).not.toBe(firstPid); + }); + + test("normal coordinated shutdown stops detached workers", async () => { + const directory = await mkdtemp(join("/tmp", "pes-")); + directories.add(directory); + const runtime = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + servers.add(runtime); + await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); + const pid = runtime.workerPids.get("demo-1"); + expect(pid).toEqual(expect.any(Number)); + + await runtime.close(); + expect(processExists(pid!)).toBe(false); + expect(runtime.workerPids.size).toBe(0); + await expect.poll(() => pathExists(runtime.socketPath)).toBe(false); + }); + + test("serializes concurrent launchers so only one server remains active", async () => { const directory = await mkdtemp(join("/tmp", "pel-")); directories.add(directory); - const generations = await Promise.all([ - startExperimentalServerGeneration({ directory, socketDirectory: directory }), - startExperimentalServerGeneration({ directory, socketDirectory: directory }), + const runtimes = await Promise.all([ + startExperimentalCoordinatedServer({ directory, socketDirectory: directory }), + startExperimentalCoordinatedServer({ directory, socketDirectory: directory }), ]); - for (const generation of generations) servers.add(generation); + for (const runtime of runtimes) servers.add(runtime); - expect(generations[0].serverId).toBe(generations[1].serverId); - const closed = await Promise.all( - generations.map((generation) => - Promise.race([ - generation.closed.then(() => true), - new Promise((resolve) => setImmediate(() => resolve(false))), - ]), - ), - ); - expect(closed.filter(Boolean)).toHaveLength(1); + expect(runtimes[0].serverId).toBe(runtimes[1].serverId); + await expect + .poll(async () => { + const closed = await Promise.all( + runtimes.map((runtime) => + Promise.race([ + runtime.closed.then(() => true), + new Promise((resolve) => setImmediate(() => resolve(false))), + ]), + ), + ); + return closed.filter(Boolean).length; + }) + .toBe(1); await expect(runExperimentalClient({ command: "client" }, { directory })).resolves.toMatchObject({ kind: "list", sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }], }); }); - test("starts a clean replacement and requires explicit reattachment", async () => { + test("shuts down the replaced server without waiting for its clients", async () => { + const directory = await mkdtemp(join("/tmp", "peg-")); + directories.add(directory); + const first = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + servers.add(first); + const client = await PiClient.connect({ + serverId: first.serverId, + transportFactory: createUnixTransportFactory({ path: first.socketPath }), + }); + await client.listSessions(); + + const replacement = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + servers.add(replacement); + await first.closed; + await expect(runExperimentalClient({ command: "client" }, { directory })).resolves.toMatchObject({ + kind: "list", + sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }], + }); + await client.dispose(); + }); + + test("discovers workers after replacing the server", async () => { const firstDirectory = await mkdtemp(join("/tmp", "per-")); - const otherDirectory = await mkdtemp(join("/tmp", "per-")); directories.add(firstDirectory); - directories.add(otherDirectory); - const first = await startExperimentalServerGeneration({ + const first = await startExperimentalCoordinatedServer({ directory: firstDirectory, socketDirectory: firstDirectory, }); - const other = await startExperimentalServerGeneration({ - directory: otherDirectory, - socketDirectory: otherDirectory, - }); servers.add(first); - servers.add(other); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory: firstDirectory }); - await runExperimentalClient({ command: "client", sessionId: "demo-2" }, { directory: otherDirectory }); const firstWorkerPid = first.workerPids.get("demo-1"); - const otherWorkerPid = other.workerPids.get("demo-2"); expect(firstWorkerPid).toEqual(expect.any(Number)); - expect(otherWorkerPid).toEqual(expect.any(Number)); - const replacement = await startExperimentalServerGeneration({ + const replacement = await startExperimentalCoordinatedServer({ directory: firstDirectory, socketDirectory: firstDirectory, }); @@ -185,11 +234,9 @@ describe("experimental durable server composition", () => { await first.closed; expect(replacement.serverId).toBe(first.serverId); - expect(replacement.workerPids.size).toBe(0); - expect(first.workerPids.size).toBe(0); - expect(processExists(firstWorkerPid!)).toBe(false); - expect(other.workerPids.get("demo-2")).toBe(otherWorkerPid); - expect(processExists(otherWorkerPid!)).toBe(true); + expect(replacement.workerPids.get("demo-1")).toBe(firstWorkerPid); + await expect.poll(() => first.workerPids.size).toBe(0); + expect(processExists(firstWorkerPid!)).toBe(true); await expect(runExperimentalClient({ command: "client" }, { directory: firstDirectory })).resolves.toMatchObject({ kind: "list", @@ -199,8 +246,36 @@ describe("experimental durable server composition", () => { ], }); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory: firstDirectory }); - expect(replacement.workerPids.get("demo-1")).toEqual(expect.any(Number)); - expect(replacement.workerPids.get("demo-1")).not.toBe(firstWorkerPid); + expect(replacement.workerPids.get("demo-1")).toBe(firstWorkerPid); + await runExperimentalClient({ command: "client", sessionId: "demo-2" }, { directory: firstDirectory }); + expect(replacement.workerPids.get("demo-2")).toEqual(expect.any(Number)); + expect(replacement.workerPids.get("demo-2")).not.toBe(firstWorkerPid); + }); + + test("restores tracked sessions that are outside the replacement catalog", async () => { + const directory = await mkdtemp(join("/tmp", "pet-")); + const emptySessionDir = await mkdtemp(join("/tmp", "pet-sessions-")); + directories.add(directory); + directories.add(emptySessionDir); + const first = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + servers.add(first); + await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); + const workerPid = first.workerPids.get("demo-1"); + + const replacement = await startExperimentalCoordinatedServer({ + directory, + socketDirectory: directory, + sessionDir: emptySessionDir, + }); + servers.add(replacement); + await first.closed; + + await expect(runExperimentalClient({ command: "client" }, { directory })).resolves.toEqual({ + kind: "list", + sessions: [{ serverId: first.serverId, sessionId: "demo-1" }], + }); + await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); + expect(replacement.workerPids.get("demo-1")).toBe(workerPid); }); test("reports missing and ambiguous session selections", async () => { @@ -232,6 +307,16 @@ describe("experimental durable server composition", () => { }); }); +async function pathExists(path: string): Promise { + try { + await lstat(path); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } +} + function processExists(pid: number): boolean { try { process.kill(pid, 0); diff --git a/packages/coding-agent/test/experimental-server-replacement.test.ts b/packages/coding-agent/test/experimental-server-replacement.test.ts index 32e918fe7..d482ec574 100644 --- a/packages/coding-agent/test/experimental-server-replacement.test.ts +++ b/packages/coding-agent/test/experimental-server-replacement.test.ts @@ -6,6 +6,7 @@ import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { afterEach, describe, expect, test } from "vitest"; import { runExperimentalClient } from "../src/cli/experimental/runtime.ts"; +import { configureExperimentalWorkerModel, createExperimentalSessions } from "./experimental-session-support.ts"; const cliPath = fileURLToPath(new URL("../src/cli.ts", import.meta.url)); const processes = new Set(); @@ -17,10 +18,10 @@ interface RunningCli { readonly errors: () => string; } -async function startServer(home: string): Promise { +async function startServer(home: string, agentDir: string): Promise { const child = spawn(process.execPath, ["--import", "tsx", cliPath, "server"], { cwd: fileURLToPath(new URL("../../..", import.meta.url)), - env: { ...process.env, HOME: home, PI_EXPERIMENTAL: "1" }, + env: { ...process.env, HOME: home, PI_CODING_AGENT_DIR: agentDir, PI_EXPERIMENTAL: "1" }, stdio: ["ignore", "pipe", "pipe"], }); processes.add(child); @@ -91,12 +92,15 @@ afterEach(async () => { }); describe.skipIf(process.platform === "win32")("experimental CLI server replacement", () => { - test("a second CLI generation starts clean and accepts explicit reattachment", async () => { + test("a second CLI server takes over the stable coordinator socket", async () => { const root = await mkdtemp(join(tmpdir(), "pcs-")); directories.add(root); const home = join(root, "long-home-segment".repeat(8)); await mkdir(home); - const first = await startServer(home); + const agentDir = join(root, "agent"); + await configureExperimentalWorkerModel(agentDir); + await createExperimentalSessions(join(agentDir, "experimental", "sessions"), ["demo-1", "demo-2"]); + const first = await startServer(home, agentDir); const firstIdentity = await waitForOutput( first, /Server: ([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})/, @@ -110,7 +114,7 @@ describe.skipIf(process.platform === "win32")("experimental CLI server replaceme connect: { transport: "unix", path: firstSocket[1]! }, }); - const replacement = await startServer(home); + const replacement = await startServer(home, agentDir); const replacementIdentity = await waitForOutput( replacement, /Server: ([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})/, diff --git a/packages/coding-agent/test/fixtures/session-worker-fixture.ts b/packages/coding-agent/test/fixtures/session-worker-fixture.ts index 710b86cef..0503ba8de 100644 --- a/packages/coding-agent/test/fixtures/session-worker-fixture.ts +++ b/packages/coding-agent/test/fixtures/session-worker-fixture.ts @@ -1,6 +1,12 @@ +import { createConnection } from "node:net"; + // Test-only child executable. Keep this independent from the real worker so // lifecycle failure paths remain deterministic as production integration evolves. -type Command = { type: "shutdown" }; +const address = process.env.PI_SESSION_WORKER_CONTROL_ADDRESS; +const token = process.env.PI_SESSION_WORKER_CONTROL_TOKEN; +const encodedSessionKey = process.env.PI_SESSION_WORKER_SESSION_KEY_BASE64; +if (!address || !token || !encodedSessionKey) throw new Error("Fixture requires worker control configuration"); +const sessionKey = Buffer.from(encodedSessionKey, "base64url").toString(); const parsedMetadata: unknown = JSON.parse(process.argv[3] ?? "null"); const sessionId = @@ -8,24 +14,38 @@ const sessionId = ? parsedMetadata.id : undefined; const mode = typeof sessionId === "string" ? sessionId.replaceAll("\0", "") : undefined; +const socket = createConnection(address); +await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); +}); +const send = (message: unknown): void => { + socket.write(`${JSON.stringify(message)}\n`); +}; +send({ type: "register_worker", protocol: 1, token, sessionKey, sessionId, pid: process.pid }); if (mode === "ready") { - process.send?.({ type: "ready", sessionId, pid: process.pid }); - process.on("message", (message: Command) => { - if (message?.type === "shutdown") process.exit(0); + send({ type: "worker_ready", sessionId, pid: process.pid }); + let buffered = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk: string) => { + buffered += chunk; + if (buffered.includes("\n") && JSON.parse(buffered.slice(0, buffered.indexOf("\n"))).type === "shutdown") { + process.exit(0); + } }); } else if (mode === "fail") { - process.send?.({ type: "failed", message: "fixture startup failed" }); + send({ type: "worker_failed", message: "fixture startup failed" }); } else if (mode === "exit") { process.exit(2); } else if (mode === "startup-hang") { - process.on("message", () => {}); + socket.resume(); } else if (mode === "hang") { - process.send?.({ type: "ready", sessionId, pid: process.pid }); - process.on("message", () => {}); + send({ type: "worker_ready", sessionId, pid: process.pid }); + socket.resume(); } else { - process.send?.({ - type: "ready", + send({ + type: "worker_ready", sessionId: sessionId === undefined ? "missing" : `${sessionId}-different`, pid: process.pid, }); diff --git a/packages/coding-agent/tsconfig.build.json b/packages/coding-agent/tsconfig.build.json index 20b0a9df9..38fbfad4d 100644 --- a/packages/coding-agent/tsconfig.build.json +++ b/packages/coding-agent/tsconfig.build.json @@ -8,7 +8,6 @@ "@earendil-works/pi-ai": ["../ai/dist/index.d.ts"], "@earendil-works/pi-ai/*": ["../ai/dist/*.d.ts", "../ai/dist/providers/*.d.ts"], "@earendil-works/pi-client": ["../client/dist/index.d.ts"], - "@earendil-works/pi-client/control": ["../client/dist/control.d.ts"], "@earendil-works/pi-client/unix": ["../client/dist/unix.d.ts"], "@earendil-works/pi-client/*": ["../client/dist/*.d.ts"], "@earendil-works/pi-server": ["../server/dist/index.d.ts"], diff --git a/packages/protocol/README.md b/packages/protocol/README.md index 057422e69..591fef1fb 100644 --- a/packages/protocol/README.md +++ b/packages/protocol/README.md @@ -6,14 +6,11 @@ Protocol version `1` currently contains the first control-plane slice: - a version handshake that identifies the logical `serverId`; - a service RPC manifest with `list()` and `attach(sessionId)`; -- a separate launcher-only server-control manifest with `drain()`; - correlated responses and bounded protocol errors. The manifest generates typed client methods and validated server dispatch. The wire uses generic `{ serverId, method, args }` calls rather than a hand-written command union. `list()` returns the durable `SessionMetadata` values from `SessionRepo.list()`. `attach()` returns only the attached `sessionId`; the real `Session` and `AgentHarness` remain hosted by the server. -`drain()` tells one server generation to close its hosted Harnesses, acknowledges successful cleanup, and begins shutdown. It does not transfer hosted-session state. The server does not know whether another generation will replace it; replacement and explicit client reattachment belong to the launcher and clients. - -Separating `ServerControlRpc` from `ServiceRpc` is an API boundary, not an authorization boundary. Transports exposing server control must authenticate the caller as an administrator. The experimental local launcher uses a mode-`0600` Unix socket and trusts same-user peers. +Server and worker lifecycle is intentionally outside this public protocol. The experimental local coordinator is only an opaque message router; each replaceable server process owns the private lifecycle protocol. Each wire frame consists of a four-byte unsigned big-endian payload length followed by one definite-length CBOR item. `encodeClientMessage()` and `encodeServerMessage()` validate and encode complete frames. `ClientMessageDecoder` and `ServerMessageDecoder` accept arbitrary stream fragmentation and coalescing. diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index 85e5d3eae..2e06962bf 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -63,31 +63,10 @@ export type ServiceRpcResultUnion = RpcResultUnion; export const ServiceRpcCallSchema = Type.Unsafe(createRpcCallSchema(ServiceRpc)); export const ServiceRpcResultSchema = Type.Unsafe(createRpcResultSchema(ServiceRpc)); -/** Administrative lifecycle operations reserved for the server launcher. */ -export const ServerControlRpc = defineRpc({ - drain: { - args: Type.Tuple([]), - result: StrictObject({}), - }, -}); -export type ServerControlRpcManifest = typeof ServerControlRpc; -export type ServerControlRpcMethod = RpcMethodName; -export type ServerControlRpcArgs = RpcArgs; -export type ServerControlRpcResult = RpcResult< - ServerControlRpcManifest, - TMethod ->; -export type ServerControlRpcCall = RpcCall; -export type ServerControlRpcResultUnion = RpcResultUnion; -export const ServerControlRpcCallSchema = Type.Unsafe(createRpcCallSchema(ServerControlRpc)); -export const ServerControlRpcResultSchema = Type.Unsafe( - createRpcResultSchema(ServerControlRpc), -); - -export type ProtocolRpcCall = ServiceRpcCall | ServerControlRpcCall; -export type ProtocolRpcResult = ServiceRpcResultUnion | ServerControlRpcResultUnion; -const ProtocolRpcCallSchema = Type.Union([ServiceRpcCallSchema, ServerControlRpcCallSchema]); -const ProtocolRpcResultSchema = Type.Union([ServiceRpcResultSchema, ServerControlRpcResultSchema]); +export type ProtocolRpcCall = ServiceRpcCall; +export type ProtocolRpcResult = ServiceRpcResultUnion; +const ProtocolRpcCallSchema = ServiceRpcCallSchema; +const ProtocolRpcResultSchema = ServiceRpcResultSchema; export const ProtocolErrorCodeSchema = Type.Union([ Type.Literal("version"), diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 918afdaef..89c063084 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -17,7 +17,6 @@ import { ProtocolValidationError, parseClientMessage, parseServerMessage, - ServerControlRpc, type ServerHello, type ServerMessage, ServerMessageDecoder, @@ -79,22 +78,6 @@ describe("RPC manifest", () => { await expect(dispatch({ method: "list", args: [] }, undefined)).rejects.toThrow(/Invalid result.*list/); }); - test("keeps launcher control methods separate from session operations", async () => { - const calls: unknown[] = []; - const control = createRpcClient(ServerControlRpc, async (call) => { - calls.push(call); - return {}; - }); - const dispatch = createRpcDispatcher(ServerControlRpc, { - drain: () => ({}), - }); - - await expect(control.drain()).resolves.toEqual({}); - await expect(dispatch({ method: "drain", args: [] }, undefined)).resolves.toEqual({}); - expect(calls).toEqual([{ method: "drain", args: [] }]); - expect("drain" in createRpcClient(ServiceRpc, async () => [])).toBe(false); - }); - test("rejects empty manifests instead of creating unusable RPC clients", () => { expect(() => defineRpc({})).toThrow(/at least one method/); }); @@ -138,7 +121,7 @@ describe("protocol validation", () => { ).toThrow(ProtocolValidationError); }); - test("validates service and launcher-control RPC calls with logical targets", () => { + test("validates service RPC calls with logical targets", () => { const list: ClientMessage = { type: "request", id: "request-1", @@ -151,15 +134,8 @@ describe("protocol validation", () => { serverId: "00000000-0000-4000-8000-000000000001", call: { method: "attach", args: ["session-1"] }, }; - const drain: ClientMessage = { - type: "request", - id: "request-3", - serverId: "00000000-0000-4000-8000-000000000001", - call: { method: "drain", args: [] }, - }; expect(parseClientMessage(list)).toEqual(list); expect(parseClientMessage(attach)).toEqual(attach); - expect(parseClientMessage(drain)).toEqual(drain); expect(() => parseClientMessage({ ...attach, call: { method: "attach", args: [] } })).toThrow( ProtocolValidationError, ); @@ -191,17 +167,6 @@ describe("protocol validation", () => { expect(parseServerMessage(message)).toEqual(message); }); - test("validates drain acknowledgements", () => { - const message: ServerMessage = { - type: "response", - id: "request-1", - ok: true, - result: {}, - }; - expect(parseServerMessage(message)).toEqual(message); - expect(() => parseServerMessage({ ...message, result: { sessionIds: [] } })).toThrow(ProtocolValidationError); - }); - test.each([ [ "empty request id", diff --git a/packages/server/README.md b/packages/server/README.md index becbaf2f8..b4dae2008 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -7,8 +7,6 @@ The current slice supports two client operations: - `list` calls `SessionRepo.list()` without opening sessions. - `attach` finds the requested metadata, passes it to the host, and retains the returned Harness handle in the server. -The separate launcher-control operation `drain` closes hosted Harnesses, acknowledges successful cleanup, and closes that server generation. It does not transfer hosted-session state, and the server does not decide whether a replacement should start. Clients explicitly reconnect and reattach after replacement. - Concurrent attachments to one session reuse one hosted Harness. Attachment is a one-shot acquisition request, so losing the client connection does not close the Harness. Server shutdown closes every hosted Harness, releasing its Session writer ownership. ```ts @@ -60,4 +58,4 @@ Applications supply a session catalog and a Harness factory. The server only cal `PiServer` composes authenticated transports through `PiServerListener`. The Unix submodule provides `createUnixListener()` and `createUnixServer()`. Low-level CBOR framing and validation come from `@earendil-works/pi-protocol`. -`ServerControlRpc` is administrative but is not itself an authorization mechanism. A transport exposing it must trust every connected peer as an administrator or authenticate control access separately. The experimental local launcher enforces mode `0600` on its Unix socket and therefore trusts same-user peers. +Server and worker lifecycle is managed by the replaceable application server rather than the public Pi protocol. The experimental coordinator only supplies stable routing between processes. diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 539f2ed5a..155072386 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -13,7 +13,6 @@ import { ProtocolValidationError, type RequestEnvelope, type ResponseEnvelope, - type ServerControlRpcResult, type ServerHello, type ServerHelloError, type ServerMessage, @@ -24,7 +23,7 @@ import { type ConnectionState, isTerminalConnection, } from "./connection.ts"; -import { INTERNAL_SERVER_ERROR_MESSAGE, PiServerError, ServerDrainingError, WrongServerError } from "./errors.ts"; +import { INTERNAL_SERVER_ERROR_MESSAGE, PiServerError, WrongServerError } from "./errors.ts"; import { HostedHarnessManager } from "./hosted-harness-manager.ts"; import type { PiServerListener } from "./listener.ts"; import type { PiServerHost, PiServerOptions } from "./types.ts"; @@ -254,19 +253,9 @@ export class PiServer { } private async handleRequest(state: ConnectionState, envelope: RequestEnvelope): Promise { - const draining = envelope.call.method === "drain"; - let ownsDrain = false; try { if (envelope.serverId !== this.serverId) throw new WrongServerError(); - let result: ProtocolRpcResult; - if (draining) { - if (this.closing) throw new ServerDrainingError(); - this.closing = true; - ownsDrain = true; - result = await this.drain(); - } else { - result = await this.sessions.executeCall(envelope.call); - } + const result: ProtocolRpcResult = await this.sessions.executeCall(envelope.call); await this.sendMessage(state, { type: "response", id: envelope.id, @@ -280,20 +269,9 @@ export class PiServer { ok: false, error: this.toProtocolError(error), } satisfies ResponseEnvelope); - } finally { - if (ownsDrain) this.scheduleClose(); } } - private async drain(): Promise> { - await this.sessions.close(); - return {}; - } - - private scheduleClose(): void { - void this.close().catch((error: unknown) => this.reportError(error)); - } - private transportClosed(connection: ConnectionState): void { if (!connection.disconnected && connection.stage !== "closing") { try { diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index d8ec9d389..6ec24834b 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -245,108 +245,6 @@ describe("list and attach protocol", () => { }); }); -describe("server draining", () => { - test("drains hosted sessions, acknowledges, and closes", async () => { - const host = new TestServerHost(); - await host.seed("session-1"); - const server = createServer(host); - const client = connect(server); - await client.hello(); - await client.request("00000000-0000-4000-8000-000000000001", { - method: "attach", - args: ["session-1"], - }); - const harness = host.latestHarness("session-1"); - - await expect( - client.request("00000000-0000-4000-8000-000000000001", { method: "drain", args: [] }), - ).resolves.toMatchObject({ - ok: true, - result: {}, - }); - await client.waitForClose(); - await server.closed; - - expect(harness.closeCount).toBe(1); - }); - - test("rejects a drain addressed to another server", async () => { - const host = new TestServerHost(); - await host.seed("session-1"); - const server = createServer(host); - const client = connect(server); - await client.hello(); - await client.request("00000000-0000-4000-8000-000000000001", { - method: "attach", - args: ["session-1"], - }); - - await expect( - client.request("00000000-0000-4000-8000-000000000002", { method: "drain", args: [] }), - ).resolves.toMatchObject({ - ok: false, - error: { code: "wrong_server" }, - }); - - expect(host.latestHarness("session-1").closeCount).toBe(0); - }); - - test("only the drain owner schedules shutdown", async () => { - const host = new TestServerHost(); - await host.seed("session-1"); - const server = createServer(host); - const owner = connect(server); - const concurrent = connect(server); - await Promise.all([owner.hello(), concurrent.hello()]); - await owner.request("00000000-0000-4000-8000-000000000001", { - method: "attach", - args: ["session-1"], - }); - const gate = host.latestHarness("session-1").gateNextClose(); - - const draining = owner.request("00000000-0000-4000-8000-000000000001", { method: "drain", args: [] }); - await gate.entered.promise; - await expect( - concurrent.request("00000000-0000-4000-8000-000000000001", { method: "drain", args: [] }), - ).resolves.toMatchObject({ - ok: false, - error: { code: "server_draining" }, - }); - expect(owner.closed).toBe(false); - - gate.release.resolve(undefined); - await expect(draining).resolves.toMatchObject({ ok: true, result: {} }); - await server.closed; - }); - - test("rejects drain acknowledgement when any hosted Harness fails to close", async () => { - const host = new TestServerHost(); - await host.seed("session-1"); - await host.seed("session-2"); - const server = createServer(host); - const client = connect(server); - await client.hello(); - await client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"] }); - await client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-2"] }); - const failedHarness = host.latestHarness("session-1"); - const closedHarness = host.latestHarness("session-2"); - failedHarness.failClose = new Error("close failed"); - - await expect( - client.request("00000000-0000-4000-8000-000000000001", { method: "drain", args: [] }), - ).resolves.toMatchObject({ - ok: false, - error: { code: "internal_error" }, - }); - await expect(server.closed).rejects.toThrow(/Failed to close hosted Harnesses/); - expect(failedHarness.closeCount).toBe(1); - expect(closedHarness.closeCount).toBe(1); - - servers.delete(server); - await failedHarness.close(); - }); -}); - describe("hosted Harness acquisition failures", () => { test("shares a Harness creation failure, releases the Session, and allows a later retry", async () => { const host = new TestServerHost(); diff --git a/tsconfig.json b/tsconfig.json index 1ad0330cd..cc2f9a85f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -25,7 +25,6 @@ "@earendil-works/pi-protocol": ["./packages/protocol/src/index.ts"], "@earendil-works/pi-protocol/*": ["./packages/protocol/src/*"], "@earendil-works/pi-client": ["./packages/client/src/index.ts"], - "@earendil-works/pi-client/control": ["./packages/client/src/control.ts"], "@earendil-works/pi-client/unix": ["./packages/client/src/unix.ts"], "@earendil-works/pi-client/*": ["./packages/client/src/*"], "@earendil-works/pi-server": ["./packages/server/src/index.ts"], diff --git a/vitest.base.ts b/vitest.base.ts index 080b2b2a0..7592113ea 100644 --- a/vitest.base.ts +++ b/vitest.base.ts @@ -12,7 +12,6 @@ export const workspaceSourcePaths = { agentNode: fileURLToPath(new URL("./packages/agent/src/node.ts", import.meta.url)), protocolIndex: fileURLToPath(new URL("./packages/protocol/src/index.ts", import.meta.url)), clientIndex: fileURLToPath(new URL("./packages/client/src/index.ts", import.meta.url)), - clientControl: fileURLToPath(new URL("./packages/client/src/control.ts", import.meta.url)), clientUnix: fileURLToPath(new URL("./packages/client/src/unix.ts", import.meta.url)), serverIndex: fileURLToPath(new URL("./packages/server/src/index.ts", import.meta.url)), serverUnix: fileURLToPath(new URL("./packages/server/src/transports/unix/index.ts", import.meta.url)), @@ -36,7 +35,6 @@ export default defineConfig({ { find: /^@earendil-works\/pi-agent-core\/node$/, replacement: workspaceSourcePaths.agentNode }, { find: /^@earendil-works\/pi-protocol$/, replacement: workspaceSourcePaths.protocolIndex }, { find: /^@earendil-works\/pi-client$/, replacement: workspaceSourcePaths.clientIndex }, - { find: /^@earendil-works\/pi-client\/control$/, replacement: workspaceSourcePaths.clientControl }, { find: /^@earendil-works\/pi-client\/unix$/, replacement: workspaceSourcePaths.clientUnix }, { find: /^@earendil-works\/pi-server$/, replacement: workspaceSourcePaths.serverIndex }, { find: /^@earendil-works\/pi-server\/unix$/, replacement: workspaceSourcePaths.serverUnix },