fix(ai-chat): «send now» во время detached-run — авторитетный серверный stop + ограниченный ретрай 409 (#396)

В автономном режиме «Interrupt and send now» во время живого detached-run делал
только локальный stop() (abort SSE), который сервер игнорирует (run живёт по
дизайну #184/#234), поэтому onFinish→flush новый POST упирался в гейт «один
активный run на чат» → 409 A_RUN_ALREADY_ACTIVE, новый turn не стартовал.
handleStop делает правильно (доп. onServerStop), sendNow — нет. Вариант A
(клиентский, горячий путь сервера не тронут):

- sendNow в автономном режиме дополнительно зовёт onServerStop(chatId) (или
  откладывает через stopPendingRef, если chatId ещё не усыновлён — как
  handleStop) и взводит one-shot supersedeRetryRef ДО stop();
- транспорт-fetch на supersede-отправке (и только на ней) ретраит РОВНО 409 с
  body.code===A_RUN_ALREADY_ACTIVE до 4 попыток с бэкоффом 150/300/600ms;
  onServerStop гарантирует осадку run → ретрай сходится. Обычная отправка (не
  взведён флаг) падает на 409 мгновенно. isRunAlreadyActive читает
  response.clone() → тело оригинала возвращается потребителю нетронутым.
  409 всегда до записи user-строки (pre-check/beginRun раньше insert) → повтор
  POST безопасен, дублей нет.

Внутренний цикл: 2 прохода. Ревью нашло CRITICAL: supersedeRetryRef застревал
взведённым, когда sendNow взвёл, но POST не ушёл (promoted head удалён →
flushNext() false; либо wasResumed-return) — следующая обычная отправка молча
ретраила настоящий 409. Починка: разоружать флаг симметрично остальным one-shot
(в ветке !flushNext() и в isStreaming-defuse-эффекте); транспорт read-and-clear
на входе каждой отправки. Мутационно: убрать disarm → strand-тест краснеет
(4 вызова вместо 1). Легаси-режим не тронут (регресс-гард).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/client/src/features/ai-chat/components/chat-thread.test.tsx

@@ -28,7 +28,10 @@ const h = vi.hoisted(() => ({
         body: Record<string, unknown>;
       }) => { body: Record<string, unknown> };
       prepareReconnectToStreamRequest?: () => { api?: string };
-      fetch?: (input: unknown, init?: { method?: string }) => Promise<unknown>;
+      fetch?: (
+        input: unknown,
+        init?: { method?: string; body?: unknown },
+      ) => Promise<unknown>;
     },
   },
 }));
@@ -200,6 +203,244 @@ describe("ChatThread — send now (#198)", () => {
   });
 });
 
+// #396: in autonomous mode a live sendNow must additionally request the
+// AUTHORITATIVE server stop of the detached run (a local abort is only a client
+// disconnect the server ignores) and arm a bounded 409 retry so the re-POST
+// converges once the one-active-run slot frees. Legacy mode is unchanged.
+describe("ChatThread — send now server-stop + supersede retry (#396)", () => {
+  beforeEach(resetState);
+  afterEach(cleanup);
+
+  // A settled assistant tail => no mount resume (attemptResumeRef false), so the
+  // "Send now" button is visible for the NEW local streaming turn while
+  // autonomous runs are enabled.
+  const settledTail = () => [
+    row("u1", "user", undefined, "hi"),
+    row("a1", "assistant", "succeeded", "done"),
+  ];
+
+  it("autonomous: sendNow during a live stream calls onServerStop with the chat id", () => {
+    const { onServerStop } = renderThread({
+      autonomousRunsEnabled: true,
+      initialRows: settledTail(),
+    });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    expect(h.state.stop).toHaveBeenCalledTimes(1);
+    expect(onServerStop).toHaveBeenCalledWith("c1");
+  });
+
+  it("legacy (autonomous off): sendNow does NOT call onServerStop and does NOT retry the send", async () => {
+    const { onServerStop } = renderThread({
+      autonomousRunsEnabled: false,
+      initialRows: settledTail(),
+    });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+    expect(onServerStop).not.toHaveBeenCalled();
+
+    // The supersede retry must NOT be armed: a POST that 409s is returned as-is
+    // (single fetch, no retry).
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(
+        new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), {
+          status: 409,
+        }),
+      );
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(res.status).toBe(409);
+  });
+
+  it("armed supersede send retries 409 A_RUN_ALREADY_ACTIVE and succeeds once the slot frees", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    // Arm the retry by performing a live sendNow (autonomous branch sets the ref).
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    const fetchMock = vi
+      .fn()
+      // First POST: the old detached run still holds the slot -> 409.
+      .mockResolvedValueOnce(
+        new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), {
+          status: 409,
+        }),
+      )
+      // Retry: the server stop settled the old run -> 200.
+      .mockResolvedValueOnce(new Response("ok", { status: 200 }));
+    vi.stubGlobal("fetch", fetchMock);
+
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(res.status).toBe(200);
+  });
+
+  it("supersede retry is one-shot: a later send (ref cleared) does NOT retry a 409", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now")); // arms the one-shot
+
+    // First armed send: immediately succeeds, consuming the arm.
+    let fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response("ok", { status: 200 }));
+    vi.stubGlobal("fetch", fetchMock);
+    await act(async () => {
+      await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      });
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    // A subsequent send is NOT armed -> a 409 is returned as-is (no retry).
+    fetchMock = vi.fn().mockResolvedValue(
+      new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), {
+        status: 409,
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(res.status).toBe(409);
+  });
+
+  it("supersede retry is bounded: exhaustion surfaces the 409 error", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    // Every attempt 409s -> after 4 attempts the last 409 surfaces.
+    const fetchMock = vi.fn().mockResolvedValue(
+      new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), {
+        status: 409,
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    // 4 attempts total (1 immediate + 3 backoff retries), then give up.
+    expect(fetchMock).toHaveBeenCalledTimes(4);
+    expect(res.status).toBe(409);
+  });
+
+  it("armed supersede send does NOT retry a non-409 status", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response("boom", { status: 500 }));
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(res.status).toBe(500);
+  });
+
+  // Strand-path regression: sendNow arms the supersede retry, but if the promoted
+  // head is removed before the abort's onFinish lands, flushNext() sends nothing
+  // (returns false) and NO re-POST consumes the arm. The arm must be disarmed on
+  // that no-send branch so the NEXT unrelated NORMAL send does not inherit it and
+  // silently retry a genuine 409 (e.g. a legitimate two-tab conflict) 4x instead
+  // of surfacing it immediately.
+  it("strand-path: a stranded supersede arm (flushNext no-send) does NOT retry a later normal 409", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    // Arm the retry via a live autonomous sendNow (promotes the head + arms).
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    // Remove the promoted head BEFORE the abort lands, so flushNext() returns
+    // false (no POST) and the arm would strand without the disarm fix.
+    fireEvent.click(screen.getByLabelText("Remove queued message"));
+
+    // The abort's onFinish now takes the flushOnAbortRef branch, calls flushNext()
+    // which finds an empty queue and returns false -> the no-send disarm must run.
+    act(() => {
+      h.state.onFinish?.({
+        message: { id: "a1", role: "assistant", parts: [] },
+        isAbort: true,
+        isDisconnect: false,
+        isError: false,
+      });
+    });
+    // No re-POST was sent (nothing to flush).
+    expect(h.state.sendMessage).not.toHaveBeenCalled();
+
+    // A subsequent NORMAL send that 409s must be returned as-is (exactly 1 fetch):
+    // the stranded arm must NOT cause the genuine 409 to be retried.
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(
+        new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), {
+          status: 409,
+        }),
+      );
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(res.status).toBe(409);
+  });
+
+  it("armed supersede send does NOT retry a 409 with a different (non-A_RUN_ALREADY_ACTIVE) body", async () => {
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+    const fetchMock = vi.fn().mockResolvedValue(
+      new Response(JSON.stringify({ code: "SOMETHING_ELSE" }), {
+        status: 409,
+      }),
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    let res!: Response;
+    await act(async () => {
+      res = (await h.state.transport!.fetch!("http://x", {
+        method: "POST",
+        body: "{}",
+      })) as Response;
+    });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(res.status).toBe(409);
+  });
+});
+
 // #388: the editor selection is snapshotted at send time and nested inside
 // openPage on the wire. The getter is read live from a ref, so each send ships a
 // fresh snapshot.

apps/client/src/features/ai-chat/components/chat-thread.tsx

@@ -70,6 +70,36 @@ const RECONNECT_MAX_ATTEMPTS = 5;
 // Backoff before attempt N (1-based): 1s, 2s, 4s, 8s, 16s.
 const RECONNECT_BASE_DELAY_MS = 1000;
 
+// #396: bounded retry for the "Interrupt and send now" re-send when it races the
+// authoritative server stop of the just-superseded detached run. The re-POST can
+// arrive before the old run has released the one-active-run slot, so the server
+// returns 409 A_RUN_ALREADY_ACTIVE. The server stop guarantees the slot frees, so
+// a few short backoffs converge. 4 total attempts: attempt 1 fires immediately,
+// then these are the waits BEFORE attempts 2, 3 and 4 (150ms, 300ms, 600ms). If
+// all 4 attempts 409, the last 409 surfaces (the banner) — acceptable per #396.
+const SUPERSEDE_RETRY_DELAYS_MS = [150, 300, 600];
+// The server error code that means "another run is already active for this chat".
+const A_RUN_ALREADY_ACTIVE = "A_RUN_ALREADY_ACTIVE";
+
+/**
+ * #396: defensively decide whether a 409 response is the one-active-run gate
+ * rejection (code A_RUN_ALREADY_ACTIVE) vs. some other 409. Reads a CLONE so the
+ * original response body stays intact for the caller when it is returned as-is.
+ * Any parse failure or unexpected shape => false (do NOT retry).
+ */
+async function isRunAlreadyActive(response: Response): Promise<boolean> {
+  try {
+    const body = (await response.clone().json()) as unknown;
+    return (
+      typeof body === "object" &&
+      body !== null &&
+      (body as { code?: unknown }).code === A_RUN_ALREADY_ACTIVE
+    );
+  } catch {
+    return false;
+  }
+}
+
 /** The page the user is currently viewing, sent as chat context. */
 export interface OpenPageContext {
   id: string;
@@ -326,6 +356,26 @@ export default function ChatThread({
   const flushOnAbortRef = useRef(false);
   const interruptNextSendRef = useRef(false);
 
+  // #396: one-shot arm for the bounded 409 A_RUN_ALREADY_ACTIVE retry on the
+  // "Interrupt and send now" re-send in autonomous mode. sendNow triggers the
+  // authoritative server stop of the detached run, but that stop and the
+  // onFinish->flushNext re-POST race: the new POST can hit the one-active-run
+  // gate before the old detached run has settled, yielding a spurious 409. When
+  // this ref is armed, the transport's send path retries that 409 with a short
+  // bounded backoff (the server stop guarantees convergence). A normal send (ref
+  // not armed) must STILL fail a 409 instantly (e.g. a genuine two-tab conflict).
+  //
+  // INVARIANT: sendNow arms this only to be consumed by the ONE re-POST that
+  // flushNext fires from onFinish. But that re-POST does not always happen (the
+  // promoted head may be gone, the finish may be a resumed turn, or the arm may
+  // race a stale finish). To keep the arm strictly one-shot it is disarmed on
+  // EVERY path where the paired interrupt one-shots (flushOnAbortRef /
+  // interruptNextSendRef) are cleared without a POST: the transport POST branch
+  // consumes it (read-and-clear), the onFinish `!flushNext()` no-send branch
+  // clears it, and the isStreaming-defuse effect clears it symmetrically. So it
+  // can never leak into a later, unrelated send and retry that send's genuine 409.
+  const supersedeRetryRef = useRef(false);
+
   // #234 F5: the user pressed Stop while streaming a BRAND-NEW chat whose server
   // chat id has not been adopted yet (the `start` chunk carrying it hadn't landed
   // when Stop was pressed). A local SSE abort alone does NOT stop the DETACHED
@@ -382,7 +432,43 @@ export default function ChatThread({
           }`,
         }),
         fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => {
-          if ((init.method ?? "GET") !== "GET") return fetch(input, init); // send path untouched
+          if ((init.method ?? "GET") !== "GET") {
+            // Send path (POST). #396: read-and-clear the one-shot supersede arm
+            // here so it is strictly scoped to THIS send. When unarmed, behave
+            // exactly as before — a single fetch, a 409 surfaces instantly (a
+            // genuine two-tab conflict must NOT be retried).
+            const supersede = supersedeRetryRef.current;
+            supersedeRetryRef.current = false;
+            if (!supersede) return fetch(input, init);
+            // Buffer a ReadableStream body once so each retry can replay it.
+            // DefaultChatTransport sends the body as a JSON STRING (replayable as
+            // is), but guard defensively in case a future SDK streams it.
+            let sendInit = init;
+            if (init.body instanceof ReadableStream) {
+              const buffered = await new Response(init.body).arrayBuffer();
+              sendInit = { ...init, body: buffered };
+            }
+            // Bounded retry: attempt 1 fires immediately, then wait between
+            // attempts per SUPERSEDE_RETRY_DELAYS_MS. Retry ONLY on a real
+            // 409 A_RUN_ALREADY_ACTIVE; any other status/body is returned as-is.
+            for (let attempt = 0; ; attempt++) {
+              const response = await fetch(input, sendInit);
+              if (
+                response.status !== 409 ||
+                attempt >= SUPERSEDE_RETRY_DELAYS_MS.length ||
+                !(await isRunAlreadyActive(response))
+              ) {
+                return response;
+              }
+              // The old detached run has not released the one-active-run slot
+              // yet; the server stop we requested guarantees it will, so back off
+              // and re-POST (the 409 fired before the user message was persisted,
+              // so re-POSTing is safe — no duplicate rows).
+              await new Promise((r) =>
+                setTimeout(r, SUPERSEDE_RETRY_DELAYS_MS[attempt]),
+              );
+            }
+          }
           // Reconnect GET: the SDK passes no AbortSignal, so wire our own controller
           // for observer Stop / unmount abort.
           const controller = new AbortController();
@@ -562,9 +648,14 @@ export default function ChatThread({
         setStopNotice(null);
         // If the promoted head vanished (e.g. the user removed it before the
         // abort landed) flushNext sends nothing — clear the one-shot interrupt
-        // tag so it can't leak onto the next unrelated send. On a real send the
-        // tag is consumed by prepareSendMessagesRequest and stays untouched.
-        if (!flushNext()) interruptNextSendRef.current = false;
+        // tag AND the #396 supersede arm so neither can leak onto the next
+        // unrelated send (no re-POST will consume the arm here). On a real send
+        // the tag is consumed by prepareSendMessagesRequest and the arm by the
+        // transport POST branch, so both stay untouched then.
+        if (!flushNext()) {
+          interruptNextSendRef.current = false;
+          supersedeRetryRef.current = false;
+        }
         return;
       }
       if (isAbort || isDisconnect || isError) return;
@@ -873,6 +964,30 @@ export default function ChatThread({
         setQueue(promoteToHead(queuedRef.current, id));
         flushOnAbortRef.current = true;
         interruptNextSendRef.current = true;
+        // #396: in autonomous mode the turn is a DETACHED run — a local stop()
+        // is only a client disconnect the server ignores, so the run keeps going.
+        // The onFinish->flushNext re-POST would then hit the one-active-run gate
+        // and get a spurious 409 A_RUN_ALREADY_ACTIVE. Mirror handleStop: request
+        // the AUTHORITATIVE server stop so the detached run settles, and arm the
+        // one-shot bounded 409 retry BEFORE stop() so the re-send converges once
+        // the slot frees. Read chatId live from chatIdRef (adopted at the `start`
+        // chunk). If it is not known yet (brand-new chat, first moment of its
+        // first turn), defer the server stop via stopPendingRef exactly as
+        // handleStop does — the onServerChatId adoption effect fires it once the
+        // id lands; the retry stays armed so the re-send still converges then.
+        if (autonomousRunsEnabled) {
+          supersedeRetryRef.current = true; // arm the bounded 409 retry
+          if (chatIdRef.current) {
+            onServerStop?.(chatIdRef.current);
+          } else {
+            // Same #234-F5 sub-window limitation documented in handleStop: if the
+            // local abort below cancels the reader before the `start` chunk lands,
+            // the adoption effect never runs and the deferred stop never fires. Not
+            // a regression; at minimum we don't strand refs (the isStreaming effect
+            // defuses stopPendingRef on the next turn start).
+            stopPendingRef.current = true;
+          }
+        }
         stop(); // -> onFinish({ isAbort: true }) flushes the promoted head
       } else {
         // Nothing to interrupt: just send it now (no interrupt note).
@@ -884,7 +999,7 @@ export default function ChatThread({
         sendMessageRef.current?.({ text: msg.text });
       }
     },
-    [setQueue, stop, setResumedTurnPair],
+    [setQueue, stop, setResumedTurnPair, autonomousRunsEnabled, onServerStop],
   );
 
   // Stop the current turn. ALWAYS abort the local SSE (`stop()`) so the composer
@@ -944,6 +1059,13 @@ export default function ChatThread({
       setStopNotice(null);
       flushOnAbortRef.current = false;
       interruptNextSendRef.current = false;
+      // #396: symmetric with the other one-shot interrupt flags — defuse a stale
+      // supersede arm that was set but whose expected re-POST never fired (the
+      // turn finished in the same tick as the click, or the promoted head was
+      // gone), so it can never leak into this (or a later) turn's send and retry
+      // that send's genuine 409. A legit arm is consumed by the transport POST
+      // branch before this new turn streams, so this does not clobber it.
+      supersedeRetryRef.current = false;
       // #234 F5: a new turn is starting — drop any pending deferred-stop from a
       // previous turn that never adopted an id, so it can never fire against this
       // (or a later) unrelated turn's run. A deferred stop for the CURRENT turn is