feat(ai-chat): interrupt agent + send queued message, keeping partial output (#198)

Add a "send now" button to queued AI-chat messages: it interrupts the
running agent and immediately sends that message, while the agent's
partial output at interruption is kept in history and the next turn is
marked as a user interrupt.

Client:
- queue-helpers: pure `promoteToHead` to move a queued message to the head.
- chat-thread: `sendNow` (promote head + abort + flush-on-abort), one-shot
  `flushOnAbortRef`/`interruptNextSendRef`, `interrupted` flag in the
  request body, and the "send now" ActionIcon in the queued list.

Server:
- `interrupted` on AiChatStreamBody; pure `isInterruptResume` confirms the
  client hint against persisted history (prev assistant turn aborted/
  streaming) before honouring it.
- prompt: INTERRUPT_NOTE injected in the context section only on a
  confirmed interrupt-resume turn so the model treats the partial answer
  above as incomplete.

Tests: promoteToHead, chat-thread send-now (abort + resend + one-shot
interrupt flag + non-streaming immediate send), isInterruptResume, and
the prompt interrupt-note injection.

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

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

@@ -0,0 +1,142 @@
+import { describe, it, expect, beforeEach, vi } from "vitest";
+import { render, screen, fireEvent, act } from "@testing-library/react";
+import { MantineProvider } from "@mantine/core";
+
+// Shared, hoisted mock state so the @ai-sdk/react and "ai" module mocks (hoisted
+// above the imports) can expose the captured useChat callbacks / transport and
+// the spies back to the test body.
+const h = vi.hoisted(() => ({
+  state: {
+    status: "streaming" as string,
+    onFinish: null as null | ((arg: Record<string, unknown>) => void),
+    sendMessage: vi.fn(),
+    stop: vi.fn(),
+    transport: null as null | {
+      prepareSendMessagesRequest: (arg: {
+        messages: unknown[];
+        body: Record<string, unknown>;
+      }) => { body: Record<string, unknown> };
+    },
+  },
+}));
+
+// Mock useChat: capture onFinish, return the spies and the controllable status.
+vi.mock("@ai-sdk/react", () => ({
+  useChat: (opts: { onFinish?: (arg: Record<string, unknown>) => void }) => {
+    h.state.onFinish = opts.onFinish ?? null;
+    return {
+      messages: [],
+      sendMessage: h.state.sendMessage,
+      status: h.state.status,
+      stop: h.state.stop,
+      error: null,
+    };
+  },
+}));
+
+// Mock "ai": deterministic ids + a transport that records its options so the test
+// can invoke prepareSendMessagesRequest and assert the `interrupted` flag.
+vi.mock("ai", () => {
+  let counter = 0;
+  return {
+    generateId: () => `gid-${counter++}`,
+    DefaultChatTransport: class {
+      constructor(opts: {
+        prepareSendMessagesRequest: (arg: {
+          messages: unknown[];
+          body: Record<string, unknown>;
+        }) => { body: Record<string, unknown> };
+      }) {
+        h.state.transport = opts;
+      }
+    },
+  };
+});
+
+// Stub the heavy children: MessageList (markdown/render) and ChatInput (the
+// composer). The ChatInput stub exposes a button that queues a message, the only
+// interaction this test needs to populate the queue while "streaming".
+vi.mock("@/features/ai-chat/components/message-list.tsx", () => ({
+  default: () => <div data-testid="message-list" />,
+}));
+vi.mock("@/features/ai-chat/components/chat-input.tsx", () => ({
+  default: ({ onQueue }: { onQueue: (text: string) => void }) => (
+    <button data-testid="queue-btn" onClick={() => onQueue("queued text")}>
+      queue
+    </button>
+  ),
+}));
+
+import ChatThread from "./chat-thread";
+
+function renderThread() {
+  const onTurnFinished = vi.fn();
+  render(
+    <MantineProvider>
+      <ChatThread chatId="c1" initialRows={[]} onTurnFinished={onTurnFinished} />
+    </MantineProvider>,
+  );
+  return { onTurnFinished };
+}
+
+describe("ChatThread — send now (#198)", () => {
+  beforeEach(() => {
+    h.state.status = "streaming";
+    h.state.onFinish = null;
+    h.state.sendMessage.mockClear();
+    h.state.stop.mockClear();
+    h.state.transport = null;
+  });
+
+  it("aborts the current turn and resends the queued message on the abort", () => {
+    renderThread();
+
+    // Queue a message while the turn is streaming.
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    const sendNowBtn = screen.getByLabelText("Send now");
+    expect(sendNowBtn).toBeTruthy();
+
+    // "Send now" interrupts the current turn (stop), but does NOT send yet —
+    // the resend happens once the abort lands in onFinish.
+    fireEvent.click(sendNowBtn);
+    expect(h.state.stop).toHaveBeenCalledTimes(1);
+    expect(h.state.sendMessage).not.toHaveBeenCalled();
+
+    // The abort we triggered reaches onFinish: the promoted head is flushed.
+    act(() => {
+      h.state.onFinish?.({
+        message: { id: "a", role: "assistant", parts: [] },
+        isAbort: true,
+        isDisconnect: false,
+        isError: false,
+      });
+    });
+    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" });
+  });
+
+  it("tags exactly the next send as interrupted (one-shot flag)", () => {
+    renderThread();
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    const prep = h.state.transport!.prepareSendMessagesRequest;
+    // The send right after "send now" carries interrupted: true...
+    expect(prep({ messages: [], body: {} }).body.interrupted).toBe(true);
+    // ...and only that one (the flag is read-and-cleared).
+    expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false);
+  });
+
+  it("sends immediately without an interrupt when not streaming", () => {
+    h.state.status = "ready";
+    renderThread();
+
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now"));
+
+    // No turn to interrupt: sent straight away, no abort, not flagged.
+    expect(h.state.stop).not.toHaveBeenCalled();
+    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" });
+    const prep = h.state.transport!.prepareSendMessagesRequest;
+    expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false);
+  });
+});

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

@@ -1,7 +1,11 @@
 import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import { generateId } from "ai";
-import { ActionIcon, Box, Group, Stack, Text } from "@mantine/core";
-import { IconClockHour4, IconX } from "@tabler/icons-react";
+import { ActionIcon, Box, Group, Stack, Text, Tooltip } from "@mantine/core";
+import {
+  IconClockHour4,
+  IconPlayerPlayFilled,
+  IconX,
+} from "@tabler/icons-react";
 import { useTranslation } from "react-i18next";
 import { useChat, type UIMessage } from "@ai-sdk/react";
 import { DefaultChatTransport } from "ai";
@@ -23,6 +27,7 @@ import { extractServerChatId } from "@/features/ai-chat/utils/adopt-chat-id.ts";
 import {
   dequeue,
   enqueueMessage,
+  promoteToHead,
   removeQueuedById,
   type QueuedMessage,
 } from "@/features/ai-chat/utils/queue-helpers.ts";
@@ -201,6 +206,16 @@ export default function ChatThread({
   // helper can call the current instance from the stable `onFinish` callback.
   const sendMessageRef = useRef<((m: { text: string }) => void) | null>(null);
 
+  // "Send now" single-flight flags. Kept in refs (not state) so they are read
+  // inside the stable `onFinish` callback and the transport closure WITHOUT a
+  // re-render or a stale closure. Both are one-shot (read-and-clear).
+  // - flushOnAbortRef: flush the promoted head on the abort WE triggered, even
+  //   though an aborted turn normally keeps the queue intact.
+  // - interruptNextSendRef: tag the next send as a user interrupt so the server
+  //   injects the "your previous answer was interrupted" note for that turn only.
+  const flushOnAbortRef = useRef(false);
+  const interruptNextSendRef = useRef(false);
+
   // FIFO dequeue + send the next queued message (no-op when the queue is empty).
   const flushNext = useCallback(() => {
     const { head, rest } = dequeue(queuedRef.current);
@@ -232,17 +247,26 @@ export default function ChatThread({
         // when null) and tell the agent which page "this page" refers to. Both
         // are read live from refs so changing chats/pages does NOT recreate the
         // transport. `openPage` is null on a non-page route.
-        prepareSendMessagesRequest: ({ messages, body }) => ({
-          body: {
-            ...body,
-            chatId: chatIdRef.current,
-            openPage: openPageRef.current,
-            // Honoured by the server only when creating a new chat; null =>
-            // universal assistant.
-            roleId: roleIdRef.current,
-            messages,
-          },
-        }),
+        prepareSendMessagesRequest: ({ messages, body }) => {
+          // Read-and-clear the interrupt flag so the "you were interrupted" note
+          // is carried by ONLY this request (the one resending the promoted
+          // message right after we aborted the previous turn). The server still
+          // confirms it against history before acting on it.
+          const interrupted = interruptNextSendRef.current;
+          interruptNextSendRef.current = false; // one-shot
+          return {
+            body: {
+              ...body,
+              chatId: chatIdRef.current,
+              openPage: openPageRef.current,
+              // Honoured by the server only when creating a new chat; null =>
+              // universal assistant.
+              roleId: roleIdRef.current,
+              interrupted,
+              messages,
+            },
+          };
+        },
       }),
     [],
   );
@@ -277,6 +301,17 @@ export default function ChatThread({
       else if (isAbort) setStopNotice("manual");
       else if (isDisconnect) setStopNotice("disconnect");
       else setStopNotice(null);
+      // "Send now": WE triggered this abort to interrupt the current turn and
+      // immediately send the promoted head. Flush it even though the turn was
+      // aborted (the normal abort path below keeps the queue intact). The
+      // interrupt note travels with this send via interruptNextSendRef.
+      if (flushOnAbortRef.current) {
+        flushOnAbortRef.current = false;
+        // Suppress the "Response stopped." flash for an intentional interrupt.
+        setStopNotice(null);
+        flushNext();
+        return;
+      }
       if (isAbort || isDisconnect || isError) return;
       flushNext();
     },
@@ -329,6 +364,31 @@ export default function ChatThread({
 
   const isStreaming = status === "submitted" || status === "streaming";
 
+  // "Send now" on a queued message: interrupt the current turn and immediately
+  // send THIS message, keeping the agent's partial output. Other queued messages
+  // stay queued and flush normally after the new turn. Reuses the existing
+  // queue/flush machinery: promote the target to the head, then abort — the
+  // onFinish flush-on-abort branch sends exactly that head, tagged as an
+  // interrupt so the server notes the previous answer was cut off.
+  const sendNow = useCallback(
+    (id: string) => {
+      if (isStreaming) {
+        // Promote to head so the onFinish -> flushNext path sends exactly it.
+        setQueue(promoteToHead(queuedRef.current, id));
+        flushOnAbortRef.current = true;
+        interruptNextSendRef.current = true;
+        stop(); // -> onFinish({ isAbort: true }) flushes the promoted head
+      } else {
+        // Nothing to interrupt: just send it now (no interrupt note).
+        const msg = queuedRef.current.find((m) => m.id === id);
+        if (!msg) return;
+        setQueue(removeQueuedById(queuedRef.current, id));
+        sendMessageRef.current?.({ text: msg.text });
+      }
+    },
+    [isStreaming, setQueue, stop],
+  );
+
   // Clear the stopped marker as soon as a new turn begins streaming.
   useEffect(() => {
     if (isStreaming) setStopNotice(null);
@@ -423,6 +483,17 @@ export default function ChatThread({
                 <Text size="xs" lineClamp={2} className={classes.queuedText}>
                   {m.text}
                 </Text>
+                <Tooltip label={t("Interrupt and send now")} withArrow>
+                  <ActionIcon
+                    size="xs"
+                    variant="subtle"
+                    color="blue"
+                    onClick={() => sendNow(m.id)}
+                    aria-label={t("Send now")}
+                  >
+                    <IconPlayerPlayFilled size={12} />
+                  </ActionIcon>
+                </Tooltip>
                 <ActionIcon
                   size="xs"
                   variant="subtle"

apps/client/src/features/ai-chat/utils/queue-helpers.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect } from "vitest";
 import {
   enqueueMessage,
   dequeue,
+  promoteToHead,
   removeQueuedById,
   type QueuedMessage,
 } from "./queue-helpers";
@@ -89,6 +90,52 @@ describe("removeQueuedById", () => {
   });
 });
 
+describe("promoteToHead", () => {
+  it("moves the matching id to the front, preserving the rest's order", () => {
+    const queue: QueuedMessage[] = [
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+      { id: "c", text: "third" },
+    ];
+    expect(promoteToHead(queue, "c")).toEqual([
+      { id: "c", text: "third" },
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ]);
+  });
+
+  it("is a no-op order-wise when the id is already the head", () => {
+    const queue: QueuedMessage[] = [
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ];
+    expect(promoteToHead(queue, "a")).toEqual([
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ]);
+  });
+
+  it("returns an equivalent list when the id is not present", () => {
+    const queue: QueuedMessage[] = [
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ];
+    expect(promoteToHead(queue, "missing")).toEqual(queue);
+  });
+
+  it("does not mutate the input queue", () => {
+    const queue: QueuedMessage[] = [
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ];
+    promoteToHead(queue, "b");
+    expect(queue).toEqual([
+      { id: "a", text: "first" },
+      { id: "b", text: "second" },
+    ]);
+  });
+});
+
 describe("FIFO order", () => {
   it("preserves order across enqueue -> dequeue", () => {
     let queue: QueuedMessage[] = [];

apps/client/src/features/ai-chat/utils/queue-helpers.ts

@@ -32,3 +32,16 @@ export function removeQueuedById(
 ): QueuedMessage[] {
   return queue.filter((m) => m.id !== id);
 }
+
+/** Move the queued message with the given id to the FRONT (returns a new array).
+ *  No-op (returns an equivalent array) when the id is absent. Pure — backs the
+ *  "send now" action: promoting a message to the head lets the existing
+ *  onFinish -> flushNext path send exactly that message on the abort we trigger. */
+export function promoteToHead(
+  queue: QueuedMessage[],
+  id: string,
+): QueuedMessage[] {
+  const target = queue.find((m) => m.id === id);
+  if (!target) return queue;
+  return [target, ...queue.filter((m) => m.id !== id)];
+}