summaryrefslogtreecommitdiffhomepage
path: root/packages/message-queue/src/pure.test.ts
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-28 21:58:29 +0900
committerAdam Malczewski <[email protected]>2026-06-28 21:58:29 +0900
commit01928fe70d22cb7b0a8d20f38b359c8a0048c34b (patch)
treee0ed7cd3ca0b35cf533fb6a7e084b9de7063fc41 /packages/message-queue/src/pure.test.ts
parent6dd9ea9b935e5011c16faed6c869c976cf5ff172 (diff)
downloaddispatch-01928fe70d22cb7b0a8d20f38b359c8a0048c34b.tar.gz
dispatch-01928fe70d22cb7b0a8d20f38b359c8a0048c34b.zip
feat(message-queue): cancel a queued steering message by id
Add the ability to cancel/close a single queued message so it never runs: while a turn is GENERATING and a user message sits in the steering queue (waiting for delivery at the next tool-result boundary or carry into a new turn), a client can remove one message by id. The cancelled message is never delivered as steering and never carried into a new turn. Complements the existing chat.queue enqueue (enqueue adds; cancel removes one). Layers (all additive; nothing existing breaks): - message-queue pure core: `cancel(state, conversationId, messageId)` — splices a single message out by id, returns the post-cancel snapshot. Idempotent (no-op if not found). Drops the key when the queue empties (mirrors drain). - message-queue service: `MessageQueueService.cancel()` — wraps the pure op, pushes a surface update ONLY on a real change (queue shrank); a missing-id cancel is a no-op with no surface push (mirrors drain's no-notify-on-empty). - session-orchestrator: `cancelQueuedMessage({ conversationId, messageId })` -> `{ cancelled, queue }`. The single entry transports call; resolves the queue lazily (same as enqueue). Degrades to `{ cancelled: false, queue: [] }` when the message-queue extension isn't loaded. `cancelled` is derived from the queue length delta (true iff a message was removed). API contract (documented for the frontend agent; see frontend-cancel-queued-message-handoff.md): HTTP: DELETE /conversations/:id/queue/:messageId -> 200 QueueCancelResponse { conversationId, cancelled, queue } (cancelled:false is a 200 idempotent no-op, not an error) WS: chat.queue.cancel { type:"chat.queue.cancel", conversationId, messageId } (additive to WsClientMessage). Fire-and-forget like chat.queue: success is confirmed by the message-queue SURFACE updating (the cancelled message leaves payload.messages). A missing-id cancel is a silent no-op (no surface update, no error). Malformed (empty conversationId/ messageId) -> chat.error. No new AgentEvent: a cancelled message never appears in the transcript (it never runs). The existing message-queue surface already reflects the post-cancel snapshot. Race-safe by construction: if the kernel drains the queue (steering) or carries it into a new turn before the cancel runs, the message is already gone -> cancel returns cancelled:false (a no-op). Version bump: @dispatch/transport-contract 0.23.0 -> 0.24.0 (additive: QueueCancelResponse + ChatQueueCancelMessage added to WsClientMessage). @dispatch/wire unchanged (QueuedMessage.id is the cancel target). No CLI command added: the CLI has no queue-listing affordance to discover a messageId, so a CLI cancel would have no input source. The HTTP DELETE is available for any non-WS client that knows the id (e.g. from a prior enqueue response's queue[]). Verification: tsc -b EXIT 0; vitest 2024 passed / 6 skipped (25 new tests); biome EXIT 0 (0 errors).
Diffstat (limited to 'packages/message-queue/src/pure.test.ts')
-rw-r--r--packages/message-queue/src/pure.test.ts79
1 files changed, 79 insertions, 0 deletions
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[] = [