diff --git a/packages/client/test/unix.test.ts b/packages/client/test/unix.test.ts index 231c8c83a..b2f2228f3 100644 --- a/packages/client/test/unix.test.ts +++ b/packages/client/test/unix.test.ts @@ -32,7 +32,7 @@ async function startServer( const path = join(directory, `${fileServerId}.sock`); const server = new PiServer( { - sessions: { list: async () => [], open: async () => Promise.reject(new Error("unused")) }, + sessions: { list: async () => [] }, createHarness: async () => Promise.reject(new Error("unused")), }, { listeners: [createUnixListener({ path })], serverId: reportedServerId }, diff --git a/packages/coding-agent/src/cli/experimental/runtime.ts b/packages/coding-agent/src/cli/experimental/runtime.ts index ed034e301..ad45a18df 100644 --- a/packages/coding-agent/src/cli/experimental/runtime.ts +++ b/packages/coding-agent/src/cli/experimental/runtime.ts @@ -80,17 +80,9 @@ export async function startExperimentalMemoryServer( const workerPids = new Map(); const host: PiServerHost = { sessions: repo, - createHarness: async (session) => { - const sessionId = session.metadata.id; + createHarness: async (metadata) => { + const sessionId = metadata.id; const worker = await startExperimentalSessionWorker(sessionId, { sessionDir }); - try { - // Prototype-only adapter: the server opened this parent-repository - // facade, while the child owns its independently restored Session. - await session.close(); - } catch (error) { - await worker.close(); - throw error; - } workerPids.set(sessionId, worker.pid); return { terminated: worker.terminated.then((error) => { diff --git a/packages/server/README.md b/packages/server/README.md index 5ef459127..becbaf2f8 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -5,7 +5,7 @@ Experimental local server for the new durable Session and Agent Harness interfac The current slice supports two client operations: - `list` calls `SessionRepo.list()` without opening sessions. -- `attach` finds the requested metadata, calls `SessionRepo.open()`, creates an `AgentHarness`, and retains that Harness in the server. +- `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. @@ -13,27 +13,50 @@ Concurrent attachments to one session reuse one hosted Harness. Attachment is a ```ts import { randomUUID } from "node:crypto"; -import { MemorySessionRepo } from "@earendil-works/pi-agent-core"; -import type { PiServerHost } from "@earendil-works/pi-server"; +import { MemorySessionRepo, type Session } from "@earendil-works/pi-agent-core"; +import { + type HostedHarnessHandle, + type PiServerHost, +} from "@earendil-works/pi-server"; import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix"; -const sessions = new MemorySessionRepo(); -const host: PiServerHost = { - sessions, - async createHarness(session) { - return createApplicationHarness({ session }); - }, -}; +async function startServer( + createHarnessForSession: (session: Session) => Promise, +) { + const sessions = new MemorySessionRepo(); + const host: PiServerHost = { + sessions, + async createHarness(metadata) { + const session = await sessions.open(metadata); + try { + return await createHarnessForSession(session); + } catch (error) { + try { + await session.close(); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + "Harness creation and Session cleanup failed", + ); + } + throw error; + } + }, + }; -const serverId = randomUUID(); -const server = createUnixServer(host, { - serverId, - path: getUnixSocketPath(serverId, "/run/user/1000/pi"), -}); -await server.start(); + const serverId = randomUUID(); + const server = createUnixServer(host, { + serverId, + path: getUnixSocketPath(serverId, "/run/user/1000/pi"), + }); + await server.start(); + return server; +} ``` -Applications supply the repository and Harness factory. `serverId` is a logical identity supplied by the launcher, not a socket address. The Unix preset requires an explicit physical `path`; `getUnixSocketPath()` derives one from a caller-selected directory. Choose a short, private runtime directory rather than deriving the route from an unbounded home-directory path. A long-lived launcher can reuse the same ID and path when replacing a server process. +Applications supply a session catalog and a Harness factory. The server only calls `list()`; the host receives the repository's concrete metadata and owns opening the Session, creating the Harness, and cleaning up failed creation. This permits the host to perform those operations in a worker process without passing an open JavaScript Session across processes. + +`serverId` is a logical identity supplied by the launcher, not a socket address. The Unix preset requires an explicit physical `path`; `getUnixSocketPath()` derives one from a caller-selected directory. Choose a short, private runtime directory rather than deriving the route from an unbounded home-directory path. A long-lived launcher can reuse the same ID and path when replacing a server process. `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`. diff --git a/packages/server/src/hosted-harness-manager.ts b/packages/server/src/hosted-harness-manager.ts index 1645a044f..46c78f173 100644 --- a/packages/server/src/hosted-harness-manager.ts +++ b/packages/server/src/hosted-harness-manager.ts @@ -1,5 +1,7 @@ +import type { SessionMetadata } from "@earendil-works/pi-agent-core"; import { createRpcDispatcher, + type SessionMetadata as ProtocolSessionMetadata, ProtocolValidationError, ServiceRpc, type ServiceRpcCall, @@ -10,30 +12,43 @@ import type { HostedHarnessHandle, PiServerHost } from "./types.ts"; class HarnessCleanupError extends AggregateError {} +function toProtocolSessionMetadata(metadata: SessionMetadata): ProtocolSessionMetadata { + return { + id: metadata.id, + createdAt: metadata.createdAt, + storageVersion: metadata.storageVersion, + ...(metadata.cwd === undefined ? {} : { cwd: metadata.cwd }), + ...(metadata.parentSessionId === undefined ? {} : { parentSessionId: metadata.parentSessionId }), + ...(metadata.legacyParentSessionPath === undefined + ? {} + : { legacyParentSessionPath: metadata.legacyParentSessionPath }), + }; +} + interface HostedSession { readonly id: string; readonly harness: HostedHarnessHandle; } -interface HostedHarnessManagerOptions { - host: PiServerHost; +interface HostedHarnessManagerOptions { + host: PiServerHost; isClosing: () => boolean; reportError: (error: unknown) => void; } -export class HostedHarnessManager { - private readonly options: HostedHarnessManagerOptions; +export class HostedHarnessManager { + private readonly options: HostedHarnessManagerOptions; private readonly hostedSessions = new Map(); private readonly openingSessions = new Map>(); private closePromise?: Promise; private readonly dispatchRpc: (call: ServiceRpcCall, context: undefined) => Promise; - constructor(options: HostedHarnessManagerOptions) { + constructor(options: HostedHarnessManagerOptions) { this.options = options; this.dispatchRpc = createRpcDispatcher( ServiceRpc, { - list: () => this.options.host.sessions.list(), + list: async () => (await this.options.host.sessions.list()).map(toProtocolSessionMetadata), attach: async (_context, sessionId) => { if (this.options.isClosing()) throw new ServerDrainingError(); const hosted = await this.acquire(sessionId); @@ -93,19 +108,7 @@ export class HostedHarnessManager { private async open(sessionId: string): Promise { const metadata = (await this.options.host.sessions.list()).find((candidate) => candidate.id === sessionId); if (!metadata) throw new SessionNotFoundError(`Unknown session: ${sessionId}`); - const session = await this.options.host.sessions.open(metadata); - let harness: HostedHarnessHandle; - try { - harness = await this.options.host.createHarness(session); - } catch (error) { - try { - await session.close(); - } catch (closeError) { - this.options.reportError(closeError); - throw new HarnessCleanupError([error, closeError], "Harness creation and Session cleanup failed"); - } - throw error; - } + const harness = await this.options.host.createHarness(metadata); if (this.options.isClosing()) { try { await harness.close(); diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 328ea56a4..539f2ed5a 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -1,3 +1,4 @@ +import type { SessionMetadata } from "@earendil-works/pi-agent-core"; import { type ClientHello, type ClientMessage, @@ -32,7 +33,7 @@ const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000; const MAX_UINT32 = 0xffff_ffff; const MAX_TIMER_DELAY_MS = 2_147_483_647; -export class PiServer { +export class PiServer { readonly serverId: string; /** Resolves after shutdown, or rejects when listener or hosted-Harness cleanup fails. */ readonly closed: Promise; @@ -42,7 +43,7 @@ export class PiServer { private readonly handshakeTimeoutMs: number; private readonly onError: ((error: Error) => void) | undefined; private readonly connections = new Set(); - private readonly sessions: HostedHarnessManager; + private readonly sessions: HostedHarnessManager; private closing = false; private closePromise?: Promise; private closedSettled = false; @@ -51,7 +52,7 @@ export class PiServer { private startPromise?: Promise; private started = false; - constructor(host: PiServerHost, options: PiServerOptions) { + constructor(host: PiServerHost, options: PiServerOptions) { const resolved = resolveOptions(options); this.listeners = options.listeners; this.serverId = options.serverId; diff --git a/packages/server/src/testing/host.ts b/packages/server/src/testing/host.ts index e58ee93f9..9bafdf7e9 100644 --- a/packages/server/src/testing/host.ts +++ b/packages/server/src/testing/host.ts @@ -73,9 +73,9 @@ interface ListDelay { export class TestServerHost implements PiServerHost { readonly repo = new MemorySessionRepo({ now: () => 1 }); readonly harnesses = new Map(); - openCount = 0; - failNextOpen?: Error; - failNextHarness?: Error; + createHarnessCount = 0; + nextCreateHarnessError?: Error; + nextHarnessCloseError?: Error; readonly sessions: PiServerHost["sessions"] = { list: async () => { const delay = this.nextListDelay; @@ -86,36 +86,38 @@ export class TestServerHost implements PiServerHost { } return this.repo.list(); }, - open: async (metadata) => { - this.openCount += 1; - const gate = this.nextOpenGate; - if (gate) { - this.nextOpenGate = undefined; - gate.entered.resolve(undefined); - await gate.release.promise; - } - if (this.failNextOpen) { - const error = this.failNextOpen; - this.failNextOpen = undefined; - throw error; - } - return this.repo.open(metadata); - }, }; private nextListDelay?: ListDelay; - private nextOpenGate?: OpenGate; + private nextCreateHarnessGate?: OpenGate; - async createHarness(session: Session): Promise { - if (this.failNextHarness) { - const error = this.failNextHarness; - this.failNextHarness = undefined; + async createHarness(metadata: SessionMetadata): Promise { + this.createHarnessCount += 1; + const gate = this.nextCreateHarnessGate; + if (gate) { + this.nextCreateHarnessGate = undefined; + gate.entered.resolve(undefined); + await gate.release.promise; + } + const session = await this.repo.open(metadata); + try { + if (this.nextCreateHarnessError) { + const error = this.nextCreateHarnessError; + this.nextCreateHarnessError = undefined; + throw error; + } + const harness = new TestHarness(session); + if (this.nextHarnessCloseError) { + harness.failClose = this.nextHarnessCloseError; + this.nextHarnessCloseError = undefined; + } + const harnesses = this.harnesses.get(metadata.id) ?? []; + harnesses.push(harness); + this.harnesses.set(metadata.id, harnesses); + return harness; + } catch (error) { + await session.close(); throw error; } - const harness = new TestHarness(session); - const harnesses = this.harnesses.get(session.metadata.id) ?? []; - harnesses.push(harness); - this.harnesses.set(session.metadata.id, harnesses); - return harness; } async seed(id = "session-1", parentSessionId?: string): Promise { @@ -131,9 +133,9 @@ export class TestServerHost implements PiServerHost { return delay; } - gateNextOpen(): OpenGate { + gateNextCreateHarness(): OpenGate { const gate = { entered: new Deferred(), release: new Deferred() }; - this.nextOpenGate = gate; + this.nextCreateHarnessGate = gate; return gate; } diff --git a/packages/server/src/transports/unix/preset.ts b/packages/server/src/transports/unix/preset.ts index 4904bc584..f2a5cfdcb 100644 --- a/packages/server/src/transports/unix/preset.ts +++ b/packages/server/src/transports/unix/preset.ts @@ -1,10 +1,14 @@ +import type { SessionMetadata } from "@earendil-works/pi-agent-core"; import { PiServer } from "../../server.ts"; import type { PiServerHost } from "../../types.ts"; import { createUnixListener } from "./listener.ts"; import type { UnixServerOptions } from "./types.ts"; /** Compose PiServer with one Unix-domain socket listener. */ -export function createUnixServer(host: PiServerHost, options: UnixServerOptions): PiServer { +export function createUnixServer( + host: PiServerHost, + options: UnixServerOptions, +): PiServer { const listener = createUnixListener({ path: options.path, mode: options.mode, diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index b21f6dc13..c4852dc4f 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -1,4 +1,4 @@ -import type { AgentHarness, Session, SessionRepo } from "@earendil-works/pi-agent-core"; +import type { AgentHarness, SessionMetadata, SessionRepo } from "@earendil-works/pi-agent-core"; import type { PiServerListener } from "./listener.ts"; export interface PiServerOptions { @@ -19,7 +19,7 @@ export interface HostedHarnessHandle extends Pick { } /** Host capabilities used directly by the list and attach control-plane operations. */ -export interface PiServerHost { - readonly sessions: Pick; - createHarness(session: Session): Promise; +export interface PiServerHost { + readonly sessions: Pick, "list">; + createHarness(metadata: TMetadata): Promise; } diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index 4dc517855..aa8ceb703 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -1,7 +1,9 @@ +import type { SessionMetadata } from "@earendil-works/pi-agent-core"; import { afterEach, describe, expect, test } from "vitest"; import type { ByteConnection, ByteConnectionHandler } from "../src/connection.ts"; import { PiServer } from "../src/server.ts"; import { ProtocolTestClient, TestServerHost, type WireChannel } from "../src/testing/index.ts"; +import type { PiServerHost } from "../src/types.ts"; const servers = new Set(); @@ -80,7 +82,54 @@ describe("list and attach protocol", () => { expect(host.harnesses.size).toBe(0); }); - test("attach opens the Session and creates one hosted Harness", async () => { + test("attach passes concrete repository metadata to the Harness host", async () => { + type BackendMetadata = SessionMetadata & { path: string; modifiedAt: number }; + const metadata: BackendMetadata = { + id: "session-1", + createdAt: 1, + storageVersion: 1, + cwd: "/workspace", + path: "/sessions/session-1.jsonl", + modifiedAt: 2, + }; + let received: BackendMetadata | undefined; + const host: PiServerHost = { + sessions: { list: async () => [metadata] }, + createHarness: async (candidate) => { + received = candidate; + return { close: async () => {} }; + }, + }; + const server = new PiServer(host, { + listeners: [], + serverId: "00000000-0000-4000-8000-000000000001", + }); + servers.add(server); + const client = connect(server); + await client.hello(); + + await expect( + client.request("00000000-0000-4000-8000-000000000001", { method: "list", args: [] }), + ).resolves.toEqual({ + type: "response", + id: "request-1", + ok: true, + result: [ + { + id: "session-1", + createdAt: 1, + storageVersion: 1, + cwd: "/workspace", + }, + ], + }); + await expect( + client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"] }), + ).resolves.toMatchObject({ ok: true, result: { sessionId: "session-1" } }); + expect(received).toBe(metadata); + }); + + test("attach creates one hosted Harness for concurrent callers", async () => { const host = new TestServerHost(); await host.seed("session-1"); const server = createServer(host); @@ -273,15 +322,15 @@ describe("server draining", () => { }); describe("hosted Harness acquisition failures", () => { - test("shares an opening failure and allows a later retry", async () => { + test("shares a Harness creation failure, releases the Session, and allows a later retry", async () => { const host = new TestServerHost(); await host.seed("session-1"); - host.failNextOpen = new Error("open failed"); + host.nextCreateHarnessError = new Error("Harness creation failed"); const server = createServer(host); const first = connect(server); const second = connect(server); await Promise.all([first.hello(), second.hello()]); - const gate = host.gateNextOpen(); + const gate = host.gateNextCreateHarness(); const firstAttach = first.request("00000000-0000-4000-8000-000000000001", { method: "attach", @@ -298,31 +347,12 @@ describe("hosted Harness acquisition failures", () => { { ok: false, error: { code: "internal_error" } }, { ok: false, error: { code: "internal_error" } }, ]); - expect(host.openCount).toBe(1); + expect(host.createHarnessCount).toBe(1); await expect( first.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"] }), ).resolves.toMatchObject({ ok: true, result: { sessionId: "session-1" } }); - expect(host.openCount).toBe(2); - expect(host.harnesses.get("session-1")).toHaveLength(1); - }); - - test("releases the opened Session when Harness creation fails", async () => { - const host = new TestServerHost(); - await host.seed("session-1"); - host.failNextHarness = new Error("harness failed"); - const server = createServer(host); - const client = connect(server); - await client.hello(); - - await expect( - client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"] }), - ).resolves.toMatchObject({ ok: false, error: { code: "internal_error" } }); - - await expect( - client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"] }), - ).resolves.toMatchObject({ ok: true, result: { sessionId: "session-1" } }); - expect(host.openCount).toBe(2); + expect(host.createHarnessCount).toBe(2); expect(host.harnesses.get("session-1")).toHaveLength(1); }); @@ -332,7 +362,7 @@ describe("hosted Harness acquisition failures", () => { const server = createServer(host); const client = connect(server); await client.hello(); - const gate = host.gateNextOpen(); + const gate = host.gateNextCreateHarness(); const attach = client.request("00000000-0000-4000-8000-000000000001", { method: "attach", args: ["session-1"], @@ -345,4 +375,31 @@ describe("hosted Harness acquisition failures", () => { await expect(attach).rejects.toThrow(/closed/i); expect(host.latestHarness("session-1").closeCount).toBe(1); }); + + test("fails shutdown when an in-flight acquisition cannot release its Harness", async () => { + const host = new TestServerHost(); + await host.seed("session-1"); + const cleanupError = new Error("close failed"); + host.nextHarnessCloseError = cleanupError; + const server = createServer(host); + const client = connect(server); + await client.hello(); + const gate = host.gateNextCreateHarness(); + const attach = client.request("00000000-0000-4000-8000-000000000001", { + method: "attach", + args: ["session-1"], + }); + await gate.entered.promise; + + const closing = server.close(); + gate.release.resolve(undefined); + + await expect(closing).rejects.toThrow(/Failed to close hosted Harnesses/); + await expect(server.closed).rejects.toThrow(/Failed to close hosted Harnesses/); + await expect(attach).rejects.toThrow(/closed/i); + expect(host.latestHarness("session-1").closeCount).toBe(1); + + servers.delete(server); + await host.latestHarness("session-1").close(); + }); });