refactor: simplify turn preemption by waiting on backend completion

src/backend/listener.ts

@@ -56,12 +56,12 @@ export class EventStreamListener {
    * misleading "version too old" string.
    */
   async subscribe(nextId: NextId): Promise<ZcodeSnapshot> {
-    // Retry transient timeouts: after a preempt (`session/stop`) the backend's
-    // main loop can be briefly busy finalising the cancelled turn (interrupting
-    // the model stream, persisting checkpoints). A subscribe issued in that
-    // window may not get a response within the per-attempt deadline. The backend
-    // recovers on its own ("wait a bit and resend works"), so a couple of
-    // retries absorb the intermittent stall instead of surfacing it to the user.
+    // Retry transient timeouts as a lightweight safety net for cold-start /
+    // network blips. The cancel-preempt path no longer needs subscribe retries
+    // to absorb a backend stop-finalization window: the turn loop now blocks
+    // until the backend emits turn.completed/turn.failed before its prompt()
+    // exits, so by the time the next prompt reaches subscribe the backend is
+    // already idle. These retries are just a last-resort cushion.
     //
     // Only `timeout` is retried — non-transient errors (reader dead, pipe
     // broken, method-not-found, session-level business error) fail fast.
@@ -76,8 +76,8 @@ export class EventStreamListener {
     // O(messages) and can take seconds-to-tens-of-seconds on long sessions:
     // observed 3.8s at 267 messages, 38s at 3500 messages) and is the primary
     // cause of subscribe timeouts. The eventSeq watermark is all we need.
-    const MAX_ATTEMPTS = 3;
-    const backoffMs = (attempt: number): number => 1000 * attempt; // 1s, 2s
+    const MAX_ATTEMPTS = 2;
+    const backoffMs = (attempt: number): number => 500 * attempt; // 0.5s
     let resp: ZcodeResponse;
     for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
       resp = await this.backend.request(
@@ -89,18 +89,18 @@ export class EventStreamListener {
           includeSnapshot: false,
           afterSeq: 0,
         },
-        10000,
+        5000,
       );
       if (!resp.error) break; // success
       const isTimeout = resp.error.message === "timeout";
       if (!isTimeout || attempt === MAX_ATTEMPTS) {
         if (isTimeout) {
-          warn(`subscribe: all ${MAX_ATTEMPTS} attempts timed out (backend unresponsive for ~30s)`);
+          warn(`subscribe: all ${MAX_ATTEMPTS} attempts timed out (backend unresponsive for ~${Math.round((MAX_ATTEMPTS * 5000 + 500) / 1000)}s)`);
         }
         throw new Error(formatSubscribeError(resp));
       }
       log(
-        `subscribe attempt ${attempt}/${MAX_ATTEMPTS} timed out, retrying in ${backoffMs(attempt)}ms (preempt/busy window)`,
+        `subscribe attempt ${attempt}/${MAX_ATTEMPTS} timed out, retrying in ${backoffMs(attempt)}ms`,
       );
       await new Promise((resolve) => setTimeout(resolve, backoffMs(attempt)));
     }

src/handlers/session.ts

@@ -34,7 +34,6 @@ import {
 import type { InternalEvent } from "../translators/index.js";
 import { log, warn } from "../utils.js";
 import type { PendingTurn, ZcodeAcpServer } from "../server.js";
-import { waitForTurnIdle } from "./extensions.js";
 import { dispatchEvent } from "./dispatch.js";
 import { sendSessionUpdate, sendTextChunk } from "./io.js";
 import { handleServerRequests } from "./server-requests.js";
@@ -542,11 +541,10 @@ class TurnFailedError extends Error {
  * `_cancel_backend_turn`: send stop with an id (some backends route by id
  * presence), never wait for a response, never throw.
  *
- * Probing the lock afterward is the caller's job — on the preempt path
- * (preemptInFlightTurn) we follow up with waitForTurnIdle to confirm release;
- * the turn-loop cancel sites call this and rely on their own event-driven
- * exit. The backend's turn loop emits turn.completed on its own (success if
- * stop didn't catch the streaming turn in time, cancelled otherwise).
+ * The turn-loop cancel site calls this once (guarded by turn.stopSent), then
+ * keeps looping until the backend emits turn.completed/turn.failed. preempt
+ * waits on pendingTurns deletion — which only happens after that backend
+ * completion event — so it transitively waits for the backend to be done.
  */
 function stopBackendTurn(server: ZcodeAcpServer, zcodeSid: string): void {
   try {
@@ -599,19 +597,22 @@ function withPreemptLock(
 
 /**
  * Cancel any other in-flight turn for this zcodeSid and wait for it to fully
- * exit (listener unregistered + pendingTurns cleaned) before returning.
+ * exit (pendingTurns cleaned) before returning.
  *
  * Must be called from inside a preempt lock section (the caller has already
  * registered itself in pendingTurns), so a concurrent prompt entering its own
  * section is guaranteed to see this caller's turn and cancel it.
  *
- * Why wait for the map entry to disappear (not just fire stop): registering
- * a second EventStreamListener overwrites the first (Map.set in client.ts),
- * so the old turn loop must have run its finally block before we subscribe.
- * The map cleanup in that finally block is the synchronization point.
+ * We do NOT fire stop here — the old turn's own loop detects turn.cancelled
+ * and fires stop itself (turn-loop cancel site), then keeps looping until the
+ * backend emits turn.completed/turn.failed. Since the turn loop now waits for
+ * that backend completion event before exiting, the pendingTurns cleanup in
+ * its finally block is the reliable "backend is done, lock released" signal.
+ * Waiting on pendingTurns deletion therefore blocks until the backend has
+ * truly finished — far more reliable than probing session/goal show (which
+ * times out during the backend's stop-finalization window).
  *
- * Best-effort: never throws. On timeout, continues anyway — session/send
- * will then hit the lock and take the existing error path.
+ * Best-effort: never throws. On timeout, continues anyway.
  */
 async function preemptInFlightTurn(
   server: ZcodeAcpServer,
@@ -623,40 +624,27 @@ async function preemptInFlightTurn(
   for (const [reqId, turn] of server.pendingTurns) {
     if (turn.zcodeSid === zcodeSid && reqId !== selfRequestId) {
       oldRequestId = reqId;
-      turn.cancelled = true; // signal the old turn loop to exit
+      turn.cancelled = true; // signal the old turn loop to fire stop + wait
       break;
     }
   }
   if (oldRequestId === undefined) return; // no in-flight turn, proceed
 
-  log(`  [preempt] in-flight turn ${oldRequestId} found, stopping it`);
-  // Fire-and-forget stop (mirrors Python's _cancel_backend_turn). The old
-  // turn loop will receive turn.completed and exit on its own.
-  stopBackendTurn(server, zcodeSid);
-
-  // The old turn is already streaming (lock held), so expectLock=false: we
-  // don't need to observe the lock appearing first, only its release. Probe
-  // session/goal show until it stops reporting "prompt is running" — this is
-  // the same mechanism waitForTurnIdle uses for compact/goal, and unlike
-  // pendingTurns polling it does NOT depend on the backend emitting a
-  // cancelled event (which never comes for a streaming turn hit by stop:
-  // it completes as success, not cancelled).
-  //
-  // pendingTurns cleanup by the old turn's finally block still happens and
-  // remains the listener-overwrite guard, but we no longer BLOCK on it.
-  const PREEMPT_TIMEOUT_MS = 35_000;
-  const released = await waitForTurnIdle(
-    server,
-    zcodeSid,
-    PREEMPT_TIMEOUT_MS,
-    "session/goal",
-    false, // old turn already holds the lock; just wait for release
-  );
-  if (!released) {
-    warn(`  [preempt] lock wait timed out for old turn ${oldRequestId}, proceeding best-effort`);
-  } else {
-    log(`  [preempt] old turn ${oldRequestId} lock released, proceeding`);
+  log(`  [preempt] in-flight turn ${oldRequestId} found, cancelling`);
+
+  // Wait for the old turn to fully exit. The turn loop's cancel handling fires
+  // stop and keeps looping until the backend emits turn.completed/turn.failed,
+  // so pendingTurns deletion only happens once the backend is truly done.
+  const PREEMPT_TIMEOUT_MS = 120_000;
+  const t0 = Date.now();
+  while (server.pendingTurns.has(oldRequestId)) {
+    if (Date.now() - t0 > PREEMPT_TIMEOUT_MS) {
+      warn(`  [preempt] timed out waiting for old turn ${oldRequestId} to exit`);
+      return;
+    }
+    await sleep(200);
   }
+  log(`  [preempt] old turn ${oldRequestId} exited, proceeding`);
 }
 
 // ---------- internals ----------
@@ -890,8 +878,20 @@ async function runEventTurn(
     }
 
     if (turn.cancelled) {
-      stopBackendTurn(server, turn.zcodeSid);
-      return { stopReason: "cancelled" };
+      // Send stop ONCE, then keep looping to wait for the backend's turn
+      // completion event. Returning immediately here lets the old turn's
+      // prompt() exit (deleting pendingTurns) BEFORE the backend finishes
+      // processing stop — the next prompt's subscribe/send then collides with
+      // the still-finalizing backend (observed 18-41s recovery window). By
+      // continuing the loop we block until the backend emits turn.completed/
+      // turn.failed (translator.turnDone check below), the real "backend done"
+      // signal. pendingTurns stays until then, so the next prompt's preempt
+      // waits on it.
+      if (!turn.stopSent) {
+        stopBackendTurn(server, turn.zcodeSid);
+        turn.stopSent = true;
+      }
+      // Fall through to pollEvent — keep receiving events until turn end.
     }
 
     const ev = await listener.pollEvent(500);

src/server.ts

@@ -31,6 +31,8 @@ export interface ClientCapabilities {
 export interface PendingTurn {
   zcodeSid: string;
   cancelled: boolean;
+  /** Set once session/stop has been fired for this turn, to avoid re-sending. */
+  stopSent?: boolean;
 }
 
 export class ZcodeAcpServer {

tests/bugfixes.test.ts

@@ -110,9 +110,8 @@ describe("subscribe retries transient timeout (preempt busy window)", () => {
     const requestSpy = vi.spyOn(fake, "request");
     requestSpy
       .mockResolvedValueOnce({ id: 1, error: { message: "timeout" } })
-      .mockResolvedValueOnce({ id: 2, error: { message: "timeout" } })
       .mockResolvedValueOnce({
-        id: 3,
+        id: 2,
         result: { eventSeq: 5, snapshot: snap },
       } as ZcodeResponse);
     // Fake timers so the backoff sleeps don't slow the test.
@@ -125,7 +124,7 @@ describe("subscribe retries transient timeout (preempt busy window)", () => {
     expect(result).toEqual(snap);
     expect(listener.subscribed).toBe(true);
     expect(listener.lastSeq).toBe(5);
-    expect(requestSpy).toHaveBeenCalledTimes(3);
+    expect(requestSpy).toHaveBeenCalledTimes(2);
   });
 
   it("does NOT retry on non-transient errors (reader dead)", async () => {