sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(server): pass session metadata to harness hosts

Browse source
This commit is contained in:
Christian Klotz 2026-08-14 23:57:14 +03:00
parent ef93d9292e
commit 95d74e8c6b
No known key found for this signature in database
9 changed files with 193 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/client/test/unix.test.ts
View file

@ -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 },

12
packages/coding-agent/src/cli/experimental/runtime.ts
View file

@ -80,17 +80,9 @@ export async function startExperimentalMemoryServer(
const workerPids = new Map<string, number>();
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) => {

57
packages/server/README.md
View file

@ -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<HostedHarnessHandle>,
) {
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`.

41
packages/server/src/hosted-harness-manager.ts
View file

@ -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<TMetadata extends SessionMetadata> {
host: PiServerHost<TMetadata>;
isClosing: () => boolean;
reportError: (error: unknown) => void;
}
export class HostedHarnessManager {
private readonly options: HostedHarnessManagerOptions;
export class HostedHarnessManager<TMetadata extends SessionMetadata = SessionMetadata> {
private readonly options: HostedHarnessManagerOptions<TMetadata>;
private readonly hostedSessions = new Map<string, HostedSession>();
private readonly openingSessions = new Map<string, Promise<HostedSession>>();
private closePromise?: Promise<void>;
private readonly dispatchRpc: (call: ServiceRpcCall, context: undefined) => Promise<ServiceRpcResultUnion>;
constructor(options: HostedHarnessManagerOptions) {
constructor(options: HostedHarnessManagerOptions<TMetadata>) {
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<HostedSession> {
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();

7
packages/server/src/server.ts
View file

@ -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<TMetadata extends SessionMetadata = SessionMetadata> {
readonly serverId: string;
/** Resolves after shutdown, or rejects when listener or hosted-Harness cleanup fails. */
readonly closed: Promise<void>;
@ -42,7 +43,7 @@ export class PiServer {
private readonly handshakeTimeoutMs: number;
private readonly onError: ((error: Error) => void) | undefined;
private readonly connections = new Set<ConnectionState>();
private readonly sessions: HostedHarnessManager;
private readonly sessions: HostedHarnessManager<TMetadata>;
private closing = false;
private closePromise?: Promise<void>;
private closedSettled = false;
@ -51,7 +52,7 @@ export class PiServer {
private startPromise?: Promise<this>;
private started = false;
constructor(host: PiServerHost, options: PiServerOptions) {
constructor(host: PiServerHost<TMetadata>, options: PiServerOptions) {
const resolved = resolveOptions(options);
this.listeners = options.listeners;
this.serverId = options.serverId;

62
packages/server/src/testing/host.ts
View file

@ -73,9 +73,9 @@ interface ListDelay {
export class TestServerHost implements PiServerHost {
readonly repo = new MemorySessionRepo({ now: () => 1 });
readonly harnesses = new Map<string, TestHarness[]>();
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<HostedHarnessHandle> {
if (this.failNextHarness) {
const error = this.failNextHarness;
this.failNextHarness = undefined;
async createHarness(metadata: SessionMetadata): Promise<HostedHarnessHandle> {
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<SessionMetadata> {
@ -131,9 +133,9 @@ export class TestServerHost implements PiServerHost {
return delay;
}
gateNextOpen(): OpenGate {
gateNextCreateHarness(): OpenGate {
const gate = { entered: new Deferred<void>(), release: new Deferred<void>() };
this.nextOpenGate = gate;
this.nextCreateHarnessGate = gate;
return gate;
}

6
packages/server/src/transports/unix/preset.ts
View file

@ -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<TMetadata extends SessionMetadata>(
host: PiServerHost<TMetadata>,
options: UnixServerOptions,
): PiServer<TMetadata> {
const listener = createUnixListener({
path: options.path,
mode: options.mode,

8
packages/server/src/types.ts
View file

@ -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<AgentHarness, "close"> {
}
/** Host capabilities used directly by the list and attach control-plane operations. */
export interface PiServerHost {
readonly sessions: Pick<SessionRepo, "list" | "open">;
createHarness(session: Session): Promise<HostedHarnessHandle>;
export interface PiServerHost<TMetadata extends SessionMetadata = SessionMetadata> {
readonly sessions: Pick<SessionRepo<TMetadata>, "list">;
createHarness(metadata: TMetadata): Promise<HostedHarnessHandle>;
}

109
packages/server/test/conformance.test.ts
View file

@ -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<PiServer>();
@ -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<BackendMetadata> = {
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();
});
});