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(coding-agent): host durable sessions in workers

Browse source
This commit is contained in:
Christian Klotz 2026-08-15 01:01:12 +03:00
parent 95d74e8c6b
commit b35b4e0e48
No known key found for this signature in database
13 changed files with 348 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

9
packages/coding-agent/src/cli/experimental/demo-sessions.ts
View file

@ -1,9 +0,0 @@
// Prototype-only shared catalog. Remove this once parent and worker resolve
// sessions from the same durable repository.
export const DEMO_SESSION_IDS = ["demo-1", "demo-2"] as const;
export type DemoSessionId = (typeof DEMO_SESSION_IDS)[number];
export function isDemoSessionId(value: string): value is DemoSessionId {
return DEMO_SESSION_IDS.some((sessionId) => sessionId === value);
}

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

@ -3,7 +3,8 @@ 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 { MemorySessionRepo } from "@earendil-works/pi-agent-core";
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";
@ -13,7 +14,6 @@ import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/u
import { getAgentDir } from "../../config.ts";
import { resolvePath } from "../../utils/paths.ts";
import type { ClientCommand } from "./commands/client.ts";
import { DEMO_SESSION_IDS } from "./demo-sessions.ts";
import { acquireExperimentalServerProfile } from "./server-profile.ts";
import { startExperimentalSessionWorker } from "./session-worker.ts";
@ -21,7 +21,7 @@ const SOCKET_RELEASE_TIMEOUT_MS = 10_000;
const SOCKET_RELEASE_POLL_MS = 10;
const EXPERIMENTAL_SOCKET_ROOT = "/tmp";
export interface ExperimentalMemoryServer {
export interface ExperimentalServer {
readonly serverId: string;
readonly sessionDir: string;
readonly socketPath: string;
@ -38,7 +38,7 @@ export type ExperimentalClientResult =
}
| { readonly kind: "attached"; readonly serverId: string; readonly sessionId: string };
export interface StartExperimentalMemoryServerOptions {
export interface StartExperimentalServerOptions {
/** Directory for server-addressed Unix sockets. Defaults to a short, private per-user runtime directory. */
readonly directory?: string;
readonly path?: string;
@ -65,24 +65,21 @@ export function resolveExperimentalSessionDirectory(sessionDir?: string): string
return resolvePath(sessionDir ?? join(getAgentDir(), "experimental", "sessions"));
}
/** Start the temporary in-memory list-and-attach server composition. */
export async function startExperimentalMemoryServer(
options: StartExperimentalMemoryServerOptions = {},
): Promise<ExperimentalMemoryServer> {
/** Start the experimental durable list-and-attach server composition. */
export async function startExperimentalServer(
options: StartExperimentalServerOptions = {},
): Promise<ExperimentalServer> {
const serverId = options.serverId ?? randomUUID();
const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir);
const repo = new MemorySessionRepo();
for (const id of DEMO_SESSION_IDS) {
const session = await repo.create({ id });
await session.close();
}
const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() });
const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir });
const workerPids = new Map<string, number>();
const host: PiServerHost = {
const host: PiServerHost<JsonlSessionMetadata> = {
sessions: repo,
createHarness: async (metadata) => {
const sessionId = metadata.id;
const worker = await startExperimentalSessionWorker(sessionId, { sessionDir });
const worker = await startExperimentalSessionWorker(metadata, { sessionDir });
workerPids.set(sessionId, worker.pid);
return {
terminated: worker.terminated.then((error) => {
@ -105,11 +102,17 @@ export async function startExperimentalMemoryServer(
await ensurePrivateSocketDirectory(socketDirectory);
socketPath = getUnixSocketPath(serverId, socketDirectory);
}
const closeCatalog = async (): Promise<void> => {
const cleanup = await Promise.allSettled([repo.close(), executionEnv.cleanup()]);
const errors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Experimental session catalog cleanup failed");
};
const server = createUnixServer(host, { serverId, path: socketPath, mode: 0o600 });
try {
await server.start();
} catch (error) {
const cleanup = await Promise.allSettled([server.close(), repo.close()]);
const cleanup = await Promise.allSettled([server.close(), closeCatalog()]);
const cleanupErrors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
if (cleanupErrors.length > 0) {
throw new AggregateError([error, ...cleanupErrors], "Experimental server startup and cleanup failed");
@ -119,10 +122,10 @@ export async function startExperimentalMemoryServer(
let closePromise: Promise<void> | undefined;
const closed = server.closed.then(
() => repo.close(),
() => closeCatalog(),
async (serverError: unknown) => {
try {
await repo.close();
await closeCatalog();
} catch (repoError) {
throw new AggregateError([serverError, repoError], "Server and repository shutdown failed");
}
@ -150,16 +153,16 @@ export async function startExperimentalMemoryServer(
/** Start the experimental local server, replacing an existing generation for the same profile. */
export async function startExperimentalServerGeneration(
options: StartExperimentalServerGenerationOptions = {},
): Promise<ExperimentalMemoryServer> {
): Promise<ExperimentalServer> {
const directory = options.directory ?? join(homedir(), ".pi", "server");
const socketDirectory = options.socketDirectory ?? getExperimentalSocketDirectory();
const { serverId, release } = await acquireExperimentalServerProfile(directory);
let runtime: ExperimentalMemoryServer;
let runtime: ExperimentalServer;
try {
await ensurePrivateSocketDirectory(socketDirectory);
const socketPath = getUnixSocketPath(serverId, socketDirectory);
await drainExistingGeneration(serverId, socketPath);
runtime = await startExperimentalMemoryServer({
runtime = await startExperimentalServer({
directory,
path: socketPath,
serverId,
@ -207,7 +210,7 @@ export async function runExperimentalClient(
command: ClientCommand,
options: RunExperimentalClientOptions = {},
): Promise<ExperimentalClientResult> {
if (command.auth !== undefined) throw new Error("Authentication is not supported by the local demo server");
if (command.auth !== undefined) throw new Error("Authentication is not supported by the experimental local server");
const routes = command.connect
? [routeFromExplicitPath(command.connect.path)]
: await discoverUnixServers({ directory: options.directory ?? getExperimentalSocketDirectory() });

145
packages/coding-agent/src/cli/experimental/session-worker-process.ts
View file

@ -1,8 +1,15 @@
import { isAbsolute } from "node:path";
import { type AgentHarness, MemorySessionRepo } from "@earendil-works/pi-agent-core";
import { isDemoSessionId } from "./demo-sessions.ts";
import {
AgentHarness,
type AgentHarness as AgentHarnessInstance,
type JsonlSessionMetadata,
JsonlSessionRepo,
type Session,
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import { ModelRuntime } from "../../core/model-runtime.ts";
// Prototype-only control protocol. Real Harness operations will require a
// Prototype-only process control. Agent operations will use the shared
// transport-independent protocol rather than additional ad hoc IPC messages.
export type SessionWorkerCommand = { type: "shutdown" };
export type SessionWorkerEvent =
@ -22,38 +29,95 @@ function send(event: SessionWorkerEvent): Promise<void> {
});
}
async function run(): Promise<void> {
// Prototype-only catalog validation. A production worker will resolve the
// requested session through durable storage.
const sessionId = process.argv[2];
if (!sessionId || !isDemoSessionId(sessionId)) throw new Error(`Unknown demo session: ${sessionId ?? ""}`);
const sessionDir = process.argv[3];
if (!sessionDir || !isAbsolute(sessionDir)) throw new Error("Session worker requires an absolute session directory");
function parseMetadata(value: string | undefined): JsonlSessionMetadata {
if (value === undefined) throw new Error("Session worker requires session metadata");
let parsed: unknown;
try {
parsed = JSON.parse(value);
} catch (error) {
throw new Error("Session worker received invalid session metadata", { cause: error });
}
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
throw new Error("Session worker received invalid session metadata");
}
const candidate = parsed as Record<string, unknown>;
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;
}
// Prototype-only isolated state. The parent and child intentionally seed
// separate repositories; this does not provide persistence or shared state.
const repo = new MemorySessionRepo();
// Seed, close, list, and reopen to model restoring an existing Session
// instead of handing the freshly created facade directly to the Harness.
const created = await repo.create({ id: sessionId });
await created.close();
const metadata = (await repo.list()).find((candidate) => candidate.id === sessionId);
if (!metadata) throw new Error(`Unknown demo session: ${sessionId}`);
const session = await repo.open(metadata);
// Prototype-only close-capable Harness placeholder. A production worker will
// construct and own the real AgentHarness for this durable Session.
const harnessOwner: Pick<AgentHarness, "close"> = { close: () => session.close() };
async function closeResources(resources: {
harness?: AgentHarnessInstance;
session?: Session<JsonlSessionMetadata>;
repo: JsonlSessionRepo;
executionEnv: NodeExecutionEnv;
}): Promise<void> {
const errors: unknown[] = [];
try {
if (resources.harness) await resources.harness.close();
else await resources.session?.close();
} catch (error) {
errors.push(error);
}
try {
await resources.repo.close();
} catch (error) {
errors.push(error);
}
try {
await resources.executionEnv.cleanup();
} catch (error) {
errors.push(error);
}
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Session worker cleanup failed");
}
async function run(): Promise<void> {
const sessionDir = process.argv[2];
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 executionEnv = new NodeExecutionEnv({ cwd: metadata.cwd });
const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir });
let session: Session<JsonlSessionMetadata> | undefined;
let harness: AgentHarnessInstance | undefined;
try {
session = await repo.open(metadata);
const modelRuntime = await ModelRuntime.create();
const model = modelRuntime.getAvailableSnapshot()[0];
if (!model) throw new Error("Session worker could not find a configured model");
({ harness } = await AgentHarness.create({ session, models: modelRuntime, model, tools: [], resources: {} }));
} catch (error) {
try {
await closeResources({ harness, session, repo, executionEnv });
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Session worker startup and cleanup failed");
}
throw error;
}
let closing: Promise<void> | undefined;
const close = (): Promise<void> => {
if (!closing) {
closing = harnessOwner.close().finally(() => repo.close());
}
closing ??= closeResources({ harness, repo, executionEnv });
return closing;
};
process.on("message", (message: SessionWorkerCommand) => {
if (message?.type !== "shutdown") return;
const closeAndExit = (): void => {
void close().then(
() => process.exit(0),
(error: unknown) => {
@ -61,12 +125,25 @@ async function run(): Promise<void> {
process.exit(1);
},
);
});
process.once("disconnect", () => void close().finally(() => process.exit(0)));
process.once("SIGTERM", () => void close().finally(() => process.exit(0)));
process.once("SIGINT", () => void close().finally(() => process.exit(0)));
};
await send({ type: "ready", sessionId, pid: process.pid });
process.on("message", (message: SessionWorkerCommand) => {
if (message?.type === "shutdown") closeAndExit();
});
process.once("disconnect", closeAndExit);
process.once("SIGTERM", closeAndExit);
process.once("SIGINT", closeAndExit);
try {
await send({ type: "ready", sessionId, pid: process.pid });
} catch (error) {
try {
await close();
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Session worker readiness and cleanup failed");
}
throw error;
}
}
void run().catch(async (error: unknown) => {

6
packages/coding-agent/src/cli/experimental/session-worker.ts
View file

@ -1,5 +1,6 @@
import { type ChildProcess, fork } from "node:child_process";
import { isAbsolute } from "node:path";
import type { JsonlSessionMetadata } from "@earendil-works/pi-agent-core";
import type { SessionWorkerCommand, SessionWorkerEvent } from "./session-worker-process.ts";
const DEFAULT_STARTUP_TIMEOUT_MS = 5_000;
@ -21,17 +22,18 @@ export interface StartExperimentalSessionWorkerOptions {
}
export async function startExperimentalSessionWorker(
sessionId: string,
metadata: JsonlSessionMetadata,
options: StartExperimentalSessionWorkerOptions,
): Promise<ExperimentalSessionWorker> {
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, [sessionId, options.sessionDir], {
const child = fork(workerUrl, [options.sessionDir, JSON.stringify(metadata)], {
execArgv: workerExecArgv(workerUrl),
stdio: ["ignore", "ignore", "inherit", "ipc"],
});

50
packages/coding-agent/test/experimental-remote-runtime.test.ts
View file

@ -1,20 +1,30 @@
import { chmod, lstat, mkdtemp, rm } from "node:fs/promises";
import { join, resolve } from "node:path";
import { afterEach, describe, expect, test } from "vitest";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import {
type ExperimentalMemoryServer,
type ExperimentalServer,
runExperimentalClient,
startExperimentalMemoryServer,
startExperimentalServer,
startExperimentalServerGeneration,
} from "../src/cli/experimental/runtime.ts";
import { configureExperimentalWorkerModel, createExperimentalSessions } from "./experimental-session-support.ts";
const servers = new Set<ExperimentalMemoryServer>();
const servers = new Set<ExperimentalServer>();
const directories = new Set<string>();
let agentDir: string;
async function makeServer(): Promise<{ directory: string; runtime: ExperimentalMemoryServer }> {
beforeEach(async () => {
agentDir = await mkdtemp(join("/tmp", "pi-experimental-agent-"));
directories.add(agentDir);
await configureExperimentalWorkerModel(agentDir);
vi.stubEnv("PI_CODING_AGENT_DIR", agentDir);
await createExperimentalSessions(join(agentDir, "experimental", "sessions"), ["demo-1", "demo-2"]);
});
async function makeServer(): Promise<{ directory: string; runtime: ExperimentalServer }> {
const directory = await mkdtemp(join("/tmp", "pes-"));
directories.add(directory);
const runtime = await startExperimentalMemoryServer({ directory });
const runtime = await startExperimentalServer({ directory });
servers.add(runtime);
return { directory, runtime };
}
@ -22,17 +32,18 @@ async function makeServer(): Promise<{ directory: string; runtime: ExperimentalM
afterEach(async () => {
await Promise.all([...servers].map((server) => server.close()));
servers.clear();
vi.unstubAllEnvs();
await Promise.all([...directories].map((directory) => rm(directory, { recursive: true, force: true })));
directories.clear();
});
describe("experimental memory server composition", () => {
describe("experimental durable server composition", () => {
test("does not change permissions on an explicit socket-path parent", async () => {
const directory = await mkdtemp(join("/tmp", "pep-"));
directories.add(directory);
await chmod(directory, 0o750);
const serverId = "00000000-0000-4000-8000-000000000001";
const runtime = await startExperimentalMemoryServer({
const runtime = await startExperimentalServer({
path: join(directory, `${serverId}.sock`),
serverId,
});
@ -41,15 +52,14 @@ describe("experimental memory server composition", () => {
expect((await lstat(directory)).mode & 0o777).toBe(0o750);
});
test("resolves the configured session directory before starting workers", async () => {
test("resolves the configured session directory", async () => {
const directory = await mkdtemp(join("/tmp", "pes-"));
directories.add(directory);
const runtime = await startExperimentalMemoryServer({ directory, sessionDir: "relative/sessions" });
const runtime = await startExperimentalServer({ directory, sessionDir: "relative/sessions" });
servers.add(runtime);
expect(runtime.sessionDir).toBe(resolve("relative/sessions"));
await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory });
expect(runtime.workerPids.get("demo-1")).toEqual(expect.any(Number));
expect(runtime.workerPids.size).toBe(0);
});
test("discovers and lists seeded sessions without hosting either session", async () => {
@ -196,8 +206,8 @@ describe("experimental memory server composition", () => {
test("reports missing and ambiguous session selections", async () => {
const sharedDirectory = await mkdtemp(join("/tmp", "ped-"));
directories.add(sharedDirectory);
const firstShared = await startExperimentalMemoryServer({ directory: sharedDirectory });
const secondShared = await startExperimentalMemoryServer({ directory: sharedDirectory });
const firstShared = await startExperimentalServer({ directory: sharedDirectory });
const secondShared = await startExperimentalServer({ directory: sharedDirectory });
servers.add(firstShared);
servers.add(secondShared);
@ -208,6 +218,18 @@ describe("experimental memory server composition", () => {
runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory: sharedDirectory }),
).rejects.toThrow("Session demo-1 is available from more than one server");
});
test("rejects a duplicate session ID within one durable repository", async () => {
await createExperimentalSessions(
join(agentDir, "experimental", "sessions"),
["demo-1"],
join(agentDir, "other-cwd"),
);
const { directory } = await makeServer();
await expect(
runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }),
).rejects.toMatchObject({ code: "session_ambiguous" });
});
});
function processExists(pid: number): boolean {

32
packages/coding-agent/test/experimental-session-support.ts Normal file
View file

@ -0,0 +1,32 @@
import { mkdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
export async function createExperimentalSessions(
sessionsRoot: string,
ids: readonly string[],
cwd = process.cwd(),
): Promise<JsonlSessionMetadata[]> {
const fileSystem = new NodeExecutionEnv({ cwd });
const repo = new JsonlSessionRepo({ fileSystem, sessionsRoot });
const metadata: JsonlSessionMetadata[] = [];
try {
for (const id of ids) {
const session = await repo.create({ id, cwd });
metadata.push(session.metadata);
await session.close();
}
return metadata;
} finally {
await repo.close();
await fileSystem.cleanup();
}
}
export async function configureExperimentalWorkerModel(agentDir: string): Promise<void> {
await mkdir(agentDir, { recursive: true });
await writeFile(join(agentDir, "auth.json"), JSON.stringify({ anthropic: { type: "api_key", key: "test-key" } }), {
mode: 0o600,
});
}

102
packages/coding-agent/test/experimental-session-worker.test.ts
View file

@ -1,21 +1,44 @@
import { afterEach, describe, expect, test } from "vitest";
import { mkdtemp, rm } from "node:fs/promises";
import { join } from "node:path";
import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import { afterEach, describe, expect, test, vi } from "vitest";
import {
type ExperimentalSessionWorker,
startExperimentalSessionWorker,
} from "../src/cli/experimental/session-worker.ts";
import { configureExperimentalWorkerModel, createExperimentalSessions } from "./experimental-session-support.ts";
const workers = new Set<ExperimentalSessionWorker>();
const fixtureUrl = new URL("fixtures/session-worker-fixture.ts", import.meta.url);
const sessionDir = "/tmp/pi-session-worker-tests";
const directories = new Set<string>();
function testMetadata(id: string): JsonlSessionMetadata {
return {
id,
createdAt: 1,
storageVersion: 1,
cwd: "/tmp",
path: `/tmp/${id}.jsonl`,
modifiedAt: 1,
};
}
afterEach(async () => {
await Promise.all([...workers].map((worker) => worker.close()));
workers.clear();
vi.unstubAllEnvs();
await Promise.all([...directories].map((directory) => rm(directory, { recursive: true, force: true })));
directories.clear();
});
describe("experimental session worker controller", () => {
test("starts a real child, validates readiness, and closes idempotently", async () => {
const worker = await startExperimentalSessionWorker("ready", { sessionDir, workerUrl: fixtureUrl });
const worker = await startExperimentalSessionWorker(testMetadata("ready"), {
sessionDir,
workerUrl: fixtureUrl,
});
workers.add(worker);
expect(worker.sessionId).toBe("ready");
@ -27,15 +50,26 @@ describe("experimental session worker controller", () => {
await expect(worker.terminated).resolves.toBeUndefined();
});
test("passes session IDs containing null bytes through serialized metadata", async () => {
const worker = await startExperimentalSessionWorker(testMetadata("ready\0"), {
sessionDir,
workerUrl: fixtureUrl,
});
workers.add(worker);
expect(worker.sessionId).toBe("ready\0");
await worker.close();
});
test("rejects a relative session directory before starting a child", async () => {
await expect(
startExperimentalSessionWorker("ready", { sessionDir: "relative", workerUrl: fixtureUrl }),
startExperimentalSessionWorker(testMetadata("ready"), { sessionDir: "relative", workerUrl: fixtureUrl }),
).rejects.toThrow("sessionDir must be absolute");
});
test("times out and kills a child that never reports readiness", async () => {
await expect(
startExperimentalSessionWorker("startup-hang", {
startExperimentalSessionWorker(testMetadata("startup-hang"), {
sessionDir,
workerUrl: fixtureUrl,
startupTimeoutMs: 10,
@ -48,13 +82,13 @@ describe("experimental session worker controller", () => {
["exit before readiness", "exit", /exited before readiness/],
["invalid ready identity", "mismatch", /invalid identity/],
] as const)("rejects %s", async (_label, sessionId, message) => {
await expect(startExperimentalSessionWorker(sessionId, { sessionDir, workerUrl: fixtureUrl })).rejects.toThrow(
message,
);
await expect(
startExperimentalSessionWorker(testMetadata(sessionId), { sessionDir, workerUrl: fixtureUrl }),
).rejects.toThrow(message);
});
test("kills a worker that ignores graceful shutdown", async () => {
const worker = await startExperimentalSessionWorker("hang", {
const worker = await startExperimentalSessionWorker(testMetadata("hang"), {
sessionDir,
workerUrl: fixtureUrl,
shutdownTimeoutMs: 10,
@ -66,7 +100,10 @@ describe("experimental session worker controller", () => {
});
test("reports an unexpected exit after readiness", async () => {
const worker = await startExperimentalSessionWorker("ready", { sessionDir, workerUrl: fixtureUrl });
const worker = await startExperimentalSessionWorker(testMetadata("ready"), {
sessionDir,
workerUrl: fixtureUrl,
});
workers.add(worker);
process.kill(worker.pid, "SIGKILL");
@ -75,6 +112,53 @@ describe("experimental session worker controller", () => {
);
expect(processExists(worker.pid)).toBe(false);
});
test("fails clearly when no model is configured", async () => {
const root = await mkdtemp(join("/tmp", "pi-session-worker-no-model-"));
directories.add(root);
const durableSessionDir = join(root, "sessions");
const [metadata] = await createExperimentalSessions(durableSessionDir, ["no-model"]);
const agentDir = join(root, "agent");
vi.stubEnv("PI_CODING_AGENT_DIR", agentDir);
await expect(startExperimentalSessionWorker(metadata!, { sessionDir: durableSessionDir })).rejects.toThrow(
"could not find a configured model",
);
});
test("opens a durable session and creates a real Harness before readiness", async () => {
const root = await mkdtemp(join("/tmp", "pi-session-worker-real-"));
directories.add(root);
const durableSessionDir = join(root, "sessions");
const [metadata] = await createExperimentalSessions(durableSessionDir, ["durable-session"]);
const agentDir = join(root, "agent");
await configureExperimentalWorkerModel(agentDir);
vi.stubEnv("PI_CODING_AGENT_DIR", agentDir);
const worker = await startExperimentalSessionWorker(metadata!, { sessionDir: durableSessionDir });
workers.add(worker);
expect(worker.sessionId).toBe("durable-session");
expect(worker.pid).not.toBe(process.pid);
expect(processExists(worker.pid)).toBe(true);
await worker.close();
await expect(worker.terminated).resolves.toBeUndefined();
const executionEnv = new NodeExecutionEnv({ cwd: metadata!.cwd });
const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: durableSessionDir });
const reopened = await repo.open(metadata!);
await expect(reopened.getRegister("lane.config", "main")).resolves.toMatchObject({
value: {
model: { provider: "anthropic" },
thinkingLevel: "off",
activeToolNames: [],
},
});
await reopened.close();
await repo.close();
await executionEnv.cleanup();
});
});
function processExists(pid: number): boolean {

1
packages/protocol/src/schemas.ts
View file

@ -93,6 +93,7 @@ export const ProtocolErrorCodeSchema = Type.Union([
Type.Literal("version"),
Type.Literal("wrong_server"),
Type.Literal("session_not_found"),
Type.Literal("session_ambiguous"),
Type.Literal("server_draining"),
Type.Literal("invalid_request"),
Type.Literal("internal_error"),

12
packages/server/src/errors.ts
View file

@ -1,6 +1,9 @@
import type { ProtocolErrorCode } from "@earendil-works/pi-protocol";
type PiServerOperationErrorCode = Extract<ProtocolErrorCode, "wrong_server" | "session_not_found" | "server_draining">;
type PiServerOperationErrorCode = Extract<
ProtocolErrorCode,
"wrong_server" | "session_not_found" | "session_ambiguous" | "server_draining"
>;
export const INTERNAL_SERVER_ERROR_MESSAGE = "Internal server error";
@ -29,6 +32,13 @@ export class SessionNotFoundError extends PiServerError {
}
}
export class SessionAmbiguousError extends PiServerError {
constructor() {
super("session_ambiguous", "Session ID matches more than one session");
this.name = "SessionAmbiguousError";
}
}
export class ServerDrainingError extends PiServerError {
constructor() {
super("server_draining", "Server is draining");

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

@ -7,7 +7,7 @@ import {
type ServiceRpcCall,
type ServiceRpcResultUnion,
} from "@earendil-works/pi-protocol";
import { ServerDrainingError, SessionNotFoundError } from "./errors.ts";
import { ServerDrainingError, SessionAmbiguousError, SessionNotFoundError } from "./errors.ts";
import type { HostedHarnessHandle, PiServerHost } from "./types.ts";
class HarnessCleanupError extends AggregateError {}
@ -106,8 +106,10 @@ export class HostedHarnessManager<TMetadata extends SessionMetadata = SessionMet
}
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 matches = (await this.options.host.sessions.list()).filter((candidate) => candidate.id === sessionId);
if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
if (matches.length > 1) throw new SessionAmbiguousError();
const metadata = matches[0]!;
const harness = await this.options.host.createHarness(metadata);
if (this.options.isClosing()) {
try {

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

@ -1,4 +1,4 @@
import type { AgentHarness, SessionMetadata, SessionRepo } from "@earendil-works/pi-agent-core";
import type { AgentHarness, SessionMetadata } from "@earendil-works/pi-agent-core";
import type { PiServerListener } from "./listener.ts";
export interface PiServerOptions {
@ -20,6 +20,6 @@ export interface HostedHarnessHandle extends Pick<AgentHarness, "close"> {
/** Host capabilities used directly by the list and attach control-plane operations. */
export interface PiServerHost<TMetadata extends SessionMetadata = SessionMetadata> {
readonly sessions: Pick<SessionRepo<TMetadata>, "list">;
readonly sessions: { list(): Promise<TMetadata[]> };
createHarness(metadata: TMetadata): Promise<HostedHarnessHandle>;
}

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

@ -185,6 +185,32 @@ describe("list and attach protocol", () => {
expect(host.harnesses.size).toBe(0);
});
test("rejects an ambiguous session ID without creating a Harness", async () => {
type BackendMetadata = SessionMetadata & { path: string };
const host: PiServerHost<BackendMetadata> = {
sessions: {
list: async () => [
{ id: "duplicate", createdAt: 1, storageVersion: 1, cwd: "/one", path: "/one/session.jsonl" },
{ id: "duplicate", createdAt: 2, storageVersion: 1, cwd: "/two", path: "/two/session.jsonl" },
],
},
createHarness: async () => {
throw new Error("must not create a Harness for an ambiguous session");
},
};
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: "attach", args: ["duplicate"] }),
).resolves.toMatchObject({ ok: false, error: { code: "session_ambiguous" } });
});
test("invalidates a terminated Harness handle and allows a later attach", async () => {
const host = new TestServerHost();
await host.seed("session-1");

2
vitest.base.ts
View file

@ -9,6 +9,7 @@ export const workspaceSourcePaths = {
aiOAuth: fileURLToPath(new URL("./packages/ai/src/oauth.ts", import.meta.url)),
aiProviders: fileURLToPath(new URL("./packages/ai/src/providers", import.meta.url)),
agentIndex: fileURLToPath(new URL("./packages/agent/src/index.ts", import.meta.url)),
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)),
@ -32,6 +33,7 @@ export default defineConfig({
replacement: `${workspaceSourcePaths.aiProviders}/$1.ts`,
},
{ find: /^@earendil-works\/pi-agent-core$/, replacement: workspaceSourcePaths.agentIndex },
{ 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 },