Merge branch 'feature/cancel-queued-message' into predev

16 files changed, 770 insertions, 4 deletions

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(