diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 1716ecd6e..dac5b6058 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/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)). diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 1ddc8761b..95c2fe853 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/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(); } } diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index 2f075b024..ad76c960d 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/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(); diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index dfebcdf61..3d7813c0c 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/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, }); diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 596d0dd13..aa71b42e2 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/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(); @@ -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((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((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((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((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); });