5 files changed, 217 insertions, 1 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;
+    },
   };
 }