summaryrefslogtreecommitdiffhomepage
path: root/packages/transport-http/src
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/transport-http/src
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/transport-http/src')
-rw-r--r--packages/transport-http/src/app.test.ts146
-rw-r--r--packages/transport-http/src/app.ts23
-rw-r--r--packages/transport-http/src/extension.ts1
3 files changed, 170 insertions, 0 deletions
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",