refactor(coding-agent): harden session action races

packages/coding-agent/CHANGELOG.md

@@ -11,6 +11,7 @@
 - Fixed long-running thinking timers to display hours and days instead of unbounded minutes.
 - Fixed overlapping daemon snapshot catch-ups closing healthy workers and preventing new sessions from starting.
 - Changed daemon and RPC session state to report literal queued actions separately from active scheduler work.
+- Changed session input scheduling to use one action lifecycle and store, fixing active work appearing queued, headless completion exiting early, `/compact` consuming itself as a successor, and incompatible clients not failing cleanly at protocol 7/schema 8.
 
 ## [0.4.0] - 2026-08-01
 

packages/coding-agent/src/core/cron-jobs.ts

@@ -87,6 +87,7 @@ export interface HeartbeatCronSessionActivity {
 	isCompacting?: boolean;
 	isRetrying?: boolean;
 	isBashRunning: boolean;
+	hasPendingSessionWork: boolean;
 	unfinishedActionCount: number;
 }
 
@@ -1357,7 +1358,8 @@ export function shouldDeferHeartbeatCronJob(job: AgentCronJob, activity: Heartbe
 		activity.isCompacting === true ||
 		activity.isRetrying === true ||
 		activity.isBashRunning ||
-		activity.unfinishedActionCount > 0;
+		activity.hasPendingSessionWork ||
+		(!activity.isStreaming && activity.unfinishedActionCount > 0);
 	if (busyBesidesStreaming) {
 		return true;
 	}

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

@@ -307,7 +307,6 @@ export interface RuntimeActivity {
 	branchMutation: boolean;
 	schedulerPauseCount: number;
 	disposing: boolean;
-	mandatoryCheckpointsComplete: boolean;
 }
 
 export function canSelectSessionAction(activity: RuntimeActivity): boolean {
@@ -322,7 +321,3 @@ export function canSelectSessionAction(activity: RuntimeActivity): boolean {
 		!activity.disposing
 	);
 }
-
-export function shouldYieldLowerRun(store: ActionStore, activity: RuntimeActivity): boolean {
-	return activity.mandatoryCheckpointsComplete && store.queuedActions("next_turn_boundary").length > 0;
-}

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

@@ -568,20 +568,16 @@ export class DaemonAgentConnection implements AgentConnection {
 		jobId: string,
 		action: AgentHeartbeatManagementAction,
 	): Promise<AgentCronJob> {
-		const hasCapability = this.client.supportsServerCapability("heartbeat_management");
-		if (!hasCapability && this.client.hello?.protocol.version !== 3) {
+		if (!this.client.supportsServerCapability("heartbeat_management")) {
 			throw new Error("Heartbeat management requires a newer Prime Agent daemon.");
 		}
 		try {
-			const command = {
+			const data = await this.requestData<{ heartbeat: AgentCronJob }>({
 				type: "heartbeat_manage",
 				activeSessionId,
 				jobId,
 				action,
-			} as const;
-			const data = hasCapability
-				? await this.requestData<{ heartbeat: AgentCronJob }>(command)
-				: await this.requestLegacyData<{ heartbeat: AgentCronJob }>(command);
+			});
 			return data.heartbeat;
 		} catch (error) {
 			if (isUnknownDaemonCommandError(error, "heartbeat_manage")) {
@@ -1380,14 +1376,6 @@ export class DaemonAgentConnection implements AgentConnection {
 		await this.requestData<unknown>(command);
 	}
 
-	private async requestLegacyData<T>(command: DaemonCommandBody, timeoutMs?: number): Promise<T> {
-		const response = await this.client.requestLegacy(command, timeoutMs);
-		if (!response.success) {
-			throw deserializeDaemonError(response);
-		}
-		return response.data as T;
-	}
-
 	private async requestData<T>(
 		command: DaemonCommandBody,
 		timeoutMs?: number,

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

@@ -328,15 +328,6 @@ export class DaemonClient {
 		);
 	}
 
-	/** One-release compatibility path for preparing and stopping a v1 daemon. */
-	async requestLegacy(
-		command: DaemonCommandBody,
-		timeoutMs = 30000,
-		options: DaemonClientRequestOptions = {},
-	): Promise<DaemonResponse> {
-		return this.requestWire(command, timeoutMs, options);
-	}
-
 	async authenticateWorker(token: string, timeoutMs = 3000): Promise<void> {
 		const legacyAuthentication = { type: "worker_auth", token } as DaemonWorkerCommandBody;
 		const response = await this.requestWire(legacyAuthentication, timeoutMs);

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

@@ -53,7 +53,6 @@ class FakeDaemonClient {
 	promptError: Error | undefined;
 	promptResponseError: string | undefined;
 	cancelPromptAdmissionStatus: "cancelled" | "owned" | "unknown" = "owned";
-	legacyHeartbeatCommandsSupported = false;
 	serverCapabilities = new Set<string>();
 	updateRestartSessions: Array<Record<string, unknown>> = [];
 	hello: DaemonHello | undefined = {
@@ -271,7 +270,7 @@ class FakeDaemonClient {
 					data: { steering: ["aborted"], followUp: ["cleared"] },
 				};
 			case "heartbeats_list":
-				return this.legacyHeartbeatCommandsSupported || this.serverCapabilities.has("heartbeat_catalog")
+				return this.serverCapabilities.has("heartbeat_catalog")
 					? { type: "response", command: command.type, success: true, data: { heartbeats: [] } }
 					: {
 							type: "response",
@@ -280,7 +279,7 @@ class FakeDaemonClient {
 							error: "Unknown daemon command: heartbeats_list",
 						};
 			case "heartbeat_manage":
-				return this.legacyHeartbeatCommandsSupported || this.serverCapabilities.has("heartbeat_management")
+				return this.serverCapabilities.has("heartbeat_management")
 					? {
 							type: "response",
 							command: command.type,
@@ -467,14 +466,6 @@ class FakeDaemonClient {
 		}
 	}
 
-	async requestLegacy(
-		command: DaemonCommand,
-		timeoutMs = 30000,
-		options: DaemonClientRequestOptions = {},
-	): Promise<DaemonResponse> {
-		return this.request(command, timeoutMs, options);
-	}
-
 	supportsServerCapability(capability: string): boolean {
 		return this.serverCapabilities.has(capability);
 	}
@@ -965,43 +956,6 @@ describe("DaemonAgentConnection", () => {
 		expect(fakeClient.requests).toEqual([]);
 	});
 
-	it("does not query the heartbeat catalog on a retained protocol-3 daemon", async () => {
-		const fakeClient = new FakeDaemonClient();
-		fakeClient.hello = {
-			type: "daemon_hello",
-			socketPath: "/tmp/prime-agent.sock",
-			protocol: { ...DAEMON_PROTOCOL_INFO, version: 3 },
-			clientId: "legacy-client",
-			serverCapabilities: [],
-		};
-		fakeClient.legacyHeartbeatCommandsSupported = true;
-		const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-original");
-
-		await expect(connection.listHeartbeats()).resolves.toEqual([]);
-		await expect(connection.manageHeartbeat("active-original", "job-1", "pause")).resolves.toMatchObject({
-			id: "job-1",
-		});
-		expect(fakeClient.requests.map((request) => request.type)).toEqual(["heartbeat_manage"]);
-	});
-
-	it("degrades heartbeat commands missing from an older protocol-3 daemon", async () => {
-		const fakeClient = new FakeDaemonClient();
-		fakeClient.hello = {
-			type: "daemon_hello",
-			socketPath: "/tmp/prime-agent.sock",
-			protocol: { ...DAEMON_PROTOCOL_INFO, version: 3 },
-			clientId: "legacy-client",
-			serverCapabilities: [],
-		};
-		const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-original");
-
-		await expect(connection.listHeartbeats()).resolves.toEqual([]);
-		await expect(connection.manageHeartbeat("active-original", "job-1", "pause")).rejects.toThrow(
-			"Heartbeat management requires a newer Prime Agent daemon.",
-		);
-		expect(fakeClient.requests.map((request) => request.type)).toEqual(["heartbeat_manage"]);
-	});
-
 	it("reattaches an open window to its restored session after an update restart", async () => {
 		const fakeClient = new FakeDaemonClient();
 		fakeClient.emitCloseOnClose = true;

packages/coding-agent/test/cron-jobs.test.ts

@@ -1352,20 +1352,23 @@ describe("shouldDeferHeartbeatCronJob", () => {
 				shouldDeferHeartbeatCronJob(job, {
 					isStreaming: true,
 					isBashRunning: false,
+					hasPendingSessionWork: false,
 					unfinishedActionCount: 0,
 				}),
 			).toBe(true);
 			expect(
 				shouldDeferHeartbeatCronJob(job, {
 					isStreaming: false,
 					isBashRunning: true,
+					hasPendingSessionWork: false,
 					unfinishedActionCount: 0,
 				}),
 			).toBe(true);
 			expect(
 				shouldDeferHeartbeatCronJob(job, {
 					isStreaming: false,
 					isBashRunning: false,
+					hasPendingSessionWork: false,
 					unfinishedActionCount: 1,
 				}),
 			).toBe(true);
@@ -1383,7 +1386,8 @@ describe("shouldDeferHeartbeatCronJob", () => {
 					shouldDeferHeartbeatCronJob(job, {
 						isStreaming: true,
 						isBashRunning: false,
-						unfinishedActionCount: 0,
+						hasPendingSessionWork: false,
+						unfinishedActionCount: 1,
 					}),
 				).toBe(false);
 			}
@@ -1398,20 +1402,23 @@ describe("shouldDeferHeartbeatCronJob", () => {
 				isStreaming: true,
 				isCompacting: true,
 				isBashRunning: false,
+				hasPendingSessionWork: false,
 				unfinishedActionCount: 0,
 			}),
 		).toBe(true);
 		expect(
 			shouldDeferHeartbeatCronJob(job, {
 				isStreaming: false,
 				isBashRunning: true,
+				hasPendingSessionWork: false,
 				unfinishedActionCount: 0,
 			}),
 		).toBe(true);
 		expect(
 			shouldDeferHeartbeatCronJob(job, {
 				isStreaming: false,
 				isBashRunning: false,
+				hasPendingSessionWork: false,
 				unfinishedActionCount: 1,
 			}),
 		).toBe(true);
@@ -1420,14 +1427,16 @@ describe("shouldDeferHeartbeatCronJob", () => {
 				isStreaming: false,
 				isRetrying: true,
 				isBashRunning: false,
+				hasPendingSessionWork: false,
 				unfinishedActionCount: 0,
 			}),
 		).toBe(true);
 		expect(
 			shouldDeferHeartbeatCronJob(job, {
-				isStreaming: false,
+				isStreaming: true,
 				isBashRunning: false,
-				unfinishedActionCount: 1,
+				hasPendingSessionWork: true,
+				unfinishedActionCount: 2,
 			}),
 		).toBe(true);
 	});
@@ -1436,7 +1445,7 @@ describe("shouldDeferHeartbeatCronJob", () => {
 		expect(
 			shouldDeferHeartbeatCronJob(
 				{ ...baseJob, source: "heartbeat" },
-				{ isStreaming: false, isBashRunning: false, unfinishedActionCount: 0 },
+				{ isStreaming: false, isBashRunning: false, hasPendingSessionWork: false, unfinishedActionCount: 0 },
 			),
 		).toBe(false);
 	});
@@ -1445,7 +1454,7 @@ describe("shouldDeferHeartbeatCronJob", () => {
 		expect(
 			shouldDeferHeartbeatCronJob(
 				{ ...baseJob, source: "cron" },
-				{ isStreaming: true, isBashRunning: true, unfinishedActionCount: 2 },
+				{ isStreaming: true, isBashRunning: true, hasPendingSessionWork: true, unfinishedActionCount: 2 },
 			),
 		).toBe(false);
 	});

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

@@ -384,25 +384,6 @@ describe("DaemonClient", () => {
 		});
 	});
 
-	it("keeps a raw one-release request path for v1 daemon handoff", async () => {
-		const client = new DaemonClient("/tmp/prime-agent.sock");
-		const connect = client.connect();
-		const socket = netMock.sockets[0]!;
-		socket.emit("connect");
-		await connect;
-
-		const response = client.requestLegacy({ type: "prepare_update_restart" });
-		const command = JSON.parse(socket.writes[0]!.trim()) as { id: string; type: string };
-		expect(command.type).toBe("prepare_update_restart");
-		expect(command).not.toHaveProperty("protocol");
-		socket.emit(
-			"data",
-			`${JSON.stringify({ id: command.id, type: "response", command: command.type, success: true })}\n`,
-		);
-		await expect(response).resolves.toMatchObject({ success: true });
-		client.close();
-	});
-
 	it("keeps durable command envelopes on the session-action protocol", async () => {
 		const client = new DaemonClient("/tmp/prime-agent.sock");
 		const connect = client.connect();

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

@@ -4620,7 +4620,7 @@ describe("daemon mode helpers", () => {
 		},
 		{
 			name: "does not enqueue another heartbeat when one is already pending",
-			activity: { isStreaming: true, unfinishedActionCount: 1 },
+			activity: { isStreaming: true, hasPendingSessionWork: true, unfinishedActionCount: 1 },
 			jobs: [{ id: "heartbeat-1", source: "heartbeat" }],
 			acceptingAgentMessage: false,
 			assertQueuedHeartbeatUntouched: true,
@@ -4742,6 +4742,8 @@ describe("daemon mode helpers", () => {
 		const sessionState = {
 			isStreaming: false,
 			isBashRunning: false,
+			hasPendingSessionWork: false,
+			unfinishedActionCount: 1,
 			sessionActions: { queuedCount: 0, steering: [], followUps: [] },
 		};
 		const prompt = vi.fn(async (_message: string, options?: { preflightResult?: (didSucceed: boolean) => void }) => {
@@ -5803,6 +5805,7 @@ type CronAdmissionActivity = Partial<{
 	isCompacting: boolean;
 	isRetrying: boolean;
 	isBashRunning: boolean;
+	hasPendingSessionWork: boolean;
 	unfinishedActionCount: number;
 }>;
 
@@ -5841,6 +5844,7 @@ function makeCronAdmissionFixture(
 			isCompacting: false,
 			isRetrying: false,
 			isBashRunning: false,
+			hasPendingSessionWork: false,
 			unfinishedActionCount: 0,
 			...activity,
 			prompt,

packages/coding-agent/test/session-action-store.test.ts

@@ -67,7 +67,6 @@ function activity(overrides: Partial<RuntimeActivity> = {}): RuntimeActivity {
 		branchMutation: false,
 		schedulerPauseCount: 0,
 		disposing: false,
-		mandatoryCheckpointsComplete: true,
 		...overrides,
 	};
 }