mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(coding-agent): reduce experimental test duplication
This commit is contained in:
Armin Ronacher
parent
8e374e4dc2
commit
abeb9e07f3
6 changed files with 40 additions and 523 deletions
|
|
@ -1,9 +1,6 @@
|
|||
import { describe, expect, test, vi } from "vitest";
|
||||
import { cli } from "../src/cli/experimental/cli.ts";
|
||||
|
||||
const UNSUPPORTED_SERVER_OPTIONS = "The experimental server command does not support existing CLI options yet";
|
||||
const UNSUPPORTED_CLIENT_OPTIONS = "The experimental client command does not support existing CLI options yet";
|
||||
|
||||
describe("experimental CLI command composition", () => {
|
||||
test("requires an experimental subcommand", () => {
|
||||
expect(cli.parse([])).toEqual({
|
||||
|
|
@ -12,15 +9,6 @@ describe("experimental CLI command composition", () => {
|
|||
});
|
||||
});
|
||||
|
||||
test.each([
|
||||
["server", "--help", UNSUPPORTED_SERVER_OPTIONS],
|
||||
["server", "--version", UNSUPPORTED_SERVER_OPTIONS],
|
||||
["client", "--help", UNSUPPORTED_CLIENT_OPTIONS],
|
||||
["client", "--version", UNSUPPORTED_CLIENT_OPTIONS],
|
||||
] as const)("rejects unsupported %s %s handling", (command, option, error) => {
|
||||
expect(cli.parse([command, option])).toEqual({ ok: false, errors: [error] });
|
||||
});
|
||||
|
||||
test("passes server options to the command action", async () => {
|
||||
const runServer = vi.fn(() => undefined);
|
||||
const result = await cli.execute(
|
||||
|
|
|
|||
|
|
@ -91,8 +91,8 @@ describe("experimental client TUI", () => {
|
|||
["continued", { command: "client" as const, continue: true }, "one", 0],
|
||||
["plugin-selected", { command: "client" as const, pluginPackages: ["./example-plugin"] }, "two", 1],
|
||||
] as const)(
|
||||
"opens a %s Session directly and renders a Harness-backed turn",
|
||||
async (_kind, command, sessionId, creates) => {
|
||||
"opens a %s Session directly and exercises the full lifecycle only for a new Session",
|
||||
async (kind, command, sessionId, creates) => {
|
||||
const directoryState = replicatedState<SessionDirectoryState>({ revision: 1, sessions: [session("one", 1)] });
|
||||
const attachment = replicatedState<SessionAttachmentState>({ status: "detached" });
|
||||
const connectionState = replicatedState<ServerConnectionState>({ status: "connected", since: "now" });
|
||||
|
|
@ -334,6 +334,9 @@ describe("experimental client TUI", () => {
|
|||
expect(component.render(80).join("\n")).not.toContain("Experimental Sessions");
|
||||
expect(component.render(80).join("\n")).not.toContain("Experimental Models");
|
||||
|
||||
// Startup selection is the only behavior specific to continue and plugin-selected Sessions.
|
||||
if (kind !== "new") return;
|
||||
|
||||
component.handleInput("hello");
|
||||
component.handleInput("\r");
|
||||
await vi.waitFor(() => expect(prompt).toHaveBeenCalledWith("hello", undefined, BACKGROUND_CONTEXT));
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import { Client } from "@earendil-works/pi-client";
|
||||
import { PROTOCOL_VERSION } from "@earendil-works/pi-protocol";
|
||||
import { Server, type ServerHost } from "@earendil-works/pi-server";
|
||||
import type { Client } from "@earendil-works/pi-client";
|
||||
import type { Server } from "@earendil-works/pi-server";
|
||||
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
|
||||
import { RadiusRelayAuthResolver } from "../src/experimental/radius-auth.ts";
|
||||
import {
|
||||
|
|
@ -161,72 +160,6 @@ describe("experimental Radius relay", () => {
|
|||
await host.close();
|
||||
});
|
||||
|
||||
test("connects the real Pi client and server through the Radius transport", async () => {
|
||||
const hostWebSockets = socketFactory();
|
||||
const clientWebSockets = socketFactory();
|
||||
const serverHost: ServerHost = {
|
||||
serverServices: {
|
||||
attachClient() {
|
||||
return {
|
||||
async invokeService() {
|
||||
throw new Error("not used");
|
||||
},
|
||||
release() {},
|
||||
};
|
||||
},
|
||||
},
|
||||
async resolveSession() {
|
||||
throw new Error("not used");
|
||||
},
|
||||
async openSession() {
|
||||
throw new Error("not used");
|
||||
},
|
||||
};
|
||||
const protocolServer = new Server(serverHost, { serverId, listeners: [] });
|
||||
await protocolServer.start();
|
||||
const statuses: string[] = [];
|
||||
const relayHost = new RadiusRelayHost({
|
||||
serverId,
|
||||
server: protocolServer,
|
||||
auth: new RadiusRelayAuthResolver({ type: "token", token: "secret" }),
|
||||
webSocketFactory: hostWebSockets.factory,
|
||||
onStatus: (status) => statuses.push(status.status),
|
||||
});
|
||||
relayHost.start();
|
||||
await vi.waitFor(() => expect(hostWebSockets.sockets).toHaveLength(1));
|
||||
const hostSocket = hostWebSockets.sockets[0]!.socket;
|
||||
hostSocket.open(hostWebSockets.sockets[0]!.options.protocol);
|
||||
await vi.waitFor(() => expect(statuses).toContain("connected"));
|
||||
hostSocket.message(JSON.stringify({ version: 1, type: "connection_open", connection_id: connectionId }));
|
||||
|
||||
const connecting = Client.connect({
|
||||
serverId,
|
||||
transportFactory: createRadiusClientTransportFactory({
|
||||
serverId,
|
||||
auth: new RadiusRelayAuthResolver({ type: "token", token: "secret" }),
|
||||
webSocketFactory: clientWebSockets.factory,
|
||||
}),
|
||||
});
|
||||
await vi.waitFor(() => expect(clientWebSockets.sockets).toHaveLength(1));
|
||||
const clientSocket = clientWebSockets.sockets[0]!.socket;
|
||||
clientSocket.onSend = (data) => {
|
||||
if (data instanceof ArrayBuffer) hostSocket.message(encodeRelayDataFrame(connectionId, new Uint8Array(data)));
|
||||
};
|
||||
hostSocket.onSend = (data) => {
|
||||
if (data instanceof ArrayBuffer) {
|
||||
const frame = parseRelayDataFrame(data);
|
||||
if (frame?.connectionId === connectionId) clientSocket.message(frame.payload);
|
||||
}
|
||||
};
|
||||
clientSocket.open(clientWebSockets.sockets[0]!.options.protocol);
|
||||
const client = await connecting;
|
||||
|
||||
expect(client.hello).toEqual({ type: "hello", version: PROTOCOL_VERSION, serverId });
|
||||
await client.dispose();
|
||||
await relayHost.close();
|
||||
await protocolServer.close();
|
||||
});
|
||||
|
||||
test("reconnects the server host after the relay connection drops", async () => {
|
||||
vi.useFakeTimers();
|
||||
const webSockets = socketFactory();
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
import { lstat, mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
|
||||
import { join, resolve } from "node:path";
|
||||
import { join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { type Context, createFacetHost, defineFacet, defineService } from "@earendil-works/chord";
|
||||
import { BACKGROUND_CONTEXT, withCancel } from "@earendil-works/chord/context";
|
||||
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";
|
||||
|
|
@ -11,7 +11,7 @@ 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 { activateServer, type RunningServer, startServer } from "../src/experimental/server.ts";
|
||||
import { type RunningServer, startServer } from "../src/experimental/server.ts";
|
||||
import { createSessionServiceSource, type SessionAttachmentState } from "../src/experimental/services/connection.ts";
|
||||
import { Models } from "../src/experimental/services/models.ts";
|
||||
import { PresentationPlugins } from "../src/experimental/services/plugins.ts";
|
||||
|
|
@ -80,16 +80,6 @@ afterEach(async () => {
|
|||
});
|
||||
|
||||
describe("experimental durable server composition", () => {
|
||||
test("resolves the configured session directory", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pes-"));
|
||||
directories.add(directory);
|
||||
const runtime = await startServer({ directory, sessionDir: "relative/sessions" });
|
||||
servers.add(runtime);
|
||||
|
||||
expect(runtime.sessionDir).toBe(resolve("relative/sessions"));
|
||||
expect(runtime.workerPids.size).toBe(0);
|
||||
});
|
||||
|
||||
test("uses PI_SERVER_DIR and PI_SERVER_ID", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pi-server-dir-"));
|
||||
directories.add(directory);
|
||||
|
|
@ -126,25 +116,6 @@ describe("experimental durable server composition", () => {
|
|||
await expect(startServer({ directory, provider: "anthropic" })).rejects.toThrow("provider requires a model");
|
||||
});
|
||||
|
||||
test("uses legacy model selection when the server model is omitted", 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 runtime = await startServer({ directory });
|
||||
servers.add(runtime);
|
||||
|
||||
const client = await attachClient(runtime, "demo-1");
|
||||
await client.dispose();
|
||||
clients.delete(client);
|
||||
await expect.poll(() => runtime.workerPids.has("demo-1")).toBe(false);
|
||||
const state = await readExperimentalSessionState(runtime.sessionDir, "demo-1");
|
||||
expect(state.model).toEqual({ provider: "anthropic", modelId: "claude-opus-4-6" });
|
||||
expect(state.activeTools).toEqual(["read", "write", "bash"]);
|
||||
});
|
||||
|
||||
test("preserves an existing Session model when the server default changes", async () => {
|
||||
await writeFile(
|
||||
join(agentDir, "settings.json"),
|
||||
|
|
@ -170,35 +141,11 @@ describe("experimental durable server composition", () => {
|
|||
expect(state.model).toEqual({ provider: "anthropic", modelId: "claude-opus-4-6" });
|
||||
});
|
||||
|
||||
test.each(["discovery", "explicit connection"] as const)(
|
||||
"rejects model options when %s selects an existing server",
|
||||
async (connection) => {
|
||||
const { directory, runtime } = await makeServer();
|
||||
await expect(
|
||||
runClient(
|
||||
{
|
||||
command: "client",
|
||||
model: "anthropic/claude-opus-4-6",
|
||||
...(connection === "explicit connection"
|
||||
? { connect: { transport: "unix" as const, path: runtime.socketPath } }
|
||||
: {}),
|
||||
},
|
||||
{ directory },
|
||||
),
|
||||
).rejects.toThrow("Model selection is only valid when automatically activating a new server");
|
||||
},
|
||||
);
|
||||
|
||||
test("rejects model options when serialized activation finds an existing server", async () => {
|
||||
const { directory, runtime } = await makeServer();
|
||||
await expect(
|
||||
activateServer({
|
||||
directory,
|
||||
requestedServerId: runtime.serverId,
|
||||
sessionDir: runtime.sessionDir,
|
||||
model: "anthropic/claude-opus-4-6",
|
||||
}),
|
||||
).rejects.toThrow("Model selection is only valid when automatically activating a new server");
|
||||
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 () => {
|
||||
|
|
@ -333,21 +280,6 @@ describe("experimental durable server composition", () => {
|
|||
});
|
||||
});
|
||||
|
||||
test("discovers and lists seeded sessions without hosting either session", async () => {
|
||||
const { directory, runtime } = await makeServer();
|
||||
|
||||
await expect(runClient({ command: "client" }, { directory })).resolves.toEqual({
|
||||
kind: "list",
|
||||
sessions: [
|
||||
{ serverId: runtime.serverId, sessionId: "demo-1" },
|
||||
{ serverId: runtime.serverId, sessionId: "demo-2" },
|
||||
],
|
||||
});
|
||||
expect(runtime.workerPids.size).toBe(0);
|
||||
const socket = await lstat(runtime.socketPath);
|
||||
expect(socket.mode & 0o777).toBe(0o600);
|
||||
});
|
||||
|
||||
test("hydrates and mutates server Session services across framed clients", async () => {
|
||||
const { runtime } = await makeServer();
|
||||
const firstClient = await Client.connect({
|
||||
|
|
@ -409,29 +341,13 @@ describe("experimental durable server composition", () => {
|
|||
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
|
||||
});
|
||||
|
||||
test("shares one worker across concurrent Session attachments", async () => {
|
||||
const { runtime } = await makeServer();
|
||||
const client = await attachClient(runtime, "demo-1");
|
||||
|
||||
expect([...runtime.workerPids.keys()]).toEqual(["demo-1"]);
|
||||
const firstPid = runtime.workerPids.get("demo-1");
|
||||
expect(firstPid).toEqual(expect.any(Number));
|
||||
await expect(attachSession(client, "demo-1")).resolves.toBeUndefined();
|
||||
expect(runtime.workerPids.get("demo-1")).toBe(firstPid);
|
||||
|
||||
const competing = await Client.connect({
|
||||
serverId: runtime.serverId,
|
||||
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
||||
});
|
||||
clients.add(competing);
|
||||
await expect(attachSession(competing, "demo-1")).resolves.toBeUndefined();
|
||||
expect(runtime.workerPids.get("demo-1")).toBe(firstPid);
|
||||
});
|
||||
|
||||
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],
|
||||
|
|
@ -441,8 +357,6 @@ describe("experimental durable server composition", () => {
|
|||
services: [Models],
|
||||
onError: (error) => errors.push(error),
|
||||
});
|
||||
expect(firstServices.attachment.value).toEqual({ status: "attaching", sessionId: "demo-1" });
|
||||
expect(secondServices.attachment.value).toEqual({ status: "attaching", sessionId: "demo-1" });
|
||||
const firstModels = firstServices.use(Models);
|
||||
const secondModels = secondServices.use(Models);
|
||||
|
||||
|
|
@ -453,23 +367,13 @@ describe("experimental durable server composition", () => {
|
|||
provider: "anthropic",
|
||||
modelId: "claude-sonnet-4-5",
|
||||
});
|
||||
expect(firstModels.state.value?.catalog.availableModels).toContainEqual(
|
||||
expect.objectContaining({ provider: "anthropic", modelId: "claude-sonnet-4-5" }),
|
||||
);
|
||||
expect(secondModels.state.value).toEqual(firstModels.state.value);
|
||||
const previousThinking = firstModels.state.value!.configuration.thinkingLevel;
|
||||
const thinkingLevels = await firstModels.getThinkingLevels(BACKGROUND_CONTEXT);
|
||||
expect(thinkingLevels).toContain(previousThinking);
|
||||
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);
|
||||
});
|
||||
await firstModels.selectThinking(previousThinking, BACKGROUND_CONTEXT);
|
||||
await vi.waitFor(() => {
|
||||
expect(firstModels.state.value!.configuration.thinkingLevel).toBe(previousThinking);
|
||||
expect(secondModels.state.value).toEqual(firstModels.state.value);
|
||||
});
|
||||
expect(errors).toEqual([]);
|
||||
|
||||
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
|
||||
|
|
@ -496,52 +400,28 @@ describe("experimental durable server composition", () => {
|
|||
]);
|
||||
const runtime = await startServer({ ...sessionWorkerModel, directory });
|
||||
servers.add(runtime);
|
||||
await attachClient(runtime, "demo-2");
|
||||
const unrelatedWorkerPid = runtime.workerPids.get("demo-2");
|
||||
const clientRuntime = await openClientRuntime(
|
||||
const clientRuntime = await openClientRuntime({ command: "client" }, { directory });
|
||||
const activated = await activateBuiltinClientServices(clientRuntime.servers[0]!);
|
||||
await activated.plugins.prepareSession(
|
||||
{
|
||||
command: "client",
|
||||
pluginPackages: [
|
||||
sessionId: "demo-1",
|
||||
packagePaths: [
|
||||
fileURLToPath(new URL("../examples/plugins/pi-example-plugin", import.meta.url)),
|
||||
secondPackagePath,
|
||||
],
|
||||
},
|
||||
{ directory },
|
||||
);
|
||||
const activated = await activateBuiltinClientServices(clientRuntime.servers[0]!);
|
||||
const selectedPackagePaths = [
|
||||
fileURLToPath(new URL("../examples/plugins/pi-example-plugin", import.meta.url)),
|
||||
secondPackagePath,
|
||||
];
|
||||
await expect(
|
||||
activated.plugins.prepareSession(
|
||||
{ sessionId: "demo-2", packagePaths: selectedPackagePaths },
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
).rejects.toThrow("active with a different plugin selection");
|
||||
expect(runtime.workerPids.get("demo-2")).toBe(unrelatedWorkerPid);
|
||||
await activated.plugins.prepareSession(
|
||||
{
|
||||
sessionId: "demo-1",
|
||||
packagePaths: selectedPackagePaths,
|
||||
},
|
||||
BACKGROUND_CONTEXT,
|
||||
);
|
||||
await activated.management.attach("demo-1", BACKGROUND_CONTEXT);
|
||||
const client = clientRuntime.servers[0]!.client;
|
||||
const services = createSessionServiceBinding(client, {
|
||||
const services = createSessionServiceBinding(clientRuntime.servers[0]!.client, {
|
||||
services: [ExampleFacetService, SecondPluginService],
|
||||
});
|
||||
try {
|
||||
await services.ready(BACKGROUND_CONTEXT);
|
||||
const example = services.use(ExampleFacetService);
|
||||
await expect(example.greet({ name: "Armin" }, BACKGROUND_CONTEXT)).resolves.toEqual({
|
||||
message: "Hello Armin from the bundled Session worker facet!!!",
|
||||
workerActivations: 1,
|
||||
});
|
||||
expect(example.workerActivations.value).toBe(1);
|
||||
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");
|
||||
expect(runtime.workerPids.get("demo-2")).toBe(unrelatedWorkerPid);
|
||||
} finally {
|
||||
await services.dispose(BACKGROUND_CONTEXT);
|
||||
await clientRuntime.dispose();
|
||||
|
|
@ -674,12 +554,12 @@ describe("experimental durable server composition", () => {
|
|||
serviceId: KeyedProbe.id,
|
||||
mode: "keyed",
|
||||
});
|
||||
const observed: { service: KeyedProbe; context: Context; value: string | undefined }[] = [];
|
||||
const observed: { service: KeyedProbe; value: string | undefined }[] = [];
|
||||
const consumer = defineFacet({
|
||||
id: "@test/keyed-probe-consumer",
|
||||
setup(env) {
|
||||
env.observe(KeyedProbe, (service, context) => {
|
||||
observed.push({ service, context, value: service.state.value?.value });
|
||||
env.observe(KeyedProbe, (service) => {
|
||||
observed.push({ service, value: service.state.value?.value });
|
||||
});
|
||||
},
|
||||
});
|
||||
|
|
@ -688,26 +568,17 @@ describe("experimental durable server composition", () => {
|
|||
expect(services.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" });
|
||||
await vi.waitFor(() => expect(observed).toHaveLength(1));
|
||||
expect(observed[0]!.value).toBe("first");
|
||||
const stale = observed[0]!.service;
|
||||
const cancellable = withCancel(BACKGROUND_CONTEXT);
|
||||
const waiting = stale.wait(cancellable.context);
|
||||
const cancellation = new Error("cancel probe wait");
|
||||
cancellable.cancel(cancellation);
|
||||
await expect(waiting).rejects.toBe(cancellation);
|
||||
const staleReplace = stale.replace;
|
||||
const staleReplace = observed[0]!.service.replace;
|
||||
await expect(staleReplace("second", BACKGROUND_CONTEXT)).resolves.toBeUndefined();
|
||||
await vi.waitFor(() => expect(observed).toHaveLength(2));
|
||||
expect(observed[0]!.context.abortSignal?.aborted).toBe(true);
|
||||
expect(observed[1]!.value).toBe("second");
|
||||
expect(() => staleReplace("late", BACKGROUND_CONTEXT)).toThrow("observation is closed");
|
||||
|
||||
const replaced = observed[1]!.service;
|
||||
const replacedReplace = replaced.replace;
|
||||
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[1]!.context.abortSignal?.aborted).toBe(true);
|
||||
expect(observed[2]!.value).toBe("first");
|
||||
expect(() => replacedReplace("late", BACKGROUND_CONTEXT)).toThrow("observation is closed");
|
||||
expect(errors).toEqual([]);
|
||||
|
|
@ -753,30 +624,6 @@ describe("experimental durable server composition", () => {
|
|||
);
|
||||
});
|
||||
|
||||
test("uses legacy provider/model and thinking-suffix resolution", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pes-"));
|
||||
directories.add(directory);
|
||||
const runtime = await startServer({ directory, model: "anthropic/claude-sonnet-4-5:high" });
|
||||
servers.add(runtime);
|
||||
await attachClient(runtime, "demo-1");
|
||||
expect(runtime.workerPids.get("demo-1")).toEqual(expect.any(Number));
|
||||
});
|
||||
|
||||
test("accepts a custom model ID through legacy model resolution", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pes-"));
|
||||
directories.add(directory);
|
||||
const runtime = await startServer({ directory, provider: "anthropic", model: "missing-model" });
|
||||
servers.add(runtime);
|
||||
const client = await Client.connect({
|
||||
serverId: runtime.serverId,
|
||||
transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
|
||||
});
|
||||
clients.add(client);
|
||||
|
||||
await expect(attachSession(client, "demo-1")).resolves.toBeUndefined();
|
||||
expect(runtime.workerPids.get("demo-1")).toEqual(expect.any(Number));
|
||||
});
|
||||
|
||||
test("stops an idle Session worker after its client disconnects", async () => {
|
||||
const { runtime } = await makeServer();
|
||||
const client = await attachClient(runtime, "demo-1");
|
||||
|
|
@ -820,70 +667,6 @@ describe("experimental durable server composition", () => {
|
|||
expect(replacementPid).not.toBe(firstPid);
|
||||
});
|
||||
|
||||
test("normal server shutdown stops detached workers", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pes-"));
|
||||
directories.add(directory);
|
||||
const runtime = await startServer({ ...sessionWorkerModel, directory });
|
||||
servers.add(runtime);
|
||||
await attachClient(runtime, "demo-1");
|
||||
const pid = runtime.workerPids.get("demo-1");
|
||||
expect(pid).toEqual(expect.any(Number));
|
||||
|
||||
await runtime.close();
|
||||
await expect.poll(() => processExists(pid!)).toBe(false);
|
||||
expect(runtime.workerPids.size).toBe(0);
|
||||
await expect.poll(() => pathExists(runtime.socketPath)).toBe(false);
|
||||
});
|
||||
|
||||
test("serializes concurrent launchers so only one server remains active", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "pel-"));
|
||||
directories.add(directory);
|
||||
const runtimes = await Promise.all([
|
||||
startServer({ ...sessionWorkerModel, directory }),
|
||||
startServer({ ...sessionWorkerModel, directory }),
|
||||
]);
|
||||
for (const runtime of runtimes) servers.add(runtime);
|
||||
|
||||
expect(runtimes[0].serverId).toBe(runtimes[1].serverId);
|
||||
await expect
|
||||
.poll(async () => {
|
||||
const closed = await Promise.all(
|
||||
runtimes.map((runtime) =>
|
||||
Promise.race([
|
||||
runtime.closed.then(() => true),
|
||||
new Promise<false>((resolve) => setImmediate(() => resolve(false))),
|
||||
]),
|
||||
),
|
||||
);
|
||||
return closed.filter(Boolean).length;
|
||||
})
|
||||
.toBe(1);
|
||||
await expect(runClient({ command: "client" }, { directory })).resolves.toMatchObject({
|
||||
kind: "list",
|
||||
sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }],
|
||||
});
|
||||
});
|
||||
|
||||
test("shuts down the replaced server without waiting for its clients", async () => {
|
||||
const directory = await mkdtemp(join("/tmp", "peg-"));
|
||||
directories.add(directory);
|
||||
const first = await startServer({ ...sessionWorkerModel, directory });
|
||||
servers.add(first);
|
||||
const client = await Client.connect({
|
||||
serverId: first.serverId,
|
||||
transportFactory: createUnixTransportFactory({ path: first.socketPath }),
|
||||
});
|
||||
|
||||
const replacement = await startServer({ ...sessionWorkerModel, directory });
|
||||
servers.add(replacement);
|
||||
await first.closed;
|
||||
await expect(runClient({ command: "client" }, { directory })).resolves.toMatchObject({
|
||||
kind: "list",
|
||||
sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }],
|
||||
});
|
||||
await client.dispose();
|
||||
});
|
||||
|
||||
test("discovers workers after replacing the server", async () => {
|
||||
const firstDirectory = await mkdtemp(join("/tmp", "per-"));
|
||||
directories.add(firstDirectory);
|
||||
|
|
|
|||
|
|
@ -1,158 +0,0 @@
|
|||
import { type ChildProcess, spawn } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import { lstat, mkdir, mkdtemp, rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { afterEach, describe, expect, test } from "vitest";
|
||||
import { runClient } from "../src/experimental/client.ts";
|
||||
import { configureExperimentalWorkerModel, createExperimentalSessions } from "./experimental-session-support.ts";
|
||||
|
||||
const cliPath = fileURLToPath(new URL("../src/cli.ts", import.meta.url));
|
||||
const processes = new Set<ChildProcess>();
|
||||
const directories = new Set<string>();
|
||||
|
||||
interface RunningCli {
|
||||
readonly child: ChildProcess;
|
||||
readonly output: () => string;
|
||||
readonly errors: () => string;
|
||||
}
|
||||
|
||||
async function startServer(home: string, agentDir: string, serverDir: string): Promise<RunningCli> {
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
["--import", "tsx", cliPath, "server", "--provider", "anthropic", "--model", "claude-sonnet-4-5"],
|
||||
{
|
||||
cwd: fileURLToPath(new URL("../../..", import.meta.url)),
|
||||
env: {
|
||||
...process.env,
|
||||
HOME: home,
|
||||
PI_CODING_AGENT_DIR: agentDir,
|
||||
PI_EXPERIMENTAL: "1",
|
||||
PI_SERVER_DIR: serverDir,
|
||||
},
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
},
|
||||
);
|
||||
processes.add(child);
|
||||
let output = "";
|
||||
let errors = "";
|
||||
child.stdout?.setEncoding("utf8");
|
||||
child.stderr?.setEncoding("utf8");
|
||||
child.stdout?.on("data", (chunk: string) => {
|
||||
output += chunk;
|
||||
});
|
||||
child.stderr?.on("data", (chunk: string) => {
|
||||
errors += chunk;
|
||||
});
|
||||
return { child, output: () => output, errors: () => errors };
|
||||
}
|
||||
|
||||
async function waitForOutput(process: RunningCli, pattern: RegExp, timeoutMs = 10_000): Promise<RegExpMatchArray> {
|
||||
const existing = process.output().match(pattern);
|
||||
if (existing) return existing;
|
||||
return new Promise((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(
|
||||
new Error(`Timed out waiting for ${pattern}; stdout: ${process.output()}; stderr: ${process.errors()}`),
|
||||
);
|
||||
}, timeoutMs);
|
||||
const inspect = (): void => {
|
||||
const match = process.output().match(pattern);
|
||||
if (!match) return;
|
||||
cleanup();
|
||||
resolve(match);
|
||||
};
|
||||
const exited = (code: number | null, signal: NodeJS.Signals | null): void => {
|
||||
cleanup();
|
||||
reject(
|
||||
new Error(
|
||||
`Server exited before ${pattern} (${signal ?? code ?? "unknown"}); stdout: ${process.output()}; stderr: ${process.errors()}`,
|
||||
),
|
||||
);
|
||||
};
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timeout);
|
||||
process.child.stdout?.off("data", inspect);
|
||||
process.child.off("exit", exited);
|
||||
};
|
||||
process.child.stdout?.on("data", inspect);
|
||||
process.child.once("exit", exited);
|
||||
});
|
||||
}
|
||||
|
||||
async function waitForExit(process: ChildProcess): Promise<void> {
|
||||
if (process.exitCode !== null || process.signalCode !== null) return;
|
||||
await once(process, "exit");
|
||||
}
|
||||
|
||||
async function stop(process: ChildProcess): Promise<void> {
|
||||
if (process.exitCode !== null || process.signalCode !== null) return;
|
||||
const exited = once(process, "exit");
|
||||
process.kill("SIGTERM");
|
||||
await exited;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all([...processes].map((process) => stop(process)));
|
||||
processes.clear();
|
||||
await Promise.all([...directories].map((directory) => rm(directory, { recursive: true, force: true })));
|
||||
directories.clear();
|
||||
});
|
||||
|
||||
describe.skipIf(process.platform === "win32")("experimental CLI server replacement", () => {
|
||||
test("a second CLI server takes over the stable coordinator socket", async () => {
|
||||
const root = await mkdtemp(join("/tmp", "pcs-"));
|
||||
directories.add(root);
|
||||
const home = join(root, "long-home-segment".repeat(8));
|
||||
await mkdir(home);
|
||||
const agentDir = join(root, "agent");
|
||||
const serverDir = join(root, "server");
|
||||
await configureExperimentalWorkerModel(agentDir);
|
||||
await createExperimentalSessions(join(agentDir, "experimental", "sessions"), ["demo-1", "demo-2"]);
|
||||
const first = await startServer(home, agentDir, serverDir);
|
||||
const firstIdentity = await waitForOutput(
|
||||
first,
|
||||
/Server: ([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})/,
|
||||
);
|
||||
const firstSocket = await waitForOutput(first, /Socket: (.+\.sock)/);
|
||||
expect(firstSocket[1]).toBe(join(serverDir, `${firstIdentity[1]}.sock`));
|
||||
expect((await lstat(join(serverDir, `control-${firstIdentity[1]}.sock`))).isSocket()).toBe(true);
|
||||
await runClient({
|
||||
command: "client",
|
||||
sessionId: "demo-1",
|
||||
connect: { transport: "unix", path: firstSocket[1]! },
|
||||
});
|
||||
|
||||
const replacement = await startServer(home, agentDir, serverDir);
|
||||
const replacementIdentity = await waitForOutput(
|
||||
replacement,
|
||||
/Server: ([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})/,
|
||||
);
|
||||
const replacementSocket = await waitForOutput(replacement, /Socket: (.+\.sock)/);
|
||||
await waitForExit(first.child);
|
||||
|
||||
expect(replacementIdentity[1]).toBe(firstIdentity[1]);
|
||||
expect(replacementSocket[1]).toBe(firstSocket[1]);
|
||||
expect(replacement.child.pid).not.toBe(first.child.pid);
|
||||
await expect(
|
||||
runClient({
|
||||
command: "client",
|
||||
connect: { transport: "unix", path: replacementSocket[1]! },
|
||||
}),
|
||||
).resolves.toMatchObject({
|
||||
kind: "list",
|
||||
sessions: [
|
||||
{ serverId: firstIdentity[1], sessionId: "demo-1" },
|
||||
{ serverId: firstIdentity[1], sessionId: "demo-2" },
|
||||
],
|
||||
});
|
||||
await expect(
|
||||
runClient({
|
||||
command: "client",
|
||||
sessionId: "demo-1",
|
||||
connect: { transport: "unix", path: replacementSocket[1]! },
|
||||
}),
|
||||
).resolves.toMatchObject({ kind: "attached", sessionId: "demo-1" });
|
||||
});
|
||||
});
|
||||
|
|
@ -259,54 +259,22 @@ describe("protocol validation", () => {
|
|||
expect(parseClientMessage(keyed)).toEqual(keyed);
|
||||
});
|
||||
|
||||
test("keeps transport RPC calls untyped while validating their envelope", () => {
|
||||
const list: ClientMessage = {
|
||||
test("keeps transport RPC payloads opaque while validating their envelope", () => {
|
||||
const message: ClientMessage = {
|
||||
type: "request",
|
||||
id: "request-1",
|
||||
target: { serverId: "00000000-0000-4000-8000-000000000001" },
|
||||
call: { serviceId: "pi.session-directory", member: "list", args: [] },
|
||||
};
|
||||
const create: ClientMessage = {
|
||||
type: "request",
|
||||
id: "request-2",
|
||||
target: { serverId: "00000000-0000-4000-8000-000000000001" },
|
||||
call: { serviceId: "pi.session-management", member: "create", args: [{}] },
|
||||
};
|
||||
const attach: ClientMessage = {
|
||||
type: "request",
|
||||
id: "request-3",
|
||||
target: { serverId: "00000000-0000-4000-8000-000000000001" },
|
||||
call: { serviceId: "pi.session-management", member: "attach", args: ["session-1"] },
|
||||
};
|
||||
const prompt: ClientMessage = {
|
||||
type: "request",
|
||||
id: "request-4",
|
||||
target: {
|
||||
serverId: "00000000-0000-4000-8000-000000000001",
|
||||
sessionId: "session-1",
|
||||
attachmentId: "attachment-1",
|
||||
},
|
||||
call: { serviceId: "pi.chat", member: "prompt", args: [["Hello"]] },
|
||||
call: {
|
||||
serviceId: "application.custom",
|
||||
member: "invoke",
|
||||
args: [{ arbitrary: true }, ["opaque"]],
|
||||
},
|
||||
};
|
||||
expect(parseClientMessage(list)).toEqual(list);
|
||||
expect(parseClientMessage(create)).toEqual(create);
|
||||
expect(parseClientMessage(attach)).toEqual(attach);
|
||||
expect(parseClientMessage(prompt)).toEqual(prompt);
|
||||
expect(parseClientMessage({ ...create, call: { ...create.call, args: [{ cwd: "" }] } })).toMatchObject({
|
||||
call: { serviceId: "pi.session-management", member: "create" },
|
||||
});
|
||||
expect(parseClientMessage({ ...attach, call: { ...attach.call, args: [] } })).toMatchObject({
|
||||
call: { serviceId: "pi.session-management", member: "attach", args: [] },
|
||||
});
|
||||
expect(
|
||||
parseClientMessage({
|
||||
...attach,
|
||||
call: { serviceId: "unknown", member: "method", args: [{ arbitrary: true }] },
|
||||
}),
|
||||
).toMatchObject({ call: { serviceId: "unknown", member: "method" } });
|
||||
expect(parseClientMessage({ ...prompt, call: { ...prompt.call, args: ["session-1", []] } })).toMatchObject({
|
||||
call: { serviceId: "pi.chat", member: "prompt" },
|
||||
});
|
||||
expect(parseClientMessage(message)).toEqual(message);
|
||||
});
|
||||
|
||||
test("validates request cancellation envelopes", () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue