Let a user stop an in-flight agent turn (CL-7201)

apps/hub/src/index.ts

@@ -73,6 +73,7 @@ import {
   createWorkbenchSubscriberRegistry,
   createWorkbenchTenancyRoutes,
   createWorkbenchTurnQueue,
+  createTurnCancelRegistry,
   createChatOrchestrator,
   createChatRoutes,
   joinRunParticipant,
@@ -1358,6 +1359,10 @@ export async function createHub(config: HubConfig) {
     claims: createInMemoryTurnClaimStore({ ttlMs: DEFAULT_TURN_CLAIM_TTL_MS }),
     publish: workbenchSubscribers.publish,
   });
+  // The live abort seam a running turn is reachable through (CL-7201) —
+  // shared the same way `turnQueue` above is, so a cancel request lands
+  // wherever a workbench's turn was actually dispatched from.
+  const turnCancellation = createTurnCancelRegistry();
   // The room timeline store (CL-6327): a workbench's own messages, held
   // as workbench data rather than platform mail.
   const roomMessages = createDrizzleRoomMessageStore(db);
@@ -1572,6 +1577,7 @@ export async function createHub(config: HubConfig) {
     clientIds: createDrizzleClientIdStore(db),
     workbenchSubscribers,
     turnQueue,
+    turnCancellation,
     requireGrant: createRequireGrant({
       grantStore: chatGrantStore,
       conditionRegistry: chatConditionRegistry,
@@ -1618,6 +1624,7 @@ export async function createHub(config: HubConfig) {
       roomMessages,
       publish: workbenchSubscribers.publish,
       turnQueue,
+      turnCancellation,
       authenticator: createWorkflowRunAuthenticator({ db }),
       tenancy: chatTenancy,
       sessionFor,
@@ -1641,6 +1648,7 @@ export async function createHub(config: HubConfig) {
     sessionFor,
     workbenchSubscribers,
     turnQueue,
+    turnCancellation,
   });
   // Tells the routine trigger popover whether a Slack-bound webhook
   // trigger is honestly offerable in this deployment — no session or
@@ -2800,6 +2808,7 @@ export async function createHub(config: HubConfig) {
               roomMessages,
               publish: workbenchSubscribers.publish,
               turnQueue,
+              turnCancellation,
             },
             {
               tenantId: input.tenantId,

apps/hub/src/slack-tag-mount.ts

@@ -26,6 +26,7 @@ import {
   type WorkbenchSubscriberRegistry,
   type WorkbenchTenancyStore,
   type WorkbenchTurnQueue,
+  type TurnCancelRegistry,
   type ChatPlatform,
   type ChatStore,
   type RoomMessageStore,
@@ -72,6 +73,9 @@ export type MountWorkbenchSlackTagDeps = {
    * send and a person's own message for the same channel serialize
    * against each other too. */
   readonly turnQueue: WorkbenchTurnQueue;
+  /** The same cancellation registry `createChatRoutes` is given
+   * (CL-7201) — shared, never a second instance. */
+  readonly turnCancellation: TurnCancelRegistry;
 };
 
 export type MountedWorkbenchSlackTag = { readonly mounted: boolean };
@@ -211,6 +215,7 @@ export async function mountWorkbenchSlackTag(
           roomMessages: deps.roomMessages,
           publish: deps.workbenchSubscribers.publish,
           turnQueue: deps.turnQueue,
+          turnCancellation: deps.turnCancellation,
         },
         {
           tenantId: input.tenantId,

docs/CHAT.md

@@ -171,6 +171,56 @@ agents replying in one room under distinct occurrences, three rapid
 messages serializing into ordered turns, and a sidecar killed
 mid-occurrence leaving both the room and the section alive.
 
+## Stopping a turn (CL-7201)
+
+Before this, the only bound on a wedged turn was the dispatch and
+wait-until-free timeouts themselves — minutes long, and no way for a
+user watching an agent go wrong to do anything but wait or reload.
+`POST /workbenches/:id/turns/cancel` (`packages/chat/src/routes.ts`)
+closes that gap, calling `cancelWorkbenchTurn`
+(`packages/chat/src/workbench-service.ts`), which runs two independent
+mechanisms together because a turn can be in either place when the user
+asks to stop it:
+
+- **Still on our own call stack.** `dispatchTurnBatch` registers one
+  `AbortController` per recipient it dispatches, via a workbench-keyed
+  `TurnCancelRegistry` (`packages/chat/src/turn-cancellation.ts`) —
+  the same "one instance, shared" pattern `WorkbenchTurnQueue` follows.
+  Its signal is composed into each `withTimeout` call (`waitUntilFree`,
+  `dispatchTurn`) as an **external signal** — `withTimeout` (CL-7193)
+  already gives `work` an `AbortSignal` the moment its own timeout wins;
+  CL-7201 extends it to also fire (with the external caller's own
+  reason, not its own timeout message) the moment that external signal
+  aborts, whichever comes first. `dispatchTurn`'s abort-close handler
+  tells a deliberate cancellation apart from a timeout by checking
+  whether the abort reason is a `TurnCancelledError`, and closes the
+  turn row `cancelled` rather than `failed`.
+- **Already off our call stack.** `sendMail` has no cancellable
+  primitive of its own (CL-7230) — once it resolves, the agent is
+  generating (or parked on a `message_response` gate somewhere in the
+  execution plane this package cannot see into) with nothing left
+  registered to abort. `cancelWorkbenchTurn` snapshots every turn
+  `AgentTurnStore.findRunningTurns` reports for the workbench _before_
+  triggering the registry above (so a row the abort path already
+  claimed is still counted), then sweeps it directly through the same
+  `finishTurn` compare-and-set.
+
+Both mechanisms race the same compare-and-set, so whichever reaches a
+given row first is the only one that ever settles it or posts a notice
+— `postCancelledNotice`, a `turnCancelled` text part distinct from
+`postUndeliveredNotice`'s `turnFailed` (a cancellation is not a
+failure, and the timeline says so). CL-7230's ceiling is honest, not
+silent: settling the row is not the same as stopping the underlying
+agent process. A late `connector.reply` that lands anyway finds no
+`running` row left to attach to, and `postReply` will not fall back to
+posting it unattached onto a 1:1 membership whose latest turn is
+`cancelled`. The composer offers a Stop affordance
+(`packages/chat-ui/src/composer.tsx`) whenever `isAwaitingReply`
+(`streaming-reply.ts`) is true — the whole in-flight phase, including
+after tokens have started streaming — independent of its own `sending`
+state. A follow-up message can still be typed and queued while a turn
+runs.
+
 ## Threads: workbench → thread → sub-thread
 
 A workbench's timeline is itself a thread — its **root thread**, one per

packages/chat-ui/src/api.ts

@@ -1318,7 +1318,7 @@ const AgentTurnWire = type({
   workbenchId: "string",
   agentAddress: "string",
   childRunId: "string",
-  status: "'running' | 'completed' | 'failed'",
+  status: "'running' | 'completed' | 'failed' | 'cancelled'",
   "replyMessageId?": "string | null",
 });
 export type AgentTurnSummary = typeof AgentTurnWire.infer;
@@ -1354,6 +1354,31 @@ export function getWorkbenchTurn(
   );
 }
 
+const CancelWorkbenchTurnWire = type({ cancelledCount: "number" });
+export type CancelWorkbenchTurnResult = typeof CancelWorkbenchTurnWire.infer;
+
+/**
+ * Stops a workbench's in-flight turn(s) (CL-7201) — `POST
+ * .../turns/cancel` in `packages/chat/src/routes.ts`. `cancelledCount`
+ * is the honest count of turns actually settled `cancelled`, not a
+ * promise that the underlying agent process stopped (see CL-7230): the
+ * composer's own Stop affordance treats any non-throwing response as
+ * "asked," and relies on the timeline's cancelled-turn notice — not this
+ * response — to clear the typing indicator.
+ */
+export function cancelWorkbenchTurn(
+  tenantId: string,
+  workbenchId: string,
+): Promise<CancelWorkbenchTurnResult> {
+  return request(
+    `${turnsPath(tenantId, workbenchId)}/cancel`,
+    CancelWorkbenchTurnWire,
+    {
+      method: "POST",
+    },
+  );
+}
+
 /**
  * The newest still-`running` turn for `agentAddress`, or `null` if none —
  * what a remounting workbench asks on mount to know whether to hydrate its

packages/chat-ui/src/chat-workspace.tsx

@@ -28,6 +28,7 @@ import type { ReactNode } from "react";
 import {
   workbenchesQueryKey,
   workbenchesQueryKeyPrefix,
+  cancelWorkbenchTurn,
   describeChatError,
   fetchRunningTurn,
   inviteAgent,
@@ -59,7 +60,11 @@ import { SLASH_COMMANDS } from "./slash-commands";
 
 import { CHAT_STRINGS } from "./strings";
 import { displayWorkbenchTitle } from "./workbench-display-title";
-import { useStreamingReply, typingAgentNames } from "./streaming-reply";
+import {
+  useStreamingReply,
+  isAwaitingReply,
+  typingAgentNames,
+} from "./streaming-reply";
 import { useTurnActivity, TurnActivityStrip } from "./turn-activity";
 import type { StreamingReplyState } from "./streaming-reply";
 import {
@@ -807,6 +812,21 @@ function ChatWorkspaceInner({
     [tenantId, activeWorkbenchId],
   );
 
+  // CL-7201: unlike the reaction/pin handlers above, this rethrows after
+  // toasting — the composer's own `onStop` awaits the returned promise
+  // and re-enables its Stop button on rejection, so a genuinely failed
+  // request (network, a denied grant) never leaves the button stuck
+  // disabled for the rest of the turn. The timeline's cancelled-turn
+  // notice, not this response, is what actually clears the typing
+  // indicator once (or if) the turn settles.
+  const handleStopTurn = useCallback(() => {
+    if (activeWorkbenchId === null) return Promise.resolve();
+    return cancelWorkbenchTurn(tenantId, activeWorkbenchId).catch((err) => {
+      toast(CHAT_STRINGS.turnCancelError);
+      throw err;
+    });
+  }, [tenantId, activeWorkbenchId]);
+
   const handlePinMessage = useCallback(
     (messageId: string) => {
       if (activeWorkbenchId === null) return;
@@ -1636,6 +1656,8 @@ function ChatWorkspaceInner({
                       bringInLoadError={bringInLoadError}
                       placeholder={composerPlaceholderFor(activeWorkbench)}
                       onSend={handleSend}
+                      running={isAwaitingReply(streamingReply)}
+                      onStop={handleStopTurn}
                       onInviteAgent={() => setInviteDialogOpen(true)}
                       onOpenAgentsSettings={() =>
                         openWorkbenchSettings("agents")

packages/chat-ui/src/composer.test.tsx

@@ -1,6 +1,7 @@
 import { describe, expect, test } from "bun:test";
 
-import { insertTextAtCaret } from "./composer";
+import { canStopComposer, insertTextAtCaret } from "./composer";
+import { isAwaitingReply } from "./streaming-reply";
 
 describe("insertTextAtCaret", () => {
   test("splices the insertion in at the caret", () => {
@@ -21,3 +22,27 @@ describe("insertTextAtCaret", () => {
     expect(result.caret).toBe(6);
   });
 });
+
+// CL-7201: the composer's stop affordance is a stand-in for "is there a
+// turn to cancel" — offered whenever the host says a turn is running,
+// independent of the composer's own `sending`/`preparing` state (queuing
+// a follow-up message while a turn runs is still allowed, so the stop
+// affordance and the send button coexist rather than one gating the
+// other).
+describe("canStopComposer", () => {
+  test("offers Stop while the host reports a turn running", () => {
+    expect(canStopComposer({ running: true })).toBe(true);
+  });
+
+  test("offers nothing when no turn is running", () => {
+    expect(canStopComposer({ running: false })).toBe(false);
+  });
+
+  test("offers Stop while awaiting a turn that already has streamed text", () => {
+    expect(
+      canStopComposer({
+        running: isAwaitingReply({ phase: "awaiting", text: "Hello" }),
+      }),
+    ).toBe(true);
+  });
+});

packages/chat-ui/src/composer.tsx

@@ -7,9 +7,10 @@
 // does not compose with an inline mention popover.
 
 import { Avatar, Button } from "@corbits/react-ui";
-import { ArrowUp, CircleNotch, Paperclip, X } from "@corbits/icons";
+import { ArrowUp, CircleNotch, Paperclip, Stop, X } from "@corbits/icons";
 import {
   forwardRef,
+  useEffect,
   useImperativeHandle,
   useLayoutEffect,
   useRef,
@@ -251,6 +252,20 @@ export function canSendComposerAction(
   return canSendComposer(text, attachments);
 }
 
+/**
+ * Whether the composer offers a Stop affordance (CL-7201) — a stand-in
+ * for "is there a turn to cancel," reported by the host from its own
+ * `isAwaitingReply` signal (the whole in-flight phase, including after
+ * tokens have started streaming — not the tokenless `isPendingReply`
+ * pulse). Deliberately independent of `sending`/
+ * `preparing`: a follow-up message can still be typed and queued while a
+ * turn runs (`turn-queue.ts` batches it), so Stop and Send coexist
+ * rather than one gating the other.
+ */
+export function canStopComposer(state: { readonly running: boolean }): boolean {
+  return state.running;
+}
+
 /** Attach stays blocked while a send or file read is in flight. */
 export function canAttachComposer(state: {
   readonly sending: boolean;
@@ -334,6 +349,20 @@ export const Composer = forwardRef<
     readonly onCreateRoutineInSpace: () => void;
     /** Defaults to the generic workbench copy — a chat passes one naming its counterpart. */
     readonly placeholder?: string;
+    /** Whether a turn is currently running for this workbench (CL-7201) —
+     * typically the host's own `isAwaitingReply(streamingReply)`. Absent
+     * or `false` renders no Stop affordance at all. */
+    readonly running?: boolean;
+    /**
+     * Cancels the running turn — `POST .../turns/cancel`. Required
+     * whenever `running` can be `true`; the composer never guesses at
+     * how to stop a turn on its own. May return a promise: a rejection
+     * re-enables the button immediately (the request itself failed —
+     * network, a denied grant — not merely a slow cancel, so there is
+     * no reason to make the person wait for `running` to change before
+     * they can try again).
+     */
+    readonly onStop?: () => void | Promise<unknown>;
   }
 >(function Composer(
   {
@@ -347,6 +376,8 @@ export const Composer = forwardRef<
     onOpenAgentsSettings,
     onCreateRoutineInSpace,
     placeholder = CHAT_STRINGS.composerPlaceholder,
+    running = false,
+    onStop,
   },
   ref,
 ) {
@@ -366,6 +397,12 @@ export const Composer = forwardRef<
   const [preparing, setPreparing] = useState(false);
   const [errorMessage, setErrorMessage] = useState<string | null>(null);
   const [focused, setFocused] = useState(false);
+  // CL-7201: guards Stop against a double-click firing two cancel
+  // requests. A second cancel is harmless server-side (compare-and-set),
+  // but there is no reason to send it. Resets once the host reports the
+  // turn is no longer running -- not on a timer, since a slow cancel
+  // (CL-7230's ceiling) must stay disabled rather than re-arm early.
+  const [stopping, setStopping] = useState(false);
   const textareaRef = useRef<HTMLTextAreaElement>(null);
   const fileInputRef = useRef<HTMLInputElement>(null);
   const attachGenerationRef = useRef(0);
@@ -377,6 +414,10 @@ export const Composer = forwardRef<
   // call in the same tick is turned away (CL-7198).
   const sendInFlightRef = useRef(false);
 
+  useEffect(() => {
+    if (!running) setStopping(false);
+  }, [running]);
+
   /** Auto-grow: the textarea reports its own content height, so the
    * measurement resets to the CSS-declared min-height before reading
    * `scrollHeight` — otherwise a shrinking draft would get stuck at its
@@ -701,6 +742,18 @@ export const Composer = forwardRef<
     void addFiles(event.target.files);
   }
 
+  function handleStop() {
+    if (stopping || onStop === undefined) return;
+    setStopping(true);
+    // CL-7201 (Critique finding): a rejected stop request is a FAILED
+    // cancel, not a slow one -- the `useEffect` above only re-enables
+    // once the host reports `running` has gone false, which never
+    // happens for a request that never reached the server. Without
+    // this catch the button stayed disabled for the rest of the turn's
+    // life with no way to retry.
+    Promise.resolve(onStop()).catch(() => setStopping(false));
+  }
+
   return (
     <div className="chat-composer">
       {slash !== null && (
@@ -907,6 +960,20 @@ export const Composer = forwardRef<
           >
             {CHAT_STRINGS.composerKeyboardHint}
           </span>
+          {canStopComposer({ running }) ? (
+            <Button
+              type="button"
+              variant="ghost"
+              size="sm"
+              className="chat-composer-icon-button"
+              disabled={stopping}
+              onClick={handleStop}
+              aria-label={CHAT_STRINGS.composerStop}
+              title={CHAT_STRINGS.composerStop}
+            >
+              <Stop aria-hidden="true" />
+            </Button>
+          ) : null}
           <Button
             type="button"
             variant={sendVisualState === "empty" ? "ghost" : "primary"}