diff --git a/src/acp/acp-adapter.test.ts b/src/acp/acp-adapter.test.ts index 346c16bc9..e59b1ab38 100644 --- a/src/acp/acp-adapter.test.ts +++ b/src/acp/acp-adapter.test.ts @@ -356,7 +356,7 @@ describe('connectAcpAdapter — handshake failure modes', () => { describe('connectAcpAdapter — per-thread session multiplexing over one connection', () => { it('resolves a separate ACP session per thread and persists each independently', async () => { const { transport } = buildFakeTransport() - const { FakeConnection, calls } = buildFakeConnection() + const { FakeConnection, calls, releasePrompts } = buildFakeConnection() const adapter = await connectAcpAdapter(remoteAgent, baseCtx(), { openTransport: async () => transport, @@ -366,11 +366,11 @@ describe('connectAcpAdapter — per-thread session multiplexing over one connect const persistedA: string[] = [] const persistedB: string[] = [] - await adapter.fetch( + const responseA = await adapter.fetch( promptInit('a'), threadCtx('thread-A', { onAcpSessionId: async (s) => void persistedA.push(s) }), ) - await adapter.fetch( + const responseB = await adapter.fetch( promptInit('b'), threadCtx('thread-B', { onAcpSessionId: async (s) => void persistedB.push(s) }), ) @@ -381,6 +381,15 @@ describe('connectAcpAdapter — per-thread session multiplexing over one connect expect(persistedA).toEqual(['sess-1']) expect(persistedB).toEqual(['sess-2']) + // Settle in-flight prompts before a follow-up send on thread-A — the adapter + // awaits the prior turn per session so we don't race a busy ACP slot. + await act(async () => { + releasePrompts() + await getClock().runAllAsync() + await readSse(responseA) + await readSse(responseB) + }) + // A second send on thread-A reuses its cached session — no extra newSession. await adapter.fetch(promptInit('a2'), threadCtx('thread-A')) expect(calls.newSession).toHaveLength(2) @@ -916,4 +925,53 @@ describe('connectAcpAdapter — Stop cancels the remote ACP turn', () => { expect(sse.join('')).toContain('"type":"finish"') expect(calls.cancel).toHaveLength(0) }) + + it('waits for an aborted prompt to settle before starting another prompt on the same session', async () => { + const { transport } = buildFakeTransport() + const { FakeConnection, calls, releasePrompts } = buildFakeConnection() + + const adapter = await connectAcpAdapter(remoteAgent, baseCtx(), { + openTransport: async () => transport, + ClientSideConnection: FakeConnection as never, + }) + + const controller = new AbortController() + const first = await adapter.fetch(promptInit('long turn', controller.signal), threadCtx('thread-A')) + expect(calls.prompt).toHaveLength(1) + + await act(async () => { + controller.abort() + await getClock().runAllAsync() + await readSse(first) + }) + + // Local stream is torn down, but the remote prompt is still gated open — + // a follow-up send must not race a second session/prompt onto the wire. + let secondStarted = false + const secondPromise = adapter + .fetch(promptInit('next'), threadCtx('thread-A', { acpSessionId: 'sess-1' })) + .then((response) => { + secondStarted = true + return response + }) + + await act(async () => { + await getClock().tickAsync(1) + }) + + expect(secondStarted).toBe(false) + expect(calls.prompt).toHaveLength(1) + + let second: Response | undefined + await act(async () => { + releasePrompts() + await getClock().runAllAsync() + second = await secondPromise + await readSse(second) + }) + + expect(secondStarted).toBe(true) + expect(calls.prompt).toHaveLength(2) + expect(calls.cancel).toHaveLength(1) + }) }) diff --git a/src/acp/acp-adapter.ts b/src/acp/acp-adapter.ts index a4af5592a..974e8ddd7 100644 --- a/src/acp/acp-adapter.ts +++ b/src/acp/acp-adapter.ts @@ -458,6 +458,11 @@ export const connectAcpAdapter = async ( readonly persistence?: Promise } const freshPending = new Map() + // In-flight `session/prompt` per ACP session. Stop tears down the local SSE + // stream immediately, but the remote turn may still hold the agent's busy + // slot until `connection.prompt` settles — the next fetch awaits this so we + // don't race a second prompt onto a still-busy session. + const inFlightPromptBySession = new Map>() const guardHandshake = (step: Promise): Promise => withHandshakeGuard(step, transport.closed, handshakeTimeoutMs) @@ -550,6 +555,11 @@ export const connectAcpAdapter = async ( const fetch = async (init: RequestInit, context: AgentAdapterContext): Promise => { const sessionId = await resolveThreadSession(context) + const previousPrompt = inFlightPromptBySession.get(sessionId) + if (previousPrompt) { + await previousPrompt + } + // First real send of a freshly-minted session: persist the id we actually // used, and (tier-3 only) seed the prior transcript as context. The marker // is keyed on "fresh session's first prompt" (not "we prepended a @@ -598,7 +608,9 @@ export const connectAcpAdapter = async ( // Stop: the AI SDK aborts the *local* stream, but the remote ACP turn keeps // running (burning tokens, still executing tool calls) until we tell the // agent. Send `session/cancel` — fire-and-forget, since teardown must not - // block on the wire — then run the same teardown. + // block on the wire — then run the same teardown. The next `fetch` still + // awaits `inFlightPromptBySession` so a follow-up send cannot race the + // agent's busy slot before this prompt settles. const onAbort = (): void => { // Fire-and-forget: teardown must not block on the wire. A rejection means // the transport is already tearing down (the turn is ending anyway), so @@ -608,6 +620,12 @@ export const connectAcpAdapter = async ( teardown() } + let settleInFlight!: () => void + const inFlight = new Promise((resolve) => { + settleInFlight = resolve + }) + inFlightPromptBySession.set(sessionId, inFlight) + // Drive the prompt off the request thread — the response stream is the // synchronous return value so the AI SDK can attach immediately. void (async () => { @@ -624,6 +642,10 @@ export const connectAcpAdapter = async ( translator.error(err instanceof Error ? err.message : String(err)) } finally { teardown() + if (inFlightPromptBySession.get(sessionId) === inFlight) { + inFlightPromptBySession.delete(sessionId) + } + settleInFlight() } })() diff --git a/src/chats/chat-instance.ts b/src/chats/chat-instance.ts index 8ceef3c6e..44fba2776 100644 --- a/src/chats/chat-instance.ts +++ b/src/chats/chat-instance.ts @@ -11,7 +11,13 @@ import { getAllSkills as defaultGetAllSkills } from '@/dal' import { isBuiltInAgent } from '@/defaults/agents' import { extractLastUserText, resolveSkillTokenInstructions } from '@/skills/resolve-skill-system-messages' import { getDb as defaultGetDb } from '@/db/database' -import { getErrorRetryable, isContentRejectionError, isContextOverflowError, isRateLimitError } from '@/lib/error-utils' +import { + getErrorRetryable, + isAcpSessionBusyError, + isContentRejectionError, + isContextOverflowError, + isRateLimitError, +} from '@/lib/error-utils' import type { HttpClient } from '@/lib/http' import { trackEvent } from '@/lib/posthog' import type { FetchFn } from '@/lib/proxy-fetch' @@ -327,6 +333,14 @@ export const createChatInstance = ( return } + // Don't auto-retry ACP SESSION_BUSY — the prior turn still owns the slot + // (common right after Stop). Blind retries worsen the race. + if (isAcpSessionBusyError(lastError)) { + lastError = null + useChatStore.getState().updateSession(id, { retriesExhausted: true }) + return + } + // Don't burn retries on errors that won't succeed on identical input: // context overflow, or anything the provider marks non-retryable (4xx // content/auth errors, unsupported content). Transient errors — 408/409, diff --git a/src/lib/error-utils.test.ts b/src/lib/error-utils.test.ts index 62c021ecd..1132cf29d 100644 --- a/src/lib/error-utils.test.ts +++ b/src/lib/error-utils.test.ts @@ -7,6 +7,7 @@ import { createHandleError, getErrorRetryable, getErrorStatusCode, + isAcpSessionBusyError, isContentRejectionError, isContextOverflowError, isRateLimitError, @@ -273,6 +274,25 @@ describe('getErrorRetryable', () => { }) }) +describe('isAcpSessionBusyError', () => { + it('detects Zeroclaw / ACP "active prompt turn" busy rejects', () => { + expect( + isAcpSessionBusyError(new Error('Session already has an active prompt turn: 50be29d3-97bd-4bcc-8012-dd7e0a160ac3')), + ).toBe(true) + expect( + isAcpSessionBusyError( + new Error(JSON.stringify({ error: 'Session already has an active prompt turn: sess-1', status: 500 })), + ), + ).toBe(true) + }) + + it('returns false for unrelated errors', () => { + expect(isAcpSessionBusyError(new Error('Network timeout'))).toBe(false) + expect(isAcpSessionBusyError(new Error(JSON.stringify({ error: 'Server error', status: 500 })))).toBe(false) + expect(isAcpSessionBusyError(null)).toBe(false) + }) +}) + describe('isContextOverflowError', () => { it("detects Anthropic's 'prompt is too long' overflow", () => { const error = new Error( diff --git a/src/lib/error-utils.ts b/src/lib/error-utils.ts index ebd27e070..cb3b4ea2a 100644 --- a/src/lib/error-utils.ts +++ b/src/lib/error-utils.ts @@ -28,6 +28,21 @@ export const isRateLimitError = (error?: Error | null): boolean => { return error.message.toLowerCase().includes('too many requests') } +/** + * ACP agents (e.g. Zeroclaw) reject a new `session/prompt` while a prior turn + * still holds the session slot — often right after Stop, when `session/cancel` + * has fired but the turn has not exited yet. Auto-retrying the identical send + * only amplifies the race; surface the error and let the user retry once idle. + */ +export const isAcpSessionBusyError = (error?: Error | null): boolean => { + if (!error?.message) { + return false + } + const parsed = parseJson(error.message) + const message = (typeof parsed?.error === 'string' ? parsed.error : error.message).toLowerCase() + return message.includes('active prompt turn') +} + /** * Extract an HTTP status code from a serialized stream/transport error, if one * is present. The frontend serializes API errors as `{"error":...,"status":N}`