mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
48dd1e2f0f
Session workers now host a durable Harness over a per-session sqlite storage. The server keeps each session in <sessionDir>/<id>/ with meta.json and session.sqlite and never opens the storage itself. Workers hold retirement while the Harness task graph has live tasks and resume interrupted work on open. Transcript serves the root conversation's durable view directly, Models follows the conversation's agent document, and AgentController maps prompt, steer, follow-up, queue cancellation, abort, and compaction onto the root conversation, plus waitForPrompt. navigate, nextRun, and resume are dropped (see experimental/services/README.md TODO). The durable TUI and the worker share harness-setup.ts. pi-server exports its own SessionMetadata and no longer depends on pi-agent-core. Chord accepts readonly JSON types in service contracts without walking them, which recursive JSON types like pi-ai's made too deep.
820 lines
34 KiB
TypeScript
820 lines
34 KiB
TypeScript
import { lstat, mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
|
|
import { join } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { type Context, createFacetHost, defineFacet, defineService } from "@earendil-works/chord";
|
|
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
|
|
import { Client, ServerError as ClientServerError } from "@earendil-works/pi-client";
|
|
import { createUnixTransportFactory } from "@earendil-works/pi-client/unix";
|
|
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
|
|
import { ExampleFacetService } from "../examples/plugins/pi-example-plugin/src/contract.ts";
|
|
import { runClient } from "../src/experimental/client.ts";
|
|
import { activateBuiltinClientServices, openClientRuntime } from "../src/experimental/client-runtime.ts";
|
|
import { createPresentationFacetLoaders } from "../src/experimental/plugins/bundled.ts";
|
|
import * as processRuntime from "../src/experimental/process.ts";
|
|
import { type RunningServer, startServer } from "../src/experimental/server.ts";
|
|
import { AgentController } from "../src/experimental/services/agent-controller.ts";
|
|
import { createSessionServiceSource, type SessionAttachmentState } from "../src/experimental/services/connection.ts";
|
|
import { Models } from "../src/experimental/services/models.ts";
|
|
import { PresentationPlugins, SessionPlugins } from "../src/experimental/services/plugins.ts";
|
|
import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts";
|
|
import { Transcript } from "../src/experimental/services/transcript.ts";
|
|
import { createServerServiceBinding, createSessionServiceBinding } from "./experimental-service-binding.ts";
|
|
import {
|
|
configureExperimentalWorkerModel,
|
|
createExperimentalSessions,
|
|
readExperimentalSessionState,
|
|
} from "./experimental-session-support.ts";
|
|
import { KeyedProbe } from "./fixtures/keyed-service.ts";
|
|
|
|
const servers = new Set<RunningServer>();
|
|
const clients = new Set<Client>();
|
|
const directories = new Set<string>();
|
|
const fauxWorkerEntryUrl = new URL("fixtures/faux-session-worker.ts", import.meta.url);
|
|
const realSpawnInternalProcess = processRuntime.spawnInternalProcess;
|
|
const sessionWorkerModel = { provider: "anthropic", model: "claude-sonnet-4-5" } as const;
|
|
const SecondPluginService = defineService<{ read(context: Context): Promise<string> }>("test.second-plugin");
|
|
let agentDir: string;
|
|
|
|
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: RunningServer }> {
|
|
const directory = await mkdtemp(join("/tmp", "pes-"));
|
|
directories.add(directory);
|
|
const runtime = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(runtime);
|
|
return { directory, runtime };
|
|
}
|
|
|
|
async function attachSession(client: Client, sessionId: string): Promise<void> {
|
|
const services = createServerServiceBinding(client, { services: [SessionManagement] });
|
|
try {
|
|
await services.ready(BACKGROUND_CONTEXT);
|
|
await services.use(SessionManagement).attach(sessionId, BACKGROUND_CONTEXT);
|
|
} finally {
|
|
await services.dispose(BACKGROUND_CONTEXT);
|
|
}
|
|
}
|
|
|
|
async function attachClient(runtime: RunningServer, sessionId: string): Promise<Client> {
|
|
const client = await Client.connect({
|
|
serverId: runtime.serverId,
|
|
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
|
});
|
|
clients.add(client);
|
|
await attachSession(client, sessionId);
|
|
return client;
|
|
}
|
|
|
|
afterEach(async () => {
|
|
await Promise.all([...clients].map((client) => client.dispose()));
|
|
clients.clear();
|
|
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 durable server composition", () => {
|
|
test("uses PI_SERVER_DIR and PI_SERVER_ID", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-server-dir-"));
|
|
directories.add(directory);
|
|
const serverId = "00000000-0000-4000-8000-000000000001";
|
|
vi.stubEnv("PI_SERVER_DIR", directory);
|
|
vi.stubEnv("PI_SERVER_ID", serverId);
|
|
const runtime = await startServer();
|
|
servers.add(runtime);
|
|
|
|
expect(runtime.serverId).toBe(serverId);
|
|
expect(runtime.socketPath).toBe(join(directory, `${serverId}.sock`));
|
|
expect((await lstat(directory)).mode & 0o777).toBe(0o700);
|
|
const publicSocket = await lstat(runtime.socketPath);
|
|
const controlSocket = await lstat(join(directory, `control-${runtime.serverId}.sock`));
|
|
expect(publicSocket.isSocket()).toBe(true);
|
|
expect(publicSocket.mode & 0o777).toBe(0o600);
|
|
expect(controlSocket.isSocket()).toBe(true);
|
|
expect(controlSocket.mode & 0o777).toBe(0o600);
|
|
const entries = await readdir(directory);
|
|
expect(entries).toHaveLength(3);
|
|
expect(entries.every((entry) => !entry.startsWith("."))).toBe(true);
|
|
expect(entries).toContain(`${serverId}.sock`);
|
|
expect(entries).toContain(`control-${serverId}.sock`);
|
|
expect(entries).toContainEqual(expect.stringMatching(new RegExp(`^server-${serverId}-[0-9a-f]{12}\\.sock$`)));
|
|
await expect(runClient({ command: "client" })).resolves.toMatchObject({
|
|
kind: "list",
|
|
sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }],
|
|
});
|
|
});
|
|
|
|
test("rejects a provider without a model", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pes-"));
|
|
directories.add(directory);
|
|
await expect(startServer({ directory, provider: "anthropic" })).rejects.toThrow("provider requires a model");
|
|
});
|
|
|
|
test("preserves an existing Session model when the server default changes", async () => {
|
|
await writeFile(
|
|
join(agentDir, "settings.json"),
|
|
JSON.stringify({ defaultProvider: "anthropic", defaultModel: "claude-opus-4-6" }),
|
|
);
|
|
const directory = await mkdtemp(join("/tmp", "pes-"));
|
|
directories.add(directory);
|
|
const first = await startServer({ directory });
|
|
servers.add(first);
|
|
const firstClient = await attachClient(first, "demo-1");
|
|
await firstClient.dispose();
|
|
clients.delete(firstClient);
|
|
await expect.poll(() => first.workerPids.has("demo-1")).toBe(false);
|
|
await first.close();
|
|
|
|
const second = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(second);
|
|
const secondClient = await attachClient(second, "demo-1");
|
|
await secondClient.dispose();
|
|
clients.delete(secondClient);
|
|
await expect.poll(() => second.workerPids.has("demo-1")).toBe(false);
|
|
const state = await readExperimentalSessionState(second.sessionDir, "demo-1");
|
|
expect(state.model).toEqual({ provider: "anthropic", modelId: "claude-opus-4-6" });
|
|
});
|
|
|
|
test("rejects model options when discovery selects an existing server", async () => {
|
|
const { directory } = await makeServer();
|
|
await expect(runClient({ command: "client", model: "anthropic/claude-opus-4-6" }, { directory })).rejects.toThrow(
|
|
"Model selection is only valid when automatically activating a new server",
|
|
);
|
|
});
|
|
|
|
test("rechecks an auto-discovered server after a version mismatch", async () => {
|
|
const { directory, runtime } = await makeServer();
|
|
const connect = Client.connect.bind(Client);
|
|
vi.spyOn(Client, "connect")
|
|
.mockRejectedValueOnce(new ClientServerError({ code: "version", message: "stale server" }))
|
|
.mockImplementation((options) => connect(options));
|
|
const clientRuntime = await openClientRuntime({ command: "client" }, { directory });
|
|
try {
|
|
expect(clientRuntime.servers.map(({ route }) => route.serverId)).toEqual([runtime.serverId]);
|
|
} finally {
|
|
await clientRuntime.dispose();
|
|
}
|
|
});
|
|
|
|
test("serializes concurrent cold activation and retires after both clients leave", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-auto-server-"));
|
|
directories.add(directory);
|
|
const serverId = "00000000-0000-4000-8000-000000000001";
|
|
vi.stubEnv("PI_SERVER_DIR", directory);
|
|
vi.stubEnv("PI_SERVER_ID", serverId);
|
|
|
|
const results = await Promise.all([runClient({ command: "client" }), runClient({ command: "client" })]);
|
|
expect(results).toEqual([
|
|
{
|
|
kind: "list",
|
|
sessions: [
|
|
{ serverId, sessionId: "demo-1" },
|
|
{ serverId, sessionId: "demo-2" },
|
|
],
|
|
},
|
|
{
|
|
kind: "list",
|
|
sessions: [
|
|
{ serverId, sessionId: "demo-1" },
|
|
{ serverId, sessionId: "demo-2" },
|
|
],
|
|
},
|
|
]);
|
|
expect(await pathExists(join(directory, `${serverId}.sock`))).toBe(true);
|
|
await expect.poll(() => pathExists(join(directory, `${serverId}.sock`)), { timeout: 5_000 }).toBe(false);
|
|
await expect.poll(() => pathExists(join(directory, `control-${serverId}.sock`)), { timeout: 5_000 }).toBe(false);
|
|
});
|
|
|
|
test("passes client plugin packages to a cold server and restores them for its next generation", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-auto-plugin-"));
|
|
directories.add(directory);
|
|
const serverId = "00000000-0000-4000-8000-000000000001";
|
|
const packagePath = fileURLToPath(new URL("../examples/plugins/pi-example-plugin", import.meta.url));
|
|
vi.stubEnv("PI_SERVER_DIR", directory);
|
|
vi.stubEnv("PI_SERVER_ID", serverId);
|
|
|
|
const first = await openClientRuntime({ command: "client", ...sessionWorkerModel });
|
|
try {
|
|
const activated = await activateBuiltinClientServices(first.servers[0]!);
|
|
const presentationPlugins = await activated.plugins.prepareSession(
|
|
{ sessionId: "demo-1", packagePaths: [packagePath] },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
await activated.management.attach("demo-1", BACKGROUND_CONTEXT);
|
|
const loaded = await createPresentationFacetLoaders(presentationPlugins)[0]!.load();
|
|
expect(loaded.facets.map(({ id }) => id)).toEqual(["@earendil-works/pi-example-plugin/tui"]);
|
|
await loaded.dispose();
|
|
} finally {
|
|
await first.dispose();
|
|
}
|
|
await expect.poll(() => pathExists(join(directory, `${serverId}.sock`)), { timeout: 5_000 }).toBe(false);
|
|
|
|
const second = await openClientRuntime({ command: "client", ...sessionWorkerModel });
|
|
try {
|
|
const activated = await activateBuiltinClientServices(second.servers[0]!);
|
|
const presentationPlugins = await activated.plugins.prepareSession(
|
|
{ sessionId: "demo-1", packagePaths: null },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
await activated.management.attach("demo-1", BACKGROUND_CONTEXT);
|
|
expect(createPresentationFacetLoaders(presentationPlugins)).toHaveLength(1);
|
|
} finally {
|
|
await second.dispose();
|
|
}
|
|
});
|
|
|
|
test("retires a cold server after its only Session attachment disconnects", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-auto-session-"));
|
|
directories.add(directory);
|
|
const serverId = "00000000-0000-4000-8000-000000000001";
|
|
vi.stubEnv("PI_SERVER_DIR", directory);
|
|
vi.stubEnv("PI_SERVER_ID", serverId);
|
|
|
|
await expect(runClient({ command: "client", sessionId: "demo-1", ...sessionWorkerModel })).resolves.toEqual({
|
|
kind: "attached",
|
|
serverId,
|
|
sessionId: "demo-1",
|
|
});
|
|
await expect.poll(() => pathExists(join(directory, `${serverId}.sock`)), { timeout: 5_000 }).toBe(false);
|
|
await expect.poll(() => pathExists(join(directory, `control-${serverId}.sock`)), { timeout: 5_000 }).toBe(false);
|
|
});
|
|
|
|
test("runs and discovers multiple logical servers from one directory", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-multi-server-"));
|
|
directories.add(directory);
|
|
const firstId = "00000000-0000-4000-8000-000000000001";
|
|
const secondId = "00000000-0000-4000-8000-000000000002";
|
|
const [first, second] = await Promise.all([
|
|
startServer({ directory, serverId: firstId }),
|
|
startServer({ directory, serverId: secondId }),
|
|
]);
|
|
servers.add(first);
|
|
servers.add(second);
|
|
|
|
expect((await lstat(join(directory, `control-${firstId}.sock`))).isSocket()).toBe(true);
|
|
expect((await lstat(join(directory, `control-${secondId}.sock`))).isSocket()).toBe(true);
|
|
await expect(runClient({ command: "client" }, { directory })).resolves.toEqual({
|
|
kind: "list",
|
|
sessions: [
|
|
{ serverId: firstId, sessionId: "demo-1" },
|
|
{ serverId: firstId, sessionId: "demo-2" },
|
|
{ serverId: secondId, sessionId: "demo-1" },
|
|
{ serverId: secondId, sessionId: "demo-2" },
|
|
],
|
|
});
|
|
|
|
await first.close();
|
|
await expect.poll(() => pathExists(first.socketPath)).toBe(false);
|
|
await expect(runClient({ command: "client" }, { directory })).resolves.toEqual({
|
|
kind: "list",
|
|
sessions: [
|
|
{ serverId: secondId, sessionId: "demo-1" },
|
|
{ serverId: secondId, sessionId: "demo-2" },
|
|
],
|
|
});
|
|
});
|
|
|
|
test("hydrates and mutates server Session services across framed clients", async () => {
|
|
const { runtime } = await makeServer();
|
|
const firstClient = await Client.connect({
|
|
serverId: runtime.serverId,
|
|
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
|
});
|
|
const secondClient = await Client.connect({
|
|
serverId: runtime.serverId,
|
|
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
|
});
|
|
clients.add(firstClient);
|
|
clients.add(secondClient);
|
|
const errors: Error[] = [];
|
|
const firstServices = createServerServiceBinding(firstClient, {
|
|
services: [SessionDirectory, SessionManagement],
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
const secondServices = createServerServiceBinding(secondClient, {
|
|
services: [SessionDirectory, SessionManagement],
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
expect(firstServices.connection.value).toMatchObject({ status: "connected" });
|
|
expect(secondServices.connection.value).toMatchObject({ status: "connected" });
|
|
await expect(firstServices.catalogue(BACKGROUND_CONTEXT)).resolves.toEqual([
|
|
{ serviceId: SessionDirectory.id, mode: "singleton" },
|
|
{ serviceId: SessionManagement.id, mode: "singleton" },
|
|
{ serviceId: PresentationPlugins.id, mode: "singleton" },
|
|
]);
|
|
const firstDirectory = firstServices.use(SessionDirectory);
|
|
const secondDirectory = secondServices.use(SessionDirectory);
|
|
const firstManagement = firstServices.use(SessionManagement);
|
|
const secondManagement = secondServices.use(SessionManagement);
|
|
|
|
await Promise.all([firstServices.ready(BACKGROUND_CONTEXT), secondServices.ready(BACKGROUND_CONTEXT)]);
|
|
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toEqual(["demo-1", "demo-2"]);
|
|
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
|
|
await firstManagement.create({ id: "demo-3" }, BACKGROUND_CONTEXT);
|
|
await vi.waitFor(() => {
|
|
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toContain("demo-3");
|
|
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
|
|
});
|
|
|
|
await Promise.all([
|
|
firstManagement.attach("demo-1", BACKGROUND_CONTEXT),
|
|
secondManagement.attach("demo-1", BACKGROUND_CONTEXT),
|
|
]);
|
|
expect(firstClient.attachment?.sessionId).toBe("demo-1");
|
|
expect(secondClient.attachment?.sessionId).toBe("demo-1");
|
|
await firstManagement.remove("demo-1", BACKGROUND_CONTEXT);
|
|
await vi.waitFor(() => {
|
|
expect(firstClient.attachment).toBeUndefined();
|
|
expect(secondClient.attachment).toBeUndefined();
|
|
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).not.toContain("demo-1");
|
|
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
|
|
});
|
|
await secondManagement.detach(BACKGROUND_CONTEXT);
|
|
expect(errors).toEqual([]);
|
|
|
|
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
|
|
});
|
|
|
|
test("hydrates and updates the Models service across concurrent framed clients", async () => {
|
|
const { runtime } = await makeServer();
|
|
const firstClient = await attachClient(runtime, "demo-1");
|
|
const workerPid = runtime.workerPids.get("demo-1");
|
|
expect(workerPid).toEqual(expect.any(Number));
|
|
const secondClient = await attachClient(runtime, "demo-1");
|
|
expect(runtime.workerPids.get("demo-1")).toBe(workerPid);
|
|
const errors: Error[] = [];
|
|
const firstServices = createSessionServiceBinding(firstClient, {
|
|
services: [Models],
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
const secondServices = createSessionServiceBinding(secondClient, {
|
|
services: [Models],
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
const firstModels = firstServices.use(Models);
|
|
const secondModels = secondServices.use(Models);
|
|
|
|
await Promise.all([firstServices.ready(BACKGROUND_CONTEXT), secondServices.ready(BACKGROUND_CONTEXT)]);
|
|
expect(firstServices.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
|
expect(secondServices.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
|
expect(firstModels.state.value?.configuration.model).toEqual({
|
|
provider: "anthropic",
|
|
modelId: "claude-sonnet-4-5",
|
|
});
|
|
expect(secondModels.state.value).toEqual(firstModels.state.value);
|
|
const previousThinking = firstModels.state.value!.configuration.thinkingLevel;
|
|
await firstModels.cycleThinking(BACKGROUND_CONTEXT);
|
|
await vi.waitFor(() => {
|
|
expect(firstModels.state.value!.configuration.thinkingLevel).not.toBe(previousThinking);
|
|
expect(secondModels.state.value).toEqual(firstModels.state.value);
|
|
});
|
|
expect(errors).toEqual([]);
|
|
|
|
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
|
|
});
|
|
|
|
test("loads conventional Session facets from multiple configured plugin packages", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pes-plugin-"));
|
|
directories.add(directory);
|
|
const secondPackagePath = join(directory, "second-plugin");
|
|
await mkdir(join(secondPackagePath, "src"), { recursive: true });
|
|
await Promise.all([
|
|
writeFile(
|
|
join(secondPackagePath, "package.json"),
|
|
`${JSON.stringify({
|
|
name: "@earendil-works/second-session-plugin",
|
|
version: "1.0.0",
|
|
peerDependencies: { "@earendil-works/chord": "^0.84.4" },
|
|
})}\n`,
|
|
),
|
|
writeFile(
|
|
join(secondPackagePath, "src", "session.ts"),
|
|
'import { defineFacet, defineService } from "@earendil-works/chord"; const Service = defineService("test.second-plugin"); export default defineFacet({ id: "second-session-plugin", setup(env) { env.provide(Service, { async read() { return "second"; } }); } });\n',
|
|
),
|
|
]);
|
|
const runtime = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(runtime);
|
|
const clientRuntime = await openClientRuntime({ command: "client" }, { directory });
|
|
const activated = await activateBuiltinClientServices(clientRuntime.servers[0]!);
|
|
await activated.plugins.prepareSession(
|
|
{
|
|
sessionId: "demo-1",
|
|
packagePaths: [
|
|
fileURLToPath(new URL("../examples/plugins/pi-example-plugin", import.meta.url)),
|
|
secondPackagePath,
|
|
],
|
|
},
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
await activated.management.attach("demo-1", BACKGROUND_CONTEXT);
|
|
const services = createSessionServiceBinding(clientRuntime.servers[0]!.client, {
|
|
services: [ExampleFacetService, SecondPluginService],
|
|
});
|
|
try {
|
|
await services.ready(BACKGROUND_CONTEXT);
|
|
await expect(
|
|
services.use(ExampleFacetService).greet({ name: "Armin" }, BACKGROUND_CONTEXT),
|
|
).resolves.toMatchObject({ workerActivations: 1 });
|
|
await expect(services.use(SecondPluginService).read(BACKGROUND_CONTEXT)).resolves.toBe("second");
|
|
} finally {
|
|
await services.dispose(BACKGROUND_CONTEXT);
|
|
await clientRuntime.dispose();
|
|
}
|
|
});
|
|
|
|
test("uses the most recently selected model for a new Session", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pes-model-default-"));
|
|
directories.add(directory);
|
|
const runtime = await startServer({ directory });
|
|
servers.add(runtime);
|
|
const firstClient = await attachClient(runtime, "demo-1");
|
|
const firstServices = createSessionServiceBinding(firstClient, { services: [Models] });
|
|
const firstModels = firstServices.use(Models);
|
|
await firstServices.ready(BACKGROUND_CONTEXT);
|
|
await firstModels.select({ provider: "anthropic", modelId: "claude-opus-4-6" }, BACKGROUND_CONTEXT);
|
|
await firstServices.dispose(BACKGROUND_CONTEXT);
|
|
await firstClient.dispose();
|
|
clients.delete(firstClient);
|
|
await expect.poll(() => runtime.workerPids.has("demo-1")).toBe(false);
|
|
|
|
const secondClient = await attachClient(runtime, "demo-2");
|
|
const secondServices = createSessionServiceBinding(secondClient, { services: [Models] });
|
|
const secondModels = secondServices.use(Models);
|
|
await secondServices.ready(BACKGROUND_CONTEXT);
|
|
expect(secondModels.state.value?.configuration.model).toEqual({
|
|
provider: "anthropic",
|
|
modelId: "claude-opus-4-6",
|
|
});
|
|
await secondServices.dispose(BACKGROUND_CONTEXT);
|
|
});
|
|
|
|
test("composes management attachment with Session service hydration", async () => {
|
|
const { runtime } = await makeServer();
|
|
const clientRuntime = await openClientRuntime({
|
|
command: "client",
|
|
connect: { transport: "unix", path: runtime.socketPath },
|
|
});
|
|
try {
|
|
const server = await activateBuiltinClientServices(clientRuntime.servers[0]!);
|
|
await server.plugins.prepareSession({ sessionId: "demo-1", packagePaths: null }, BACKGROUND_CONTEXT);
|
|
await server.management.attach("demo-1", BACKGROUND_CONTEXT);
|
|
|
|
expect(server.session.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
|
expect(server.models.state.value?.configuration.model).toEqual({
|
|
provider: "anthropic",
|
|
modelId: "claude-sonnet-4-5",
|
|
});
|
|
} finally {
|
|
await clientRuntime.dispose();
|
|
}
|
|
});
|
|
|
|
test("fences superseded attachment hydration by attachment generation", async ({ onTestFinished }) => {
|
|
const { runtime } = await makeServer();
|
|
const client = await Client.connect({
|
|
serverId: runtime.serverId,
|
|
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
|
});
|
|
clients.add(client);
|
|
const errors: Error[] = [];
|
|
const services = createSessionServiceBinding(client, {
|
|
services: [Models],
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
const models = services.use(Models);
|
|
await services.ready(BACKGROUND_CONTEXT);
|
|
const states: SessionAttachmentState[] = [];
|
|
const removeStateListener = services.attachment.subscribe((state) => states.push(state));
|
|
let releaseDelay!: () => void;
|
|
const delayed = new Promise<void>((resolve) => {
|
|
releaseDelay = resolve;
|
|
});
|
|
const subscribeService = client.subscribeService.bind(client);
|
|
const subscribe = vi
|
|
.spyOn(client, "subscribeService")
|
|
.mockImplementation(async (target, serviceId, mode, listener, signal) => {
|
|
if ("sessionId" in target && target.sessionId === "demo-2" && serviceId === Models.id) {
|
|
await delayed;
|
|
}
|
|
return subscribeService(target, serviceId, mode, listener, signal);
|
|
});
|
|
onTestFinished(() => subscribe.mockRestore());
|
|
|
|
try {
|
|
await attachSession(client, "demo-2");
|
|
expect(services.attachment.value).toEqual({ status: "attaching", sessionId: "demo-2" });
|
|
await attachSession(client, "demo-1");
|
|
await services.whenAttached("demo-1", BACKGROUND_CONTEXT);
|
|
expect(services.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
|
expect(models.state.value?.configuration.model).toEqual({
|
|
provider: "anthropic",
|
|
modelId: "claude-sonnet-4-5",
|
|
});
|
|
} finally {
|
|
releaseDelay();
|
|
removeStateListener();
|
|
await services.dispose(BACKGROUND_CONTEXT);
|
|
}
|
|
|
|
const latestAttach = states
|
|
.map((state) => state.status === "attaching" && state.sessionId === "demo-1")
|
|
.lastIndexOf(true);
|
|
expect(latestAttach).toBeGreaterThanOrEqual(0);
|
|
expect(states.slice(latestAttach)).not.toContainEqual({ status: "attached", sessionId: "demo-2" });
|
|
expect(states.slice(latestAttach)).not.toContainEqual({ status: "degraded", sessionId: "demo-2" });
|
|
expect(errors).toEqual([]);
|
|
});
|
|
|
|
test("observes keyed service instances and fences replacement generations over framed transport", async ({
|
|
onTestFinished,
|
|
}) => {
|
|
const spawn = vi
|
|
.spyOn(processRuntime, "spawnInternalProcess")
|
|
.mockImplementation((role, args, options) =>
|
|
realSpawnInternalProcess(
|
|
role,
|
|
args,
|
|
role === "session-worker" ? { ...options, entryUrl: fauxWorkerEntryUrl } : options,
|
|
),
|
|
);
|
|
onTestFinished(() => spawn.mockRestore());
|
|
const { runtime } = await makeServer();
|
|
const client = await attachClient(runtime, "demo-1");
|
|
const errors: Error[] = [];
|
|
const services = createSessionServiceSource(client, {
|
|
onError: (error) => errors.push(error),
|
|
});
|
|
await expect(services.catalogue(BACKGROUND_CONTEXT)).resolves.toContainEqual({
|
|
serviceId: KeyedProbe.id,
|
|
mode: "keyed",
|
|
});
|
|
const observed: { service: KeyedProbe; value: string | undefined }[] = [];
|
|
const consumer = defineFacet({
|
|
id: "@test/keyed-probe-consumer",
|
|
setup(env) {
|
|
env.observe(KeyedProbe, (service) => {
|
|
observed.push({ service, value: service.state.value?.value });
|
|
});
|
|
},
|
|
});
|
|
const facetHost = await createFacetHost({ facets: [consumer], serviceSources: [services] });
|
|
|
|
expect(services.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
|
await vi.waitFor(() => expect(observed).toHaveLength(1));
|
|
expect(observed[0]!.value).toBe("first");
|
|
const staleReplace = observed[0]!.service.replace;
|
|
await expect(staleReplace("second", BACKGROUND_CONTEXT)).resolves.toBeUndefined();
|
|
await vi.waitFor(() => expect(observed).toHaveLength(2));
|
|
expect(observed[1]!.value).toBe("second");
|
|
expect(() => staleReplace("late", BACKGROUND_CONTEXT)).toThrow("observation is closed");
|
|
|
|
const replacedReplace = observed[1]!.service.replace;
|
|
await expect(attachSession(client, "demo-2")).resolves.toBeUndefined();
|
|
await services.whenAttached("demo-2", BACKGROUND_CONTEXT);
|
|
expect(observed).toHaveLength(3);
|
|
expect(services.attachment.value).toEqual({ status: "attached", sessionId: "demo-2" });
|
|
expect(observed[2]!.value).toBe("first");
|
|
expect(() => replacedReplace("late", BACKGROUND_CONTEXT)).toThrow("observation is closed");
|
|
expect(errors).toEqual([]);
|
|
|
|
await facetHost.dispose();
|
|
await expect(services.dispose(BACKGROUND_CONTEXT)).resolves.toBeUndefined();
|
|
});
|
|
|
|
test("answers a client prompt through the worker-owned service provider", async ({ onTestFinished }) => {
|
|
const spawn = vi
|
|
.spyOn(processRuntime, "spawnInternalProcess")
|
|
.mockImplementation((role, args, options) =>
|
|
realSpawnInternalProcess(
|
|
role,
|
|
args,
|
|
role === "session-worker" ? { ...options, entryUrl: fauxWorkerEntryUrl } : options,
|
|
),
|
|
);
|
|
onTestFinished(() => spawn.mockRestore());
|
|
const { directory } = await makeServer();
|
|
|
|
const result = await runClient({ command: "client", sessionId: "demo-1", prompt: "question" }, { directory });
|
|
|
|
expect(result).toMatchObject({ kind: "prompted", text: "deterministic remote answer" });
|
|
});
|
|
|
|
test("replicates the transcript after consecutive prompts", async ({ onTestFinished }) => {
|
|
const spawn = vi
|
|
.spyOn(processRuntime, "spawnInternalProcess")
|
|
.mockImplementation((role, args, options) =>
|
|
realSpawnInternalProcess(
|
|
role,
|
|
args,
|
|
role === "session-worker" ? { ...options, entryUrl: fauxWorkerEntryUrl } : options,
|
|
),
|
|
);
|
|
onTestFinished(() => spawn.mockRestore());
|
|
const { runtime } = await makeServer();
|
|
const client = await attachClient(runtime, "demo-1");
|
|
const services = createSessionServiceBinding(client, { services: [AgentController, SessionPlugins, Transcript] });
|
|
const controller = services.use(AgentController);
|
|
const transcript = services.use(Transcript);
|
|
await services.ready(BACKGROUND_CONTEXT);
|
|
await services.use(SessionPlugins).reload(BACKGROUND_CONTEXT);
|
|
try {
|
|
for (const message of ["first question", "second question"]) {
|
|
const response = await controller.prompt({ message, images: null }, BACKGROUND_CONTEXT);
|
|
expect(response).toMatchObject({ accepted: true, operationId: expect.any(String) });
|
|
if (!response.accepted) throw new Error(response.error.message);
|
|
await expect(controller.waitForPrompt(response.operationId, BACKGROUND_CONTEXT)).resolves.toEqual({
|
|
status: "done",
|
|
text: "deterministic remote answer",
|
|
reason: null,
|
|
});
|
|
await vi.waitFor(() => {
|
|
expect(transcript.state.value?.entries.filter((entry) => entry.kind === "pi.assistant")).toHaveLength(
|
|
message === "first question" ? 1 : 2,
|
|
);
|
|
});
|
|
}
|
|
} finally {
|
|
await services.dispose(BACKGROUND_CONTEXT);
|
|
}
|
|
});
|
|
|
|
test("stops an idle Session worker after its client disconnects", async () => {
|
|
const { runtime } = await makeServer();
|
|
const client = await attachClient(runtime, "demo-1");
|
|
const pid = runtime.workerPids.get("demo-1");
|
|
expect(pid).toEqual(expect.any(Number));
|
|
|
|
await client.dispose();
|
|
clients.delete(client);
|
|
await expect.poll(() => runtime.workerPids.has("demo-1")).toBe(false);
|
|
expect(processExists(pid!)).toBe(false);
|
|
});
|
|
|
|
test("starts one process per attached session and stops them during shutdown", async () => {
|
|
const { runtime } = await makeServer();
|
|
await Promise.all([attachClient(runtime, "demo-1"), attachClient(runtime, "demo-2")]);
|
|
|
|
const pids = [...runtime.workerPids.values()];
|
|
expect(pids).toHaveLength(2);
|
|
expect(new Set(pids).size).toBe(2);
|
|
for (const pid of pids) expect(processExists(pid)).toBe(true);
|
|
|
|
await runtime.close();
|
|
expect(runtime.workerPids.size).toBe(0);
|
|
await Promise.all(pids.map((pid) => expect.poll(() => processExists(pid)).toBe(false)));
|
|
});
|
|
|
|
test("server runtime replaces an exited worker on the next attach", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pew-"));
|
|
directories.add(directory);
|
|
const runtime = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(runtime);
|
|
const client = await attachClient(runtime, "demo-1");
|
|
const firstPid = runtime.workerPids.get("demo-1");
|
|
expect(firstPid).toEqual(expect.any(Number));
|
|
|
|
process.kill(firstPid!, "SIGKILL");
|
|
await expect.poll(() => runtime.workerPids.has("demo-1")).toBe(false);
|
|
await attachSession(client, "demo-1");
|
|
const replacementPid = runtime.workerPids.get("demo-1");
|
|
expect(replacementPid).toEqual(expect.any(Number));
|
|
expect(replacementPid).not.toBe(firstPid);
|
|
});
|
|
|
|
test("discovers workers after replacing the server", async () => {
|
|
const firstDirectory = await mkdtemp(join("/tmp", "per-"));
|
|
directories.add(firstDirectory);
|
|
const first = await startServer({ ...sessionWorkerModel, directory: firstDirectory });
|
|
servers.add(first);
|
|
await attachClient(first, "demo-1");
|
|
const firstWorkerPid = first.workerPids.get("demo-1");
|
|
expect(firstWorkerPid).toEqual(expect.any(Number));
|
|
|
|
const replacement = await startServer({ ...sessionWorkerModel, directory: firstDirectory });
|
|
servers.add(replacement);
|
|
await first.closed;
|
|
|
|
expect(replacement.serverId).toBe(first.serverId);
|
|
expect(replacement.workerPids.get("demo-1")).toBe(firstWorkerPid);
|
|
await expect.poll(() => first.workerPids.size).toBe(0);
|
|
expect(processExists(firstWorkerPid!)).toBe(true);
|
|
|
|
await expect(runClient({ command: "client" }, { directory: firstDirectory })).resolves.toMatchObject({
|
|
kind: "list",
|
|
sessions: [
|
|
{ serverId: first.serverId, sessionId: "demo-1" },
|
|
{ serverId: first.serverId, sessionId: "demo-2" },
|
|
],
|
|
});
|
|
await attachClient(replacement, "demo-1");
|
|
expect(replacement.workerPids.get("demo-1")).toBe(firstWorkerPid);
|
|
await attachClient(replacement, "demo-2");
|
|
expect(replacement.workerPids.get("demo-2")).toEqual(expect.any(Number));
|
|
expect(replacement.workerPids.get("demo-2")).not.toBe(firstWorkerPid);
|
|
});
|
|
|
|
test("retires an unclaimed idle worker after replacement demand expires", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pi-orphan-worker-"));
|
|
directories.add(directory);
|
|
vi.stubEnv("__PI_SESSION_WORKER_ORPHAN_DEMAND_GRACE_MS", "50");
|
|
const first = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(first);
|
|
await attachClient(first, "demo-1");
|
|
const workerPid = first.workerPids.get("demo-1");
|
|
expect(workerPid).toEqual(expect.any(Number));
|
|
|
|
const replacement = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(replacement);
|
|
await first.closed;
|
|
expect(replacement.workerPids.get("demo-1")).toBe(workerPid);
|
|
|
|
await expect.poll(() => replacement.workerPids.has("demo-1"), { timeout: 5_000 }).toBe(false);
|
|
expect(processExists(workerPid!)).toBe(false);
|
|
});
|
|
|
|
test("restores tracked sessions that are outside the replacement catalog", async () => {
|
|
const directory = await mkdtemp(join("/tmp", "pet-"));
|
|
const emptySessionDir = await mkdtemp(join("/tmp", "pet-sessions-"));
|
|
directories.add(directory);
|
|
directories.add(emptySessionDir);
|
|
const first = await startServer({ ...sessionWorkerModel, directory });
|
|
servers.add(first);
|
|
await attachClient(first, "demo-1");
|
|
const workerPid = first.workerPids.get("demo-1");
|
|
|
|
const replacement = await startServer({
|
|
...sessionWorkerModel,
|
|
directory,
|
|
sessionDir: emptySessionDir,
|
|
});
|
|
servers.add(replacement);
|
|
await first.closed;
|
|
|
|
await expect(runClient({ command: "client" }, { directory })).resolves.toEqual({
|
|
kind: "list",
|
|
sessions: [{ serverId: first.serverId, sessionId: "demo-1" }],
|
|
});
|
|
await attachClient(replacement, "demo-1");
|
|
expect(replacement.workerPids.get("demo-1")).toBe(workerPid);
|
|
});
|
|
|
|
test("reports missing and ambiguous session selections", async () => {
|
|
const sharedDirectory = await mkdtemp(join("/tmp", "ped-"));
|
|
directories.add(sharedDirectory);
|
|
const firstShared = await startServer({
|
|
...sessionWorkerModel,
|
|
directory: sharedDirectory,
|
|
serverId: "00000000-0000-4000-8000-000000000001",
|
|
});
|
|
const secondShared = await startServer({
|
|
...sessionWorkerModel,
|
|
directory: sharedDirectory,
|
|
serverId: "00000000-0000-4000-8000-000000000002",
|
|
});
|
|
servers.add(firstShared);
|
|
servers.add(secondShared);
|
|
|
|
await expect(
|
|
runClient({ command: "client", sessionId: "missing" }, { directory: sharedDirectory }),
|
|
).rejects.toThrow("No discovered server contains session missing");
|
|
await expect(
|
|
runClient({ 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 session directory", async () => {
|
|
await expect(
|
|
createExperimentalSessions(
|
|
join(agentDir, "experimental", "sessions"),
|
|
["demo-1"],
|
|
join(agentDir, "other-cwd"),
|
|
),
|
|
).rejects.toThrow("Session demo-1 already exists");
|
|
});
|
|
});
|
|
|
|
async function pathExists(path: string): Promise<boolean> {
|
|
try {
|
|
await lstat(path);
|
|
return true;
|
|
} catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code === "ENOENT") return false;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
function processExists(pid: number): boolean {
|
|
try {
|
|
process.kill(pid, 0);
|
|
return true;
|
|
} catch (error) {
|
|
return (error as NodeJS.ErrnoException).code === "EPERM";
|
|
}
|
|
}
|