summaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
-rw-r--r--frontend-cancel-queued-message-handoff.md140
-rw-r--r--packages/message-queue/src/index.ts1
-rw-r--r--packages/message-queue/src/pure.test.ts79
-rw-r--r--packages/message-queue/src/pure.ts36
-rw-r--r--packages/message-queue/src/service.test.ts69
-rw-r--r--packages/message-queue/src/service.ts33
-rw-r--r--packages/session-orchestrator/src/index.ts1
-rw-r--r--packages/session-orchestrator/src/orchestrator.ts39
-rw-r--r--packages/session-orchestrator/src/queue.test.ts180
-rw-r--r--packages/transport-contract/package.json2
-rw-r--r--packages/transport-contract/src/index.ts42
-rw-r--r--packages/transport-http/src/app.test.ts146
-rw-r--r--packages/transport-http/src/app.ts23
-rw-r--r--packages/transport-http/src/extension.ts1
-rw-r--r--packages/transport-ws/src/extension.ts17
-rw-r--r--packages/transport-ws/src/router.test.ts55
-rw-r--r--packages/transport-ws/src/router.ts50
17 files changed, 910 insertions, 4 deletions
diff --git a/frontend-cancel-queued-message-handoff.md b/frontend-cancel-queued-message-handoff.md
new file mode 100644
index 0000000..b92abd4
--- /dev/null
+++ b/frontend-cancel-queued-message-handoff.md
@@ -0,0 +1,140 @@
+# FE handoff — cancel a queued message
+
+Courier this to `../frontend` (cross-repo contract change; `lsp references` does
+not span repos — ORCHESTRATOR §7). All changes are ADDITIVE — nothing existing
+breaks.
+
+## What shipped (backend)
+
+A per-message **cancel** for the steering message queue: while a turn is
+GENERATING and a user message is sitting in the queue (waiting to be delivered
+as steering at the next tool-result boundary, or carried into a new turn), the
+client can **cancel a single queued message by id** so it never runs. The
+message is removed from the queue and is never delivered as steering and never
+carried into a new turn.
+
+This complements the existing `chat.queue` enqueue (see
+`frontend-message-queue-handoff.md`). Enqueue adds; cancel removes one.
+
+Versions: `@dispatch/transport-contract` `0.23.0 → 0.24.0`. Bump the pinned
+`file:` dep. (`@dispatch/wire` is unchanged — `QueuedMessage` already has the
+`id` the cancel targets; no new wire type was needed.)
+
+## The two entry points (pick one — same backend behavior)
+
+### 1. WebSocket op: `chat.queue.cancel` (what the FE should use)
+
+```ts
+interface ChatQueueCancelMessage {
+ readonly type: "chat.queue.cancel";
+ readonly conversationId: string;
+ readonly messageId: string; // the stable QueuedMessage.id (from the queue surface / enqueue response)
+}
+```
+(additive to `WsClientMessage`.)
+
+- **Fire-and-forget**, exactly like `chat.queue`. On success the server emits
+ NOTHING back — the `message-queue` SURFACE updates (the cancelled message
+ leaves the `payload.messages` snapshot). On failure (missing/empty
+ `conversationId` or `messageId`) the server replies `chat.error`
+ (`{ type: "chat.error"; conversationId?; message }`).
+- **Idempotent:** cancelling a message that is no longer queued (already
+ drained/delivered as steering, or already cancelled, or never existed, or an
+ unknown conversation) is a **silent no-op** — no surface update, no error. So
+ a client may optimistically remove the row from its queue UI on click and
+ fire-and-forget the cancel; if the message was already gone, nothing breaks.
+- **`messageId`** is the stable client-visible `QueuedMessage.id` you already
+ render from the queue surface snapshot (or got back from `chat.queue` /
+ `POST /conversations/:id/queue`'s `queue[]`).
+
+### 2. HTTP path (for the CLI / non-WS clients; the FE uses the WS op above)
+
+`DELETE /conversations/:id/queue/:messageId` (no request body) → `QueueCancelResponse`:
+
+```ts
+interface QueueCancelResponse {
+ readonly conversationId: string;
+ readonly cancelled: boolean; // true = a message was found + removed
+ readonly queue: readonly QueuedMessage[]; // post-cancel snapshot
+}
+```
+- `cancelled: true` — the message was in the queue and has been removed (it will
+ never run).
+- `cancelled: false` — the message was NOT in the queue (already
+ drained/delivered, never existed, unknown conversation) OR the message-queue
+ extension isn't loaded (degraded). Still HTTP **200** (idempotent — not an
+ error).
+- `queue` is the post-cancel snapshot (empty when no queue extension is loaded).
+
+## How the FE confirms a cancel
+
+The queue is control/state on the **surface** channel (NOT the chat stream), so
+cancel is confirmed the same way enqueue is — by the `message-queue` surface
+updating:
+
+1. You already **subscribe to the `message-queue` surface** (scope
+ `conversation`) and render `payload.messages` (`QueuedMessage[]`) with the
+ `rendererId: "message-queue"` custom renderer (per
+ `frontend-message-queue-handoff.md`).
+2. On cancel, the surface pushes a **full new spec** whose `payload.messages`
+ no longer contains the cancelled id. The cancelled message simply leaves the
+ queue list — render the new snapshot.
+3. **No new `AgentEvent`** is emitted for a cancel. The cancelled message never
+ appears in the transcript (it was never delivered as steering — that's the
+ point). If a message was already drained (delivered as a `steering` bubble or
+ carried into a new turn's `user-message`) before the cancel arrived, the
+ cancel is a no-op (`cancelled: false`) and the transcript is unchanged.
+
+## UX suggestion
+
+- Render a **cancel (×) affordance** on each pending row in the queue UI (the
+ surface snapshot gives you the `id` to target). On click → send
+ `chat.queue.cancel { conversationId, messageId }`.
+- Optimistically remove the row from the queue UI on click; the surface update
+ will confirm it (or, if the message was already drained, the surface already
+ shows it gone — no harm).
+- Expect a `chat.error` only for a malformed send (empty `conversationId` /
+ `messageId`) — in practice a client that sends the id it just rendered will
+ never hit this.
+
+## Race notes (safe by construction)
+
+- **Cancel vs. drain (steering delivery):** if the kernel drains the queue at a
+ tool-result boundary in the instant between the user clicking cancel and the
+ server processing it, the message is already gone — the cancel returns
+ `cancelled: false` (a no-op). The drained message was delivered as a
+ `steering` event and is in the transcript; the cancel correctly did nothing.
+ No double-delivery, no error.
+- **Cancel vs. post-seal carry:** same — if the turn sealed and the queue was
+ carried into a new turn before the cancel ran, the message is already the new
+ turn's opening `user-message`; the cancel is a no-op.
+- **Cancel is scoped per conversation:** cancelling on conversation A never
+ touches conversation B's queue.
+
+## What we need the FE to do
+
+1. **Bump pinned dep:** `@dispatch/transport-contract` → `0.24.0`.
+2. **Add `chat.queue.cancel`** to the FE's `WsClientMessage` union (it is
+ additive — no exhaustive switch breaks; if the FE has one, add the
+ `chat.queue.cancel` case to its WS dispatcher).
+3. **Cancel affordance per queued row:** a × / cancel button on each pending
+ message in the queue UI that sends
+ `chat.queue.cancel { conversationId, messageId }` using the row's `id`.
+ Optimistically remove the row; reconcile from the surface update.
+4. **No new event handling** — the existing `message-queue` surface subscription
+ already reflects the post-cancel snapshot; just render it. No `steering` /
+ transcript change for a cancelled message (it never runs).
+
+## Notes / known gaps
+
+- **No CLI command.** The CLI's `send --queue` enqueues but there is no CLI
+ command to list the queue or obtain a `messageId`, so a CLI `cancel` was not
+ added (it would have no way to discover a message id). The HTTP
+ `DELETE /conversations/:id/queue/:messageId` is available for any non-WS
+ client that already knows the id (e.g. from a prior
+ `POST /conversations/:id/queue` response's `queue[]`).
+- **Close-with-queued-messages** (the open product question noted in
+ `frontend-message-queue-handoff.md`) is unchanged by this feature: an
+ explicit `POST /conversations/:id/close` still aborts the in-flight turn and
+ the carry still fires. Cancel is a separate, per-message affordance that does
+ not touch the turn.
diff --git a/packages/message-queue/src/index.ts b/packages/message-queue/src/index.ts
index 11467e1..ae0868b 100644
--- a/packages/message-queue/src/index.ts
+++ b/packages/message-queue/src/index.ts
@@ -10,6 +10,7 @@ export type { QueuedMessage, QueuePayload } from "@dispatch/wire";
export { extension, manifest } from "./extension.js";
export {
buildQueueSpec,
+ cancel,
combine,
drain,
enqueue,
diff --git a/packages/message-queue/src/pure.test.ts b/packages/message-queue/src/pure.test.ts
index 3fd6039..4fca0fa 100644
--- a/packages/message-queue/src/pure.test.ts
+++ b/packages/message-queue/src/pure.test.ts
@@ -2,6 +2,7 @@ import type { QueuedMessage } from "@dispatch/wire";
import { describe, expect, it } from "vitest";
import {
buildQueueSpec,
+ cancel,
combine,
drain,
enqueue,
@@ -98,6 +99,84 @@ describe("drain", () => {
});
});
+describe("cancel", () => {
+ it("removes the matching message and returns the post-cancel snapshot", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps); // id-1
+ enqueue(state, "c1", "b", deps); // id-2
+ enqueue(state, "c1", "c", deps); // id-3
+
+ const snapshot = cancel(state, "c1", "id-2");
+ expect(snapshot.map((m) => m.id)).toEqual(["id-1", "id-3"]);
+ expect(snapshot.map((m) => m.text)).toEqual(["a", "c"]);
+ // live state reflects the removal
+ expect(getQueue(state, "c1").map((m) => m.id)).toEqual(["id-1", "id-3"]);
+ });
+
+ it("removing the only message drops the key (queue is empty + clean)", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "only", deps); // id-1
+
+ const snapshot = cancel(state, "c1", "id-1");
+ expect(snapshot).toEqual([]);
+ expect(getQueue(state, "c1")).toEqual([]);
+ // key removed so a fresh getQueue is a clean empty (not a lingering [] key)
+ expect(state.has("c1")).toBe(false);
+ });
+
+ it("returns a COPY — mutating the snapshot does not affect live state", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps);
+ enqueue(state, "c1", "b", deps);
+
+ const snapshot = cancel(state, "c1", "id-1");
+ snapshot.push({ id: "evil", text: "mutate", queuedAt: 0 });
+ expect(getQueue(state, "c1")).toHaveLength(1);
+ });
+
+ it("is idempotent — cancelling a missing id is a no-op (returns snapshot without it)", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps); // id-1
+
+ // unknown message id
+ const snapshot = cancel(state, "c1", "nope");
+ expect(snapshot.map((m) => m.id)).toEqual(["id-1"]);
+ expect(getQueue(state, "c1")).toHaveLength(1);
+
+ // a second cancel of the already-removed id-1 (re-add then cancel twice)
+ cancel(state, "c1", "id-1");
+ expect(getQueue(state, "c1")).toEqual([]);
+ expect(cancel(state, "c1", "id-1")).toEqual([]); // already gone — no-op
+ });
+
+ it("is scoped per conversation — cancelling on one conversation does not affect another", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps); // id-1
+ enqueue(state, "c2", "b", deps); // id-2
+
+ const snapshot = cancel(state, "c1", "id-1");
+ expect(snapshot).toEqual([]);
+ expect(getQueue(state, "c1")).toEqual([]);
+ // c2 untouched
+ expect(getQueue(state, "c2").map((m) => m.id)).toEqual(["id-2"]);
+ });
+
+ it("cancelling on an unknown / empty conversation is a no-op (returns [])", () => {
+ const state: MessageQueueState = new Map();
+ expect(cancel(state, "unknown", "anything")).toEqual([]);
+ // unknown id on a conversation that exists but is empty post-drain
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps);
+ drain(state, "c1"); // empties + deletes the key
+ expect(cancel(state, "c1", "id-1")).toEqual([]);
+ });
+});
+
describe("combine", () => {
it("combine joins texts with blank-line separator", () => {
const msgs: QueuedMessage[] = [
diff --git a/packages/message-queue/src/pure.ts b/packages/message-queue/src/pure.ts
index 981e005..ffb9dc2 100644
--- a/packages/message-queue/src/pure.ts
+++ b/packages/message-queue/src/pure.ts
@@ -75,6 +75,42 @@ export function drain(state: MessageQueueState, conversationId: string): QueuedM
}
/**
+ * Cancel: remove a SINGLE queued message by id from a conversation's queue so
+ * it never runs (never delivered as steering, never carried into a new turn).
+ * Returns the post-cancel queue snapshot (a fresh array copy). Idempotent — if
+ * no message with `messageId` exists in the conversation's queue (already
+ * drained/delivered, never existed, unknown conversation) the queue is
+ * unchanged and the returned snapshot simply does not contain it; the caller
+ * distinguishes "removed" from "not found" via the `cancelWithFlag` helper
+ * (this returns the snapshot only, like the other pure ops).
+ *
+ * The cancelled message is dropped entirely — it is NOT returned (the caller
+ * does not need it; the surface re-renders from the snapshot). Mutates `state`
+ * in place (splices the message out of the conversation's array).
+ */
+export function cancel(
+ state: MessageQueueState,
+ conversationId: string,
+ messageId: string,
+): QueuedMessage[] {
+ const existing = state.get(conversationId);
+ if (existing === undefined || existing.length === 0) {
+ return getQueue(state, conversationId);
+ }
+ const idx = existing.findIndex((m) => m.id === messageId);
+ if (idx === -1) {
+ return getQueue(state, conversationId);
+ }
+ existing.splice(idx, 1);
+ // If the queue is now empty, drop the key so `getQueue` stays a clean empty
+ // array (mirrors `drain` deleting the key on empty).
+ if (existing.length === 0) {
+ state.delete(conversationId);
+ }
+ return getQueue(state, conversationId);
+}
+
+/**
* Combine drained messages' texts into a single steering string, joined by a
* blank line (`\n\n`). Pure — the session-orchestrator builds the final
* ChatMessage from this.
diff --git a/packages/message-queue/src/service.test.ts b/packages/message-queue/src/service.test.ts
index aa59dd3..086414e 100644
--- a/packages/message-queue/src/service.test.ts
+++ b/packages/message-queue/src/service.test.ts
@@ -99,3 +99,72 @@ describe("message-queue service", () => {
expect(combine(drained)).toBe("alpha\n\nbeta");
});
});
+
+describe("message-queue service cancel", () => {
+ it("cancel removes the message and pushes a surface update (queue shrank)", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "a"); // q-1
+ svc.enqueue("c1", "b"); // q-2
+ svc.enqueue("c1", "c"); // q-3
+ expect(deps.calls.value).toBe(3); // three enqueues notified
+
+ const snapshot = svc.cancel("c1", "q-2");
+ expect(deps.calls.value).toBe(4); // cancel pushed a surface update
+ expect(snapshot.map((m) => m.id)).toEqual(["q-1", "q-3"]);
+
+ // live state reflects the removal
+ expect(svc.getQueue("c1").map((m) => m.id)).toEqual(["q-1", "q-3"]);
+ });
+
+ it("cancel of the only message empties the queue + pushes a surface update", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "only"); // q-1
+ expect(deps.calls.value).toBe(1);
+
+ const snapshot = svc.cancel("c1", "q-1");
+ expect(deps.calls.value).toBe(2); // surface update (queue → empty)
+ expect(snapshot).toEqual([]);
+ expect(svc.getQueue("c1")).toEqual([]);
+ });
+
+ it("cancel of a missing id does NOT push a surface update (no change)", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "a"); // q-1
+ expect(deps.calls.value).toBe(1);
+
+ // unknown message id — no-op, no notify
+ const snapshot = svc.cancel("c1", "nope");
+ expect(deps.calls.value).toBe(1); // unchanged — no change
+ expect(snapshot.map((m) => m.id)).toEqual(["q-1"]);
+
+ // unknown conversation — also a no-op, no notify
+ expect(svc.cancel("nope", "q-1")).toEqual([]);
+ expect(deps.calls.value).toBe(1); // still unchanged
+ });
+
+ it("cancel after a drain (queue empty) is a no-op with no surface update", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "a"); // q-1
+ svc.drain("c1");
+ expect(deps.calls.value).toBe(2); // enqueue + drain
+
+ expect(svc.cancel("c1", "q-1")).toEqual([]);
+ expect(deps.calls.value).toBe(2); // no notify — queue was already empty
+ });
+
+ it("cancel is scoped per conversation", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "a"); // q-1
+ svc.enqueue("c2", "b"); // q-2
+
+ const snapshot = svc.cancel("c1", "q-1");
+ expect(snapshot).toEqual([]);
+ expect(svc.getQueue("c1")).toEqual([]);
+ expect(svc.getQueue("c2").map((m) => m.id)).toEqual(["q-2"]);
+ });
+});
diff --git a/packages/message-queue/src/service.ts b/packages/message-queue/src/service.ts
index 97e270d..db12fd8 100644
--- a/packages/message-queue/src/service.ts
+++ b/packages/message-queue/src/service.ts
@@ -11,7 +11,12 @@
import { defineService, type Logger, type ServiceHandle } from "@dispatch/kernel";
import type { QueuedMessage } from "@dispatch/wire";
import type { MessageQueueState, QueueDeps } from "./pure.js";
-import { drain as drainQueue, enqueue as enqueueMessage, getQueue as readQueue } from "./pure.js";
+import {
+ cancel as cancelMessage,
+ drain as drainQueue,
+ enqueue as enqueueMessage,
+ getQueue as readQueue,
+} from "./pure.js";
/**
* The message-queue service interface. Obtained via
@@ -29,6 +34,14 @@ export interface MessageQueueService {
* was empty (and then NO surface update is pushed — no change).
*/
drain(conversationId: string): QueuedMessage[];
+ /**
+ * Cancel: remove a SINGLE queued message by id so it never runs (never
+ * delivered as steering, never carried into a new turn). Returns the
+ * post-cancel queue snapshot. A surface update is pushed ONLY when a message
+ * was actually removed (the queue shrank); cancelling a missing id is a
+ * no-op that pushes nothing (no change). Idempotent.
+ */
+ cancel(conversationId: string, messageId: string): QueuedMessage[];
}
/**
@@ -84,5 +97,23 @@ export function createMessageQueueService(deps: MessageQueueDeps): MessageQueueS
deps.notify();
return drained;
},
+ cancel(conversationId, messageId) {
+ // Notify ONLY on a real change (the queue shrank). Compare the pre-cancel
+ // length to the post-cancel snapshot — a missing id is a no-op that pushes
+ // no surface update (mirrors drain's no-notify-on-empty rule).
+ const beforeLen = readQueue(state, conversationId).length;
+ const snapshot = cancelMessage(state, conversationId, messageId);
+ if (snapshot.length === beforeLen) {
+ // nothing removed — no change, no surface push
+ return snapshot;
+ }
+ deps.logger?.debug("message-queue: cancelled", {
+ conversationId,
+ messageId,
+ queueLen: snapshot.length,
+ });
+ deps.notify();
+ return snapshot;
+ },
};
}
diff --git a/packages/session-orchestrator/src/index.ts b/packages/session-orchestrator/src/index.ts
index 3d3e816..360e8d2 100644
--- a/packages/session-orchestrator/src/index.ts
+++ b/packages/session-orchestrator/src/index.ts
@@ -1,5 +1,6 @@
export { extension, manifest } from "./extension.js";
export {
+ type CancelQueuedMessageResult,
type CompactionService,
type ConversationClosedPayload,
type ConversationCompactedPayload,
diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts
index 6e5cfca..85a31ff 100644
--- a/packages/session-orchestrator/src/orchestrator.ts
+++ b/packages/session-orchestrator/src/orchestrator.ts
@@ -170,6 +170,20 @@ export interface EnqueueResult {
readonly queue: readonly QueuedMessage[];
}
+/**
+ * Result of `SessionOrchestrator.cancelQueuedMessage`. `cancelled` is true when
+ * a message with the given id was found in the conversation's queue and removed
+ * (it will never run — never delivered as steering, never carried into a new
+ * turn). `cancelled` is false when the message was not in the queue (already
+ * drained/delivered, never existed, unknown conversation) OR when the
+ * message-queue extension isn't loaded (degraded — feature off). `queue` is the
+ * post-cancel snapshot (empty when no queue extension is loaded).
+ */
+export interface CancelQueuedMessageResult {
+ readonly cancelled: boolean;
+ readonly queue: readonly QueuedMessage[];
+}
+
export type TurnEventListener = (event: AgentEvent) => void;
interface ActiveTurn {
@@ -331,6 +345,18 @@ export interface SessionOrchestrator {
* returned snapshot is empty (degraded — feature off).
*/
enqueue(input: EnqueueInput): EnqueueResult;
+ /**
+ * Cancel (remove) a SINGLE queued message by id so it never runs. The single
+ * entry transports call to cancel a queued steering message. Resolves the
+ * message-queue service lazily (same as `enqueue`); when the extension isn't
+ * loaded the call degrades to `{ cancelled: false, queue: [] }`. Idempotent —
+ * cancelling a message that is no longer queued (already drained/delivered)
+ * returns `{ cancelled: false, ... }` without error.
+ */
+ cancelQueuedMessage(input: {
+ readonly conversationId: string;
+ readonly messageId: string;
+ }): CancelQueuedMessageResult;
subscribe(conversationId: string, listener: TurnEventListener): () => void;
isActive(conversationId: string): boolean;
/**
@@ -1133,6 +1159,19 @@ export function createSessionOrchestrator(
return { startedTurn: false, queue: snapshot };
},
+ cancelQueuedMessage({ conversationId, messageId }) {
+ // When the message-queue extension isn't loaded this degrades: nothing to
+ // cancel, empty snapshot (feature off). Mirrors `enqueue`'s degraded path.
+ const queue = deps.resolveQueue?.();
+ if (queue === undefined) {
+ return { cancelled: false, queue: [] };
+ }
+ const beforeLen = queue.getQueue(conversationId).length;
+ const snapshot = queue.cancel(conversationId, messageId);
+ const cancelled = snapshot.length < beforeLen;
+ return { cancelled, queue: snapshot };
+ },
+
subscribe(conversationId, listener) {
let listeners = subscribers.get(conversationId);
if (listeners === undefined) {
diff --git a/packages/session-orchestrator/src/queue.test.ts b/packages/session-orchestrator/src/queue.test.ts
index a09a441..65cc06f 100644
--- a/packages/session-orchestrator/src/queue.test.ts
+++ b/packages/session-orchestrator/src/queue.test.ts
@@ -593,3 +593,183 @@ describe("enqueue", () => {
await sealed;
});
});
+
+// --- cancelQueuedMessage facade (remove a single queued message by id) ---
+
+describe("cancelQueuedMessage", () => {
+ it("removes a queued message by id → cancelled:true + post-cancel snapshot", () => {
+ const store = createInMemoryStore();
+ const queue = createTestQueue();
+ queue.enqueue("conv-cancel", "a");
+ queue.enqueue("conv-cancel", "b");
+ const second = queue.getQueue("conv-cancel")[1];
+ if (second === undefined) throw new Error("expected a second enqueued message");
+ const secondId = second.id;
+ expect(queue.getQueue("conv-cancel")).toHaveLength(2);
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => simpleProvider(),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn,
+ resolveQueue: () => queue,
+ });
+
+ const result = orchestrator.cancelQueuedMessage({
+ conversationId: "conv-cancel",
+ messageId: secondId,
+ });
+ expect(result.cancelled).toBe(true);
+ expect(result.queue.map((m) => m.id)).not.toContain(secondId);
+ expect(result.queue).toHaveLength(1);
+ expect(queue.getQueue("conv-cancel").map((m) => m.id)).not.toContain(secondId);
+ });
+
+ it("cancel of the only message empties the queue (cancelled:true)", () => {
+ const store = createInMemoryStore();
+ const queue = createTestQueue();
+ queue.enqueue("conv-only", "solo");
+ const onlyId = queue.getQueue("conv-only")[0]?.id;
+ if (onlyId === undefined) throw new Error("expected an enqueued message");
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => simpleProvider(),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn,
+ resolveQueue: () => queue,
+ });
+
+ const result = orchestrator.cancelQueuedMessage({
+ conversationId: "conv-only",
+ messageId: onlyId,
+ });
+ expect(result.cancelled).toBe(true);
+ expect(result.queue).toEqual([]);
+ expect(queue.getQueue("conv-only")).toEqual([]);
+ });
+
+ it("cancel of a missing id → cancelled:false, queue unchanged (idempotent)", () => {
+ const store = createInMemoryStore();
+ const queue = createTestQueue();
+ queue.enqueue("conv-miss", "a");
+ queue.enqueue("conv-miss", "b");
+ const before = queue.getQueue("conv-miss");
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => simpleProvider(),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn,
+ resolveQueue: () => queue,
+ });
+
+ const result = orchestrator.cancelQueuedMessage({
+ conversationId: "conv-miss",
+ messageId: "does-not-exist",
+ });
+ expect(result.cancelled).toBe(false);
+ expect(result.queue.map((m) => m.id)).toEqual(before.map((m) => m.id));
+ // live state unchanged
+ expect(queue.getQueue("conv-miss").map((m) => m.id)).toEqual(before.map((m) => m.id));
+ });
+
+ it("cancel on an unknown conversation → cancelled:false, empty queue", () => {
+ const store = createInMemoryStore();
+ const queue = createTestQueue();
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => simpleProvider(),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn,
+ resolveQueue: () => queue,
+ });
+
+ const result = orchestrator.cancelQueuedMessage({
+ conversationId: "never-existed",
+ messageId: "anything",
+ });
+ expect(result.cancelled).toBe(false);
+ expect(result.queue).toEqual([]);
+ });
+
+ it("no queue ext (resolveQueue undefined) → cancelled:false, empty queue (degraded)", () => {
+ const store = createInMemoryStore();
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => simpleProvider(),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn,
+ // resolveQueue intentionally omitted — feature degrades off
+ });
+
+ const result = orchestrator.cancelQueuedMessage({
+ conversationId: "conv-noqueue",
+ messageId: "whatever",
+ });
+ expect(result.cancelled).toBe(false);
+ expect(result.queue).toEqual([]);
+ });
+
+ it("cancelled message is NOT delivered as steering (never runs)", async () => {
+ const store = createInMemoryStore();
+ const queue = createTestQueue();
+ queue.enqueue("conv-steer", "keep-me");
+ const cancelId = queue.enqueue("conv-steer", "cancel-me")[1]?.id;
+ if (cancelId === undefined) throw new Error("expected a second enqueued message");
+ // the first message id (kept)
+ const keepId = queue.getQueue("conv-steer")[0]?.id;
+ if (keepId === undefined) throw new Error("expected a kept message");
+
+ const { captured, drainedMessages, runTurn: captureRunTurn } = createDrainingCaptureRunTurn();
+
+ const { orchestrator } = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => ({ id: "p", stream: async function* () {} }),
+ resolveTools: noTools,
+ applyToolsFilter: identityApplyToolsFilter,
+ runTurn: captureRunTurn,
+ resolveQueue: () => queue,
+ });
+
+ // Cancel the second message BEFORE the turn drains.
+ const cancelResult = orchestrator.cancelQueuedMessage({
+ conversationId: "conv-steer",
+ messageId: cancelId,
+ });
+ expect(cancelResult.cancelled).toBe(true);
+ expect(cancelResult.queue.map((m) => m.id)).toEqual([keepId]);
+
+ const events: AgentEvent[] = [];
+ const unsub = orchestrator.subscribe("conv-steer", (e) => events.push(e));
+
+ orchestrator.startTurn({ conversationId: "conv-steer", text: "go" });
+ await waitForSealed(orchestrator, "conv-steer");
+ unsub();
+
+ // The steering drain combined ONLY the kept message — the cancelled one
+ // is absent from the drained text.
+ expect(drainedMessages).toHaveLength(1);
+ const steerMsg = drainedMessages[0];
+ if (steerMsg === undefined) throw new Error("expected a drained message");
+ const chunk = steerMsg.chunks[0];
+ if (chunk === undefined || chunk.type !== "text") throw new Error("expected text chunk");
+ expect(chunk.text).toBe("keep-me");
+ expect(chunk.text).not.toContain("cancel-me");
+
+ // drainSteering was wired + the queue is now empty (the kept one drained).
+ expect(captured[0]?.drainSteering).toBeDefined();
+ expect(queue.getQueue("conv-steer")).toHaveLength(0);
+
+ // The steering event carries only the kept text.
+ const steering = events.find(isSteering);
+ expect(steering?.text).toBe("keep-me");
+ });
+});
diff --git a/packages/transport-contract/package.json b/packages/transport-contract/package.json
index 660898a..0f65b7b 100644
--- a/packages/transport-contract/package.json
+++ b/packages/transport-contract/package.json
@@ -1,6 +1,6 @@
{
"name": "@dispatch/transport-contract",
- "version": "0.23.0",
+ "version": "0.24.0",
"type": "module",
"private": true,
"main": "dist/index.js",
diff --git a/packages/transport-contract/src/index.ts b/packages/transport-contract/src/index.ts
index 797ad22..766abb6 100644
--- a/packages/transport-contract/src/index.ts
+++ b/packages/transport-contract/src/index.ts
@@ -468,6 +468,27 @@ export interface QueueResponse {
readonly queue: readonly QueuedMessage[];
}
+/**
+ * Response body for
+ * `DELETE /conversations/:id/queue/:messageId` — cancel (remove) a single
+ * queued steering message by id so it never runs.
+ *
+ * `cancelled` is `true` when a message with the given id was found in the
+ * conversation's queue and removed (it will never be delivered as steering nor
+ * carried into a new turn). `cancelled` is `false` when the message was not in
+ * the queue (already drained/delivered, never existed, unknown conversation)
+ * OR when the message-queue extension isn't loaded (degraded — feature off).
+ * `queue` is the post-cancel snapshot (empty when no queue extension is
+ * loaded). Idempotent — cancelling a message that is no longer queued returns
+ * `cancelled: false` with HTTP 200 (not an error), so a client may optimistically
+ * fire-and-forget a cancel and reconcile from the surface.
+ */
+export interface QueueCancelResponse {
+ readonly conversationId: string;
+ readonly cancelled: boolean;
+ readonly queue: readonly QueuedMessage[];
+}
+
// ─── Per-conversation LSP status ──────────────────────────────────────────────
/** The connection state of a single language server for a workspace. */
@@ -688,6 +709,24 @@ export interface ChatQueueMessage {
}
/**
+ * Client → server: cancel (remove) a SINGLE queued steering message by id so
+ * it never runs. The WebSocket counterpart of the HTTP
+ * `DELETE /conversations/:id/queue/:messageId` (`QueueCancelResponse`).
+ * Fire-and-forget: success is confirmed by the message-queue SURFACE updating
+ * (the cancelled message leaves the snapshot); a failure (missing/empty
+ * `conversationId` or `messageId`) arrives as a `chat.error`. Idempotent —
+ * cancelling a message that is no longer queued (already drained/delivered) is
+ * a silent no-op (no surface update, no error). `messageId` is the stable
+ * client-visible `QueuedMessage.id` (obtained from the queue surface snapshot
+ * or the enqueue response).
+ */
+export interface ChatQueueCancelMessage {
+ readonly type: "chat.queue.cancel";
+ readonly conversationId: string;
+ readonly messageId: string;
+}
+
+/**
* Every client → server WS message: surface ops (`@dispatch/ui-contract`) + chat
* ops. A server discriminates on `type`.
*/
@@ -696,7 +735,8 @@ export type WsClientMessage =
| ChatSendMessage
| ChatSubscribeMessage
| ChatUnsubscribeMessage
- | ChatQueueMessage;
+ | ChatQueueMessage
+ | ChatQueueCancelMessage;
/**
* Every server → client WS message: surface ops (`@dispatch/ui-contract`) + chat
diff --git a/packages/transport-http/src/app.test.ts b/packages/transport-http/src/app.test.ts
index 03f1959..f7afed7 100644
--- a/packages/transport-http/src/app.test.ts
+++ b/packages/transport-http/src/app.test.ts
@@ -15,6 +15,7 @@ import { DEFAULT_TEMPLATE } from "@dispatch/system-prompt";
import { createThroughputStore, dayKeyOf } from "@dispatch/throughput-store";
import type {
DeleteWorkspaceResponse,
+ QueueCancelResponse,
QueuedMessage,
QueueResponse,
SystemPromptVariable,
@@ -273,6 +274,9 @@ function createFakeOrchestrator(events: AgentEvent[]): SessionOrchestrator {
enqueue() {
return { startedTurn: false, queue: [] };
},
+ cancelQueuedMessage() {
+ return { cancelled: false, queue: [] };
+ },
closeConversation() {
return { abortedTurn: false };
},
@@ -309,6 +313,9 @@ function createCapturingOrchestrator(): SessionOrchestrator & {
enqueue() {
return { startedTurn: false, queue: [] };
},
+ cancelQueuedMessage() {
+ return { cancelled: false, queue: [] };
+ },
closeConversation() {
return { abortedTurn: false };
},
@@ -335,6 +342,9 @@ function createThrowingOrchestrator(error: Error): SessionOrchestrator {
enqueue() {
return { startedTurn: false, queue: [] };
},
+ cancelQueuedMessage() {
+ return { cancelled: false, queue: [] };
+ },
closeConversation() {
return { abortedTurn: false };
},
@@ -2069,6 +2079,142 @@ describe("POST /conversations/:id/queue", () => {
});
});
+describe("DELETE /conversations/:id/queue/:messageId", () => {
+ it("when a message is cancelled → 200 + QueueCancelResponse (cancelled:true + post-cancel queue)", async () => {
+ const remaining: readonly QueuedMessage[] = [
+ { id: "q1", text: "kept", queuedAt: 1700000000000 },
+ ];
+ let received: { conversationId: string; messageId: string } | undefined;
+ const orchestrator: SessionOrchestrator = {
+ ...createFakeOrchestrator([]),
+ cancelQueuedMessage(input) {
+ received = input;
+ return { cancelled: true, queue: remaining };
+ },
+ };
+ const app = createApp({
+ conversationStore: createFakeConversationStore(),
+ orchestrator,
+ credentialStore: createFakeCredentialStore([]),
+ logger: noopLogger,
+ });
+
+ const res = await app.request("/conversations/conv1/queue/q2", {
+ method: "DELETE",
+ });
+
+ expect(res.status).toBe(200);
+ const body = (await res.json()) as QueueCancelResponse;
+ expect(body.conversationId).toBe("conv1");
+ expect(body.cancelled).toBe(true);
+ expect(body.queue).toEqual(remaining);
+ // forwards the path conversationId + messageId
+ expect(received?.conversationId).toBe("conv1");
+ expect(received?.messageId).toBe("q2");
+ });
+
+ it("when the message is not in the queue → 200 cancelled:false (idempotent, not an error)", async () => {
+ const queue: readonly QueuedMessage[] = [
+ { id: "q1", text: "still-queued", queuedAt: 1700000000000 },
+ ];
+ const orchestrator: SessionOrchestrator = {
+ ...createFakeOrchestrator([]),
+ cancelQueuedMessage() {
+ return { cancelled: false, queue };
+ },
+ };
+ const app = createApp({
+ conversationStore: createFakeConversationStore(),
+ orchestrator,
+ credentialStore: createFakeCredentialStore([]),
+ logger: noopLogger,
+ });
+
+ const res = await app.request("/conversations/conv1/queue/missing", {
+ method: "DELETE",
+ });
+
+ expect(res.status).toBe(200);
+ const body = (await res.json()) as QueueCancelResponse;
+ expect(body.cancelled).toBe(false);
+ expect(body.queue).toEqual(queue);
+ });
+
+ it("when the queue ext is not loaded → 200 cancelled:false, empty queue (degraded)", async () => {
+ const orchestrator: SessionOrchestrator = {
+ ...createFakeOrchestrator([]),
+ cancelQueuedMessage() {
+ return { cancelled: false, queue: [] };
+ },
+ };
+ const app = createApp({
+ conversationStore: createFakeConversationStore(),
+ orchestrator,
+ credentialStore: createFakeCredentialStore([]),
+ logger: noopLogger,
+ });
+
+ const res = await app.request("/conversations/conv1/queue/whatever", {
+ method: "DELETE",
+ });
+
+ expect(res.status).toBe(200);
+ const body = (await res.json()) as QueueCancelResponse;
+ expect(body.cancelled).toBe(false);
+ expect(body.queue).toEqual([]);
+ });
+
+ it("delegates the cancel to the orchestrator (never reads the body)", async () => {
+ let calls = 0;
+ const orchestrator: SessionOrchestrator = {
+ ...createFakeOrchestrator([]),
+ cancelQueuedMessage() {
+ calls += 1;
+ return { cancelled: true, queue: [] };
+ },
+ };
+ const app = createApp({
+ conversationStore: createFakeConversationStore(),
+ orchestrator,
+ credentialStore: createFakeCredentialStore([]),
+ logger: noopLogger,
+ });
+
+ // No Content-Type / body — the endpoint takes the messageId from the path.
+ const res = await app.request("/conversations/conv-x/queue/m1", {
+ method: "DELETE",
+ });
+
+ expect(res.status).toBe(200);
+ expect(calls).toBe(1);
+ });
+
+ it("logs an info line on success and never logs the message text", async () => {
+ const logger = createFakeLogger();
+ const orchestrator: SessionOrchestrator = {
+ ...createFakeOrchestrator([]),
+ cancelQueuedMessage() {
+ return { cancelled: true, queue: [] };
+ },
+ };
+ const app = createApp({
+ conversationStore: createFakeConversationStore(),
+ orchestrator,
+ credentialStore: createFakeCredentialStore([]),
+ logger,
+ });
+
+ await app.request("/conversations/conv1/queue/q-secret", { method: "DELETE" });
+
+ const infoLogs = logger.records.filter((r) => r.level === "info");
+ expect(infoLogs).toHaveLength(1);
+ expect(infoLogs[0]?.msg).toBe("conversations: cancelled queued message");
+ expect(infoLogs[0]?.attrs?.conversationId).toBe("conv1");
+ expect(infoLogs[0]?.attrs?.messageId).toBe("q-secret");
+ expect(infoLogs[0]?.attrs?.cancelled).toBe(true);
+ });
+});
+
describe("GET /conversations/:id/cwd", () => {
it("returns null when unset", async () => {
const app = createApp({
diff --git a/packages/transport-http/src/app.ts b/packages/transport-http/src/app.ts
index 656be9d..bce9125 100644
--- a/packages/transport-http/src/app.ts
+++ b/packages/transport-http/src/app.ts
@@ -29,6 +29,7 @@ import type {
ModelResponse,
ModelsResponse,
OpenConversationResponse,
+ QueueCancelResponse,
QueueResponse,
ReasoningEffortResponse,
SetCompactPercentRequest,
@@ -796,6 +797,28 @@ export function createApp(opts: CreateServerOptions): Hono {
return c.json(response, 200);
});
+ app.delete("/conversations/:id/queue/:messageId", (c) => {
+ const conversationId = c.req.param("id");
+ const messageId = c.req.param("messageId");
+
+ // `cancelQueuedMessage` is synchronous and owns the lookup + removal (no
+ // separate race — the pure `cancel` is idempotent). It does not throw for an
+ // unknown/idle conversation, which instead returns cancelled:false. Mirrors
+ // the direct sync call used by `POST /conversations/:id/queue`.
+ const { cancelled, queue } = opts.orchestrator.cancelQueuedMessage({
+ conversationId,
+ messageId,
+ });
+ log.info("conversations: cancelled queued message", {
+ conversationId,
+ messageId,
+ cancelled,
+ queueLength: queue.length,
+ });
+ const response: QueueCancelResponse = { conversationId, cancelled, queue };
+ return c.json(response, 200);
+ });
+
app.get("/conversations/:id/cwd", async (c) => {
const conversationId = c.req.param("id");
try {
diff --git a/packages/transport-http/src/extension.ts b/packages/transport-http/src/extension.ts
index f424e42..effbadd 100644
--- a/packages/transport-http/src/extension.ts
+++ b/packages/transport-http/src/extension.ts
@@ -56,6 +56,7 @@ export const manifest: Manifest = {
"/conversations/:id/mcp",
"/conversations/:id/open",
"/conversations/:id/queue",
+ "/conversations/:id/queue/:messageId",
"/conversations/:id/reasoning-effort",
"/conversations/:id/status",
"/conversations/:id/stop",
diff --git a/packages/transport-ws/src/extension.ts b/packages/transport-ws/src/extension.ts
index d26712b..88f721e 100644
--- a/packages/transport-ws/src/extension.ts
+++ b/packages/transport-ws/src/extension.ts
@@ -345,6 +345,23 @@ export function createTransportWsExtension(): Extension {
break;
}
+ case "chat-queue-cancel": {
+ // Fire-and-forget: success is confirmed by the message-queue
+ // SURFACE updating (the cancelled message leaves the snapshot),
+ // NOT by a reply here. Cancelling a message that is no longer
+ // queued is a silent no-op (no surface update, no error).
+ const cancelResult = orchestrator.cancelQueuedMessage({
+ conversationId: result.conversationId,
+ messageId: result.messageId,
+ });
+ logger.info?.("transport-ws: chat.queue.cancel accepted", {
+ conversationId: result.conversationId,
+ messageId: result.messageId,
+ cancelled: cancelResult.cancelled,
+ });
+ break;
+ }
+
case "chat-error": {
logger.warn?.("transport-ws: malformed chat.send", {
reason: result.errorMessage,
diff --git a/packages/transport-ws/src/router.test.ts b/packages/transport-ws/src/router.test.ts
index 3c3e70b..19b5bb5 100644
--- a/packages/transport-ws/src/router.test.ts
+++ b/packages/transport-ws/src/router.test.ts
@@ -604,6 +604,59 @@ describe("routeClientMessage", () => {
});
});
+ describe("chat.queue.cancel", () => {
+ it("routes a valid chat.queue.cancel → { kind: 'chat-queue-cancel', conversationId, messageId }", () => {
+ const registry = fakeRegistry([]);
+ const connSubs = new Set<string>();
+
+ const result = routeClientMessage(registry, connSubs, {
+ type: "chat.queue.cancel",
+ conversationId: "conv-1",
+ messageId: "q-42",
+ });
+
+ expect(result).toEqual({
+ kind: "chat-queue-cancel",
+ conversationId: "conv-1",
+ messageId: "q-42",
+ });
+ });
+
+ it("rejects empty conversationId → chat-error (no cancel signal)", () => {
+ const registry = fakeRegistry([]);
+ const connSubs = new Set<string>();
+
+ const result = routeClientMessage(registry, connSubs, {
+ type: "chat.queue.cancel",
+ conversationId: "",
+ messageId: "q-42",
+ });
+
+ expect(result.kind).toBe("chat-error");
+ if (result.kind !== "chat-error") throw new Error("expected chat-error");
+ expect(result.errorMessage).toContain("non-empty string");
+ expect(result.errorMessage).toContain("conversationId");
+ });
+
+ it("rejects empty messageId → chat-error (no cancel signal)", () => {
+ const registry = fakeRegistry([]);
+ const connSubs = new Set<string>();
+
+ for (const messageId of ["", undefined as unknown as string]) {
+ const result = routeClientMessage(registry, connSubs, {
+ type: "chat.queue.cancel",
+ conversationId: "conv-1",
+ messageId,
+ });
+
+ expect(result.kind).toBe("chat-error");
+ if (result.kind !== "chat-error") throw new Error("expected chat-error");
+ expect(result.errorMessage).toContain("non-empty string");
+ expect(result.errorMessage).toContain("messageId");
+ }
+ });
+ });
+
describe("exhaustive switch (regression guard for Wave-0 fan-out)", () => {
// Every WsClientMessage variant must route to a defined result with a
// known kind — no fall-through / undefined return. If the union is
@@ -622,6 +675,7 @@ describe("routeClientMessage", () => {
{ type: "chat.subscribe", conversationId: "c1" },
{ type: "chat.unsubscribe", conversationId: "c1" },
{ type: "chat.queue", conversationId: "c1", text: "steer" },
+ { type: "chat.queue.cancel", conversationId: "c1", messageId: "m1" },
];
const validKinds = new Set<RouteResult["kind"]>([
@@ -631,6 +685,7 @@ describe("routeClientMessage", () => {
"chat-subscribe",
"chat-unsubscribe",
"chat-queue",
+ "chat-queue-cancel",
]);
for (const msg of samples) {
diff --git a/packages/transport-ws/src/router.ts b/packages/transport-ws/src/router.ts
index 0caf305..014db96 100644
--- a/packages/transport-ws/src/router.ts
+++ b/packages/transport-ws/src/router.ts
@@ -9,6 +9,7 @@
import type { SurfaceContext, SurfaceRegistry } from "@dispatch/surface-registry";
import type {
+ ChatQueueCancelMessage,
ChatQueueMessage,
ChatSendMessage,
ChatSubscribeMessage,
@@ -99,6 +100,20 @@ export interface ChatQueueRouteResult {
readonly workspaceId?: string;
}
+/**
+ * The effect a validated chat.queue.cancel should produce. The shell calls
+ * `orchestrator.cancelQueuedMessage({ conversationId, messageId })` and emits
+ * NOTHING back (fire-and-forget): success is confirmed by the message-queue
+ * SURFACE updating (the cancelled message leaves the snapshot). Cancelling a
+ * message that is no longer queued is a silent no-op (no surface update, no
+ * error). Mirrors `ChatQueueRouteResult`'s fire-and-forget style.
+ */
+export interface ChatQueueCancelRouteResult {
+ readonly kind: "chat-queue-cancel";
+ readonly conversationId: string;
+ readonly messageId: string;
+}
+
/** The effect any client WS message should produce. */
export type RouteResult =
| SurfaceRouteResult
@@ -106,7 +121,8 @@ export type RouteResult =
| ChatRouteError
| ChatSubscribeRouteResult
| ChatUnsubscribeRouteResult
- | ChatQueueRouteResult;
+ | ChatQueueRouteResult
+ | ChatQueueCancelRouteResult;
// ── Helpers ─────────────────────────────────────────────────────────────────
@@ -152,6 +168,8 @@ export function routeClientMessage(
return handleChatUnsubscribe(msg);
case "chat.queue":
return handleChatQueue(msg);
+ case "chat.queue.cancel":
+ return handleChatQueueCancel(msg);
}
}
@@ -253,6 +271,36 @@ function handleChatQueue(msg: ChatQueueMessage): ChatQueueRouteResult | ChatRout
};
}
+/**
+ * Validate a chat.queue.cancel: both `conversationId` and `messageId` must be
+ * non-empty strings. Invalid → `chat-error` (the shell replies with `chat.error`,
+ * same style as a malformed `chat.queue`; the orchestrator is never called).
+ * Valid → `chat-queue-cancel` (the shell calls `orchestrator.cancelQueuedMessage`).
+ */
+function handleChatQueueCancel(
+ msg: ChatQueueCancelMessage,
+): ChatQueueCancelRouteResult | ChatRouteError {
+ if (typeof msg.conversationId !== "string" || msg.conversationId.length === 0) {
+ return {
+ kind: "chat-error",
+ conversationId: msg.conversationId,
+ errorMessage: "chat.queue.cancel requires a non-empty string `conversationId`",
+ };
+ }
+ if (typeof msg.messageId !== "string" || msg.messageId.length === 0) {
+ return {
+ kind: "chat-error",
+ conversationId: msg.conversationId,
+ errorMessage: "chat.queue.cancel requires a non-empty string `messageId`",
+ };
+ }
+ return {
+ kind: "chat-queue-cancel",
+ conversationId: msg.conversationId,
+ messageId: msg.messageId,
+ };
+}
+
// ── Per-message handlers ────────────────────────────────────────────────────
function handleSubscribe(