From 1625449a85a1b93644a94a1b866a2982cce4c484 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:07:42 -0700 Subject: [PATCH 01/22] Add tests for withTimeout's external cancellation signal --- packages/chat/src/with-timeout.test.ts | 77 ++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/packages/chat/src/with-timeout.test.ts b/packages/chat/src/with-timeout.test.ts index d2fd66017..15d2daed2 100644 --- a/packages/chat/src/with-timeout.test.ts +++ b/packages/chat/src/with-timeout.test.ts @@ -72,4 +72,81 @@ describe("withTimeout", () => { await Bun.sleep(60); expect(sawAbort).toBe(false); }); + + // CL-7201: a caller with its own reason to give up (a user cancelling a + // turn) must be able to cut `work` short exactly like a timeout does — + // without waiting for the timeout's own clock, and without work having + // to guess which of two signals it was handed. + describe("an external signal (CL-7201)", () => { + test("aborts work's signal immediately when the external signal fires, well before the deadline", async () => { + const external = new AbortController(); + let reason: unknown; + + const promise = withTimeout( + (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { + reason = signal.reason; + reject(signal.reason); + }); + }), + 10_000, + "did not settle within 10000ms", + external.signal, + ); + + const cancelReason = new Error("cancelled by user"); + external.abort(cancelReason); + + await expect(promise).rejects.toThrow("cancelled by user"); + expect(reason).toBe(cancelReason); + }); + + test("an external signal that is already aborted aborts work immediately", async () => { + const external = new AbortController(); + external.abort(new Error("already gone")); + let aborted = false; + + await expect( + withTimeout( + (signal) => + new Promise((_resolve, reject) => { + if (signal.aborted) { + aborted = true; + reject(signal.reason); + return; + } + signal.addEventListener("abort", () => reject(signal.reason)); + }), + 10_000, + "did not settle within 10000ms", + external.signal, + ), + ).rejects.toThrow("already gone"); + expect(aborted).toBe(true); + }); + + test("the timeout still wins when the external signal never fires", async () => { + const external = new AbortController(); + await expect( + withTimeout( + () => new Promise(() => {}), + 10, + "did not settle within 10ms", + external.signal, + ), + ).rejects.toThrow("did not settle within 10ms"); + }); + + test("work settling on its own is unaffected by an external signal that never fires", async () => { + const external = new AbortController(); + const result = await withTimeout( + async () => "ok", + 50, + "timed out", + external.signal, + ); + expect(result).toBe("ok"); + }); + }); }); From 70b07e28823066dd789ec9509f21564785be5cb7 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:07:42 -0700 Subject: [PATCH 02/22] withTimeout: accept an external AbortSignal, so a caller's own cancellation cuts work short --- packages/chat/src/with-timeout.ts | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/packages/chat/src/with-timeout.ts b/packages/chat/src/with-timeout.ts index a162b3469..80ace0d11 100644 --- a/packages/chat/src/with-timeout.ts +++ b/packages/chat/src/with-timeout.ts @@ -12,10 +12,20 @@ // poll loop, close a row it opened) can react instead of running to // completion unobserved. A `work` that ignores the signal keeps // today's abandon-on-timeout behavior. +// +// CL-7201: a caller can also hand in its own `externalSignal` — a user +// cancelling the turn this `work` belongs to, say — that cuts `work` +// short exactly like the timeout does, without waiting for the timeout's +// own clock. `work` only ever sees ONE signal regardless of which of the +// two fired, so it never has to reason about "which deadline was this." +// The rejection carries the external signal's own `reason` rather than +// this module's timeout `Error`, so a caller downstream (`dispatchTurn`) +// can tell a real cancellation apart from a genuine timeout. export function withTimeout( work: (signal: AbortSignal) => Promise, ms: number, message: string, + externalSignal?: AbortSignal, ): Promise { const controller = new AbortController(); return new Promise((resolve, reject) => { @@ -23,13 +33,29 @@ export function withTimeout( controller.abort(new Error(message)); reject(new Error(message)); }, ms); + + const onExternalAbort = () => { + clearTimeout(timer); + controller.abort(externalSignal?.reason); + reject(externalSignal?.reason); + }; + if (externalSignal?.aborted === true) { + onExternalAbort(); + } else { + externalSignal?.addEventListener("abort", onExternalAbort, { + once: true, + }); + } + work(controller.signal).then( (value) => { clearTimeout(timer); + externalSignal?.removeEventListener("abort", onExternalAbort); resolve(value); }, (cause: unknown) => { clearTimeout(timer); + externalSignal?.removeEventListener("abort", onExternalAbort); reject(cause); }, ); From 123b1a34c1c1ade056a895bbd8de302d2035d1f5 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:09:28 -0700 Subject: [PATCH 03/22] Add tests for cancelled-turn status, findRunningTurns, and late-answer CAS safety --- packages/chat/src/agent-turns.test.ts | 88 +++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/packages/chat/src/agent-turns.test.ts b/packages/chat/src/agent-turns.test.ts index 34044fd2c..7fb0bdc4d 100644 --- a/packages/chat/src/agent-turns.test.ts +++ b/packages/chat/src/agent-turns.test.ts @@ -286,6 +286,94 @@ describe("createInMemoryAgentTurnStore", () => { expect(stored?.status).toBe("failed"); expect(stored?.replyMessageId).toBeNull(); }); + + // CL-7201: a turn cancelled by the user must never be resurrected by + // a real reply that lands after the fact -- exactly the same + // guarantee CL-7193 already gives a timed-out turn against a late + // reply, now exercised for the cancel outcome specifically, since + // cancellation is a NEW way to close a turn out from under work + // that is still in flight (a `message_response` gate answer, an + // ordinary reply) and has no code of its own to fall back on if + // this compare-and-set ever regressed. + test("a turn cancelled while a real reply is in flight cannot be reopened by that reply landing late", async () => { + const store = createInMemoryAgentTurnStore(); + const opened = await store.startTurn(BASE); + + const cancelled = await store.finishTurn({ + tenantId: BASE.tenantId, + turnId: opened.id, + status: "cancelled", + error: "Cancelled by user", + }); + expect(cancelled?.status).toBe("cancelled"); + + // The agent (or a message_response gate answer correlating to + // this turn) finishes its work anyway -- CL-7230's known ceiling + // is that nothing here can stop it -- and tries to close the same + // turn as a normal completed reply. + const lateReply = await store.finishTurn({ + tenantId: BASE.tenantId, + turnId: opened.id, + status: "completed", + replyMessageId: "msg_late_reply", + }); + expect(lateReply).toBeUndefined(); + + const stored = await store.getTurn({ + tenantId: BASE.tenantId, + turnId: opened.id, + }); + expect(stored?.status).toBe("cancelled"); + expect(stored?.replyMessageId).toBeNull(); + }); + }); + + // CL-7201: the cancel endpoint's fallback sweep -- settling a turn + // whose dispatch has already moved off the caller's own call stack + // (its `sendMail` resolved, an abort signal has nothing left to + // reach) -- must see every currently-running turn for a workbench, + // not just the newest one. `listTurns` alone is the wrong primitive + // for that sweep: it has no status filter and pages to + // `AGENT_TURNS_PAGE_SIZE`, so a busy workbench could silently leave + // older running turns unsettled. + describe("findRunningTurns (CL-7201)", () => { + test("returns every running turn for a workbench, across agents, excluding settled ones", async () => { + const store = createInMemoryAgentTurnStore(); + const first = await store.startTurn(BASE); + const second = await store.startTurn({ + ...BASE, + agentAddress: "ins_echo2@acme.example", + }); + const settled = await store.startTurn({ + ...BASE, + agentAddress: "ins_echo3@acme.example", + }); + await store.finishTurn({ + tenantId: BASE.tenantId, + turnId: settled.id, + status: "completed", + replyMessageId: "msg_done", + }); + await store.startTurn({ ...BASE, workbenchId: "wb_other" }); + + const running = await store.findRunningTurns({ + tenantId: BASE.tenantId, + workbenchId: BASE.workbenchId, + }); + + expect(new Set(running.map((turn) => turn.id))).toEqual( + new Set([first.id, second.id]), + ); + }); + + test("returns nothing for a workbench with no running turns", async () => { + const store = createInMemoryAgentTurnStore(); + const running = await store.findRunningTurns({ + tenantId: BASE.tenantId, + workbenchId: BASE.workbenchId, + }); + expect(running).toEqual([]); + }); }); }); From aaf399164669bd982e496966f5edabbffb518d4c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:09:28 -0700 Subject: [PATCH 04/22] agent-turns: add a cancelled status and findRunningTurns for the cancel sweep --- packages/chat/src/agent-turns.ts | 45 ++++++++++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/packages/chat/src/agent-turns.ts b/packages/chat/src/agent-turns.ts index a5080384b..ec3880acf 100644 --- a/packages/chat/src/agent-turns.ts +++ b/packages/chat/src/agent-turns.ts @@ -21,7 +21,7 @@ import { agentTurns } from "./schema"; import type { ChatDb } from "./store"; import { CHAT_TURN_TIMEOUT_MS } from "./turn-claims"; -export type AgentTurnStatus = "running" | "completed" | "failed"; +export type AgentTurnStatus = "running" | "completed" | "failed" | "cancelled"; export interface AgentTurn { readonly id: string; @@ -100,6 +100,20 @@ export interface AgentTurnStore { readonly workbenchId: string; readonly agentAddress: string; }): Promise; + /** + * Every `running` turn for a workbench, across every agent — the + * cancel endpoint's own read (CL-7201): a workbench can have more than + * one agent turn in flight at once (`dispatchTurnBatch` fans out + * concurrently), and settling "the" in-flight turn means settling all + * of them. Deliberately not built on `listTurns`: that method has no + * status filter and pages to `AGENT_TURNS_PAGE_SIZE`, so a busy + * workbench could silently leave an older running turn off the page + * and therefore unsettled. + */ + findRunningTurns(input: { + readonly tenantId: string; + readonly workbenchId: string; + }): Promise; /** * Resolves once (workbench, agent) has no `running` turn — immediately * if none is running right now, otherwise when the current one closes @@ -348,6 +362,16 @@ export function createInMemoryAgentTurnStore( return runningTurn(input); }, + async findRunningTurns(input) { + expireStaleTurns(); + return [...rows.values()].filter( + (turn) => + turn.tenantId === input.tenantId && + turn.workbenchId === input.workbenchId && + turn.status === "running", + ); + }, + async waitUntilFree(input, signal) { for (;;) { if (signal?.aborted) throw signal.reason; @@ -403,7 +427,9 @@ function toAgentTurn(row: AgentTurnRow): AgentTurn { ) : []; const status: AgentTurnStatus = - row.status === "completed" || row.status === "failed" + row.status === "completed" || + row.status === "failed" || + row.status === "cancelled" ? row.status : "running"; return { @@ -551,6 +577,21 @@ export function createDrizzleAgentTurnStore< return resolveRunningTurn(input); }, + async findRunningTurns(input) { + await expireStaleTurns(input); + const rows = await db + .select() + .from(agentTurns) + .where( + and( + eq(agentTurns.tenantId, input.tenantId), + eq(agentTurns.workbenchId, input.workbenchId), + eq(agentTurns.status, "running"), + ), + ); + return rows.map((row) => toAgentTurn(row as AgentTurnRow)); + }, + async waitUntilFree(input, signal) { for (;;) { if (signal?.aborted) throw signal.reason; From 63ad8035cb26a6c43d32825ffa80001afdaa6f6b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:10:13 -0700 Subject: [PATCH 05/22] Add tests for the turn-cancellation registry --- packages/chat/src/turn-cancellation.test.ts | 67 +++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 packages/chat/src/turn-cancellation.test.ts diff --git a/packages/chat/src/turn-cancellation.test.ts b/packages/chat/src/turn-cancellation.test.ts new file mode 100644 index 000000000..61feba776 --- /dev/null +++ b/packages/chat/src/turn-cancellation.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, test } from "bun:test"; + +import { + createTurnCancelRegistry, + TurnCancelledError, +} from "./turn-cancellation"; + +describe("createTurnCancelRegistry (CL-7201)", () => { + test("cancel aborts every controller registered for that workbench", () => { + const registry = createTurnCancelRegistry(); + const a = registry.register("wb_1"); + const b = registry.register("wb_1"); + + const cancelled = registry.cancel("wb_1"); + + expect(cancelled).toBe(true); + expect(a.signal.aborted).toBe(true); + expect(b.signal.aborted).toBe(true); + expect(a.signal.reason).toBeInstanceOf(TurnCancelledError); + }); + + test("cancel never touches a controller registered for a different workbench", () => { + const registry = createTurnCancelRegistry(); + const other = registry.register("wb_2"); + + registry.cancel("wb_1"); + + expect(other.signal.aborted).toBe(false); + }); + + test("cancel on a workbench with nothing registered is a harmless no-op", () => { + const registry = createTurnCancelRegistry(); + expect(registry.cancel("wb_never_ran")).toBe(false); + }); + + test("unregister removes a controller so a later cancel no longer reaches it", () => { + const registry = createTurnCancelRegistry(); + const controller = registry.register("wb_1"); + registry.unregister("wb_1", controller); + + const cancelled = registry.cancel("wb_1"); + + expect(cancelled).toBe(false); + expect(controller.signal.aborted).toBe(false); + }); + + test("unregistering one of several controllers leaves the rest cancellable", () => { + const registry = createTurnCancelRegistry(); + const a = registry.register("wb_1"); + const b = registry.register("wb_1"); + registry.unregister("wb_1", a); + + registry.cancel("wb_1"); + + expect(a.signal.aborted).toBe(false); + expect(b.signal.aborted).toBe(true); + }); + + test("a second cancel call is a harmless no-op once nothing is left registered", () => { + const registry = createTurnCancelRegistry(); + const controller = registry.register("wb_1"); + registry.cancel("wb_1"); + registry.unregister("wb_1", controller); + + expect(registry.cancel("wb_1")).toBe(false); + }); +}); From 0d08b6340b7cf57d0e49eabe6e0328507e846068 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:10:13 -0700 Subject: [PATCH 06/22] Add a workbench-keyed turn cancellation registry --- packages/chat/src/turn-cancellation.ts | 79 ++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 packages/chat/src/turn-cancellation.ts diff --git a/packages/chat/src/turn-cancellation.ts b/packages/chat/src/turn-cancellation.ts new file mode 100644 index 000000000..75f32820d --- /dev/null +++ b/packages/chat/src/turn-cancellation.ts @@ -0,0 +1,79 @@ +// CL-7201: the live abort seam a running turn is reachable through while +// it is still on our own call stack — the `waitUntilFree` wait, the +// `dispatchTurn` call around `sendMail` — as opposed to `agent-turns.ts`'s +// projection, which records what a turn's outcome WAS after the fact. +// This registry is what lets a cancel request reach the actual in-flight +// work instead of only closing its row once whatever it was doing +// eventually finishes on its own. +// +// Deliberately process-local and workbench-keyed, mirroring +// `createTurnFreedSignal` (`./agent-turns.ts`): a workbench can have more +// than one agent turn running at once (`dispatchTurnBatch` fans out +// concurrently), so cancelling "the" in-flight turn for a workbench means +// aborting every controller registered for it, not just one. + +/** The abort reason `TurnCancelRegistry.cancel` gives every controller it + * aborts — the one thing `dispatchTurn`'s own abort-close handler checks + * to tell a user's deliberate cancellation apart from a dispatch deadline + * timing out, so it can settle the turn row `cancelled` rather than + * `failed`. */ +export class TurnCancelledError extends Error { + constructor(message = "Cancelled by user") { + super(message); + this.name = "TurnCancelledError"; + } +} + +export interface TurnCancelRegistry { + /** Registers a fresh controller for `workbenchId`'s in-flight work. + * The caller owns unregistering it (typically in a `finally`) once + * that work settles on its own, win or lose. */ + register(workbenchId: string): AbortController; + /** Removes a controller this workbench no longer needs cancelled — + * its own work already settled, so a later `cancel` must not abort a + * controller nothing is listening on any more. */ + unregister(workbenchId: string, controller: AbortController): void; + /** + * Aborts every controller currently registered for `workbenchId` with + * a `TurnCancelledError`, and returns whether anything was actually + * reachable this way. `false` does not mean nothing was running — a + * turn whose `sendMail` has already resolved and moved off our call + * stack entirely has nothing left registered here at all; settling + * that turn's row is `cancelWorkbenchTurn`'s (`./workbench-service.ts`) + * separate sweep, not this registry's job. + */ + cancel(workbenchId: string): boolean; +} + +export function createTurnCancelRegistry(): TurnCancelRegistry { + const controllersByWorkbench = new Map>(); + + return { + register(workbenchId) { + const controller = new AbortController(); + const controllers = + controllersByWorkbench.get(workbenchId) ?? new Set(); + controllers.add(controller); + controllersByWorkbench.set(workbenchId, controllers); + return controller; + }, + + unregister(workbenchId, controller) { + const controllers = controllersByWorkbench.get(workbenchId); + if (controllers === undefined) return; + controllers.delete(controller); + if (controllers.size === 0) { + controllersByWorkbench.delete(workbenchId); + } + }, + + cancel(workbenchId) { + const controllers = controllersByWorkbench.get(workbenchId); + if (controllers === undefined || controllers.size === 0) return false; + for (const controller of controllers) { + controller.abort(new TurnCancelledError()); + } + return true; + }, + }; +} From 2cfe145ac5c6f8811a68737d291b8b25289a2be3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:22:54 -0700 Subject: [PATCH 07/22] Add tests for the cancel-turn dispatch path and its HTTP route --- packages/chat/test/turn-cancel-route.test.ts | 76 ++++++ .../test/turn-cancellation-dispatch.test.ts | 255 ++++++++++++++++++ 2 files changed, 331 insertions(+) create mode 100644 packages/chat/test/turn-cancel-route.test.ts create mode 100644 packages/chat/test/turn-cancellation-dispatch.test.ts diff --git a/packages/chat/test/turn-cancel-route.test.ts b/packages/chat/test/turn-cancel-route.test.ts new file mode 100644 index 000000000..5d5f4a67f --- /dev/null +++ b/packages/chat/test/turn-cancel-route.test.ts @@ -0,0 +1,76 @@ +// CL-7201: the HTTP surface for cancelling a workbench's in-flight +// turn(s). `turn-cancellation-dispatch.test.ts` exercises +// `cancelWorkbenchTurn`'s own logic directly; this file proves the route +// wires it up correctly — grant enforcement, 404 on an unknown +// workbench, and the shape of the response. +import { describe, expect, test } from "bun:test"; + +import { createInMemoryAgentTurnStore } from "../src/agent-turns"; +import { createChatRoutes } from "../src/routes"; +import { + buildDeps, + createWorkbench, + fakePlatform, + mountAs, +} from "./test-support"; + +describe("POST /workbenches/:id/turns/cancel (CL-7201)", () => { + test("cancels a stuck turn and clears it via the response body", async () => { + const platform = fakePlatform({ + invitable: [{ id: "wfd_echo", name: "echo" }], + }); + platform.sendMail = () => new Promise(() => {}); + const agentTurns = createInMemoryAgentTurnStore(); + const deps = buildDeps({ + platform, + agentTurns, + turnDispatchTimeoutMs: 60_000, + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "chat", + definitionId: "wfd_echo", + }); + + void app.request(`/workbenches/${workbench.id}/messages`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ parts: [{ kind: "text", text: "hello" }] }), + }); + await Bun.sleep(5); + + const response = await app.request( + `/workbenches/${workbench.id}/turns/cancel`, + { method: "POST" }, + ); + expect(response.status).toBe(200); + const body = await response.json(); + expect(body).toEqual({ cancelledCount: 1 }); + }); + + test("404s for a workbench that doesn't exist", async () => { + const deps = buildDeps(); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + + const response = await app.request( + "/workbenches/run_does_not_exist/turns/cancel", + { method: "POST" }, + ); + + expect(response.status).toBe(404); + }); + + test("denies the request when the write grant is refused", async () => { + const deps = buildDeps({ + requireGrant: () => async (c) => + c.json({ error: { code: "forbidden" } }, 403), + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + + const response = await app.request("/workbenches/run_1/turns/cancel", { + method: "POST", + }); + + expect(response.status).toBe(403); + }); +}); diff --git a/packages/chat/test/turn-cancellation-dispatch.test.ts b/packages/chat/test/turn-cancellation-dispatch.test.ts new file mode 100644 index 000000000..e3212afa6 --- /dev/null +++ b/packages/chat/test/turn-cancellation-dispatch.test.ts @@ -0,0 +1,255 @@ +// CL-7201: cancellation has two independent ways to reach a running +// turn, and both must settle the row honestly (never claim `failed` +// for something the user asked for) with exactly one notice on the +// timeline: +// +// 1. The turn is still reachable on our own call stack (`waitUntilFree` +// still waiting, or `dispatchTurn`'s `sendMail` still in flight) -- +// the cancellation registry's `AbortSignal` cuts it short directly. +// 2. The turn's dispatch has already resolved and moved off our call +// stack entirely (the agent is generating, or parked on a +// `message_response` gate somewhere in the execution plane this +// package has no visibility into) -- `cancelWorkbenchTurn`'s sweep +// is the only thing left that can settle the row, per CL-7230's +// ceiling: it can record the outcome, not stop the underlying work. +import { describe, expect, test } from "bun:test"; + +import { createInMemoryAgentTurnStore } from "../src/agent-turns"; +import { createChatRoutes } from "../src/routes"; +import { createTurnCancelRegistry } from "../src/turn-cancellation"; +import { createWorkbenchSubscriberRegistry } from "../src/workbench-events"; +import { cancelWorkbenchTurn } from "../src/workbench-service"; +import { + buildDeps, + createWorkbench, + fakePlatform, + mountAs, + settleFanout, + TENANT, + timelineOf, +} from "./test-support"; + +async function postHello(app: ReturnType, workbenchId: string) { + return app.request(`/workbenches/${workbenchId}/messages`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ parts: [{ kind: "text", text: "hello" }] }), + }); +} + +describe("cancelWorkbenchTurn (CL-7201)", () => { + test("cancelling a turn stuck inside dispatchTurn's sendMail settles it cancelled, not failed, with one honest notice", async () => { + const platform = fakePlatform({ + invitable: [{ id: "wfd_echo", name: "echo" }], + }); + platform.sendMail = () => new Promise(() => {}); + const agentTurns = createInMemoryAgentTurnStore(); + const turnCancellation = createTurnCancelRegistry(); + const workbenchSubscribers = createWorkbenchSubscriberRegistry(); + + // Generous on purpose: proves cancellation beats the deadline rather + // than racing it. + const deps = buildDeps({ + platform, + agentTurns, + turnCancellation, + workbenchSubscribers, + turnDispatchTimeoutMs: 60_000, + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "chat", + definitionId: "wfd_echo", + }); + + void postHello(app, workbench.id); + // Let dispatchTurn actually start (startTurn + the stuck sendMail + // call) before cancelling -- otherwise there is nothing in flight + // yet to cancel. + await Bun.sleep(5); + + const result = await cancelWorkbenchTurn( + { + agentTurns, + turnCancellation, + roomMessages: deps.roomMessages, + publish: workbenchSubscribers.publish, + }, + { tenantId: TENANT.id, workbenchId: workbench.id }, + ); + await settleFanout(); + + expect(result.cancelledCount).toBe(1); + + const turns = await agentTurns.listTurns({ + tenantId: TENANT.id, + workbenchId: workbench.id, + }); + expect(turns).toHaveLength(1); + expect(turns[0]?.status).toBe("cancelled"); + + const timeline = await timelineOf(deps, workbench.id); + const cancelledNotices = timeline.filter((message) => + message.parts.some( + (part) => part.kind === "text" && part.turnCancelled === true, + ), + ); + expect(cancelledNotices).toHaveLength(1); + + const undeliveredNotices = timeline.filter((message) => + message.parts.some( + (part) => part.kind === "text" && part.turnFailed === true, + ), + ); + expect(undeliveredNotices).toHaveLength(0); + }); + + test("cancelling a turn whose dispatch already resolved off our call stack still settles it via the sweep", async () => { + const platform = fakePlatform({ + invitable: [{ id: "wfd_echo", name: "echo" }], + }); + const agentTurns = createInMemoryAgentTurnStore(); + const turnCancellation = createTurnCancelRegistry(); + const workbenchSubscribers = createWorkbenchSubscriberRegistry(); + const deps = buildDeps({ + platform, + agentTurns, + turnCancellation, + workbenchSubscribers, + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "chat", + definitionId: "wfd_echo", + }); + + await postHello(app, workbench.id); + await settleFanout(); + + // Sanity: nothing in this harness closes the row on its own -- + // proves the later "cancelled" status came from the sweep, not from + // some other path already having settled it. + const beforeCancel = await agentTurns.listTurns({ + tenantId: TENANT.id, + workbenchId: workbench.id, + }); + expect(beforeCancel[0]?.status).toBe("running"); + + const result = await cancelWorkbenchTurn( + { + agentTurns, + turnCancellation, + roomMessages: deps.roomMessages, + publish: workbenchSubscribers.publish, + }, + { tenantId: TENANT.id, workbenchId: workbench.id }, + ); + + expect(result.cancelledCount).toBe(1); + const after = await agentTurns.listTurns({ + tenantId: TENANT.id, + workbenchId: workbench.id, + }); + expect(after[0]?.status).toBe("cancelled"); + + const timeline = await timelineOf(deps, workbench.id); + const cancelledNotices = timeline.filter((message) => + message.parts.some( + (part) => part.kind === "text" && part.turnCancelled === true, + ), + ); + expect(cancelledNotices).toHaveLength(1); + }); + + test("a late reply landing after cancellation cannot reopen the turn", async () => { + const platform = fakePlatform({ + invitable: [{ id: "wfd_echo", name: "echo" }], + }); + const agentTurns = createInMemoryAgentTurnStore(); + const turnCancellation = createTurnCancelRegistry(); + const workbenchSubscribers = createWorkbenchSubscriberRegistry(); + const deps = buildDeps({ + platform, + agentTurns, + turnCancellation, + workbenchSubscribers, + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "chat", + definitionId: "wfd_echo", + }); + + await postHello(app, workbench.id); + await settleFanout(); + + await cancelWorkbenchTurn( + { + agentTurns, + turnCancellation, + roomMessages: deps.roomMessages, + publish: workbenchSubscribers.publish, + }, + { tenantId: TENANT.id, workbenchId: workbench.id }, + ); + + const [turn] = await agentTurns.listTurns({ + tenantId: TENANT.id, + workbenchId: workbench.id, + }); + if (turn === undefined) throw new Error("expected a turn to be running"); + const lateReply = await agentTurns.finishTurn({ + tenantId: TENANT.id, + turnId: turn.id, + status: "completed", + replyMessageId: "msg_late", + }); + expect(lateReply).toBeUndefined(); + + const stored = await agentTurns.getTurn({ + tenantId: TENANT.id, + turnId: turn.id, + }); + expect(stored?.status).toBe("cancelled"); + }); + + test("cancelling a workbench with nothing running is a harmless no-op", async () => { + const agentTurns = createInMemoryAgentTurnStore(); + const turnCancellation = createTurnCancelRegistry(); + const workbenchSubscribers = createWorkbenchSubscriberRegistry(); + const deps = buildDeps({ + agentTurns, + turnCancellation, + workbenchSubscribers, + }); + + const result = await cancelWorkbenchTurn( + { + agentTurns, + turnCancellation, + roomMessages: deps.roomMessages, + publish: workbenchSubscribers.publish, + }, + { tenantId: TENANT.id, workbenchId: "wb_never_ran" }, + ); + + expect(result.cancelledCount).toBe(0); + }); + + test("cancelling without an agentTurns store configured never throws", async () => { + const turnCancellation = createTurnCancelRegistry(); + const workbenchSubscribers = createWorkbenchSubscriberRegistry(); + const deps = buildDeps({ turnCancellation, workbenchSubscribers }); + + const result = await cancelWorkbenchTurn( + { + turnCancellation, + roomMessages: deps.roomMessages, + publish: workbenchSubscribers.publish, + }, + { tenantId: TENANT.id, workbenchId: "wb_1" }, + ); + + expect(result.cancelledCount).toBe(0); + }); +}); From 41e0b20d8bf5dc249460e70cb9c873bdc7e2cf82 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:22:55 -0700 Subject: [PATCH 08/22] Add a cancel-turn endpoint that propagates cancellation through dispatch --- packages/chat/src/parts.ts | 7 + packages/chat/src/routes.ts | 53 +++++ packages/chat/src/workbench-service.ts | 217 +++++++++++++++++- .../chat/src/workflow-participant-routes.ts | 5 + .../test/workflow-participant-routes.test.ts | 2 + 5 files changed, 282 insertions(+), 2 deletions(-) diff --git a/packages/chat/src/parts.ts b/packages/chat/src/parts.ts index 6a6a79165..0c428784b 100644 --- a/packages/chat/src/parts.ts +++ b/packages/chat/src/parts.ts @@ -14,6 +14,13 @@ export const TextPart = type({ * (`PrFailedTurnStrip`) instead of a plain bubble. Absent on every * other text part. */ "turnFailed?": "boolean", + /** Set only on the cancelled-turn notice `postCancelledNotice` + * (`./workbench-service.ts`) posts in the cancelled agent's own voice + * (CL-7201) — distinct from `turnFailed`: a user cancelling a turn is + * not a failure, and the frontend renders it with its own honest copy + * rather than `FailedTurnStrip`'s "didn't reply" framing. Absent on + * every other text part. */ + "turnCancelled?": "boolean", }); export type TextPart = typeof TextPart.infer; diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index 4d3f348d3..b67c15d17 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -76,6 +76,7 @@ import { findExistingAgentChat, removeWorkbenchParticipant, sendWorkbenchMessage, + cancelWorkbenchTurn, } from "./workbench-service"; import { bridgeWorkbenchStream, @@ -92,6 +93,10 @@ import { createWorkbenchTurnQueue, type WorkbenchTurnQueue, } from "./turn-queue"; +import { + createTurnCancelRegistry, + type TurnCancelRegistry, +} from "./turn-cancellation"; import type { ChatPlatform } from "./platform-port"; import type { ChatStore } from "./store"; import { @@ -296,6 +301,15 @@ export type CreateChatRoutesDeps = { * consumer of turn-claim state to share it with. */ turnQueue?: WorkbenchTurnQueue; + /** + * The live abort seam a running turn is reachable through (CL-7201) — + * see `./turn-cancellation.ts` and `SendWorkbenchMessageDeps`'s field + * of the same name in `./workbench-service.ts`. Defaults to a fresh, + * router-scoped registry when omitted, the same "construct one + * instance, share it everywhere" pattern `turnQueue` follows — it is + * process-local, in-memory state with no cost to always have. + */ + turnCancellation?: TurnCancelRegistry; /** * Slack-Connect-style workbench projection (CL-5882) — see * `./workbench-share.ts`. Omitted entirely, every `/workbenches/:id/shares*` @@ -921,6 +935,7 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { claims: createInMemoryTurnClaimStore({ ttlMs: deps.turnTimeoutMs }), publish, }); + const turnCancellation = deps.turnCancellation ?? createTurnCancelRegistry(); app.post( "/workbenches", @@ -2103,6 +2118,7 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { roomMessages: deps.roomMessages, publish, turnQueue, + turnCancellation, ...(deps.agentTurns !== undefined ? { agentTurns: deps.agentTurns } : {}), @@ -2297,6 +2313,7 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { roomMessages: deps.roomMessages, publish, turnQueue, + turnCancellation, ...(deps.agentTurns !== undefined ? { agentTurns: deps.agentTurns } : {}), @@ -3700,6 +3717,42 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { }, ); + // CL-7201: the escape hatch for a wedged or wrongly-guessed turn — the + // only bound before this route existed was the dispatch/wait-until-free + // timeouts, measured in minutes. Write, not read: cancelling changes + // the workbench's state, exactly like posting a message does. + app.post( + "/workbenches/:id/turns/cancel", + deps.requireGrant(idResource("room", "id"), "write"), + async (c) => { + const tenant = c.get("tenant"); + const principal = c.get("principal"); + const workbenchId = c.req.param("id"); + const access = await resolveWorkbenchAccess( + deps, + tenant.id, + workbenchId, + principal.id, + principal.refId, + ); + if (access === undefined) { + return c.json(ErrorEnvelope("not_found", "workbench not found"), 404); + } + const result = await cancelWorkbenchTurn( + { + turnCancellation, + roomMessages: deps.roomMessages, + publish, + ...(deps.agentTurns !== undefined + ? { agentTurns: deps.agentTurns } + : {}), + }, + { tenantId: access.ownerTenantId, workbenchId }, + ); + return c.json(result); + }, + ); + app.get( "/workbenches/:id/turns/:turnId", deps.requireGrant(idResource("room", "id"), "read"), diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 4904eb723..29f87b5fa 100644 --- a/packages/chat/src/workbench-service.ts +++ b/packages/chat/src/workbench-service.ts @@ -23,6 +23,10 @@ import { } from "./turn-context"; import type { AgentTurnStore } from "./agent-turns"; import type { ThreadStore } from "./threads"; +import { + TurnCancelledError, + type TurnCancelRegistry, +} from "./turn-cancellation"; import { addParticipant, handleFromName, @@ -1234,6 +1238,18 @@ export type SendWorkbenchMessageDeps = { * that claim releases. See `./turn-queue.ts`. */ readonly turnQueue: WorkbenchTurnQueue; + /** + * The live abort seam a running turn is reachable through while still + * on our own call stack (CL-7201) — see `./turn-cancellation.ts`. + * `dispatchTurnBatch` registers one controller per recipient it + * dispatches and composes it into each `withTimeout` call so a + * cancellation lands exactly like a timeout does, distinguished only + * by its `TurnCancelledError` reason. Required, not optional: it is + * process-local, in-memory state with no cost to always have — every + * composition gets real cancellation propagation, not just the ones + * that remember to wire it up. + */ + readonly turnCancellation: TurnCancelRegistry; /** * The turn projection (CL-6329). `dispatchTurn` opens a row before it * touches the execution plane, so an in-flight turn is visible from @@ -1645,6 +1661,7 @@ async function dispatchTurnBatch( | "turnDispatchTimeoutMs" | "waitUntilFreeTimeoutMs" | "agentTurns" + | "turnCancellation" >, tenantId: string, workbenchId: string, @@ -1692,6 +1709,16 @@ async function dispatchTurnBatch( // delay every agent mentioned after it. await Promise.all( recipients.map(async (agentAddress) => { + // CL-7201: one controller for this recipient's whole attempt — + // spanning both the `waitUntilFree` wait below and the + // `dispatchTurn` call after it — so a cancel request lands + // wherever this recipient actually is, not just one of the two + // phases. Registered before either `withTimeout` call and always + // unregistered once this recipient's own attempt settles, win or + // lose, so a cancel arriving after the fact never reaches (or + // leaks a reference to) a controller nothing is listening on any + // more. + const cancelController = deps.turnCancellation.register(workbenchId); try { // CL-6670: wait under its own deadline, separate from the // per-hop deadline below. An agent that already has a turn @@ -1724,6 +1751,7 @@ async function dispatchTurnBatch( ), waitUntilFreeTimeoutMs, waitUntilFreeTimeoutMessage(agentAddress, waitUntilFreeTimeoutMs), + cancelController.signal, ); } // CL-6644: one deadline around the whole turn, not another @@ -1764,8 +1792,24 @@ async function dispatchTurnBatch( ), turnDispatchTimeoutMs, turnDispatchTimeoutMessage(agentAddress, turnDispatchTimeoutMs), + cancelController.signal, ); } catch (err) { + // CL-7201: a deliberate cancellation is not a failure — the + // user asked for exactly this. `dispatchTurn`'s own abort-close + // handler (if this recipient was still inside its `sendMail` + // call) or `cancelWorkbenchTurn`'s sweep (if it wasn't) already + // settled the turn row and posted the one honest notice this + // gets; reporting it here too would double-post and misfile a + // user action as an operational error. + if (err instanceof TurnCancelledError) { + fanoutLog.info( + "Turn dispatch for {agentAddress} on workbench {workbenchId} " + + "was cancelled by the user for message(s) {messageIds}", + { agentAddress, workbenchId, messageIds }, + ); + return; + } const refId = reportError(err, { operation: "chat.dispatchTurn", tenantId, @@ -1792,6 +1836,8 @@ async function dispatchTurnBatch( cause: err, refId, }); + } finally { + deps.turnCancellation.unregister(workbenchId, cancelController); } }), ); @@ -1838,9 +1884,21 @@ export type DispatchTurnInput = { * row to attach to — `finishTurn`'s compare-and-set means whichever of * the abort close and the late reply's own close reaches the row first * is the only one that applies. + * + * CL-7201: the same abort can also be a deliberate cancellation rather + * than a timeout (`signal.reason instanceof TurnCancelledError`) — this + * closes the row `cancelled` instead of `failed` in that case, and, only + * if this close actually wins `finishTurn`'s compare-and-set (this + * recipient's controller was still reachable when the user cancelled), + * posts the one cancelled notice the timeline gets for it. Losing that + * race means `cancelWorkbenchTurn`'s own sweep already settled the row + * and posted the notice instead — exactly one of the two ever does. */ export async function dispatchTurn( - deps: Pick, + deps: Pick< + SendWorkbenchMessageDeps, + "platform" | "agentTurns" | "roomMessages" | "publish" + >, input: DispatchTurnInput, signal?: AbortSignal, ): Promise { @@ -1853,16 +1911,25 @@ export async function dispatchTurn( const closeAsTimedOut = () => { if (turn === undefined) return; + const cancelled = signal?.reason instanceof TurnCancelledError; deps.agentTurns ?.finishTurn({ tenantId: input.tenantId, turnId: turn.id, - status: "failed", + status: cancelled ? "cancelled" : "failed", error: signal?.reason instanceof Error ? signal.reason.message : "turn dispatch timed out", }) + .then((finished) => { + if (finished === undefined || !cancelled) return; + return postCancelledNotice(deps, { + tenantId: input.tenantId, + workbenchId: input.workbenchId, + agentAddress: input.agentAddress, + }); + }) .catch((err: unknown) => { reportError(err, { operation: "chat.dispatchTurn.closeAsTimedOut", @@ -1988,3 +2055,149 @@ async function postUndeliveredNotice( ); } } + +const CANCELLED_NOTICE = "This turn was cancelled."; + +/** + * The timeline's own record of a cancellation (CL-7201) — deliberately + * distinct from `postUndeliveredNotice`: a user stopping a turn is not a + * failure, and reusing `turnFailed`'s "didn't reply" framing (with its + * Retry action) would misrepresent something the user asked for as + * something that went wrong on its own. Posted in the cancelled agent's + * own voice for the same reason `postUndeliveredNotice` is: the person + * reading the room sees who stopped replying and why, without a system + * message breaking the conversation's voice. Swallows its own failure + * exactly like `postUndeliveredNotice` — if the timeline is unreachable + * there is nowhere left to say so. + */ +async function postCancelledNotice( + deps: Pick, + input: { + readonly tenantId: string; + readonly workbenchId: string; + readonly agentAddress: string; + }, +): Promise { + try { + await postRoomMessage(deps, { + tenantId: input.tenantId, + workbenchId: input.workbenchId, + sender: { name: null, address: input.agentAddress }, + runId: localPartOf(input.agentAddress), + parts: [ + { + kind: "text", + text: CANCELLED_NOTICE, + turnCancelled: true, + }, + ], + }); + } catch (err) { + // report-error-ignore: mirrors postUndeliveredNotice just above — + // the timeline being unreachable here is the same pre-existing, + // already-tracked class of failure with nowhere left to report to. + fanoutLog.error( + "Could not post the cancelled-turn notice for {agentAddress} onto " + + "workbench {workbenchId}'s timeline: {err}", + { agentAddress: input.agentAddress, workbenchId: input.workbenchId, err }, + ); + } +} + +export type CancelWorkbenchTurnResult = { + /** How many turns were running for this workbench the moment cancel + * was asked for, and are now settled `cancelled` — the honest answer + * to "what did cancelling stop," per CL-7230's ceiling. Counted by + * outcome, not by which of the two mechanisms below actually + * performed the write: a turn `dispatchTurn`'s own abort-close + * settled a moment before this call's sweep reached it still counts + * here as cancelled. A workbench with nothing running returns 0. */ + readonly cancelledCount: number; +}; + +/** + * Stops a workbench's in-flight turn(s) (CL-7201) — the cancel route's + * own logic, kept here rather than in `routes.ts` so it is testable + * without an HTTP round trip. + * + * Two independent mechanisms, run together, because a turn can be in + * either of two places when the user asks to stop it: + * + * 1. Still reachable on our own call stack — `dispatchTurnBatch`'s + * per-recipient controller (`./turn-cancellation.ts`) is aborted, + * which cuts the `waitUntilFree` wait or the `dispatchTurn` call + * short exactly like a timeout does, synchronously, the moment + * `cancel` below fires. That recipient's own abort-close handler + * settles its row and posts the notice — frequently winning the + * race against this function's own sweep, below, for any turn that + * was still reachable this way. + * 2. Already off our call stack entirely — `sendMail` already resolved, + * the agent is generating (or parked on a `message_response` gate + * somewhere in the execution plane this package cannot see into). + * Nothing is registered to abort any more, so the row is found via + * `findRunningTurns` (snapshotted BEFORE `cancel` below, so a row + * path 1 already claimed is still counted) and settled directly. + * + * Both paths call the same `finishTurn` compare-and-set, so whichever + * reaches a given row first is the only one that ever settles or + * notifies for it — never both, never neither; the loser's own + * `finishTurn` call harmlessly returns `undefined`. CL-7230's known + * ceiling applies to path 2 specifically: settling the row is not the + * same as stopping the underlying agent process, which this cannot + * reach. + */ +export async function cancelWorkbenchTurn( + deps: Pick< + SendWorkbenchMessageDeps, + "agentTurns" | "turnCancellation" | "roomMessages" | "publish" + >, + input: { readonly tenantId: string; readonly workbenchId: string }, +): Promise { + if (deps.agentTurns === undefined) { + deps.turnCancellation.cancel(input.workbenchId); + return { cancelledCount: 0 }; + } + const agentTurns = deps.agentTurns; + + // Snapshotted before `cancel` fires: path 1's abort-close can (and + // often does) win a row's compare-and-set synchronously inside + // `cancel` itself, before this sweep's own `finishTurn` call ever + // runs — this list is what lets the sweep still count, and settle, + // whatever it didn't personally win. + const running = await agentTurns.findRunningTurns({ + tenantId: input.tenantId, + workbenchId: input.workbenchId, + }); + + deps.turnCancellation.cancel(input.workbenchId); + + await Promise.all( + running.map(async (turn) => { + const finished = await agentTurns.finishTurn({ + tenantId: input.tenantId, + turnId: turn.id, + status: "cancelled", + error: "Cancelled by user", + }); + // `undefined` means path 1's abort-close already won this row — + // it already posted its own notice, so posting a second one here + // would double it up. + if (finished === undefined) return; + await postCancelledNotice(deps, { + tenantId: input.tenantId, + workbenchId: input.workbenchId, + agentAddress: finished.agentAddress, + }); + }), + ); + + const settled = await Promise.all( + running.map((turn) => + agentTurns.getTurn({ tenantId: input.tenantId, turnId: turn.id }), + ), + ); + return { + cancelledCount: settled.filter((turn) => turn?.status === "cancelled") + .length, + }; +} diff --git a/packages/chat/src/workflow-participant-routes.ts b/packages/chat/src/workflow-participant-routes.ts index 9d2a1f7f2..998eebb08 100644 --- a/packages/chat/src/workflow-participant-routes.ts +++ b/packages/chat/src/workflow-participant-routes.ts @@ -126,6 +126,10 @@ export type CreateWorkflowParticipantRoutesDeps = { * workflow-child message and a person's own message for the same * workbench serialize against each other too. */ readonly turnQueue: SendWorkbenchMessageDeps["turnQueue"]; + /** The same cancellation registry `createChatRoutes` is given + * (CL-7201) — shared, never a second instance, so a cancel request + * reaches a controller registered from either entry point. */ + readonly turnCancellation: SendWorkbenchMessageDeps["turnCancellation"]; readonly authenticator: WorkflowRunAuthenticator; readonly tenancy: Pick< WorkbenchTenancyStore, @@ -405,6 +409,7 @@ export function createWorkflowParticipantRoutes( roomMessages: deps.roomMessages, publish: deps.publish, turnQueue: deps.turnQueue, + turnCancellation: deps.turnCancellation, }, { tenantId: scope.tenantId, diff --git a/packages/chat/test/workflow-participant-routes.test.ts b/packages/chat/test/workflow-participant-routes.test.ts index 6b43fb30e..1127a2787 100644 --- a/packages/chat/test/workflow-participant-routes.test.ts +++ b/packages/chat/test/workflow-participant-routes.test.ts @@ -18,6 +18,7 @@ import { } from "../src/workflow-participant-routes"; import { createInMemoryTurnClaimStore } from "../src/turn-claims"; import { createWorkbenchTurnQueue } from "../src/turn-queue"; +import { createTurnCancelRegistry } from "../src/turn-cancellation"; import { createInMemoryWorkbenchTenancyStore } from "../src/workbench-tenancy"; import { fakePlatform, TENANT } from "./test-support"; @@ -54,6 +55,7 @@ function buildApp( claims: createInMemoryTurnClaimStore({ ttlMs: 60_000 }), publish, }), + turnCancellation: overrides.turnCancellation ?? createTurnCancelRegistry(), authenticator: overrides.authenticator ?? authenticateAsRun, tenancy: overrides.tenancy ?? createInMemoryWorkbenchTenancyStore(), sessionFor: overrides.sessionFor ?? (async () => ["session=test"]), From 210e4b3a5f87cf0bae1c0e3f0c58d685e854cb4e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:25:31 -0700 Subject: [PATCH 09/22] Wire the cancel-turn registry into the hub's chat, workflow, and Slack send paths --- apps/hub/src/index.ts | 9 +++++++++ apps/hub/src/slack-tag-mount.ts | 5 +++++ packages/chat/src/index.ts | 7 +++++++ 3 files changed, 21 insertions(+) diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 6b0d5c2b3..b3292a08b 100644 --- a/apps/hub/src/index.ts +++ b/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, diff --git a/apps/hub/src/slack-tag-mount.ts b/apps/hub/src/slack-tag-mount.ts index 341b455ea..a3af480eb 100644 --- a/apps/hub/src/slack-tag-mount.ts +++ b/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, diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index d97b7c446..7bc393aeb 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -141,6 +141,11 @@ export type { WorkbenchTurnQueue, WorkbenchTurnQueueDeps, } from "./turn-queue"; +export { + createTurnCancelRegistry, + TurnCancelledError, +} from "./turn-cancellation"; +export type { TurnCancelRegistry } from "./turn-cancellation"; export type { WorkbenchEvents, WorkbenchLauncher, @@ -239,6 +244,7 @@ export { sendWorkbenchMessage, startWorkflowCommand, provisionSpaceWorkbench, + cancelWorkbenchTurn, } from "./workbench-service"; export type { LaunchAndJoinAgentDeps, @@ -253,6 +259,7 @@ export type { SendWorkbenchMessageDeps, SendWorkbenchMessageInput, SendWorkbenchMessageResult, + CancelWorkbenchTurnResult, StartWorkflowCommandDeps, StartWorkflowCommandInput, StartWorkflowCommandResult, From 0e19addfe4b31279d8dc77b150f14ac2836cc9f9 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:34:46 -0700 Subject: [PATCH 10/22] Add tests for the composer's stop affordance and the cancelled-turn typing signal --- packages/chat-ui/src/composer.test.tsx | 18 +++++++- packages/chat-ui/src/streaming-reply.test.ts | 44 ++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/packages/chat-ui/src/composer.test.tsx b/packages/chat-ui/src/composer.test.tsx index 19b3e3ac9..f9a6717ef 100644 --- a/packages/chat-ui/src/composer.test.tsx +++ b/packages/chat-ui/src/composer.test.tsx @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; -import { insertTextAtCaret } from "./composer"; +import { canStopComposer, insertTextAtCaret } from "./composer"; describe("insertTextAtCaret", () => { test("splices the insertion in at the caret", () => { @@ -21,3 +21,19 @@ 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); + }); +}); diff --git a/packages/chat-ui/src/streaming-reply.test.ts b/packages/chat-ui/src/streaming-reply.test.ts index 2605a2f7e..6a99e457f 100644 --- a/packages/chat-ui/src/streaming-reply.test.ts +++ b/packages/chat-ui/src/streaming-reply.test.ts @@ -217,6 +217,50 @@ describe("nextStreamingReplyState (CL-6376: the typing pulse clears on a dispatc }); }); +// CL-7201: a user-cancelled turn clears the same pulse a failed one does +// — `postCancelledNotice` carries `turnCancelled`, not `turnFailed`, so +// this is its own case rather than reusing the failure fixture above. +describe("nextStreamingReplyState (CL-7201: the typing pulse clears on a user cancellation too)", () => { + test("a chat.message carrying a turnCancelled part clears a pending reply", () => { + const state = awaiting(""); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + parts: [ + { + kind: "text", + text: "This turn was cancelled.", + turnCancelled: true, + }, + ], + }, + }), + ).toBeNull(); + }); + + test("a chat.message from the cancelled agent's own address is never mistaken for a rendered reply", () => { + const state = awaiting(""); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + sender: { name: null, address: MYRA.address }, + parts: [ + { + kind: "text", + text: "This turn was cancelled.", + turnCancelled: true, + }, + ], + }, + }), + ).toBeNull(); + }); +}); + describe("nextStreamingReplyState (CL-false-no-reply: rendered content, not a lifecycle event, ends the turn)", () => { test("a chat.message from the awaiting turn's agent moves straight to replied — the reply already rendered, connector.reply or not", () => { const state = awaiting("Full answer."); From 10b8ef00317fed0a093455d4997d9ad830ab8c0f Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:34:46 -0700 Subject: [PATCH 11/22] Add a composer stop affordance and an honest cancelled-turn notice in the timeline --- packages/chat-ui/src/api.ts | 27 +++++++++++- packages/chat-ui/src/chat-workspace.tsx | 20 ++++++++- packages/chat-ui/src/composer.tsx | 55 ++++++++++++++++++++++++- packages/chat-ui/src/streaming-reply.ts | 45 ++++++++++++++++---- packages/chat-ui/src/strings.ts | 3 ++ packages/chat-ui/src/styles.css | 14 +++++++ packages/chat-ui/src/timeline.tsx | 37 +++++++++++++++++ packages/icons/src/index.tsx | 1 + 8 files changed, 192 insertions(+), 10 deletions(-) diff --git a/packages/chat-ui/src/api.ts b/packages/chat-ui/src/api.ts index 2a6e39f62..761b631c9 100644 --- a/packages/chat-ui/src/api.ts +++ b/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 { + 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 diff --git a/packages/chat-ui/src/chat-workspace.tsx b/packages/chat-ui/src/chat-workspace.tsx index fd7d4a2d3..a7b618572 100644 --- a/packages/chat-ui/src/chat-workspace.tsx +++ b/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, + isPendingReply, + typingAgentNames, +} from "./streaming-reply"; import { useTurnActivity, TurnActivityStrip } from "./turn-activity"; import type { StreamingReplyState } from "./streaming-reply"; import { @@ -807,6 +812,17 @@ function ChatWorkspaceInner({ [tenantId, activeWorkbenchId], ); + // CL-7201: fire-and-forget, matching the reaction/pin handlers above — + // the composer's own `stopping` state is the user-visible feedback, + // and the timeline's cancelled-turn notice (not this response) is + // what actually clears the typing indicator once the turn settles. + const handleStopTurn = useCallback(() => { + if (activeWorkbenchId === null) return; + cancelWorkbenchTurn(tenantId, activeWorkbenchId).catch(() => + toast(CHAT_STRINGS.turnCancelError), + ); + }, [tenantId, activeWorkbenchId]); + const handlePinMessage = useCallback( (messageId: string) => { if (activeWorkbenchId === null) return; @@ -1636,6 +1652,8 @@ function ChatWorkspaceInner({ bringInLoadError={bringInLoadError} placeholder={composerPlaceholderFor(activeWorkbench)} onSend={handleSend} + running={isPendingReply(streamingReply)} + onStop={handleStopTurn} onInviteAgent={() => setInviteDialogOpen(true)} onOpenAgentsSettings={() => openWorkbenchSettings("agents") diff --git a/packages/chat-ui/src/composer.tsx b/packages/chat-ui/src/composer.tsx index 96bb5686c..3b29d3e0e 100644 --- a/packages/chat-ui/src/composer.tsx +++ b/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,18 @@ 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 + * `isPendingReply`-style signal. 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 +347,14 @@ 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 `isPendingReply(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. */ + readonly onStop?: () => void; } >(function Composer( { @@ -347,6 +368,8 @@ export const Composer = forwardRef< onOpenAgentsSettings, onCreateRoutineInSpace, placeholder = CHAT_STRINGS.composerPlaceholder, + running = false, + onStop, }, ref, ) { @@ -366,6 +389,12 @@ export const Composer = forwardRef< const [preparing, setPreparing] = useState(false); const [errorMessage, setErrorMessage] = useState(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(null); const fileInputRef = useRef(null); const attachGenerationRef = useRef(0); @@ -377,6 +406,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 +734,12 @@ export const Composer = forwardRef< void addFiles(event.target.files); } + function handleStop() { + if (stopping || onStop === undefined) return; + setStopping(true); + onStop(); + } + return (
{slash !== null && ( @@ -907,6 +946,20 @@ export const Composer = forwardRef< > {CHAT_STRINGS.composerKeyboardHint} + {canStopComposer({ running }) ? ( + + ) : null}