diff options
| author | Adam Malczewski <[email protected]> | 2026-06-28 21:58:29 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-28 21:58:29 +0900 |
| commit | 01928fe70d22cb7b0a8d20f38b359c8a0048c34b (patch) | |
| tree | e0ed7cd3ca0b35cf533fb6a7e084b9de7063fc41 /packages/message-queue/src/service.ts | |
| parent | 6dd9ea9b935e5011c16faed6c869c976cf5ff172 (diff) | |
| download | dispatch-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/service.ts')
| -rw-r--r-- | packages/message-queue/src/service.ts | 33 |
1 files changed, 32 insertions, 1 deletions
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; + }, }; } |
