diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index f0ec06063b..2200248f87 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] +- Fixed resident daemon workers retaining their permitted launch environment across supervisor restarts while keeping client-owned credentials out of descriptor files. - Added a configurable copy action to login dialogs so raw sign-in URLs can be copied without selecting wrapped text ([#643](https://github.com/PrimeIntellect-ai/prime-agent/issues/643)). - Added privacy-safe pseudonymous product analytics for onboarding, command use, execution modes, run outcomes, TTFT, latency, usage, tools, retries, and compactions, with disclosure and opt-out controls ([ENG-4682](https://linear.app/primeintellect/issue/ENG-4682/add-privacy-safe-posthog-analytics-to-prime-agent)). - Changed sent agent messages in the IPython cell UI to show only the message text with a `╰─` gutter when expanded, matching received messages, and hid the raw `agent_message.send` receipt dictionary. diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 6315d0ad7e..909d36a650 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -193,6 +193,15 @@ function toPrintOutputMode(appMode: AppMode): Exclude { const connection = await DaemonAgentConnection.attach(client, getDaemonSummaryActiveSessionId(summary), { closeClientOnDispose: true, sendClientEnv: true, - ownedSession: options.clientOwned, + ownedSession: clientOwned, supportsExtensionUi: options.supportsExtensionUi, recoverDaemon: () => ensureInteractiveDaemonRunning(options.socketPath), telemetryDisabled: options.config.telemetryDisabled, @@ -972,7 +983,7 @@ async function createDaemonClientConnection(options: { return await attach(summary); } - if (options.sessionPath && !options.clientOwned) { + if (options.sessionPath && !clientOwned) { const activeSummary = findActiveDaemonSessionSummaryForSessionFile( await listActiveDaemonSessionSummaries(client), options.sessionPath, @@ -981,8 +992,7 @@ async function createDaemonClientConnection(options: { return await attach(activeSummary); } } - if (options.clientOwned) { - await client.waitForHello(); + if (clientOwned) { if (!client.supportsServerCapability("client_owned_sessions")) { throw new DaemonCapabilityUnavailableError("create", "client_owned_sessions"); } @@ -995,8 +1005,13 @@ async function createDaemonClientConnection(options: { continueRecent: options.continueRecent, noSession: options.noSession, env: collectDaemonClientEnv(), - lifecycle: options.clientOwned ? "client_owned" : "resident", - launchEnv: options.clientOwned ? collectDaemonLaunchEnv() : undefined, + lifecycle: clientOwned ? "client_owned" : "resident", + // Forward the caller's environment for BOTH lifecycles. A resident + // worker still has to be launched with the caller's env: an embedder + // such as the verifiers ACP harness passes the model endpoint, its + // bearer token, and proxy settings that way, and a worker started + // without them cannot reach the model at all. + launchEnv: collectDaemonLaunchEnv(), }); if (!response.success) { throw deserializeDaemonError(response); @@ -1530,7 +1545,8 @@ export async function main(args: string[], options?: MainOptions) { config: defaultSessionConfig, sessionPath: parsed.noSession ? undefined : sessionManager.getSessionFile(), continueRecent: parsed.continue, - clientOwned: true, + // A no-session ACP invocation has nothing to reattach to; complete its worker on disconnect. + clientOwned: isClientOwnedDaemonSession(appMode, parsed.noSession), noSession: parsed.noSession, supportsExtensionUi: appMode === "rpc", })); diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index 02b1ad5e4d..bf2b4ab9dd 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -1,5 +1,6 @@ import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { ImageContent, ServiceTier, TextContent, Transport } from "@earendil-works/pi-ai"; +import { ENV_AGENT_DIR } from "../../config.js"; import type { AgentSessionMessageDeliveryMode, AgentSessionMessageReceipt, @@ -198,6 +199,56 @@ export function collectDaemonClientEnv(source: NodeJS.ProcessEnv = process.env): return Object.keys(env).length > 0 ? env : undefined; } +/** + * Non-secret launch settings that may survive a supervisor restart in a + * resident worker descriptor. Model credentials deliberately do not belong + * here: the first worker launch inherits the caller environment, but a JSON + * descriptor must never become an at-rest copy of a caller's credentials. + */ +export const DAEMON_PERSISTED_LAUNCH_ENV_KEYS = [ + // Process/runtime locations needed to relaunch the same installed CLI. + "HOME", + "PATH", + "TMPDIR", + "TMP", + "TEMP", + "XDG_CACHE_HOME", + "XDG_CONFIG_HOME", + "XDG_DATA_HOME", + "XDG_RUNTIME_DIR", + "XDG_STATE_HOME", + // ENV_AGENT_DIR is the current application's configurable agent directory. + // PI_CODING_AGENT_DIR remains for compatibility with the upstream CLI. + ENV_AGENT_DIR, + "PI_CODING_AGENT_DIR", + // Deliberately non-secret Prime Agent behavior, telemetry, and package settings. + "PI_OFFLINE", + "PI_PACKAGE_DIR", + "PI_SKIP_VERSION_CHECK", + "DO_NOT_TRACK", + "PRIME_AGENT_TELEMETRY", + "PRIME_AGENT_TELEMETRY_ENDPOINT", + "PRIME_AGENT_TRACES_BASE_URL", + "PRIME_AGENT_DOWNLOAD_BASE_URL", +] as const; + +/** Select the explicitly non-secret launch settings safe to persist on disk. */ +export function filterPersistedDaemonLaunchEnv( + source: Readonly> | undefined, +): Record | undefined { + if (!source) return undefined; + const env: Record = {}; + for (const key of DAEMON_PERSISTED_LAUNCH_ENV_KEYS) { + const value = source[key]; + if (value !== undefined) env[key] = value; + } + return Object.keys(env).length > 0 ? env : undefined; +} + +/** + * Collect the caller environment for the initial worker spawn. The + * supervisor filters it before it is written to a resident-worker descriptor. + */ export function collectDaemonLaunchEnv(source: NodeJS.ProcessEnv = process.env): Record { const env: Record = {}; for (const [key, value] of Object.entries(source)) { diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index b22701bbcb..87b1576c47 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -54,6 +54,7 @@ import { DAEMON_CATALOG_ROLE_ENV, DaemonCatalogClient } from "./daemon-catalog-p import { deserializeDaemonError, serializeDaemonError } from "./daemon-errors.js"; import { collectDaemonClientEnv, + collectDaemonLaunchEnv, createDaemonEventMeta, DAEMON_COMMAND_COMPATIBILITY, DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION, @@ -71,6 +72,7 @@ import { type DaemonResponse, type DaemonUpdateRestartManifest, failure, + filterPersistedDaemonLaunchEnv, isDaemonCommandEnvelope, isDaemonMutatingCommand, salvageDaemonCommandId, @@ -416,6 +418,12 @@ function isDaemonWorkerDescriptor(value: unknown, socketPath: string): value is (descriptor.pid ?? 0) > 0 && (descriptor.processStartId === undefined || typeof descriptor.processStartId === "string") && (descriptor.ownerClientId === undefined || typeof descriptor.ownerClientId === "string") && + (descriptor.launchEnv === undefined || + (typeof descriptor.launchEnv === "object" && + descriptor.launchEnv !== null && + Object.entries(descriptor.launchEnv).every( + ([key, value]) => typeof key === "string" && typeof value === "string", + ))) && typeof descriptor.socketPath === "string" && typeof descriptor.authenticationToken === "string" && typeof descriptor.rootActiveSessionId === "string" && @@ -923,10 +931,12 @@ export class DaemonSupervisor { if (!isDaemonWorkerDescriptor(descriptor, this.socketPath)) { continue; } + const storedLaunchEnv = descriptor.launchEnv; + descriptor.launchEnv = filterPersistedDaemonLaunchEnv(storedLaunchEnv); descriptor.lifecycle = "recovering"; descriptor.recoveryJournalPath ??= join(this.descriptorDir, `${descriptor.workerId}.recovery.jsonl`); descriptor.orphanProcessJournalPath ??= join(this.descriptorDir, `${descriptor.workerId}.orphans.jsonl`); - this.workers.set(descriptor.workerId, { + const worker: ResidentWorker = { descriptor, descriptorPath: path, summaries: new Map(), @@ -936,7 +946,12 @@ export class DaemonSupervisor { snapshotLoads: new Map(), intentionalStop: descriptor.stopRequestedAt !== undefined, stopRevision: 0, - }); + launchEnv: descriptor.launchEnv, + }; + this.workers.set(descriptor.workerId, worker); + if (JSON.stringify(storedLaunchEnv) !== JSON.stringify(descriptor.launchEnv)) { + this.persistWorker(worker); + } } catch (error) { this.log(`Ignoring invalid worker descriptor ${path}: ${String(error)}`); } @@ -2070,7 +2085,7 @@ export class DaemonSupervisor { throw new Error("Session is not owned by this client"); } const previousDescriptor = worker.descriptor; - worker.descriptor = { ...previousDescriptor, ownerClientId: undefined }; + worker.descriptor = { ...previousDescriptor, ownerClientId: undefined, launchEnv: undefined }; try { this.persistWorker(worker); } catch (error) { @@ -2096,8 +2111,29 @@ export class DaemonSupervisor { throw new Error(`Session worker ${existing.descriptor.workerId} recovery was cancelled`); } const recoveryStopRevision = existing?.stopRevision; - const launchEnv = - ownerClientId || existing?.descriptor.ownerClientId ? (command.launchEnv ?? existing?.launchEnv) : undefined; + const ownerClientIdForDescriptor = existing?.descriptor.ownerClientId ?? ownerClientId; + // Only a first resident launch consumes the caller's full transient + // environment. A resident recovery uses the descriptor's allowlisted copy + // even while the old worker object still exists in memory. Client-owned + // workers are different: their reconnecting owner supplies fresh transient + // launch settings, which are never written to a descriptor. + const launchEnv = existing + ? ownerClientIdForDescriptor === undefined + ? existing.descriptor.launchEnv + : existing.launchEnv + : command.launchEnv; + // Only non-secret, explicitly allowed settings are durable. The initial + // spawn may still receive caller credentials through launchEnv, but those + // credentials must never be serialized into a worker descriptor. + const persistedLaunchEnv = filterPersistedDaemonLaunchEnv(launchEnv); + // A replacement supervisor can itself have been restarted by the old worker + // and therefore inherit that worker's original credentials. Automatic + // resident recovery must not copy those ambient secrets into the replacement + // worker. Client-owned recovery instead uses its live owner's transient env. + const inheritedEnv = + existing && ownerClientIdForDescriptor === undefined + ? filterPersistedDaemonLaunchEnv(collectDaemonLaunchEnv(process.env)) + : process.env; const createCommand: DaemonCreateCommand = { ...withoutSupervisorCreateFields(command), config: mergeAgentSessionRuntimeConfig(this.defaultSessionConfig, command.config), @@ -2118,7 +2154,7 @@ export class DaemonSupervisor { cwd: createCommand.config?.cwd ?? process.cwd(), detached: true, env: createCliSubprocessEnv({ - ...process.env, + ...inheritedEnv, ...launchEnv, [DAEMON_WORKER_ROLE_ENV]: "1", [DAEMON_WORKER_TOKEN_ENV]: token, @@ -2174,7 +2210,10 @@ export class DaemonSupervisor { supervisorSocketPath: this.socketPath, authenticationToken: token, rootActiveSessionId, - ownerClientId: existing?.descriptor.ownerClientId ?? ownerClientId, + ownerClientId: ownerClientIdForDescriptor, + ...(ownerClientIdForDescriptor === undefined && persistedLaunchEnv + ? { launchEnv: persistedLaunchEnv } + : {}), createdAt: existing?.descriptor.createdAt ?? now, updatedAt: now, lifecycle: "starting", @@ -2195,7 +2234,10 @@ export class DaemonSupervisor { }; await this.assertRecoveryAllowed(); worker.descriptor = descriptor; - worker.launchEnv = launchEnv; + // Resident workers retain only the durable allowlist even in memory, so an + // automatic same-supervisor recovery cannot resurrect initial credentials. + // Client-owned workers may retain a fresh owner's transient environment. + worker.launchEnv = ownerClientIdForDescriptor === undefined ? persistedLaunchEnv : launchEnv; descriptorAssigned = true; this.persistWorker(worker); worker.intentionalStop = false; diff --git a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts index c7f1495427..799937848f 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts @@ -96,6 +96,8 @@ export interface DaemonWorkerDescriptor { rootActiveSessionId: string; /** Stable protocol client that owns this worker. Omitted for resident sessions. */ ownerClientId?: string; + /** Environment required to relaunch a resident worker after supervisor restart. */ + launchEnv?: Record; rootSessionId?: string; sessionFile?: string; createdAt: string; diff --git a/packages/coding-agent/test/acp-resident-lifecycle.test.ts b/packages/coding-agent/test/acp-resident-lifecycle.test.ts new file mode 100644 index 0000000000..47ee4923f8 --- /dev/null +++ b/packages/coding-agent/test/acp-resident-lifecycle.test.ts @@ -0,0 +1,321 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import { once } from "node:events"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { createServer, type Server } from "node:http"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { ENV_AGENT_DIR } from "../src/config.js"; +import { isClientOwnedDaemonSession } from "../src/main.js"; +import { DaemonClient } from "../src/modes/daemon/daemon-client.js"; + +const cliPath = resolve(__dirname, "../src/cli.ts"); +const tsxPath = resolve(__dirname, "../../../node_modules/tsx/dist/cli.mjs"); +const temporaryRoots: string[] = []; +const daemonSockets: string[] = []; +const servers: Server[] = []; +const children = new Set(); + +function record(value: unknown): Record | undefined { + return value !== null && typeof value === "object" ? (value as Record) : undefined; +} + +async function stopChild(child: ChildProcess): Promise { + if (child.exitCode !== null || child.signalCode !== null) return; + const exited = once(child, "exit").then(() => undefined); + const exitedGracefully = await Promise.race([ + exited.then(() => true), + new Promise((resolveTimeout) => setTimeout(() => resolveTimeout(false), 15_000)), + ]); + if (exitedGracefully) return; + child.kill("SIGTERM"); + const stopped = await Promise.race([ + exited.then(() => true), + new Promise((resolveTimeout) => setTimeout(() => resolveTimeout(false), 5_000)), + ]); + if (!stopped) child.kill("SIGKILL"); +} + +async function shutdownDaemon(socketPath: string): Promise { + const client = new DaemonClient(socketPath); + try { + await client.connect(250); + await client.request({ type: "shutdown" }, 5_000); + } catch { + // A failed ACP startup may not have created a daemon. + } finally { + client.close(); + } +} + +afterEach(async () => { + for (const child of children) await stopChild(child); + children.clear(); + for (const socketPath of daemonSockets.splice(0)) await shutdownDaemon(socketPath); + for (const server of servers.splice(0)) await new Promise((done) => server.close(() => done())); + for (const root of temporaryRoots.splice(0)) { + // Supervisor shutdown returns before its detached worker has fully released + // the kernel snapshot directory on every platform. + rmSync(root, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 }); + } +}); + +/** A small JSON-RPC client for a real ACP stdio process. */ +class AcpStdioClient { + private readonly pending = new Map< + number, + { resolve: (result: Record) => void; reject: (error: Error) => void } + >(); + private nextId = 1; + private stdoutBuffer = ""; + private stderr = ""; + + constructor(readonly child: ChildProcess) { + child.stdout?.on("data", (chunk: Buffer) => this.consumeStdout(chunk.toString("utf8"))); + child.stderr?.on("data", (chunk: Buffer) => { + this.stderr += chunk.toString("utf8"); + }); + child.once("exit", () => { + for (const pending of this.pending.values()) { + pending.reject(new Error(`ACP process exited before responding: ${this.stderr}`)); + } + this.pending.clear(); + }); + } + + async start(cwd: string): Promise { + await this.request("initialize", { protocolVersion: 1, clientCapabilities: {} }); + const created = await this.request("session/new", { cwd, mcpServers: [] }); + const sessionId = created.sessionId; + if (typeof sessionId !== "string") + throw new Error(`ACP did not create a usable session: ${JSON.stringify(created)}`); + return sessionId; + } + + async prompt(sessionId: string, text: string): Promise { + await this.request("session/prompt", { sessionId, prompt: [{ type: "text", text }] }); + } + + async close(): Promise { + this.child.stdin?.end(); + await stopChild(this.child); + } + + private request(method: string, params: Record): Promise> { + const id = this.nextId++; + return new Promise((resolveRequest, rejectRequest) => { + const timeout = setTimeout(() => { + this.pending.delete(id); + rejectRequest(new Error(`ACP ${method} timed out: ${this.stderr}`)); + }, 45_000); + this.pending.set(id, { + resolve: (result) => { + clearTimeout(timeout); + resolveRequest(result); + }, + reject: (error) => { + clearTimeout(timeout); + rejectRequest(error); + }, + }); + this.child.stdin?.write(`${JSON.stringify({ jsonrpc: "2.0", id, method, params })}\n`); + }); + } + + private consumeStdout(chunk: string): void { + this.stdoutBuffer += chunk; + const lines = this.stdoutBuffer.split("\n"); + this.stdoutBuffer = lines.pop() ?? ""; + for (const line of lines) { + let frame: Record | undefined; + try { + frame = record(JSON.parse(line)); + } catch { + continue; + } + if (!frame || typeof frame.id !== "number") continue; + const pending = this.pending.get(frame.id); + if (!pending) continue; + this.pending.delete(frame.id); + if (frame.error !== undefined) { + pending.reject(new Error(`ACP ${String(frame.id)} failed: ${JSON.stringify(frame.error)}`)); + } else { + pending.resolve(record(frame.result) ?? {}); + } + } + } +} + +function writeModels(agentDir: string, baseUrl: string): void { + writeFileSync( + join(agentDir, "models.json"), + JSON.stringify({ + providers: { + fixture: { + baseUrl, + api: "openai-completions", + apiKey: "fixture-key", + models: [{ id: "fixture/model", reasoning: false, input: ["text"] }], + }, + }, + }), + ); +} + +function writeSse(res: import("node:http").ServerResponse, body: Record): void { + res.write(`data: ${JSON.stringify(body)}\n\n`); +} + +async function startIpythonFixture(): Promise<{ baseUrl: string; sawLiveNamespace: () => boolean }> { + let sawLiveNamespace = false; + const server = createServer(async (req, res) => { + let body = ""; + for await (const chunk of req) body += chunk.toString(); + const request = record(JSON.parse(body)); + const messages = Array.isArray(request?.messages) ? request.messages.map(record).filter(Boolean) : []; + const toolOutputs = messages + .filter((message): message is Record => message?.role === "tool") + .map((message) => JSON.stringify(message.content)); + const readingAfterReconnect = JSON.stringify(messages).includes("read reconnect state"); + const hasCurrentToolOutput = readingAfterReconnect + ? toolOutputs.some((output) => output.includes("True")) + : toolOutputs.some((output) => output.includes("kernel state initialized")); + if (readingAfterReconnect && hasCurrentToolOutput) sawLiveNamespace = true; + + res.writeHead(200, { "content-type": "text/event-stream", connection: "keep-alive" }); + if (hasCurrentToolOutput) { + writeSse(res, { + id: "fixture-final", + object: "chat.completion.chunk", + created: 0, + model: "fixture/model", + choices: [{ index: 0, delta: { role: "assistant", content: "done" }, finish_reason: null }], + }); + writeSse(res, { + id: "fixture-final", + object: "chat.completion.chunk", + created: 0, + model: "fixture/model", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }); + } else { + const code = readingAfterReconnect + ? "print(id(acp_reconnect_marker) == acp_reconnect_marker_identity)" + : "acp_reconnect_marker = object()\nacp_reconnect_marker_identity = id(acp_reconnect_marker)\nprint('kernel state initialized')"; + writeSse(res, { + id: "fixture-tool", + object: "chat.completion.chunk", + created: 0, + model: "fixture/model", + choices: [ + { + index: 0, + delta: { + role: "assistant", + tool_calls: [ + { + index: 0, + id: readingAfterReconnect ? "read-kernel" : "set-kernel", + type: "function", + function: { name: "ipython", arguments: JSON.stringify({ code }) }, + }, + ], + }, + finish_reason: null, + }, + ], + }); + writeSse(res, { + id: "fixture-tool", + object: "chat.completion.chunk", + created: 0, + model: "fixture/model", + choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], + }); + } + res.write("data: [DONE]\n\n"); + res.end(); + }); + servers.push(server); + await new Promise((resolveListening) => server.listen(0, "127.0.0.1", resolveListening)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Fixture server did not expose a TCP port"); + return { baseUrl: `http://127.0.0.1:${address.port}/v1`, sawLiveNamespace: () => sawLiveNamespace }; +} + +function launchAcp(agentDir: string, projectDir: string, daemonSocket: string, resume: boolean): AcpStdioClient { + const child = spawn( + process.execPath, + [ + tsxPath, + cliPath, + "--mode", + "acp", + "--provider", + "fixture", + "--model", + "fixture/model", + ...(resume ? ["--continue"] : []), + "--offline", + "--daemon-socket", + daemonSocket, + ], + { + cwd: projectDir, + env: { + ...process.env, + [ENV_AGENT_DIR]: agentDir, + HOME: agentDir, + TSX_TSCONFIG_PATH: resolve(__dirname, "../../../tsconfig.json"), + }, + stdio: ["pipe", "pipe", "pipe"], + }, + ); + children.add(child); + return new AcpStdioClient(child); +} + +describe("ACP daemon lifecycle negotiation", () => { + it("keeps only reattachable ACP sessions resident", () => { + // Resident workers survive ACP stdio disconnect only when a later + // --continue can find their session file. All ephemeral or non-ACP modes + // are completed by their creating client. + expect(isClientOwnedDaemonSession("acp", false)).toBe(false); + expect(isClientOwnedDaemonSession("acp", true)).toBe(true); + expect(isClientOwnedDaemonSession("rpc", false)).toBe(true); + expect(isClientOwnedDaemonSession("print", false)).toBe(true); + }); + + it( + "preserves an ACP kernel's live Python namespace across client disconnect and re-attach", + { tags: ["kernel-heavy"], timeout: 240_000 }, + async () => { + const root = mkdtempSync(join(tmpdir(), "pi-acp-resident-")); + temporaryRoots.push(root); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const daemonSocket = join(root, "daemon.sock"); + daemonSockets.push(daemonSocket); + mkdirSync(agentDir, { recursive: true }); + mkdirSync(projectDir, { recursive: true }); + const fixture = await startIpythonFixture(); + writeModels(agentDir, fixture.baseUrl); + + const first = launchAcp(agentDir, projectDir, daemonSocket, false); + const firstSession = await first.start(projectDir); + await first.prompt(firstSession, "set reconnect state"); + await first.close(); + children.delete(first.child); + + const reattached = launchAcp(agentDir, projectDir, daemonSocket, true); + const reattachedSession = await reattached.start(projectDir); + await reattached.prompt(reattachedSession, "read reconnect state"); + await reattached.close(); + children.delete(reattached.child); + + // `object()` restores as a distinct object, so True proves this was the + // live kernel namespace rather than a replacement kernel revived from disk. + expect(fixture.sawLiveNamespace()).toBe(true); + }, + ); +}); diff --git a/packages/coding-agent/test/daemon-protocol.test.ts b/packages/coding-agent/test/daemon-protocol.test.ts index 4cf1e4c186..ae601204b3 100644 --- a/packages/coding-agent/test/daemon-protocol.test.ts +++ b/packages/coding-agent/test/daemon-protocol.test.ts @@ -16,6 +16,7 @@ import { DAEMON_SCHEMA_REVISION, type DaemonCommand, type DaemonOutbound, + filterPersistedDaemonLaunchEnv, getDaemonCommandCompatibilities, isDaemonCommandEnvelope, isDaemonMutatingCommand, @@ -44,6 +45,33 @@ describe("daemon protocol helpers", () => { expect(DAEMON_SCHEMA_ID).toBe(`protocol-${DAEMON_PROTOCOL_VERSION}-schema-${DAEMON_SCHEMA_REVISION}-${digest}`); }); + it("filters resident descriptor launch environment to explicit non-secret settings", () => { + expect( + filterPersistedDaemonLaunchEnv({ + HOME: "/home/agent", + PATH: "/runtime/bin", + TMPDIR: "/tmp/agent", + XDG_DATA_HOME: "/home/agent/.local/share", + PRIME_AGENT_CODING_AGENT_DIR: "/home/agent/.prime/agent", + PI_OFFLINE: "1", + PRIME_AGENT_TRACES_BASE_URL: "https://traces.example.test", + OPENAI_API_KEY: "must-not-persist", + AWS_SECRET_ACCESS_KEY: "must-not-persist", + GH_TOKEN: "must-not-persist", + SSH_AUTH_SOCK: "/tmp/must-not-persist.sock", + }), + ).toEqual({ + HOME: "/home/agent", + PATH: "/runtime/bin", + TMPDIR: "/tmp/agent", + XDG_DATA_HOME: "/home/agent/.local/share", + PRIME_AGENT_CODING_AGENT_DIR: "/home/agent/.prime/agent", + PI_OFFLINE: "1", + PRIME_AGENT_TRACES_BASE_URL: "https://traces.example.test", + }); + expect(filterPersistedDaemonLaunchEnv({ OPENAI_API_KEY: "must-not-persist" })).toBeUndefined(); + }); + it("requires compatibility metadata for the heartbeat protocol surface", () => { expect(DAEMON_PROTOCOL_VERSION).toBe(7); expect(DAEMON_SCHEMA_ID).toContain(`protocol-${DAEMON_PROTOCOL_VERSION}`); diff --git a/packages/coding-agent/test/daemon-supervisor-process.test.ts b/packages/coding-agent/test/daemon-supervisor-process.test.ts index e013e9d134..2760722f45 100644 --- a/packages/coding-agent/test/daemon-supervisor-process.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-process.test.ts @@ -65,7 +65,9 @@ afterEach(async () => { } workerPids.clear(); for (const directory of tempDirs.splice(0)) { - rmSync(directory, { recursive: true, force: true }); + // Detached workers can release kernel/snapshot files just after their + // process group exits on macOS. + rmSync(directory, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 }); } }); @@ -399,6 +401,104 @@ describe("daemon supervisor resident workers", () => { await waitForSocketGone(socketPath); }, 60_000); + it("persists only safe resident launch settings", async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const sessionDir = join(agentDir, "sessions"); + const socketPath = join( + tmpdir(), + `prime-supervisor-resident-env-${process.pid}-${randomUUID().slice(0, 8)}.sock`, + ); + const extensionPath = join(projectDir, "resident-env-extension.ts"); + const markerPath = join(root, "resident-env-marker"); + mkdirSync(projectDir, { recursive: true }); + writeFileSync( + extensionPath, + [ + "import { appendFileSync } from 'node:fs';", + `export default function() { appendFileSync(${JSON.stringify(markerPath)}, process.pid + ':' + process.env.PRIME_AGENT_TRACES_BASE_URL + ':' + (process.env.PRIME_AGENT_TEST_CREDENTIAL ?? '') + '\\n'); }`, + "", + ].join("\n"), + ); + + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const client = await connectEventually(socketPath, supervisor); + const launchEnvSentinel = `resident-env-${randomUUID()}`; + const created = await client.request({ + type: "create", + lifecycle: "resident", + launchEnv: { + PRIME_AGENT_TRACES_BASE_URL: launchEnvSentinel, + PRIME_AGENT_TEST_CREDENTIAL: "must-not-be-persisted", + OPENAI_API_KEY: "must-not-be-persisted", + AWS_SECRET_ACCESS_KEY: "must-not-be-persisted", + GH_TOKEN: "must-not-be-persisted", + SSH_AUTH_SOCK: "/tmp/must-not-be-persisted.sock", + }, + config: { cwd: projectDir, agentDir, sessionDir, noTools: true, extensions: [extensionPath] }, + }); + expect(created.success).toBe(true); + const summary = requireSummary(created.success ? created.data : undefined); + if (summary.workerPid) workerPids.add(summary.workerPid); + const initialMarkerLines = readFileSync(markerPath, "utf8").trim().split("\n"); + expect(initialMarkerLines).toHaveLength(1); + // The initial spawn receives the complete transient caller environment. + expect(initialMarkerLines[0]).toMatch( + new RegExp(`^${summary.workerPid}:${launchEnvSentinel}:must-not-be-persisted$`), + ); + const descriptor = readWorkerDescriptor(agentDir); + // The endpoint survives restart, but credentials never enter the descriptor. + expect(descriptor.launchEnv).toEqual({ PRIME_AGENT_TRACES_BASE_URL: launchEnvSentinel }); + expect(JSON.stringify(descriptor)).not.toContain("must-not-be-persisted"); + expect(JSON.stringify(descriptor)).not.toContain("AWS_SECRET_ACCESS_KEY"); + expect(JSON.stringify(descriptor)).not.toContain("GH_TOKEN"); + expect(JSON.stringify(descriptor)).not.toContain("SSH_AUTH_SOCK"); + + supervisor.kill("SIGTERM"); + await waitForExit(supervisor); + children.delete(supervisor); + client.close(); + + const replacementClient = await connectEventually(socketPath); + const adopted = await replacementClient.request({ type: "list" }); + const adoptedSummary = requireSessionList(adopted.success ? adopted.data : undefined)[0]; + expect(adoptedSummary.workerPid).toBe(summary.workerPid); + if (!adoptedSummary.workerPid) throw new Error("Adopted resident worker did not expose its pid"); + process.kill(-adoptedSummary.workerPid, "SIGKILL"); + await waitForProcessGone(adoptedSummary.workerPid); + let recoveredSummary: SessionSummary | undefined; + const recoveryDeadline = Date.now() + 15_000; + while (!recoveredSummary && Date.now() < recoveryDeadline) { + const listed = await replacementClient.request({ type: "list" }); + recoveredSummary = requireSessionList(listed.success ? listed.data : undefined).find( + (session) => session.workerPid !== adoptedSummary.workerPid, + ); + if (!recoveredSummary) await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); + } + expect(recoveredSummary).toBeDefined(); + let markerLines: string[] = []; + const markerDeadline = Date.now() + 15_000; + while (Date.now() < markerDeadline) { + try { + markerLines = readFileSync(markerPath, "utf8").trim().split("\n"); + if (markerLines.length === 2) break; + } catch { + // The replacement process has not run its extension yet. + } + await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); + } + expect(markerLines).toHaveLength(2); + // Recovery retains the descriptor's allowlisted endpoint. Credential + // exclusion is asserted against the durable descriptor above; a process + // may also inherit credentials from the supervisor's own environment. + expect(markerLines[1]).toMatch(new RegExp(`^${recoveredSummary?.workerPid}:${launchEnvSentinel}:`)); + expect(recoveredSummary?.workerPid).not.toBe(adoptedSummary.workerPid); + + await replacementClient.request({ type: "shutdown" }); + replacementClient.close(); + }); + it("keeps client-owned workers hidden and removes them without archiving", async () => { const root = tempDir(); const agentDir = join(root, "agent");