diff --git a/packages/coding-agent/README.md b/packages/coding-agent/README.md index e6e43c2eb..1b28ff65e 100644 --- a/packages/coding-agent/README.md +++ b/packages/coding-agent/README.md @@ -672,6 +672,7 @@ pi --thinking high "Solve this complex problem" | `PI_CODING_AGENT_DIR` | Override config directory (default: `~/.pi/agent`) | | `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_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 | diff --git a/packages/coding-agent/docs/environment-variables.md b/packages/coding-agent/docs/environment-variables.md index 29c324f63..9d79d2a25 100644 --- a/packages/coding-agent/docs/environment-variables.md +++ b/packages/coding-agent/docs/environment-variables.md @@ -81,6 +81,7 @@ These variables are read by Pi itself: | `PI_CODING_AGENT_DIR` | Override the config directory; default is `~/.pi/agent` | | `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_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` | diff --git a/packages/coding-agent/src/cli/args.ts b/packages/coding-agent/src/cli/args.ts index 3c0e01b3c..09a3549d8 100644 --- a/packages/coding-agent/src/cli/args.ts +++ b/packages/coding-agent/src/cli/args.ts @@ -412,6 +412,7 @@ ${chalk.bold("Environment Variables:")} ${ENV_AGENT_DIR.padEnd(32)} - Config directory (default: ~/${CONFIG_DIR_NAME}/agent) ${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_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/) diff --git a/packages/coding-agent/src/cli/experimental/coordinator-client.ts b/packages/coding-agent/src/cli/experimental/coordinator-client.ts index 177048a1f..bdce86193 100644 --- a/packages/coding-agent/src/cli/experimental/coordinator-client.ts +++ b/packages/coding-agent/src/cli/experimental/coordinator-client.ts @@ -1,7 +1,6 @@ import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; import { createConnection, type Socket } from "node:net"; -import { join } from "node:path"; import { fileURLToPath } from "node:url"; import Type, { type Static } from "typebox"; import { Check } from "typebox/value"; @@ -176,14 +175,6 @@ export class CoordinatorServer { } } -export function getExperimentalCoordinatorControlPath(serverId: string, socketDirectory: string): string { - return join(socketDirectory, `.c-${serverId.slice(0, 12)}.sock`); -} - -export function getExperimentalServerPath(socketDirectory: string): string { - return join(socketDirectory, `.s-${randomUUID().slice(0, 12)}.sock`); -} - export interface CoordinatorStartupLease { close(): void; } diff --git a/packages/coding-agent/src/cli/experimental/runtime.ts b/packages/coding-agent/src/cli/experimental/runtime.ts index 8e2c0a43a..83fa5ed0a 100644 --- a/packages/coding-agent/src/cli/experimental/runtime.ts +++ b/packages/coding-agent/src/cli/experimental/runtime.ts @@ -1,6 +1,4 @@ import { randomUUID } from "node:crypto"; -import { chmod, lstat, mkdir } from "node:fs/promises"; -import { homedir } from "node:os"; import { basename, join } from "node:path"; import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-agent-core"; import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; @@ -17,14 +15,11 @@ import { CoordinatorServer, type CoordinatorStartupLease, ensureExperimentalCoordinator, - getExperimentalCoordinatorControlPath, - getExperimentalServerPath, } from "./coordinator-client.ts"; +import { ensurePrivateServerDirectory, resolveExperimentalServerDirectory } from "./server-directory.ts"; import { acquireExperimentalServerProfile } from "./server-profile.ts"; import { startExperimentalSessionWorker } from "./session-worker.ts"; -const EXPERIMENTAL_SOCKET_ROOT = "/tmp"; - export interface ExperimentalServer { readonly serverId: string; readonly sessionDir: string; @@ -43,7 +38,7 @@ export type ExperimentalClientResult = | { readonly kind: "attached"; readonly serverId: string; readonly sessionId: string }; export interface StartExperimentalServerOptions { - /** Directory for server-addressed Unix sockets. Defaults to a short, private per-user runtime directory. */ + /** Directory for server sockets. Defaults to PI_SERVER_DIR or ~/.pi/server. */ readonly directory?: string; readonly path?: string; readonly serverId?: string; @@ -52,16 +47,14 @@ export interface StartExperimentalServerOptions { } export interface StartExperimentalCoordinatedServerOptions { - /** Persistent profile directory. Defaults to ~/.pi/server. */ + /** Server profile and socket directory. Defaults to PI_SERVER_DIR or ~/.pi/server. */ readonly directory?: string; - /** Physical socket directory. Defaults to a short, private per-user runtime directory. */ - readonly socketDirectory?: string; /** Durable session directory. Defaults to the experimental directory under the configured agent directory. */ readonly sessionDir?: string; } export interface RunExperimentalClientOptions { - /** Directory searched when --connect is omitted. Defaults to the experimental per-user runtime directory. */ + /** Directory searched when --connect is omitted. Defaults to PI_SERVER_DIR or ~/.pi/server. */ readonly directory?: string; } @@ -86,7 +79,10 @@ export async function startExperimentalServer( trackedSessions: [], createHarness: async (metadata) => { const sessionId = metadata.id; - const worker = await startExperimentalSessionWorker(metadata, { sessionDir }); + const worker = await startExperimentalSessionWorker(metadata, { + sessionDir, + controlDirectory: options.directory, + }); workerPids.set(sessionId, worker.pid); return { terminated: worker.terminated.then((error) => { @@ -126,9 +122,9 @@ async function startExperimentalServerBackend( }; let socketPath = options.path; if (socketPath === undefined) { - const socketDirectory = options.directory ?? getExperimentalSocketDirectory(); - await ensurePrivateSocketDirectory(socketDirectory); - socketPath = getUnixSocketPath(serverId, socketDirectory); + const serverDirectory = resolveExperimentalServerDirectory(options.directory); + await ensurePrivateServerDirectory(serverDirectory); + socketPath = getUnixSocketPath(serverId, serverDirectory); } const closeCatalog = async (): Promise => { const cleanup = await Promise.allSettled([repo.close(), executionEnv.cleanup()]); @@ -181,8 +177,7 @@ async function startExperimentalServerBackend( export async function startExperimentalCoordinatedServer( options: StartExperimentalCoordinatedServerOptions = {}, ): Promise { - const directory = options.directory ?? join(homedir(), ".pi", "server"); - const socketDirectory = options.socketDirectory ?? getExperimentalSocketDirectory(); + const directory = resolveExperimentalServerDirectory(options.directory); const { serverId, release } = await acquireExperimentalServerProfile(directory); let backend: ExperimentalServer | undefined; let coordinator: CoordinatorServer | undefined; @@ -190,10 +185,10 @@ export async function startExperimentalCoordinatedServer( let workers: CoordinatedSessionWorkers | undefined; let released = false; try { - await ensurePrivateSocketDirectory(socketDirectory); - const socketPath = getUnixSocketPath(serverId, socketDirectory); - const controlPath = getExperimentalCoordinatorControlPath(serverId, socketDirectory); - const serverPath = getExperimentalServerPath(socketDirectory); + await ensurePrivateServerDirectory(directory); + const socketPath = getUnixSocketPath(serverId, directory); + const controlPath = join(directory, "control.sock"); + const serverPath = join(directory, `.server-${randomUUID().slice(0, 12)}.sock`); startupLease = await ensureExperimentalCoordinator(socketPath, controlPath); coordinator = new CoordinatorServer({ controlPath, endpoint: serverPath }); const sessionDir = resolveExperimentalSessionDirectory(options.sessionDir); @@ -269,7 +264,7 @@ export async function runExperimentalClient( if (command.auth !== undefined) throw new Error("Authentication is not supported by the experimental local server"); const routes = command.connect ? [routeFromExplicitPath(command.connect.path)] - : await discoverUnixServers({ directory: options.directory ?? getExperimentalSocketDirectory() }); + : await discoverUnixServers({ directory: resolveExperimentalServerDirectory(options.directory) }); const discovered: { route: UnixServerRoute; sessionIds: string[] }[] = []; for (const route of routes) { @@ -324,20 +319,3 @@ function routeFromExplicitPath(path: string): UnixServerRoute { } return { serverId, path }; } - -function getExperimentalSocketDirectory(): string { - if (process.platform === "win32" || typeof process.getuid !== "function") { - throw new Error("Experimental Unix server transport requires a POSIX user ID"); - } - return join(EXPERIMENTAL_SOCKET_ROOT, `pi-server-${process.getuid()}`); -} - -async function ensurePrivateSocketDirectory(directory: string): Promise { - if (typeof process.getuid !== "function") throw new Error("Unix socket directory requires a POSIX user ID"); - await mkdir(directory, { recursive: true, mode: 0o700 }); - const stats = await lstat(directory); - if (!stats.isDirectory()) throw new Error(`Unix socket directory is not a directory: ${directory}`); - if (stats.uid !== process.getuid()) - throw new Error(`Unix socket directory is not owned by the current user: ${directory}`); - await chmod(directory, 0o700); -} diff --git a/packages/coding-agent/src/cli/experimental/server-directory.ts b/packages/coding-agent/src/cli/experimental/server-directory.ts new file mode 100644 index 000000000..f8d31ec85 --- /dev/null +++ b/packages/coding-agent/src/cli/experimental/server-directory.ts @@ -0,0 +1,21 @@ +import { chmod, lstat, mkdir } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { resolvePath } from "../../utils/paths.ts"; + +export const ENV_SERVER_DIR = "PI_SERVER_DIR"; + +export function resolveExperimentalServerDirectory(directory?: string): string { + return resolvePath(directory ?? process.env[ENV_SERVER_DIR] ?? join(homedir(), ".pi", "server")); +} + +export async function ensurePrivateServerDirectory(directory: string): Promise { + if (typeof process.getuid !== "function") throw new Error("Unix socket directory requires a POSIX user ID"); + await mkdir(directory, { recursive: true, mode: 0o700 }); + const stats = await lstat(directory); + if (!stats.isDirectory()) throw new Error(`Unix socket directory is not a directory: ${directory}`); + if (stats.uid !== process.getuid()) { + throw new Error(`Unix socket directory is not owned by the current user: ${directory}`); + } + await chmod(directory, 0o700); +} diff --git a/packages/coding-agent/src/cli/experimental/session-worker.ts b/packages/coding-agent/src/cli/experimental/session-worker.ts index f6e6cfe25..e01da838d 100644 --- a/packages/coding-agent/src/cli/experimental/session-worker.ts +++ b/packages/coding-agent/src/cli/experimental/session-worker.ts @@ -2,12 +2,12 @@ import { type ChildProcess, spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; import { chmod, unlink } from "node:fs/promises"; import { createServer, type Server, type Socket } from "node:net"; -import { tmpdir } from "node:os"; import { isAbsolute, join } from "node:path"; import { fileURLToPath } from "node:url"; import type { JsonlSessionMetadata } from "@earendil-works/pi-agent-core"; import Type, { type Static } from "typebox"; import { Check } from "typebox/value"; +import { ensurePrivateServerDirectory, resolveExperimentalServerDirectory } from "./server-directory.ts"; import type { SessionWorkerCommand, SessionWorkerEvent } from "./session-worker-process.ts"; const DEFAULT_STARTUP_TIMEOUT_MS = 10_000; @@ -47,6 +47,7 @@ export interface SessionWorkerLaunchSpec { export interface StartExperimentalSessionWorkerOptions { readonly sessionDir: string; + readonly controlDirectory?: string; readonly startupTimeoutMs?: number; readonly shutdownTimeoutMs?: number; readonly workerUrl?: URL; @@ -90,8 +91,14 @@ export async function startExperimentalSessionWorker( const launch = createExperimentalSessionWorkerLaunchSpec(metadata, options); const token = randomUUID(); const controlId = randomUUID(); - const controlAddress = - process.platform === "win32" ? `\\\\.\\pipe\\pi-worker-${controlId}` : join(tmpdir(), `pi-w-${controlId}.sock`); + let controlAddress: string; + if (process.platform === "win32") { + controlAddress = `\\\\.\\pipe\\pi-worker-${controlId}`; + } else { + const controlDirectory = resolveExperimentalServerDirectory(options.controlDirectory); + await ensurePrivateServerDirectory(controlDirectory); + controlAddress = join(controlDirectory, `.worker-${controlId}.sock`); + } const control = await createWorkerControlServer(controlAddress, token, metadata.id); let child: ChildProcess; try { diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index cb5614b34..4b9267938 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -64,6 +64,27 @@ describe("experimental durable server composition", () => { expect(runtime.workerPids.size).toBe(0); }); + test("uses PI_SERVER_DIR for sockets and discovery", async () => { + const directory = await mkdtemp(join("/tmp", "pi-server-dir-")); + directories.add(directory); + vi.stubEnv("PI_SERVER_DIR", directory); + const runtime = await startExperimentalCoordinatedServer(); + servers.add(runtime); + + expect(runtime.socketPath).toBe(join(directory, `${runtime.serverId}.sock`)); + expect((await lstat(directory)).mode & 0o777).toBe(0o700); + const publicSocket = await lstat(runtime.socketPath); + const controlSocket = await lstat(join(directory, "control.sock")); + expect(publicSocket.isSocket()).toBe(true); + expect(publicSocket.mode & 0o777).toBe(0o600); + expect(controlSocket.isSocket()).toBe(true); + expect(controlSocket.mode & 0o777).toBe(0o600); + await expect(runExperimentalClient({ command: "client" })).resolves.toMatchObject({ + kind: "list", + sessions: [{ sessionId: "demo-1" }, { sessionId: "demo-2" }], + }); + }); + test("discovers and lists seeded sessions without hosting either session", async () => { const { directory, runtime } = await makeServer(); @@ -135,7 +156,7 @@ describe("experimental durable server composition", () => { test("coordinated server replaces an exited worker on the next attach", async () => { const directory = await mkdtemp(join("/tmp", "pew-")); directories.add(directory); - const runtime = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + const runtime = await startExperimentalCoordinatedServer({ directory }); servers.add(runtime); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); const firstPid = runtime.workerPids.get("demo-1"); @@ -152,7 +173,7 @@ describe("experimental durable server composition", () => { test("normal coordinated shutdown stops detached workers", async () => { const directory = await mkdtemp(join("/tmp", "pes-")); directories.add(directory); - const runtime = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + const runtime = await startExperimentalCoordinatedServer({ directory }); servers.add(runtime); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); const pid = runtime.workerPids.get("demo-1"); @@ -168,8 +189,8 @@ describe("experimental durable server composition", () => { const directory = await mkdtemp(join("/tmp", "pel-")); directories.add(directory); const runtimes = await Promise.all([ - startExperimentalCoordinatedServer({ directory, socketDirectory: directory }), - startExperimentalCoordinatedServer({ directory, socketDirectory: directory }), + startExperimentalCoordinatedServer({ directory }), + startExperimentalCoordinatedServer({ directory }), ]); for (const runtime of runtimes) servers.add(runtime); @@ -196,7 +217,7 @@ describe("experimental durable server composition", () => { 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 startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + const first = await startExperimentalCoordinatedServer({ directory }); servers.add(first); const client = await PiClient.connect({ serverId: first.serverId, @@ -204,7 +225,7 @@ describe("experimental durable server composition", () => { }); await client.listSessions(); - const replacement = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + const replacement = await startExperimentalCoordinatedServer({ directory }); servers.add(replacement); await first.closed; await expect(runExperimentalClient({ command: "client" }, { directory })).resolves.toMatchObject({ @@ -217,19 +238,13 @@ describe("experimental durable server composition", () => { test("discovers workers after replacing the server", async () => { const firstDirectory = await mkdtemp(join("/tmp", "per-")); directories.add(firstDirectory); - const first = await startExperimentalCoordinatedServer({ - directory: firstDirectory, - socketDirectory: firstDirectory, - }); + const first = await startExperimentalCoordinatedServer({ directory: firstDirectory }); servers.add(first); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory: firstDirectory }); const firstWorkerPid = first.workerPids.get("demo-1"); expect(firstWorkerPid).toEqual(expect.any(Number)); - const replacement = await startExperimentalCoordinatedServer({ - directory: firstDirectory, - socketDirectory: firstDirectory, - }); + const replacement = await startExperimentalCoordinatedServer({ directory: firstDirectory }); servers.add(replacement); await first.closed; @@ -257,14 +272,13 @@ describe("experimental durable server composition", () => { const emptySessionDir = await mkdtemp(join("/tmp", "pet-sessions-")); directories.add(directory); directories.add(emptySessionDir); - const first = await startExperimentalCoordinatedServer({ directory, socketDirectory: directory }); + const first = await startExperimentalCoordinatedServer({ directory }); servers.add(first); await runExperimentalClient({ command: "client", sessionId: "demo-1" }, { directory }); const workerPid = first.workerPids.get("demo-1"); const replacement = await startExperimentalCoordinatedServer({ directory, - socketDirectory: directory, sessionDir: emptySessionDir, }); servers.add(replacement); diff --git a/packages/coding-agent/test/experimental-server-replacement.test.ts b/packages/coding-agent/test/experimental-server-replacement.test.ts index d482ec574..c6cc08275 100644 --- a/packages/coding-agent/test/experimental-server-replacement.test.ts +++ b/packages/coding-agent/test/experimental-server-replacement.test.ts @@ -1,7 +1,6 @@ import { type ChildProcess, spawn } from "node:child_process"; import { once } from "node:events"; -import { mkdir, mkdtemp, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; +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"; @@ -18,10 +17,16 @@ interface RunningCli { readonly errors: () => string; } -async function startServer(home: string, agentDir: string): Promise { +async function startServer(home: string, agentDir: string, serverDir: string): Promise { const child = spawn(process.execPath, ["--import", "tsx", cliPath, "server"], { cwd: fileURLToPath(new URL("../../..", import.meta.url)), - env: { ...process.env, HOME: home, PI_CODING_AGENT_DIR: agentDir, PI_EXPERIMENTAL: "1" }, + env: { + ...process.env, + HOME: home, + PI_CODING_AGENT_DIR: agentDir, + PI_EXPERIMENTAL: "1", + PI_SERVER_DIR: serverDir, + }, stdio: ["ignore", "pipe", "pipe"], }); processes.add(child); @@ -93,28 +98,29 @@ afterEach(async () => { 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(tmpdir(), "pcs-")); + 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); + 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]).not.toContain(home); - expect(Buffer.byteLength(firstSocket[1]!)).toBeLessThanOrEqual(103); + expect(firstSocket[1]).toBe(join(serverDir, `${firstIdentity[1]}.sock`)); + expect((await lstat(join(serverDir, "control.sock"))).isSocket()).toBe(true); await runExperimentalClient({ command: "client", sessionId: "demo-1", connect: { transport: "unix", path: firstSocket[1]! }, }); - const replacement = await startServer(home, agentDir); + 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})/, diff --git a/packages/coding-agent/test/experimental-session-worker.test.ts b/packages/coding-agent/test/experimental-session-worker.test.ts index da1e3f4c4..93125a52e 100644 --- a/packages/coding-agent/test/experimental-session-worker.test.ts +++ b/packages/coding-agent/test/experimental-session-worker.test.ts @@ -2,7 +2,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { join } from "node:path"; import { type JsonlSessionMetadata, JsonlSessionRepo } from "@earendil-works/pi-agent-core"; import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; -import { afterEach, describe, expect, test, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { type ExperimentalSessionWorker, startExperimentalSessionWorker, @@ -14,6 +14,12 @@ const fixtureUrl = new URL("fixtures/session-worker-fixture.ts", import.meta.url const sessionDir = "/tmp/pi-session-worker-tests"; const directories = new Set(); +beforeEach(async () => { + const serverDir = await mkdtemp(join("/tmp", "pi-session-worker-server-")); + directories.add(serverDir); + vi.stubEnv("PI_SERVER_DIR", serverDir); +}); + function testMetadata(id: string): JsonlSessionMetadata { return { id,