From b35b4e0e489a96a8a83b1d533ee9bf6235749814 Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Sat, 15 Aug 2026 01:01:12 +0300 Subject: [PATCH] feat(coding-agent): host durable sessions in workers --- .../src/cli/experimental/demo-sessions.ts | 9 -- .../src/cli/experimental/runtime.ts | 47 +++--- .../experimental/session-worker-process.ts | 145 ++++++++++++++---- .../src/cli/experimental/session-worker.ts | 6 +- .../test/experimental-remote-runtime.test.ts | 50 ++++-- .../test/experimental-session-support.ts | 32 ++++ .../test/experimental-session-worker.test.ts | 102 ++++++++++-- packages/protocol/src/schemas.ts | 1 + packages/server/src/errors.ts | 12 +- packages/server/src/hosted-harness-manager.ts | 8 +- packages/server/src/types.ts | 4 +- packages/server/test/conformance.test.ts | 26 ++++ vitest.base.ts | 2 + 13 files changed, 348 insertions(+), 96 deletions(-) delete mode 100644 packages/coding-agent/src/cli/experimental/demo-sessions.ts create mode 100644 packages/coding-agent/test/experimental-session-support.ts diff --git a/packages/coding-agent/src/cli/experimental/demo-sessions.ts b/packages/coding-agent/src/cli/experimental/demo-sessions.ts deleted file mode 100644 index 78bd37d46..000000000 --- a/packages/coding-agent/src/cli/experimental/demo-sessions.ts +++ /dev/null @@ -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); -} diff --git a/packages/coding-agent/src/cli/experimental/runtime.ts b/packages/coding-agent/src/cli/experimental/runtime.ts index ad45a18df..297a50487 100644 --- a/packages/coding-agent/src/cli/experimental/runtime.ts +++ b/packages/coding-agent/src/cli/experimental/runtime.ts @@ -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 { +/** 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 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(); - const host: PiServerHost = { + const host: PiServerHost = { 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 => { + 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 | 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 { +): Promise { 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 { - 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() }); 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 8e7a25a7d..e5fba0d14 100644 --- a/packages/coding-agent/src/cli/experimental/session-worker-process.ts +++ b/packages/coding-agent/src/cli/experimental/session-worker-process.ts @@ -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 { }); } -async function run(): Promise { - // 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; + 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 = { close: () => session.close() }; +async function closeResources(resources: { + harness?: AgentHarnessInstance; + session?: Session; + repo: JsonlSessionRepo; + executionEnv: NodeExecutionEnv; +}): Promise { + 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 { + 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 | 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 | undefined; const close = (): Promise => { - 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 { 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) => { diff --git a/packages/coding-agent/src/cli/experimental/session-worker.ts b/packages/coding-agent/src/cli/experimental/session-worker.ts index 8765e981c..ceee791b9 100644 --- a/packages/coding-agent/src/cli/experimental/session-worker.ts +++ b/packages/coding-agent/src/cli/experimental/session-worker.ts @@ -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 { 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"], }); diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index 063da3f7a..edd6ac501 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -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(); +const servers = new Set(); const directories = new Set(); +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 { diff --git a/packages/coding-agent/test/experimental-session-support.ts b/packages/coding-agent/test/experimental-session-support.ts new file mode 100644 index 000000000..25d5b027d --- /dev/null +++ b/packages/coding-agent/test/experimental-session-support.ts @@ -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 { + 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 { + await mkdir(agentDir, { recursive: true }); + await writeFile(join(agentDir, "auth.json"), JSON.stringify({ anthropic: { type: "api_key", key: "test-key" } }), { + mode: 0o600, + }); +} diff --git a/packages/coding-agent/test/experimental-session-worker.test.ts b/packages/coding-agent/test/experimental-session-worker.test.ts index cf6c1bb78..da1e3f4c4 100644 --- a/packages/coding-agent/test/experimental-session-worker.test.ts +++ b/packages/coding-agent/test/experimental-session-worker.test.ts @@ -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(); const fixtureUrl = new URL("fixtures/session-worker-fixture.ts", import.meta.url); const sessionDir = "/tmp/pi-session-worker-tests"; +const directories = new Set(); + +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 { diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index 81529dfd5..85e5d3eae 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -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"), diff --git a/packages/server/src/errors.ts b/packages/server/src/errors.ts index 330a17e1c..ccd240163 100644 --- a/packages/server/src/errors.ts +++ b/packages/server/src/errors.ts @@ -1,6 +1,9 @@ import type { ProtocolErrorCode } from "@earendil-works/pi-protocol"; -type PiServerOperationErrorCode = Extract; +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"); diff --git a/packages/server/src/hosted-harness-manager.ts b/packages/server/src/hosted-harness-manager.ts index 46c78f173..93815d13c 100644 --- a/packages/server/src/hosted-harness-manager.ts +++ b/packages/server/src/hosted-harness-manager.ts @@ -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 { - 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 { diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index c4852dc4f..f325a8e8d 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -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 { /** Host capabilities used directly by the list and attach control-plane operations. */ export interface PiServerHost { - readonly sessions: Pick, "list">; + readonly sessions: { list(): Promise }; createHarness(metadata: TMetadata): Promise; } diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index aa8ceb703..d8ec9d389 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -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 = { + 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"); diff --git a/vitest.base.ts b/vitest.base.ts index 3d6a8df7a..080b2b2a0 100644 --- a/vitest.base.ts +++ b/vitest.base.ts @@ -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 },