diff options
| -rw-r--r-- | frontend-cancel-queued-message-handoff.md | 140 | ||||
| -rw-r--r-- | packages/message-queue/src/index.ts | 1 | ||||
| -rw-r--r-- | packages/message-queue/src/pure.test.ts | 79 | ||||
| -rw-r--r-- | packages/message-queue/src/pure.ts | 36 | ||||
| -rw-r--r-- | packages/message-queue/src/service.test.ts | 69 | ||||
| -rw-r--r-- | packages/message-queue/src/service.ts | 33 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/index.ts | 1 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.ts | 39 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/queue.test.ts | 180 | ||||
| -rw-r--r-- | packages/transport-contract/package.json | 2 | ||||
| -rw-r--r-- | packages/transport-contract/src/index.ts | 42 | ||||
| -rw-r--r-- | packages/transport-http/src/app.test.ts | 146 | ||||
| -rw-r--r-- | packages/transport-http/src/app.ts | 23 | ||||
| -rw-r--r-- | packages/transport-http/src/extension.ts | 1 | ||||
| -rw-r--r-- | packages/transport-ws/src/extension.ts | 17 | ||||
| -rw-r--r-- | packages/transport-ws/src/router.test.ts | 55 | ||||
| -rw-r--r-- | packages/transport-ws/src/router.ts | 50 |
17 files changed, 910 insertions, 4 deletions
diff --git a/frontend-cancel-queued-message-handoff.md b/frontend-cancel-queued-message-handoff.md new file mode 100644 index 0000000..b92abd4 --- /dev/null +++ b/frontend-cancel-queued-message-handoff.md @@ -0,0 +1,140 @@ +# FE handoff — cancel a queued message + +Courier this to `../frontend` (cross-repo contract change; `lsp references` does +not span repos — ORCHESTRATOR §7). All changes are ADDITIVE — nothing existing +breaks. + +## What shipped (backend) + +A per-message **cancel** for the steering message queue: while a turn is +GENERATING and a user message is sitting in the queue (waiting to be delivered +as steering at the next tool-result boundary, or carried into a new turn), the +client can **cancel a single queued message by id** so it never runs. The +message is removed from the queue and is never delivered as steering and never +carried into a new turn. + +This complements the existing `chat.queue` enqueue (see +`frontend-message-queue-handoff.md`). Enqueue adds; cancel removes one. + +Versions: `@dispatch/transport-contract` `0.23.0 → 0.24.0`. Bump the pinned +`file:` dep. (`@dispatch/wire` is unchanged — `QueuedMessage` already has the +`id` the cancel targets; no new wire type was needed.) + +## The two entry points (pick one — same backend behavior) + +### 1. WebSocket op: `chat.queue.cancel` (what the FE should use) + +```ts +interface ChatQueueCancelMessage { + readonly type: "chat.queue.cancel"; + readonly conversationId: string; + readonly messageId: string; // the stable QueuedMessage.id (from the queue surface / enqueue response) +} +``` +(additive to `WsClientMessage`.) + +- **Fire-and-forget**, exactly like `chat.queue`. On success the server emits + NOTHING back — the `message-queue` SURFACE updates (the cancelled message + leaves the `payload.messages` snapshot). On failure (missing/empty + `conversationId` or `messageId`) the server replies `chat.error` + (`{ type: "chat.error"; conversationId?; message }`). +- **Idempotent:** cancelling a message that is no longer queued (already + drained/delivered as steering, or already cancelled, or never existed, or an + unknown conversation) is a **silent no-op** — no surface update, no error. So + a client may optimistically remove the row from its queue UI on click and + fire-and-forget the cancel; if the message was already gone, nothing breaks. +- **`messageId`** is the stable client-visible `QueuedMessage.id` you already + render from the queue surface snapshot (or got back from `chat.queue` / + `POST /conversations/:id/queue`'s `queue[]`). + +### 2. HTTP path (for the CLI / non-WS clients; the FE uses the WS op above) + +`DELETE /conversations/:id/queue/:messageId` (no request body) → `QueueCancelResponse`: + +```ts +interface QueueCancelResponse { + readonly conversationId: string; + readonly cancelled: boolean; // true = a message was found + removed + readonly queue: readonly QueuedMessage[]; // post-cancel snapshot +} +``` +- `cancelled: true` — the message was in the queue and has been removed (it will + never run). +- `cancelled: false` — the message was NOT in the queue (already + drained/delivered, never existed, unknown conversation) OR the message-queue + extension isn't loaded (degraded). Still HTTP **200** (idempotent — not an + error). +- `queue` is the post-cancel snapshot (empty when no queue extension is loaded). + +## How the FE confirms a cancel + +The queue is control/state on the **surface** channel (NOT the chat stream), so +cancel is confirmed the same way enqueue is — by the `message-queue` surface +updating: + +1. You already **subscribe to the `message-queue` surface** (scope + `conversation`) and render `payload.messages` (`QueuedMessage[]`) with the + `rendererId: "message-queue"` custom renderer (per + `frontend-message-queue-handoff.md`). +2. On cancel, the surface pushes a **full new spec** whose `payload.messages` + no longer contains the cancelled id. The cancelled message simply leaves the + queue list — render the new snapshot. +3. **No new `AgentEvent`** is emitted for a cancel. The cancelled message never + appears in the transcript (it was never delivered as steering — that's the + point). If a message was already drained (delivered as a `steering` bubble or + carried into a new turn's `user-message`) before the cancel arrived, the + cancel is a no-op (`cancelled: false`) and the transcript is unchanged. + +## UX suggestion + +- Render a **cancel (×) affordance** on each pending row in the queue UI (the + surface snapshot gives you the `id` to target). On click → send + `chat.queue.cancel { conversationId, messageId }`. +- Optimistically remove the row from the queue UI on click; the surface update + will confirm it (or, if the message was already drained, the surface already + shows it gone — no harm). +- Expect a `chat.error` only for a malformed send (empty `conversationId` / + `messageId`) — in practice a client that sends the id it just rendered will + never hit this. + +## Race notes (safe by construction) + +- **Cancel vs. drain (steering delivery):** if the kernel drains the queue at a + tool-result boundary in the instant between the user clicking cancel and the + server processing it, the message is already gone — the cancel returns + `cancelled: false` (a no-op). The drained message was delivered as a + `steering` event and is in the transcript; the cancel correctly did nothing. + No double-delivery, no error. +- **Cancel vs. post-seal carry:** same — if the turn sealed and the queue was + carried into a new turn before the cancel ran, the message is already the new + turn's opening `user-message`; the cancel is a no-op. +- **Cancel is scoped per conversation:** cancelling on conversation A never + touches conversation B's queue. + +## What we need the FE to do + +1. **Bump pinned dep:** `@dispatch/transport-contract` → `0.24.0`. +2. **Add `chat.queue.cancel`** to the FE's `WsClientMessage` union (it is + additive — no exhaustive switch breaks; if the FE has one, add the + `chat.queue.cancel` case to its WS dispatcher). +3. **Cancel affordance per queued row:** a × / cancel button on each pending + message in the queue UI that sends + `chat.queue.cancel { conversationId, messageId }` using the row's `id`. + Optimistically remove the row; reconcile from the surface update. +4. **No new event handling** — the existing `message-queue` surface subscription + already reflects the post-cancel snapshot; just render it. No `steering` / + transcript change for a cancelled message (it never runs). + +## Notes / known gaps + +- **No CLI command.** The CLI's `send --queue` enqueues but there is no CLI + command to list the queue or obtain a `messageId`, so a CLI `cancel` was not + added (it would have no way to discover a message id). The HTTP + `DELETE /conversations/:id/queue/:messageId` is available for any non-WS + client that already knows the id (e.g. from a prior + `POST /conversations/:id/queue` response's `queue[]`). +- **Close-with-queued-messages** (the open product question noted in + `frontend-message-queue-handoff.md`) is unchanged by this feature: an + explicit `POST /conversations/:id/close` still aborts the in-flight turn and + the carry still fires. Cancel is a separate, per-message affordance that does + not touch the turn. 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( |
