sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(coding-agent): isolate coordinators by server ID

Browse source
This commit is contained in:
Armin Ronacher 2026-08-15 12:20:05 +02:00
parent d96144c6a3
commit 5876bc49f9
13 changed files with 164 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/coding-agent/README.md
View file

@ -673,6 +673,7 @@ pi --thinking high "Solve this complex problem"
| `PI_CODING_AGENT_SESSION_DIR` | Override session storage directory (overridden by `--session-dir`) |
| `PI_PACKAGE_DIR` | Override package directory (useful for Nix/Guix where store paths tokenize poorly) |
| `PI_SERVER_DIR` | Override experimental server profile and socket directory (default: `~/.pi/server`) |
| `PI_SERVER_ID` | Select the logical experimental server ID when `--server-id` is omitted |
| `PI_OFFLINE` | Disable startup network operations, including update checks, package update checks, and install/update telemetry |
| `PI_SKIP_VERSION_CHECK` | Skip the Pi version update check at startup. This prevents the `pi.dev` latest-version request |
| `PI_TELEMETRY` | Override install/update telemetry and provider attribution headers. Use `1`/`true`/`yes` to enable or `0`/`false`/`no` to disable. This does not disable update checks |

1
packages/coding-agent/docs/environment-variables.md
View file

@ -82,6 +82,7 @@ These variables are read by Pi itself:
| `PI_CODING_AGENT_SESSION_DIR` | Override session storage; overridden by `--session-dir` |
| `PI_PACKAGE_DIR` | Override the package directory, useful for Nix/Guix store paths |
| `PI_SERVER_DIR` | Override the experimental server profile and socket directory; default is `~/.pi/server` |
| `PI_SERVER_ID` | Select the logical experimental server ID when `--server-id` is omitted |
| `PI_OFFLINE` | Disable startup network operations, including update checks, package updates, and install/update telemetry |
| `PI_SKIP_VERSION_CHECK` | Disable the `pi.dev` latest-version request |
| `PI_TELEMETRY` | Override install/update telemetry and provider attribution headers: `1`/`true`/`yes` or `0`/`false`/`no` |

1
packages/coding-agent/src/cli/args.ts
View file

@ -413,6 +413,7 @@ ${chalk.bold("Environment Variables:")}
${ENV_SESSION_DIR.padEnd(32)} - Session storage directory (overridden by --session-dir)
PI_PACKAGE_DIR - Override package directory (for Nix/Guix store paths)
PI_SERVER_DIR - Experimental server profile and socket directory (default: ~/.pi/server)
PI_SERVER_ID - Logical experimental server ID (overridden by --server-id)
PI_OFFLINE - Disable startup network operations when set to 1/true/yes
PI_TELEMETRY - Override install telemetry when set to 1/true/yes or 0/false/no
PI_SHARE_VIEWER_URL - Base URL for /share command (default: https://pi.dev/session/)

12
packages/coding-agent/src/cli/experimental/commands/server.ts
View file

@ -1,5 +1,6 @@
import { isServerId, type ServerId } from "@earendil-works/pi-protocol";
import type { AuthInput } from "../auth.ts";
import { Command, stringOption } from "../command.ts";
import { Command, stringOption, valueOption } from "../command.ts";
import {
authTokenFileOption,
authTokenOption,
@ -14,6 +15,7 @@ export interface ServerCommand {
readonly command: "server";
readonly auth?: AuthInput;
readonly listen?: readonly TransportAddress[];
readonly serverId?: ServerId;
readonly sessionDir?: string;
}
@ -22,16 +24,23 @@ export interface ServerCommandContext {
}
const listenOption = transportOption("--listen");
const serverIdOption = valueOption("--server-id", (value) =>
isServerId(value)
? { ok: true, value }
: { ok: false, error: `Invalid --server-id "${value}"; expected a lowercase UUIDv4` },
);
const sessionDirOption = stringOption("--session-dir");
export const serverCommand = new Command<ServerCommand, ServerCommandContext>("server")
.option(listenOption)
.option(serverIdOption)
.option(sessionDirOption)
.option(authTokenOption)
.option(authTokenFileOption)
.build((input) => {
const { auth, errors: authErrors } = parseAuth(input);
const listen = input.values(listenOption);
const serverId = input.value(serverIdOption);
const sessionDir = input.value(sessionDirOption);
const { errors: optionErrors } = parseLegacyOptions(input);
const errors = [...authErrors, ...optionErrors, ...unsupportedLegacyOptions("server", input)];
@ -42,6 +51,7 @@ export const serverCommand = new Command<ServerCommand, ServerCommandContext>("s
command: "server",
...(auth === undefined ? {} : { auth }),
...(listen.length === 0 ? {} : { listen }),
...(serverId === undefined ? {} : { serverId }),
...(sessionDir === undefined ? {} : { sessionDir }),
},
};

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

@ -4,7 +4,7 @@ import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import { PiClient } from "@earendil-works/pi-client";
import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix";
import { isServerId } from "@earendil-works/pi-protocol";
import { isServerId, type ServerId } from "@earendil-works/pi-protocol";
import type { HostedHarnessHandle, PiServer, PiServerHost } from "@earendil-works/pi-server";
import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix";
import { getAgentDir } from "../../config.ts";
@ -16,7 +16,7 @@ import {
type CoordinatorStartupLease,
ensureExperimentalCoordinator,
} from "./coordinator-client.ts";
import { ensurePrivateServerDirectory, resolveExperimentalServerDirectory } from "./server-directory.ts";
import { ENV_SERVER_ID, ensurePrivateServerDirectory, resolveExperimentalServerDirectory } from "./server-directory.ts";
import { acquireExperimentalServerProfile } from "./server-profile.ts";
import { startExperimentalSessionWorker } from "./session-worker.ts";
@ -41,7 +41,7 @@ export interface StartExperimentalServerOptions {
/** Directory for server sockets. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
readonly path?: string;
readonly serverId?: string;
readonly serverId?: ServerId;
/** Durable session directory. Defaults to the experimental directory under the configured agent directory. */
readonly sessionDir?: string;
}
@ -49,6 +49,8 @@ export interface StartExperimentalServerOptions {
export interface StartExperimentalCoordinatedServerOptions {
/** Server profile and socket directory. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
/** Logical service ID. Defaults to PI_SERVER_ID or the directory's default-server-id. */
readonly serverId?: ServerId;
/** Durable session directory. Defaults to the experimental directory under the configured agent directory. */
readonly sessionDir?: string;
}
@ -178,7 +180,10 @@ export async function startExperimentalCoordinatedServer(
options: StartExperimentalCoordinatedServerOptions = {},
): Promise<ExperimentalServer> {
const directory = resolveExperimentalServerDirectory(options.directory);
const { serverId, release } = await acquireExperimentalServerProfile(directory);
const { serverId, release } = await acquireExperimentalServerProfile(
directory,
options.serverId ?? process.env[ENV_SERVER_ID],
);
let backend: ExperimentalServer | undefined;
let coordinator: CoordinatorServer | undefined;
let startupLease: CoordinatorStartupLease | undefined;
@ -187,8 +192,8 @@ export async function startExperimentalCoordinatedServer(
try {
await ensurePrivateServerDirectory(directory);
const socketPath = getUnixSocketPath(serverId, directory);
const controlPath = join(directory, "control.sock");
const serverPath = join(directory, `.server-${randomUUID().slice(0, 12)}.sock`);
const controlPath = join(directory, `control-${serverId}.sock`);
const serverPath = join(directory, `.server-${serverId}-${randomUUID().slice(0, 12)}.sock`);
startupLease = await ensureExperimentalCoordinator(socketPath, controlPath);
coordinator = new CoordinatorServer({ controlPath, endpoint: serverPath });
const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir);

1
packages/coding-agent/src/cli/experimental/server-directory.ts
View file

@ -4,6 +4,7 @@ import { join } from "node:path";
import { resolvePath } from "../../utils/paths.ts";
export const ENV_SERVER_DIR = "PI_SERVER_DIR";
export const ENV_SERVER_ID = "PI_SERVER_ID";
export function resolveExperimentalServerDirectory(directory?: string): string {
return resolvePath(directory ?? process.env[ENV_SERVER_DIR] ?? join(homedir(), ".pi", "server"));

70
packages/coding-agent/src/cli/experimental/server-profile.ts
View file

@ -7,55 +7,57 @@ import lockfile from "proper-lockfile";
const LOCK_STALE_MS = 30_000;
const LOCK_RETRY_MS = 25;
const LOCK_WAIT_MS = 30_000;
const SERVER_ID_FILE = "server-id";
const DEFAULT_SERVER_ID_FILE = "default-server-id";
export interface ExperimentalServerProfile {
readonly serverId: ServerId;
release(): Promise<void>;
}
/** Lock one experimental server profile and load or create its stable identity. */
export async function acquireExperimentalServerProfile(directory: string): Promise<ExperimentalServerProfile> {
/** Lock one logical server ID in a shared experimental server directory. */
export async function acquireExperimentalServerProfile(
directory: string,
requestedServerId?: string,
): Promise<ExperimentalServerProfile> {
await mkdir(directory, { recursive: true, mode: 0o700 });
const release = await lockfile.lock(directory, {
let serverId: ServerId;
if (requestedServerId !== undefined) {
if (!isServerId(requestedServerId)) throw new Error(`Invalid experimental server ID: ${requestedServerId}`);
serverId = requestedServerId;
} else {
const path = join(directory, DEFAULT_SERVER_ID_FILE);
try {
const value = (await readFile(path, "utf8")).trim();
if (!isServerId(value)) throw new Error(`Invalid default experimental server identity in ${path}`);
serverId = value;
} catch (error) {
const code = error instanceof Error && "code" in error ? error.code : undefined;
if (code !== "ENOENT") throw error;
const candidate = randomUUID();
try {
await writeFile(path, candidate, { encoding: "utf8", mode: 0o600, flag: "wx" });
serverId = candidate;
} catch (writeError) {
const writeCode = writeError instanceof Error && "code" in writeError ? writeError.code : undefined;
if (writeCode !== "EEXIST") throw writeError;
const value = (await readFile(path, "utf8")).trim();
if (!isServerId(value)) throw new Error(`Invalid default experimental server identity in ${path}`);
serverId = value;
}
}
}
const release = await lockfile.lock(join(directory, `.launcher-${serverId}`), {
realpath: false,
lockfilePath: join(directory, ".launcher.lock"),
stale: LOCK_STALE_MS,
update: LOCK_STALE_MS / 3,
retries: {
retries: Math.ceil(LOCK_WAIT_MS / LOCK_RETRY_MS),
factor: 1,
minTimeout: LOCK_RETRY_MS,
maxTimeout: LOCK_RETRY_MS,
maxRetryTime: LOCK_WAIT_MS,
},
});
try {
return { serverId: await loadOrCreateServerId(directory), release };
} catch (error) {
try {
await release();
} catch (releaseError) {
throw new AggregateError([error, releaseError], "Server profile loading and lock release failed");
}
throw error;
}
}
async function loadOrCreateServerId(directory: string): Promise<ServerId> {
const path = join(directory, SERVER_ID_FILE);
try {
const value = (await readFile(path, "utf8")).trim();
if (!isServerId(value)) throw new Error(`Invalid experimental server identity in ${path}`);
return value;
} catch (error) {
if (!isErrorCode(error, "ENOENT")) throw error;
}
const serverId = randomUUID();
await writeFile(path, serverId, { encoding: "utf8", mode: 0o600, flag: "wx" });
return serverId;
}
function isErrorCode(error: unknown, code: string): boolean {
return error instanceof Error && "code" in error && error.code === code;
return { serverId, release };
}

5
packages/coding-agent/src/main.ts
View file

@ -590,7 +590,10 @@ async function waitForTermination(serverClosed: Promise<void>): Promise<void> {
async function runExperimentalServerCommand(command: ServerCommand): Promise<void> {
if (command.auth !== undefined) throw new Error("Authentication is not supported by the local demo server");
if (command.listen !== undefined) throw new Error("The local demo server uses its server-addressed Unix socket");
const runtime = await startExperimentalCoordinatedServer({ sessionDir: command.sessionDir });
const runtime = await startExperimentalCoordinatedServer({
serverId: command.serverId,
sessionDir: command.sessionDir,
});
console.log(`Server: ${runtime.serverId}`);
console.log(`Socket: ${runtime.socketPath}`);
try {

12
packages/coding-agent/test/experimental-cli-command.test.ts
View file

@ -38,6 +38,16 @@ describe("experimental CLI commands", () => {
});
});
test("parses an experimental server ID", () => {
expect(experimentalCli.parse(["server", "--server-id", "00000000-0000-4000-8000-000000000001"])).toEqual({
ok: true,
command: {
command: "server",
serverId: "00000000-0000-4000-8000-000000000001",
},
});
});
test("parses an experimental server session directory", () => {
expect(experimentalCli.parse(["server", "--session-dir", "~/pi-sessions"])).toEqual({
ok: true,
@ -149,6 +159,8 @@ describe("experimental CLI commands", () => {
"The experimental server command does not support existing CLI options yet",
],
[["client", "--connect", "ws://localhost:8080"], 'Unsupported --connect transport "ws:"'],
[["server", "--server-id", "not-a-uuid"], "Invalid --server-id"],
[["server", "--server-id"], "--server-id requires a value"],
[["server", "--session-dir"], "--session-dir requires a value"],
[
["server", "--session-dir", "/tmp/first", "--session-dir=/tmp/second"],

27
packages/coding-agent/test/experimental-cli-resolution.test.ts
View file

@ -80,19 +80,24 @@ describe("experimental CLI command composition", () => {
});
});
test("passes the server session directory to the command action", async () => {
test("passes server options to the command action", async () => {
const runServer = vi.fn(() => undefined);
const result = await experimentalCli.execute(["server", "--session-dir", "./sessions"], {
runPi: vi.fn(() => undefined),
runServer,
runClient: vi.fn(() => undefined),
});
const result = await experimentalCli.execute(
["server", "--server-id", "00000000-0000-4000-8000-000000000001", "--session-dir", "./sessions"],
{
runPi: vi.fn(() => undefined),
runServer,
runClient: vi.fn(() => undefined),
},
);
expect(result).toEqual({
ok: true,
command: { command: "server", sessionDir: "./sessions" },
});
expect(runServer).toHaveBeenCalledWith({ command: "server", sessionDir: "./sessions" });
const command = {
command: "server" as const,
serverId: "00000000-0000-4000-8000-000000000001",
sessionDir: "./sessions",
};
expect(result).toEqual({ ok: true, command });
expect(runServer).toHaveBeenCalledWith(command);
});
test.each(["pi", "server", "client"] as const)("executes the parsed %s command", async (name) => {

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

@ -64,17 +64,20 @@ describe("experimental durable server composition", () => {
expect(runtime.workerPids.size).toBe(0);
});
test("uses PI_SERVER_DIR for sockets and discovery", async () => {
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 startExperimentalCoordinatedServer();
servers.add(runtime);
expect(runtime.socketPath).toBe(join(directory, `${runtime.serverId}.sock`));
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.sock"));
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);
@ -85,6 +88,41 @@ describe("experimental durable server composition", () => {
});
});
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([
startExperimentalCoordinatedServer({ directory, serverId: firstId }),
startExperimentalCoordinatedServer({ 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(runExperimentalClient({ 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(runExperimentalClient({ command: "client" }, { directory })).resolves.toEqual({
kind: "list",
sessions: [
{ serverId: secondId, sessionId: "demo-1" },
{ serverId: secondId, sessionId: "demo-2" },
],
});
});
test("discovers and lists seeded sessions without hosting either session", async () => {
const { directory, runtime } = await makeServer();

34
packages/coding-agent/test/experimental-server-profile.test.ts
View file

@ -1,10 +1,12 @@
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, test } from "vitest";
import { acquireExperimentalServerProfile } from "../src/cli/experimental/server-profile.ts";
const directories = new Set<string>();
const FIRST_SERVER_ID = "00000000-0000-4000-8000-000000000001";
const SECOND_SERVER_ID = "00000000-0000-4000-8000-000000000002";
async function makeDirectory(): Promise<string> {
const directory = await mkdtemp(join(tmpdir(), "psp-"));
@ -34,10 +36,22 @@ describe("experimental server profile", () => {
const second = await pendingSecond;
expect(secondAcquired).toBe(true);
expect(second.serverId).toBe(first.serverId);
expect((await readFile(join(directory, "default-server-id"), "utf8")).trim()).toBe(first.serverId);
await second.release();
});
test("does not serialize or share identity across server profiles", async () => {
test("does not serialize different server IDs in one directory", async () => {
const directory = await makeDirectory();
const [first, second] = await Promise.all([
acquireExperimentalServerProfile(directory, FIRST_SERVER_ID),
acquireExperimentalServerProfile(directory, SECOND_SERVER_ID),
]);
expect(first.serverId).toBe(FIRST_SERVER_ID);
expect(second.serverId).toBe(SECOND_SERVER_ID);
await Promise.all([first.release(), second.release()]);
});
test("does not share the default identity across server directories", async () => {
const firstDirectory = await makeDirectory();
const secondDirectory = await makeDirectory();
@ -49,11 +63,19 @@ describe("experimental server profile", () => {
await Promise.all([first.release(), second.release()]);
});
test("rejects corrupt identity and releases the launcher lock", async () => {
test("rejects a corrupt default identity", async () => {
const directory = await makeDirectory();
await writeFile(join(directory, "server-id"), "invalid\n");
await writeFile(join(directory, "default-server-id"), "invalid\n");
await expect(acquireExperimentalServerProfile(directory)).rejects.toThrow(/Invalid experimental server identity/);
await expect(access(join(directory, ".launcher.lock"))).rejects.toMatchObject({ code: "ENOENT" });
await expect(acquireExperimentalServerProfile(directory)).rejects.toThrow(
/Invalid default experimental server identity/,
);
});
test("rejects an invalid explicit server ID", async () => {
const directory = await makeDirectory();
await expect(acquireExperimentalServerProfile(directory, "invalid")).rejects.toThrow(
"Invalid experimental server ID",
);
});
});

2
packages/coding-agent/test/experimental-server-replacement.test.ts
View file

@ -113,7 +113,7 @@ describe.skipIf(process.platform === "win32")("experimental CLI server replaceme
);
const firstSocket = await waitForOutput(first, /Socket: (.+\.sock)/);
expect(firstSocket[1]).toBe(join(serverDir, `${firstIdentity[1]}.sock`));
expect((await lstat(join(serverDir, "control.sock"))).isSocket()).toBe(true);
expect((await lstat(join(serverDir, `control-${firstIdentity[1]}.sock`))).isSocket()).toBe(true);
await runExperimentalClient({
command: "client",
sessionId: "demo-1",