diff --git a/packages/coding-agent/test/experimental-cli-resolution.test.ts b/packages/coding-agent/test/experimental-cli-resolution.test.ts index 985b7e5f7..b9ceaeb29 100644 --- a/packages/coding-agent/test/experimental-cli-resolution.test.ts +++ b/packages/coding-agent/test/experimental-cli-resolution.test.ts @@ -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( diff --git a/packages/coding-agent/test/experimental-client-tui.test.ts b/packages/coding-agent/test/experimental-client-tui.test.ts index 40819c7e1..321a5734c 100644 --- a/packages/coding-agent/test/experimental-client-tui.test.ts +++ b/packages/coding-agent/test/experimental-client-tui.test.ts @@ -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({ revision: 1, sessions: [session("one", 1)] }); const attachment = replicatedState({ status: "detached" }); const connectionState = replicatedState({ 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)); diff --git a/packages/coding-agent/test/experimental-radius-relay.test.ts b/packages/coding-agent/test/experimental-radius-relay.test.ts index 7ef6795ce..1048798c0 100644 --- a/packages/coding-agent/test/experimental-radius-relay.test.ts +++ b/packages/coding-agent/test/experimental-radius-relay.test.ts @@ -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(); diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index bb44b56e5..dd7acaa7a 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -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((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); diff --git a/packages/coding-agent/test/experimental-server-replacement.test.ts b/packages/coding-agent/test/experimental-server-replacement.test.ts deleted file mode 100644 index 2f33f7fd7..000000000 --- a/packages/coding-agent/test/experimental-server-replacement.test.ts +++ /dev/null @@ -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(); -const directories = new Set(); - -interface RunningCli { - readonly child: ChildProcess; - readonly output: () => string; - readonly errors: () => string; -} - -async function startServer(home: string, agentDir: string, serverDir: string): Promise { - 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 { - 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 { - if (process.exitCode !== null || process.signalCode !== null) return; - await once(process, "exit"); -} - -async function stop(process: ChildProcess): Promise { - 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" }); - }); -}); diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index fe4bda4cf..51435a554 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -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", () => {