fix(acp): wait for settled prompt after Stop; skip SESSION_BUSY auto-retry

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)
+  })
 })

src/acp/acp-adapter.ts

@@ -458,6 +458,11 @@ export const connectAcpAdapter = async (
     readonly persistence?: Promise<void>
   }
   const freshPending = new Map<string, FreshSessionState>()
+  // 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<string, Promise<void>>()
 
   const guardHandshake = <T>(step: Promise<T>): Promise<T> =>
     withHandshakeGuard(step, transport.closed, handshakeTimeoutMs)
@@ -550,6 +555,11 @@ export const connectAcpAdapter = async (
   const fetch = async (init: RequestInit, context: AgentAdapterContext): Promise<Response> => {
     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<void>((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()
       }
     })()
 

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,

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(

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}`