fix(coding-agent): restore attach/catch-up ordering invariants and sweep snapshot atomicity

- Finding 22: async snapshot creation exposed clients before attach initialization; publish client attachment only after the snapshot resolves. origin: PR #587
- Finding 23: catch-up validity was checked only before async snapshot creation; revalidate session and client attachment after the await. origin: PR #587
- Finding 25: deterministic ambiguous selectors entered the 30-second remote-send retry loop; fail fast for Ambiguous errors. origin: PR #588
- Finding 28: active rows ignored stable file-keyed schedule pins during active-id rebinding; consult both id- and file-keyed heartbeat/cron sets. origin: PR #588
- Finding 29: passivation sampled activity before an asynchronous passive-registry walk; complete the walk before atomically sampling summary and admission markers. origin: PR #589

Author: snimu

packages/coding-agent/src/modes/daemon/daemon-mode.ts

@@ -2289,6 +2289,7 @@ export class AgentDaemon {
 		state: ActiveSessionState,
 		passiveRlmSubagents?: readonly PassiveRlmSubagent[],
 	): Promise<SessionPassivationSnapshot> {
+		const passiveDescendants = passiveRlmSubagents ?? (await this.listPassiveRlmSubagents());
 		const summary = summaryForActiveSession(state);
 		const sessionFile = state.runtime.session.sessionFile;
 		const jobs = this.cronStore
@@ -2299,7 +2300,6 @@ export class AgentDaemon {
 					job.status !== "cancelled" &&
 					job.status !== "completed",
 			);
-		const passiveDescendants = passiveRlmSubagents ?? (await this.listPassiveRlmSubagents());
 		const hasPendingAdmission =
 			this.agentMessageAcceptingTargets.has(state.activeSessionId) ||
 			this.agentMessagePreparingTargets.has(state.activeSessionId) ||
@@ -3264,20 +3264,18 @@ export class AgentDaemon {
 				const snapshotSignal = streamsSnapshot
 					? markClientSnapshotStreaming(client, state.activeSessionId)
 					: undefined;
-				state.clients.add(client);
-				client.attachedActiveSessionIds.add(state.activeSessionId);
 				let result: DaemonAttachResult;
 				try {
 					result = await this.createAttachResult(client, state, command);
 				} catch (error) {
-					state.clients.delete(client);
-					client.attachedActiveSessionIds.delete(state.activeSessionId);
 					removeDaemonClientSessionCapabilities(client, state.activeSessionId);
 					if (streamsSnapshot) {
 						finishClientSnapshotStreaming(client, state.activeSessionId);
 					}
 					throw error;
 				}
+				state.clients.add(client);
+				client.attachedActiveSessionIds.add(state.activeSessionId);
 				if (deferClientEnv && clientEnv) {
 					this.updateRestart?.deferredClientEnv.push({ client, state, env: clientEnv });
 				}
@@ -4733,7 +4731,10 @@ export class AgentDaemon {
 				return response.data as AgentSessionMessageReceipt;
 			} catch (error) {
 				lastError = error;
-				if (error instanceof Error && error.message.startsWith("Unknown active session:")) {
+				if (
+					error instanceof Error &&
+					(error.message.startsWith("Unknown active session:") || error.message.startsWith("Ambiguous"))
+				) {
 					throw error;
 				}
 			} finally {
@@ -5489,6 +5490,9 @@ export class AgentDaemon {
 				type: "attach",
 				activeSessionId,
 			});
+			if (this.sessions.get(activeSessionId) !== state || !state.clients.has(client)) {
+				continue;
+			}
 			if (
 				client.transport === "private-framed" &&
 				daemonClientCapabilitiesForSession(client, activeSessionId).has("chunked_snapshot")

packages/coding-agent/src/modes/daemon/daemon-session-list.ts

@@ -144,8 +144,10 @@ export function buildSessionList(
 					activeSession,
 					savedSession,
 					heartbeatSessionIds.has(activeSession.activeSessionId),
-					registeredHeartbeatSessionIds.has(activeSession.activeSessionId),
-					registeredCronSessionIds.has(activeSession.activeSessionId),
+					registeredHeartbeatSessionIds.has(activeSession.activeSessionId) ||
+						registeredHeartbeatSessionFiles.has(sessionFile),
+					registeredCronSessionIds.has(activeSession.activeSessionId) ||
+						registeredCronSessionFiles.has(sessionFile),
 				),
 			);
 			seenActiveSessionIds.add(activeSession.activeSessionId);
@@ -162,13 +164,17 @@ export function buildSessionList(
 
 	for (const activeSession of activeSessions) {
 		if (!seenActiveSessionIds.has(activeSession.activeSessionId)) {
+			const sessionFile = activeSession.runtime.session.sessionFile;
+			const resolvedSessionFile = sessionFile ? resolve(sessionFile) : undefined;
 			entries.push(
 				summaryForActiveSession(
 					activeSession,
 					undefined,
 					heartbeatSessionIds.has(activeSession.activeSessionId),
-					registeredHeartbeatSessionIds.has(activeSession.activeSessionId),
-					registeredCronSessionIds.has(activeSession.activeSessionId),
+					registeredHeartbeatSessionIds.has(activeSession.activeSessionId) ||
+						(resolvedSessionFile !== undefined && registeredHeartbeatSessionFiles.has(resolvedSessionFile)),
+					registeredCronSessionIds.has(activeSession.activeSessionId) ||
+						(resolvedSessionFile !== undefined && registeredCronSessionFiles.has(resolvedSessionFile)),
 				),
 			);
 		}

packages/coding-agent/test/daemon-mode.test.ts

@@ -1,6 +1,6 @@
 import { EventEmitter } from "node:events";
 import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
-import type { Socket } from "node:net";
+import { createServer, type Socket } from "node:net";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { Api, Model } from "@earendil-works/pi-ai";
@@ -31,9 +31,13 @@ import {
 } from "../src/modes/daemon/daemon-mode.js";
 import {
 	createDaemonCommandEnvelope,
+	DAEMON_PROTOCOL_INFO,
+	DAEMON_SCHEMA_REVISION,
 	type DaemonAttachResult,
 	type DaemonCommand,
+	failure,
 } from "../src/modes/daemon/daemon-protocol.js";
+import { DAEMON_WORKER_SUPERVISOR_SOCKET_ENV } from "../src/modes/daemon/daemon-worker-protocol.js";
 
 describe("daemon mode helpers", () => {
 	it("preserves envelope client identity while registering prompt admission", () => {
@@ -1018,6 +1022,64 @@ describe("daemon mode helpers", () => {
 		expect(sendRemoteAgentSessionMessage).toHaveBeenCalledWith(source, "deleted-child", "continue", undefined);
 	});
 
+	it("does not retry permanent ambiguity errors from the supervisor", async () => {
+		const tempDir = mkdtempSync(join(tmpdir(), "pa-ambiguous-"));
+		const socketPath = join(tempDir, "s");
+		let requestCount = 0;
+		const server = createServer((socket) => {
+			socket.write(
+				`${JSON.stringify({
+					type: "daemon_hello",
+					socketPath,
+					protocol: DAEMON_PROTOCOL_INFO,
+					schemaRevision: DAEMON_SCHEMA_REVISION,
+					serverCapabilities: [],
+				})}\n`,
+			);
+			let buffered = "";
+			socket.on("data", (chunk: Buffer) => {
+				buffered += chunk.toString("utf8");
+				const newline = buffered.indexOf("\n");
+				if (newline < 0 || requestCount > 0) return;
+				const command = JSON.parse(buffered.slice(0, newline)) as { id?: string };
+				requestCount++;
+				socket.write(
+					`${JSON.stringify(failure(command.id, "send_message", new Error('Ambiguous session selector "duplicate"')))}\n`,
+				);
+			});
+		});
+		const previousSocketPath = process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV];
+		try {
+			await new Promise<void>((resolve, reject) => {
+				server.once("error", reject);
+				server.listen(socketPath, resolve);
+			});
+			process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV] = socketPath;
+			const daemon = new AgentDaemon(join(tempDir, "worker.sock"), {
+				defaultSessionConfig: { agentDir: tempDir, cwd: tempDir },
+				createRuntime: vi.fn(),
+			});
+			const source = makeState("source");
+			const internals = daemon as unknown as {
+				sendRemoteAgentSessionMessage(
+					fromState: ActiveSessionState,
+					targetSelector: string,
+					message: string,
+				): Promise<unknown>;
+			};
+
+			await expect(internals.sendRemoteAgentSessionMessage(source, "duplicate", "hello")).rejects.toThrow(
+				'Ambiguous session selector "duplicate"',
+			);
+			expect(requestCount).toBe(1);
+		} finally {
+			if (previousSocketPath === undefined) delete process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV];
+			else process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV] = previousSocketPath;
+			await new Promise<void>((resolve) => server.close(() => resolve()));
+			rmSync(tempDir, { recursive: true, force: true });
+		}
+	});
+
 	it("reports queued status when a direct accept races into the queue", async () => {
 		const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", {
 			defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" },
@@ -3136,6 +3198,84 @@ describe("daemon mode helpers", () => {
 		expect(client.catchupActiveSessionIds).toEqual(new Set());
 	});
 
+	it("does not attach a non-chunked client until its snapshot is ready", async () => {
+		const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", {
+			defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" },
+			createRuntime: vi.fn(),
+		});
+		const state = makeState("active");
+		const client = makeClient("client-1", state.activeSessionId);
+		client.attachedActiveSessionIds.clear();
+		let releaseSnapshot!: () => void;
+		const snapshotGate = new Promise<void>((resolve) => {
+			releaseSnapshot = resolve;
+		});
+		const result = {
+			activeSessionId: state.activeSessionId,
+			snapshot: { summary: {}, state: {}, messages: [] },
+			lastEventSequence: 0,
+		} as unknown as DaemonAttachResult;
+		const internals = daemon as unknown as {
+			sessions: Map<string, ActiveSessionState>;
+			createAttachResult: ReturnType<typeof vi.fn>;
+			handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise<unknown>;
+		};
+		internals.sessions.set(state.activeSessionId, state);
+		internals.createAttachResult = vi.fn(async () => {
+			await snapshotGate;
+			return result;
+		});
+
+		const attach = internals.handleCommand(client, { type: "attach", activeSessionId: state.activeSessionId });
+		await vi.waitFor(() => expect(internals.createAttachResult).toHaveBeenCalledOnce());
+		expect(state.clients).not.toContain(client);
+		expect(client.attachedActiveSessionIds).not.toContain(state.activeSessionId);
+		releaseSnapshot();
+		await attach;
+		expect(state.clients).toContain(client);
+		expect(client.attachedActiveSessionIds).toContain(state.activeSessionId);
+	});
+
+	it("drops a backpressure catch-up when the client detaches during snapshot creation", async () => {
+		const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", {
+			defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" },
+			createRuntime: vi.fn(),
+		});
+		const state = makeState("active");
+		const write = vi.fn(() => true);
+		const client = makeClient("client-1", state.activeSessionId);
+		client.socket = { destroyed: false, write } as unknown as Socket;
+		client.catchupActiveSessionIds = new Set([state.activeSessionId]);
+		state.clients.add(client);
+		let releaseSnapshot!: () => void;
+		const snapshotGate = new Promise<void>((resolve) => {
+			releaseSnapshot = resolve;
+		});
+		const result = {
+			activeSessionId: state.activeSessionId,
+			snapshot: { summary: {}, state: {}, messages: [], lastEventSequence: 0 },
+			lastEventSequence: 0,
+		} as unknown as DaemonAttachResult;
+		const internals = daemon as unknown as {
+			sessions: Map<string, ActiveSessionState>;
+			createAttachResult: ReturnType<typeof vi.fn>;
+			drainBackpressuredClientCatchups(client: DaemonSocketClient): Promise<void>;
+		};
+		internals.sessions.set(state.activeSessionId, state);
+		internals.createAttachResult = vi.fn(async () => {
+			await snapshotGate;
+			return result;
+		});
+
+		const catchup = internals.drainBackpressuredClientCatchups(client);
+		await vi.waitFor(() => expect(internals.createAttachResult).toHaveBeenCalledOnce());
+		state.clients.delete(client);
+		client.attachedActiveSessionIds.delete(state.activeSessionId);
+		releaseSnapshot();
+		await catchup;
+		expect(write).not.toHaveBeenCalled();
+	});
+
 	it("marks a chunked attach as snapshotting before deferred streaming", async () => {
 		const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-snapshot-order-"));
 		try {
@@ -4034,6 +4174,41 @@ describe("daemon mode helpers", () => {
 		}
 	});
 
+	it("does not passivate a child that starts streaming during the fence snapshot", async () => {
+		const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-passivation-stream-race-"));
+		try {
+			const fixture = makePersistedRlmDaemonFixture(tempDir);
+			const internals = fixture.daemon as unknown as {
+				createRuntime(command: Extract<DaemonCommand, { type: "create" }>): Promise<ActiveSessionState>;
+				listPassiveRlmSubagents: ReturnType<typeof vi.fn>;
+				passivateIdleChildren(threshold: number, now: number, limit: number): Promise<number>;
+			};
+			const parentState = await internals.createRuntime({ type: "create", sessionPath: fixture.parentSessionFile });
+			const childState = await internals.createRuntime({ type: "create", sessionPath: fixture.childSessionFile });
+			const childSession = childState.runtime.session as unknown as {
+				isStreaming: boolean;
+				isSessionActive: boolean;
+				abort: ReturnType<typeof vi.fn>;
+			};
+			let passiveListCalls = 0;
+			internals.listPassiveRlmSubagents = vi.fn(async () => {
+				passiveListCalls++;
+				if (passiveListCalls === 2) {
+					childSession.isStreaming = true;
+					childSession.isSessionActive = true;
+				}
+				return [];
+			});
+
+			await expect(internals.passivateIdleChildren(90, Date.parse("2036-08-01T12:00:00Z"), 1)).resolves.toBe(0);
+			expect(passiveListCalls).toBe(2);
+			expect(childSession.abort).not.toHaveBeenCalled();
+			expect(parentState.runtime.session.releaseFinishedRlmChildSession).not.toHaveBeenCalled();
+		} finally {
+			rmSync(tempDir, { recursive: true, force: true });
+		}
+	});
+
 	it("limits each worker sweep and leaves non-leaf children resident", async () => {
 		const daemon = new AgentDaemon("/tmp/prime-agent-passivation-cap.sock", {
 			defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" },

packages/coding-agent/test/daemon-session-list.test.ts

@@ -167,6 +167,30 @@ describe("buildSessionList", () => {
 		});
 	});
 
+	it("keeps file-keyed schedule pins on active rows while active ids are being rebound", () => {
+		const sessionFile = "/tmp/child.jsonl";
+		const [entry] = buildSessionList(
+			[makeState({ activeSessionId: "new-active-id", sessionFile })],
+			[makeSessionInfo({ id: "child-session", path: sessionFile })],
+			[
+				makeCronJob({
+					id: "stale-heartbeat",
+					activeSessionId: "old-active-id",
+					sessionFile,
+					source: "heartbeat",
+				}),
+				makeCronJob({
+					id: "stale-cron",
+					activeSessionId: "old-active-id",
+					sessionFile,
+					source: "cron",
+				}),
+			],
+		);
+
+		expect(entry).toMatchObject({ hasRegisteredHeartbeat: true, hasRegisteredCronJob: true });
+	});
+
 	it("reports accepted in-flight prompts as active with no queued work", () => {
 		const oneMessage = [{ role: "user", content: "hi" }] as unknown as AgentMessage[];
 		const summary = summaryForActiveSession(