1 files changed, 79 insertions, 48 deletions

diff --git a/packages/message-queue/src/service.ts b/packages/message-queue/src/service.ts
index 91d0fc5..db12fd8 100644
--- a/packages/message-queue/src/service.ts
+++ b/packages/message-queue/src/service.ts
@@ -11,24 +11,37 @@
 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
  * `host.getService(messageQueueHandle)`.
  */
 export interface MessageQueueService {
-	/** Append a message; return the CURRENT queue snapshot (post-append). */
-	enqueue(conversationId: string, text: string): QueuedMessage[];
-	/** Current queue snapshot (empty array if none / unknown conversation). */
-	getQueue(conversationId: string): QueuedMessage[];
-	/**
-	 * Drain: return all queued messages, CLEAR the queue, and push a surface
-	 * update (empty). Returns the raw `QueuedMessage[]` (NOT a ChatMessage —
-	 * the caller combines + builds the ChatMessage). Empty array if the queue
-	 * was empty (and then NO surface update is pushed — no change).
-	 */
-	drain(conversationId: string): QueuedMessage[];
+  /** Append a message; return the CURRENT queue snapshot (post-append). */
+  enqueue(conversationId: string, text: string): QueuedMessage[];
+  /** Current queue snapshot (empty array if none / unknown conversation). */
+  getQueue(conversationId: string): QueuedMessage[];
+  /**
+   * Drain: return all queued messages, CLEAR the queue, and push a surface
+   * update (empty). Returns the raw `QueuedMessage[]` (NOT a ChatMessage —
+   * the caller combines + builds the ChatMessage). Empty array if the queue
+   * 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[];
 }
 
 /**
@@ -36,19 +49,19 @@ export interface MessageQueueService {
  * session-orchestrator imports to reach the queue — no string-keyed lookup.
  */
 export const messageQueueHandle: ServiceHandle<MessageQueueService> =
-	defineService<MessageQueueService>("message-queue");
+  defineService<MessageQueueService>("message-queue");
 
 /** Deps for the service shell — the pure core's deps plus the surface notifier. */
 export interface MessageQueueDeps extends QueueDeps {
-	/**
-	 * Notify the surface-registry that the queue changed, so the transport
-	 * re-fetches + pushes a full new SurfaceUpdate. Called ONLY on a real
-	 * change (enqueue → grew; drain-non-empty → emptied); a no-op drain of an
-	 * already-empty queue does NOT notify (no change → no surface push).
-	 */
-	readonly notify: () => void;
-	/** Injected host logger (optional in tests). Logs counts/lengths at debug, never bodies. */
-	readonly logger?: Logger;
+  /**
+   * Notify the surface-registry that the queue changed, so the transport
+   * re-fetches + pushes a full new SurfaceUpdate. Called ONLY on a real
+   * change (enqueue → grew; drain-non-empty → emptied); a no-op drain of an
+   * already-empty queue does NOT notify (no change → no surface push).
+   */
+  readonly notify: () => void;
+  /** Injected host logger (optional in tests). Logs counts/lengths at debug, never bodies. */
+  readonly logger?: Logger;
 }
 
 /**
@@ -58,31 +71,49 @@ export interface MessageQueueDeps extends QueueDeps {
  * in this closure, reachable only through the returned service.
  */
 export function createMessageQueueService(deps: MessageQueueDeps): MessageQueueService {
-	const state: MessageQueueState = new Map();
+  const state: MessageQueueState = new Map();
 
-	return {
-		enqueue(conversationId, text) {
-			const snapshot = enqueueMessage(state, conversationId, text, deps);
-			deps.logger?.debug("message-queue: enqueued", {
-				conversationId,
-				queueLen: snapshot.length,
-				textLen: text.length,
-			});
-			deps.notify();
-			return snapshot;
-		},
-		getQueue(conversationId) {
-			return readQueue(state, conversationId);
-		},
-		drain(conversationId) {
-			const drained = drainQueue(state, conversationId);
-			if (drained.length === 0) return drained;
-			deps.logger?.debug("message-queue: drained", {
-				conversationId,
-				count: drained.length,
-			});
-			deps.notify();
-			return drained;
-		},
-	};
+  return {
+    enqueue(conversationId, text) {
+      const snapshot = enqueueMessage(state, conversationId, text, deps);
+      deps.logger?.debug("message-queue: enqueued", {
+        conversationId,
+        queueLen: snapshot.length,
+        textLen: text.length,
+      });
+      deps.notify();
+      return snapshot;
+    },
+    getQueue(conversationId) {
+      return readQueue(state, conversationId);
+    },
+    drain(conversationId) {
+      const drained = drainQueue(state, conversationId);
+      if (drained.length === 0) return drained;
+      deps.logger?.debug("message-queue: drained", {
+        conversationId,
+        count: drained.length,
+      });
+      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;
+    },
+  };
 }