fix(coding-agent): answer ACP prompts only once the session admits the next turn

packages/coding-agent/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 - Added ACP `_meta.quiescence` to report outstanding subagents and remaining autonomous continuations; `session/prompt` rejects if its live child-roster read fails.
+- Fixed ACP rejecting an immediate follow-up prompt with "Agent is already processing" when injected work (subagent replies, heartbeats) restarted the session after the previous turn; the follow-up now queues behind the in-flight work instead of failing, and `session/cancel` drops a queued follow-up that has not started.
 - 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)).

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

@@ -4464,12 +4464,44 @@ export class AgentSession {
 		const outcome = this._agentMessageOutcome(agentMessageId);
 		outcome.completion = createAgentMessageDeferred();
 		const completion = outcome.completion.promise;
+		// Admission-time signal checks cannot see an abort that lands once a
+		// followUp prompt is already queued behind busy work. Cancel the not-yet-
+		// started action here so promptAndWait rejects and the caller can report
+		// the cancellation instead of leaving the queued prompt to run later.
+		const signal = options?.signal;
+		let cancelQueuedPrompt: (() => void) | undefined;
+		const detachCancelQueuedPrompt = () => {
+			if (signal && cancelQueuedPrompt) {
+				signal.removeEventListener("abort", cancelQueuedPrompt);
+			}
+		};
 		try {
 			await this.promptUntilAccepted(text, { ...options, agentMessageId });
+			if (signal) {
+				cancelQueuedPrompt = () => {
+					const error = new Error("Prompt was cancelled before it started.");
+					const cancelled = this._cancelSessionActions(
+						(action) =>
+							action.agentMessageId === agentMessageId &&
+							action.payload.kind === "turn" &&
+							(action.lifecycle.state === "queued" || action.lifecycle.state === "selected"),
+						error,
+					);
+					if (cancelled.length > 0) {
+						this._settleAgentMessage(agentMessageId, "completion", error);
+					}
+				};
+				// Register before re-checking aborted: an abort between the check
+				// and registration would otherwise never invoke the listener.
+				signal.addEventListener("abort", cancelQueuedPrompt, { once: true });
+				if (signal.aborted) cancelQueuedPrompt();
+			}
 			await completion;
 		} catch (error) {
 			this._settleAgentMessage(agentMessageId, "completion", this._asError(error));
 			throw error;
+		} finally {
+			detachCancelQueuedPrompt();
 		}
 	}
 

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

@@ -387,7 +387,21 @@ export async function runAcpModeWithConnection(
 				// rebuild the transcript mid-turn, so record the pre-turn messages
 				// themselves rather than how many there were.
 				const priorMessages = turnBoundary(await connection.getMessages());
-				await connection.promptAndWait(text, images.length > 0 ? { images } : undefined);
+				// A follow-up prompt can arrive while injected work (subagent replies,
+				// heartbeats) keeps the resident session busy. ACP has no native queue
+				// field, so queue the host turn behind that work with follow-up
+				// semantics instead of rejecting it as "Agent is already processing".
+				// promptAndWait then resolves once the queued turn has actually run,
+				// which is also what keeps the stop reason below attributed to the
+				// turn this prompt gates.
+				await connection.promptAndWait(text, {
+					...(images.length > 0 ? { images } : {}),
+					streamingBehavior: "followUp",
+					queueIfBusy: true,
+					// session/cancel must drop a prompt that is still waiting in the
+					// queue, not just the turn currently streaming.
+					signal: abort.signal,
+				});
 				// Autonomous gates continue inside this same prompt turn: the turn is
 				// only over once the gate loop settles.
 				const status = await connection.waitForHeadlessCompletion();

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

@@ -3742,6 +3742,9 @@ export class AgentDaemon {
 					});
 				}
 				if (admission.status === "owned") {
+					// The prompt is session-owned; still propagate the cancel so a
+					// queued not-yet-started prompt is dropped rather than run later.
+					admission.controller?.abort();
 					return success(command.id, command.type, {
 						status: "owned" as const,
 					});

packages/coding-agent/test/suite/acp-mode.test.ts

@@ -62,6 +62,7 @@ function fakeAcpConnection(
 			return snapshot;
 		},
 		promptAndWait: async () => {},
+		waitForIdle: async () => {},
 		waitForHeadlessCompletion: async () => ({
 			enabled: false,
 			continuationsUsed: 0,
@@ -75,6 +76,31 @@ function fakeAcpConnection(
 	};
 }
 
+/**
+ * Real injected work with production timing: after the first headless-completion
+ * idle observation, this starts a genuine session turn (a subagent reply has the
+ * same shape) and blocks until the session is streaming it. Returns a checker
+ * for whether the injection has happened.
+ */
+function injectWorkAfterHeadlessCompletion(connection: any, session: any, text: string): () => boolean {
+	const waitForHeadlessCompletion = connection.waitForHeadlessCompletion.bind(connection);
+	let injected = false;
+	connection.waitForHeadlessCompletion = async () => {
+		const status = await waitForHeadlessCompletion();
+		if (!injected) {
+			injected = true;
+			void session.prompt(text);
+			const deadline = Date.now() + 5_000;
+			while (!session.isStreaming && Date.now() < deadline) {
+				await new Promise((resolve) => setTimeout(resolve, 1));
+			}
+			expect(session.isStreaming).toBe(true);
+		}
+		return status;
+	};
+	return () => injected;
+}
+
 function connectAcpClient(connection: any): ClientHarness {
 	// Two web streams crossed over: agent's stdout is the client's stdin.
 	const toAgent = new TransformStream<Uint8Array, Uint8Array>();
@@ -294,4 +320,202 @@ describe("ACP mode end to end", () => {
 		expect(quiescence.update._meta[PRIME_AGENT_META_NAMESPACE].quiescence.outstandingSubagents).toBe(1);
 		close();
 	});
+	// Production race: after ACP reported a completed turn, injected work (a
+	// subagent reply, heartbeat, goal message) restarted the resident session, so
+	// the client's next prompt was rejected with "Agent is already processing".
+	// Each regression injects real work after the first idle observation and
+	// drives a real ACP client, rather than asserting call ordering.
+
+	it("queues a follow-up prompt behind injected work instead of rejecting it", async () => {
+		const harness = await createHarness();
+		let releaseInjected!: () => void;
+		const injectedHeld = new Promise<any>((resolve) => {
+			releaseInjected = () => resolve(fauxAssistantMessage("injected work done"));
+		});
+		harness.setResponses([
+			fauxAssistantMessage("turn one done"),
+			() => injectedHeld,
+			fauxAssistantMessage("turn two done"),
+		]);
+		const connection = new InProcessAgentConnection(runtimeHostFor(harness.session));
+		const injected = injectWorkAfterHeadlessCompletion(connection, harness.session, "injected work");
+		const { client, updates } = connectAcpClient(connection);
+		await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} });
+		const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] });
+
+		const first = await client.request("session/prompt", {
+			sessionId: session.sessionId,
+			prompt: [{ type: "text", text: "First turn" }],
+		});
+		expect(first.stopReason).toBe("end_turn");
+		expect(injected()).toBe(true);
+		// The injected turn is still streaming, so the session is busy when the
+		// follow-up prompt arrives, exactly as in the production campaign.
+		expect(harness.session.isStreaming).toBe(true);
+
+		let secondSettled = false;
+		const second = client
+			.request("session/prompt", {
+				sessionId: session.sessionId,
+				prompt: [{ type: "text", text: "Second turn" }],
+			})
+			.finally(() => {
+				secondSettled = true;
+			});
+		// The queued turn must neither resolve early nor reject with
+		// "Agent is already processing".
+		await new Promise((resolve) => setTimeout(resolve, 50));
+		expect(secondSettled).toBe(false);
+		expect(harness.session.isStreaming).toBe(true);
+		releaseInjected();
+		await expect(second).resolves.toMatchObject({ stopReason: "end_turn" });
+		expect(harness.session.isStreaming).toBe(false);
+
+		const text = updates
+			.filter((u) => u.update?.sessionUpdate === "agent_message_chunk")
+			.map((u) => u.update.content.text)
+			.join("");
+		expect(text).toContain("turn two done");
+
+		harness.cleanup();
+	}, 5_000);
+
+	it("reports the queued turn's stop reason from a fresh autonomous status", async () => {
+		const harness = await createHarness({
+			autonomous: {
+				enabled: true,
+				maxTurns: 2,
+				maxContinuations: 3,
+				maxTokens: 80_000,
+				gates: { commands: ["true"], maxRetries: 3 },
+			},
+		});
+		let releaseInjected!: () => void;
+		const injectedHeld = new Promise<any>((resolve) => {
+			releaseInjected = () => resolve(fauxAssistantMessage("injected work done"));
+		});
+		harness.setResponses([
+			fauxAssistantMessage("turn one done"),
+			() => injectedHeld,
+			fauxAssistantMessage("turn two done"),
+		]);
+		const connection = new InProcessAgentConnection(runtimeHostFor(harness.session));
+		injectWorkAfterHeadlessCompletion(connection, harness.session, "injected work");
+		const { client } = connectAcpClient(connection);
+		await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} });
+		const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] });
+
+		const first = await client.request("session/prompt", {
+			sessionId: session.sessionId,
+			prompt: [{ type: "text", text: "First turn" }],
+		});
+		expect(first.stopReason).toBe("end_turn");
+		expect(harness.session.getAutonomousStatus().turnsUsed).toBe(1);
+
+		const second = client.request("session/prompt", {
+			sessionId: session.sessionId,
+			prompt: [{ type: "text", text: "Second turn" }],
+		});
+		await new Promise((resolve) => setTimeout(resolve, 50));
+		expect(harness.session.isStreaming).toBe(true);
+		releaseInjected();
+		// The second prompt's response gates its own queued turn, so the stop
+		// reason comes from the status captured after that turn ran, not from a
+		// snapshot taken before the injected work consumed the autonomous budget.
+		await expect(second).resolves.toMatchObject({ stopReason: "max_turn_requests" });
+		expect(harness.session.getAutonomousStatus().turnsUsed).toBeGreaterThanOrEqual(
+			harness.session.getAutonomousStatus().limits.maxTurns,
+		);
+		harness.cleanup();
+	}, 5_000);
+
+	it("does not hold the prompt response open for detached work", async () => {
+		const harness = await createHarness();
+		let releaseInjected!: () => void;
+		const injectedHeld = new Promise<any>((resolve) => {
+			releaseInjected = () => resolve(fauxAssistantMessage("injected work done"));
+		});
+		harness.setResponses([fauxAssistantMessage("turn one done"), () => injectedHeld]);
+		const connection = new InProcessAgentConnection(runtimeHostFor(harness.session));
+		const injected = injectWorkAfterHeadlessCompletion(connection, harness.session, "injected work");
+		const { client } = connectAcpClient(connection);
+		await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} });
+		const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] });
+
+		let settled = false;
+		const prompt = client
+			.request("session/prompt", {
+				sessionId: session.sessionId,
+				prompt: [{ type: "text", text: "First turn" }],
+			})
+			.finally(() => {
+				settled = true;
+			});
+		const deadline = Date.now() + 5_000;
+		while (!injected() && Date.now() < deadline) {
+			await new Promise((resolve) => setTimeout(resolve, 1));
+		}
+		expect(injected()).toBe(true);
+		// The response boundary reports quiescence rather than waiting for it:
+		// detached work must not hold the session/prompt response open, which is
+		// exactly the quiescence-blocking #806 rejected.
+		await new Promise((resolve) => setTimeout(resolve, 100));
+		expect(settled).toBe(true);
+		expect(harness.session.isStreaming).toBe(true);
+		releaseInjected();
+		await expect(prompt).resolves.toMatchObject({ stopReason: "end_turn" });
+		harness.cleanup();
+	}, 5_000);
+	it("cancels a prompt that is still queued behind busy work", async () => {
+		const harness = await createHarness();
+		let releaseInjected!: () => void;
+		const injectedHeld = new Promise<any>((resolve) => {
+			releaseInjected = () => resolve(fauxAssistantMessage("injected work done"));
+		});
+		harness.setResponses([
+			fauxAssistantMessage("turn one done"),
+			() => injectedHeld,
+			fauxAssistantMessage("queued turn done"),
+		]);
+		const connection = new InProcessAgentConnection(runtimeHostFor(harness.session));
+		const injected = injectWorkAfterHeadlessCompletion(connection, harness.session, "injected work");
+		const { client } = connectAcpClient(connection);
+		await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} });
+		const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] });
+
+		const first = await client.request("session/prompt", {
+			sessionId: session.sessionId,
+			prompt: [{ type: "text", text: "First turn" }],
+		});
+		expect(first.stopReason).toBe("end_turn");
+		expect(injected()).toBe(true);
+		expect(harness.session.isStreaming).toBe(true);
+
+		const queued = client.request("session/prompt", {
+			sessionId: session.sessionId,
+			prompt: [{ type: "text", text: "Second turn" }],
+		});
+		await new Promise((resolve) => setTimeout(resolve, 50));
+		expect(harness.session.isStreaming).toBe(true);
+
+		// session/cancel must drop the queued prompt, not leave it to run once
+		// the busy turn finishes. The notification handler may wait on the
+		// in-flight turn, so do not await it here.
+		void client.notify("session/cancel", { sessionId: session.sessionId });
+		await expect(queued).resolves.toMatchObject({ stopReason: "cancelled" });
+		const texts = () =>
+			harness.session.messages
+				.filter((m) => m.role === "assistant")
+				.map((m: any) =>
+					Array.isArray(m.content)
+						? m.content.map((p: any) => (p.type === "text" ? p.text : p.type)).join(" ")
+						: String(m.content),
+				);
+		expect(texts().join("|")).not.toContain("queued turn done");
+		// Releasing the busy turn must not let the cancelled prompt execute.
+		releaseInjected();
+		await new Promise((resolve) => setTimeout(resolve, 300));
+		expect(texts().join("|")).not.toContain("queued turn done");
+		harness.cleanup();
+	}, 5_000);
 });