feat(acp): preserve causal prompt correlation

packages/agent/src/agent-loop.ts

@@ -3,6 +3,7 @@
  * Transforms to Message[] only at the LLM call boundary.
  */
 
+import { randomBytes } from "node:crypto";
 import {
 	type AssistantMessage,
 	type AssistantMessageEvent,
@@ -26,6 +27,17 @@ import type {
 export type AgentEventSink = (event: AgentEvent) => Promise<void> | void;
 
 const ABORT_ERROR_MESSAGE = "Request was aborted";
+const PRIME_AGENT_RELAY_ID_HEADER = "X-Prime-Agent-Relay-ID";
+
+function withPrimeAgentRelayId(headers: Record<string, string> | undefined): Record<string, string> {
+	const filteredHeaders = Object.fromEntries(
+		Object.entries(headers ?? {}).filter(
+			([name]) => name.toLowerCase() !== PRIME_AGENT_RELAY_ID_HEADER.toLowerCase(),
+		),
+	);
+	return { ...filteredHeaders, [PRIME_AGENT_RELAY_ID_HEADER]: randomBytes(16).toString("hex") };
+}
+
 const EMPTY_USAGE: AssistantMessage["usage"] = {
 	input: 0,
 	output: 0,
@@ -516,6 +528,7 @@ async function streamAssistantResponse(
 				...config,
 				apiKey: resolvedApiKey,
 				signal,
+				...(config.model.provider === "intercept" ? { headers: withPrimeAgentRelayId(config.headers) } : {}),
 			}),
 			signal,
 		);

packages/agent/test/agent-loop.test.ts

@@ -9,7 +9,14 @@ import {
 import { Type } from "typebox";
 import { describe, expect, it, vi } from "vitest";
 import { agentLoop, agentLoopContinue, runAgentLoop } from "../src/agent-loop.js";
-import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool } from "../src/types.js";
+import type {
+	AgentContext,
+	AgentEvent,
+	AgentLoopConfig,
+	AgentMessage,
+	AgentTool,
+	StreamFn,
+} from "../src/types.js";
 
 // Mock stream for testing - mimics MockAssistantStream
 class MockAssistantStream extends EventStream<AssistantMessageEvent, AssistantMessage> {
@@ -110,6 +117,95 @@ function identityConverter(messages: AgentMessage[]): Message[] {
 	return messages.filter((m) => m.role === "user" || m.role === "assistant" || m.role === "toolResult") as Message[];
 }
 
+describe("Agent relay identity", () => {
+	it("generates a distinct protected relay ID per intercept completion and keeps it through downstream retries", async () => {
+		const seenRelayIds: string[] = [];
+		const retryRelayIds: string[] = [];
+		const eventPayloads: string[] = [];
+		const streamFn: StreamFn = vi.fn((_model, _context, options) => {
+			const relayId = options?.headers?.["X-Prime-Agent-Relay-ID"];
+			if (!relayId) throw new Error("missing relay ID");
+			seenRelayIds.push(relayId);
+
+			// The provider SDK owns retries; it receives this one logical request's unchanged headers.
+			for (let attempt = 0; attempt < 2; attempt++) {
+				retryRelayIds.push(options?.headers?.["X-Prime-Agent-Relay-ID"] ?? "");
+			}
+
+			const stream = new MockAssistantStream();
+			queueMicrotask(() => {
+				stream.push({
+					type: "done",
+					reason: "stop",
+					message: createAssistantMessage([{ type: "text", text: "Response" }]),
+				});
+			});
+			return stream;
+		});
+		const config: AgentLoopConfig = {
+			model: { ...createModel(), provider: "intercept" },
+			convertToLlm: identityConverter,
+			headers: {
+				"X-Extra-Auth": "preserved",
+				"x-prime-agent-relay-id": "caller-must-not-control-this",
+			},
+		};
+		const run = async () => {
+			const events: AgentEvent[] = [];
+			await runAgentLoop(
+				[createUserMessage("Hello")],
+				{ systemPrompt: "You are helpful.", messages: [], tools: [] },
+				config,
+				(event) => {
+					events.push(event);
+				},
+				undefined,
+				streamFn,
+			);
+			eventPayloads.push(JSON.stringify(events));
+		};
+
+		await run();
+		await run();
+
+		expect(streamFn).toHaveBeenCalledTimes(2);
+		expect(seenRelayIds).toHaveLength(2);
+		expect(seenRelayIds[0]).toMatch(/^[0-9a-f]{32}$/);
+		expect(seenRelayIds[1]).toMatch(/^[0-9a-f]{32}$/);
+		expect(seenRelayIds[0]).not.toBe(seenRelayIds[1]);
+		expect(retryRelayIds).toEqual([seenRelayIds[0], seenRelayIds[0], seenRelayIds[1], seenRelayIds[1]]);
+		expect(eventPayloads.join("\n")).not.toContain(seenRelayIds[0]!);
+		expect(eventPayloads.join("\n")).not.toContain(seenRelayIds[1]!);
+	});
+
+	it("does not add a relay ID for non-intercept providers", async () => {
+		let headers: Record<string, string> | undefined;
+		const stream = new MockAssistantStream();
+		const streamFn: StreamFn = vi.fn((_model, _context, options) => {
+			headers = options?.headers;
+			queueMicrotask(() => {
+				stream.push({
+					type: "done",
+					reason: "stop",
+					message: createAssistantMessage([{ type: "text", text: "Response" }]),
+				});
+			});
+			return stream;
+		});
+
+		await runAgentLoop(
+			[createUserMessage("Hello")],
+			{ systemPrompt: "You are helpful.", messages: [], tools: [] },
+			{ model: createModel(), convertToLlm: identityConverter, headers: { "X-Extra-Auth": "preserved" } },
+			() => {},
+			undefined,
+			streamFn,
+		);
+
+		expect(headers).toEqual({ "X-Extra-Auth": "preserved" });
+	});
+});
+
 describe("agentLoop with AgentMessage", () => {
 	it("should preserve a terminal response when abort fires after done", async () => {
 		const context: AgentContext = {

packages/coding-agent/src/cli/command-registry.ts

@@ -87,10 +87,15 @@ export const COMMAND_SPECS: readonly CommandSpec[] = [
 	},
 	{
 		path: ["shutdown"],
-		usage: "shutdown [--force] [--json]",
-		summary: "Stop every agent and background service",
-		description: "Without --force, an interactive confirmation is required. --force also kills unresponsive workers.",
-		options: ["--force  Skip confirmation and kill unresponsive processes", "--json   Print JSON"],
+		usage: "shutdown [--force] [--json] [--daemon-socket <path>]",
+		summary: "Stop agents and background services",
+		description:
+			"Without --daemon-socket, stops every service after confirmation. With --daemon-socket, stops only the verified daemon at that exact path and cannot be combined with --force.",
+		options: [
+			"--force                 Skip confirmation and kill unresponsive processes",
+			"--daemon-socket <path>  Stop only the verified daemon at this exact socket",
+			"--json                  Print JSON",
+		],
 	},
 	{
 		path: ["package"],

packages/coding-agent/src/cli/daemon-launch.ts

@@ -265,6 +265,37 @@ export async function shutdownDaemonAndWait(socketPath: string, timeoutMs = 5000
 	}
 }
 
+/**
+ * Shut down exactly one current daemon addressed by `socketPath`.
+ *
+ * Unlike the machine-wide shutdown/reap commands this performs no discovery. It
+ * also refuses stale, foreign, or same-version/different-build daemons before
+ * sending a shutdown request. Callers use this for rollout-owned daemon sockets.
+ */
+export async function shutdownExactDaemonAndWait(socketPath: string, timeoutMs = 10_000): Promise<boolean> {
+	const client = new DaemonClient(socketPath);
+	try {
+		await client.connect(1000);
+		const hello = await client.waitForHello(2000);
+		const expectedRuntime = getDaemonRuntimeIdentity();
+		const compatible =
+			hello.protocol.version === DAEMON_PROTOCOL_VERSION &&
+			hello.schemaId === DAEMON_SCHEMA_ID &&
+			hello.appVersion === VERSION &&
+			hello.runtime?.buildId === expectedRuntime.buildId;
+		if (!compatible) {
+			throw new Error(`Refusing exact shutdown of an unverified daemon on ${socketPath}`);
+		}
+		return await shutdownConnectedDaemonAndWait(client, socketPath, timeoutMs, hello);
+	} catch (error) {
+		client.close();
+		if (!existsSync(socketPath)) {
+			return true;
+		}
+		throw error;
+	}
+}
+
 // activeSessions is undefined when the daemon is reachable but its sessions couldn't
 // be listed — callers must treat that as "possibly busy", not idle.
 export type RunningDaemonProbe =

packages/coding-agent/src/cli/public-command.ts

@@ -13,6 +13,7 @@ import {
 	REMOVED_COMMAND_NAMES,
 } from "./command-registry.js";
 import { handleDaemonCommand } from "./daemon-command.js";
+import { shutdownExactDaemonAndWait } from "./daemon-launch.js";
 import { runPs, runReap, runShutdownAll } from "./daemon-ps.js";
 import { DAEMON_UPDATE_RESTART_COORDINATOR_FLAG } from "./daemon-update-restart.js";
 
@@ -259,8 +260,38 @@ async function runDoctor(args: string[]): Promise<PublicCommandResult> {
 }
 
 async function runShutdown(args: string[]): Promise<PublicCommandResult> {
-	const options = parseBooleanOptions(args, new Set(["--force", "--json"]), "shutdown");
+	let socketPath: string | undefined;
+	const booleanOptions: string[] = [];
+	for (let index = 0; index < args.length; index++) {
+		const arg = args[index]!;
+		if (arg === "--daemon-socket") {
+			const value = args[++index];
+			if (!value || value.startsWith("-")) {
+				return fail("--daemon-socket requires a path");
+			}
+			if (socketPath !== undefined) {
+				return fail("--daemon-socket may only be supplied once");
+			}
+			socketPath = value;
+			continue;
+		}
+		booleanOptions.push(arg);
+	}
+	const options = parseBooleanOptions(booleanOptions, new Set(["--force", "--json"]), "shutdown");
 	if (!options) return HANDLED;
+	if (socketPath !== undefined) {
+		if (options.has("--force")) {
+			return fail("--force cannot be combined with exact --daemon-socket shutdown");
+		}
+		const stopped = await shutdownExactDaemonAndWait(socketPath);
+		if (!stopped) {
+			throw new Error(`Exact daemon shutdown was not confirmed for ${socketPath}`);
+		}
+		if (options.has("--json")) {
+			console.log(JSON.stringify({ socketPath, stopped: true }));
+		}
+		return HANDLED;
+	}
 	await runShutdownAll(options.has("--json"), options.has("--force"));
 	return HANDLED;
 }

packages/coding-agent/src/core/agent-session.ts

@@ -938,6 +938,7 @@ interface RlmChildRun {
 	prompt: string;
 	sessionName: string;
 	sessionDir: string;
+	model: Model<Api>;
 	status: RlmChildAgentStatus;
 	error?: string;
 	abort: () => void;
@@ -3567,7 +3568,9 @@ export class AgentSession {
 			// Check for retryable errors first (overloaded, rate limit, server errors)
 			const concreteAuthFailure = this._isConcreteProviderAuthFailure(msg);
 			const retryConcreteAuthFailure =
-				concreteAuthFailure && !this._isStructuredPermanentProviderRetryExhausted(msg);
+				msg.provider !== "intercept" &&
+				concreteAuthFailure &&
+				!this._isStructuredPermanentProviderRetryExhausted(msg);
 			if (this._isRetryableError(msg) || retryConcreteAuthFailure) {
 				if (retryConcreteAuthFailure) {
 					this._captureRetryAuthFailureSource(msg);
@@ -9535,6 +9538,60 @@ export class AgentSession {
 		return unsubscribe;
 	}
 
+	/**
+	 * Live recursive child roster for connection snapshots.
+	 *
+	 * The run registry is authoritative while a child is queued or running; retained
+	 * sessions preserve completed children and expose any nested work that outlives
+	 * their direct parent run. This deliberately does not reconstruct state from
+	 * observer events, which can predate a newly attached connection.
+	 */
+	getRlmChildSnapshots(): RlmChildAgentSnapshot[] {
+		const snapshots: RlmChildAgentSnapshot[] = [];
+		const recorded = new Set<string>();
+		const traversed = new Set<string>();
+		for (const run of this._activeRlmChildRuns.values()) {
+			const hidden =
+				run.detachedDeletion || this._deletingRlmChildren.has(run.id) || this._deletedRlmChildIds.has(run.id);
+			const child = run.session;
+			if (!hidden) {
+				snapshots.push({
+					id: run.id,
+					parentId: this._rlmParentNodeId,
+					sessionName: child?.sessionName ?? run.sessionName,
+					model: `${(child?.model ?? run.model).provider}/${(child?.model ?? run.model).id}`,
+					label: rlmChildLabel(run.prompt),
+					status: run.status,
+					sessionDir: run.sessionDir,
+				});
+				recorded.add(run.id);
+			}
+			if (child) {
+				traversed.add(run.id);
+				snapshots.push(...child.getRlmChildSnapshots());
+			}
+		}
+		for (const [childId, child] of this._rlmChildSessions) {
+			if (recorded.has(childId) || traversed.has(childId)) continue;
+			const hidden = this._deletingRlmChildren.has(childId) || this._deletedRlmChildIds.has(childId);
+			if (!hidden) {
+				snapshots.push({
+					id: childId,
+					parentId: this._rlmParentNodeId,
+					sessionName: child.sessionName,
+					model: child.model ? `${child.model.provider}/${child.model.id}` : undefined,
+					label: child.sessionName ?? "child agent",
+					// A failed delete retains the session solely for cleanup retry. Preserve
+					// its cancellation truth in snapshots rather than reviving it as done.
+					status: this._rlmChildCleanupFailures.has(childId) ? "cancelled" : "done",
+					sessionDir: child._rlmSessionDir ?? child.sessionManager.getSessionDir(),
+				});
+			}
+			snapshots.push(...child.getRlmChildSnapshots());
+		}
+		return snapshots;
+	}
+
 	/** True when any direct or nested subagent is still running or queued. */
 	hasRunningRlmChildren(): boolean {
 		for (const run of this._activeRlmChildRuns.values()) {
@@ -9732,6 +9789,7 @@ export class AgentSession {
 			prompt,
 			sessionName,
 			sessionDir: childSessionDir,
+			model: modelSelection.model,
 			status: "queued",
 			settled: false,
 			abort: noopRlmChildAbort,
@@ -10059,6 +10117,13 @@ export class AgentSession {
 			return false;
 		}
 
+		// The intercept provider adds an idempotency key at the logical request boundary.
+		// Retrying here would call agent.continue() and mint a new key; provider retries
+		// remain below that boundary and retain the original key.
+		if (message.provider === "intercept") {
+			return false;
+		}
+
 		if (this._isAgentLifecycleFailure(message)) {
 			return false;
 		}