diff options
26 files changed, 1282 insertions, 7 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/cli/src/args.test.ts b/packages/cli/src/args.test.ts index e6b43cf..14c4ffc 100644 --- a/packages/cli/src/args.test.ts +++ b/packages/cli/src/args.test.ts @@ -216,6 +216,49 @@ describe("parseArgs", () => { expect(result.kind).toBe("error"); if (result.kind === "error") expect(result.message).toContain("--workspace requires a value"); }); + + it("parses --title flag", () => { + const result = parseArgs(["m", "--text", "x", "--title", "My Task"], { defaultServer }); + expect(result).toEqual({ + kind: "chat", + server: "http://localhost:24203", + modelName: "m", + text: "x", + file: undefined, + cwd: undefined, + conversationId: undefined, + reasoningEffort: undefined, + showReasoning: false, + open: false, + title: "My Task", + }); + }); + + it("parses --title with --workspace together", () => { + const result = parseArgs(["m", "--text", "x", "--workspace", "ws", "--title", "T"], { + defaultServer, + }); + expect(result.kind).toBe("chat"); + if (result.kind === "chat") { + expect(result.workspaceId).toBe("ws"); + expect(result.title).toBe("T"); + } + }); + + it("omits title when --title is not given", () => { + const result = parseArgs(["m", "--text", "x"], { defaultServer }); + expect(result.kind).toBe("chat"); + if (result.kind === "chat") { + expect(result.title).toBeUndefined(); + expect(result).not.toHaveProperty("title"); + } + }); + + it("errors when --title has no value", () => { + const result = parseArgs(["m", "--text", "x", "--title"], { defaultServer }); + expect(result.kind).toBe("error"); + if (result.kind === "error") expect(result.message).toContain("--title requires a value"); + }); }); describe("list", () => { diff --git a/packages/cli/src/args.ts b/packages/cli/src/args.ts index 52f1fba..581d4c2 100644 --- a/packages/cli/src/args.ts +++ b/packages/cli/src/args.ts @@ -27,6 +27,7 @@ export type ParsedCommand = readonly showReasoning: boolean; readonly open: boolean; readonly workspaceId?: string | undefined; + readonly title?: string | undefined; } | { readonly kind: "list"; @@ -307,6 +308,7 @@ export function parseArgs(argv: readonly string[], opts: ParseOpts): ParsedComma let open = false; let server = opts.defaultServer; let workspaceId: string | undefined; + let title: string | undefined; for (let i = 1; i < argv.length; i++) { const arg = argv[i] as string; @@ -338,6 +340,10 @@ export function parseArgs(argv: readonly string[], opts: ParseOpts): ParsedComma case "--open": open = true; break; + case "--title": + if (i + 1 >= argv.length) return { kind: "error", message: "--title requires a value" }; + title = argv[++i]; + break; case "--effort": if (i + 1 >= argv.length) return { @@ -383,5 +389,6 @@ export function parseArgs(argv: readonly string[], opts: ParseOpts): ParsedComma showReasoning, open, ...(workspaceId !== undefined && { workspaceId }), + ...(title !== undefined && { title }), }; } diff --git a/packages/cli/src/main.ts b/packages/cli/src/main.ts index 9fca347..c4fff1e 100644 --- a/packages/cli/src/main.ts +++ b/packages/cli/src/main.ts @@ -30,7 +30,7 @@ const USAGE = `Usage: dispatch read <conversationId> [--server <url>] dispatch open <conversationId> [--server <url>] dispatch send <conversationId> --text "..." [--file <path>] [--queue] [--open] [--cwd <dir>] [--effort <level>] [--workspace <id>] [--server <url>] - dispatch <modelName> --text "..." [--file <path>] [--cwd <dir>] [--conversation <id>] [--effort <level>] [--workspace <id>] [--server <url>] [--show-reasoning] [--open] + dispatch <modelName> --text "..." [--file <path>] [--cwd <dir>] [--conversation <id>] [--effort <level>] [--workspace <id>] [--title <title>] [--server <url>] [--show-reasoning] [--open] dispatch --help Effort levels: low, medium, high (default), xhigh, max`; diff --git a/packages/cli/src/message.test.ts b/packages/cli/src/message.test.ts index 536d64f..2deb197 100644 --- a/packages/cli/src/message.test.ts +++ b/packages/cli/src/message.test.ts @@ -111,6 +111,22 @@ describe("buildChatRequest", () => { ); expect(req).not.toHaveProperty("workspaceId"); }); + + it("includes title when provided", () => { + const req = buildChatRequest( + { modelName: "m", text: "x", title: "My Task", showReasoning: false }, + { cwd: "/work", message: "x" }, + ); + expect(req.title).toBe("My Task"); + }); + + it("omits title when not provided", () => { + const req = buildChatRequest( + { modelName: "m", text: "x", showReasoning: false }, + { cwd: "/work", message: "x" }, + ); + expect(req).not.toHaveProperty("title"); + }); }); describe("workspace flag → ChatRequest", () => { @@ -145,3 +161,26 @@ describe("workspace flag → ChatRequest", () => { expect(req.workspaceId).toBe("shorthand"); }); }); + +describe("title flag → ChatRequest", () => { + const defaultServer = "http://localhost:24203"; + + it("--title flag sets title on request", () => { + const parsed = parseArgs(["my-model", "--text", "hi", "--title", "My Task"], { + defaultServer, + }); + expect(parsed.kind).toBe("chat"); + if (parsed.kind !== "chat") return; + const req = buildChatRequest(parsed, { cwd: "/work", message: "hi" }); + expect(req.title).toBe("My Task"); + }); + + it("--title flag omitted sends no title", () => { + const parsed = parseArgs(["my-model", "--text", "hi"], { defaultServer }); + expect(parsed.kind).toBe("chat"); + if (parsed.kind !== "chat") return; + const req = buildChatRequest(parsed, { cwd: "/work", message: "hi" }); + expect(req.title).toBeUndefined(); + expect(req).not.toHaveProperty("title"); + }); +}); diff --git a/packages/cli/src/message.ts b/packages/cli/src/message.ts index ddbec6b..5db9966 100644 --- a/packages/cli/src/message.ts +++ b/packages/cli/src/message.ts @@ -39,6 +39,7 @@ interface ChatCmd { readonly conversationId?: string | undefined; readonly reasoningEffort?: ReasoningEffort | undefined; readonly workspaceId?: string | undefined; + readonly title?: string | undefined; readonly showReasoning: boolean; } @@ -55,5 +56,6 @@ export function buildChatRequest(cmd: ChatCmd, ctx: BuildCtx): ChatRequest { ...(cmd.cwd !== undefined ? { cwd: cmd.cwd } : { cwd: ctx.cwd }), ...(cmd.reasoningEffort !== undefined && { reasoningEffort: cmd.reasoningEffort }), ...(cmd.workspaceId !== undefined && { workspaceId: cmd.workspaceId }), + ...(cmd.title !== undefined && { title: cmd.title }), }; } 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 33f53c9..ffc5d58 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; /** @@ -1151,6 +1177,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 216359f..e5746c3 100644 --- a/packages/session-orchestrator/src/queue.test.ts +++ b/packages/session-orchestrator/src/queue.test.ts @@ -607,3 +607,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/contract.types.test.ts b/packages/transport-contract/src/contract.types.test.ts index 34ff544..3cc1b1e 100644 --- a/packages/transport-contract/src/contract.types.test.ts +++ b/packages/transport-contract/src/contract.types.test.ts @@ -68,6 +68,17 @@ const _chatWithHttpImage: ChatRequest = { images: [{ url: "https://example.com/diagram.png" }], }; +// ─── ChatRequest.title (additive optional) ─────────────────────────────────── + +const _chatWithTitle: ChatRequest = { + message: "implement the feature", + title: "Summon: add --title flag", +}; + +const _chatWithoutTitle: ChatRequest = { + message: "hello", +}; + // ─── Computer list / single response ───────────────────────────────────────── const _computer: Computer = { @@ -285,6 +296,16 @@ describe("transport-contract types compile and are exported", () => { expect(_chatWithHttpImage.images?.[0]?.mimeType).toBeUndefined(); }); + // ─── ChatRequest.title (additive optional) ──────────────────────────────── + + it("ChatRequest: title is additive optional (omittable)", () => { + expect(_chatWithoutTitle.title).toBeUndefined(); + }); + + it("ChatRequest: carries title when set", () => { + expect(_chatWithTitle.title).toBe("Summon: add --title flag"); + }); + it("ModelsResponse: ModelMetadata carries optional vision flag", () => { const resp: ModelsResponse = { models: ["umans/kimi-k2.7", "umans/glm-5.2"], diff --git a/packages/transport-contract/src/index.ts b/packages/transport-contract/src/index.ts index 797ad22..015b385 100644 --- a/packages/transport-contract/src/index.ts +++ b/packages/transport-contract/src/index.ts @@ -121,6 +121,23 @@ export interface ChatRequest { * defaultCwd = null). */ readonly workspaceId?: string; + + /** + * A human-readable title for the conversation tab — set at creation time + * (before the turn starts) via the conversation store's + * `setConversationTitle`, so the tab shows it immediately instead of the + * default derived from the first message (`"Untitled"` until the first + * append). Omit to keep the auto-derived title. When present, the value is + * trimmed server-side; a whitespace-only value is treated as absent + * (auto-derive). A non-string value → HTTP 400 `{ error }`. + * + * Backward compatible — clients that omit it are unaffected. Mirrors the + * dedicated `PUT /conversations/:id/title` endpoint but is atomic with + * conversation creation (no second round-trip), so the title is persisted + * before the turn's first message is appended (and thus before the tab is + * opened with `--open`). + */ + readonly title?: string; } /** @@ -468,6 +485,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 +726,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 +752,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..9b1480d 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 }; }, @@ -789,6 +799,154 @@ describe("POST /chat", () => { expect(cap.received?.modelName).toBeUndefined(); expect(cap.received?.cwd).toBeUndefined(); }); + + it("sets the conversation title from the request before the turn", async () => { + const calls: { conversationId: string; title: string }[] = []; + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle(conversationId, title) { + calls.push({ conversationId, title }); + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1", title: "My Task" }), + }); + + expect(res.status).toBe(200); + expect(calls).toEqual([{ conversationId: "conv1", title: "My Task" }]); + }); + + it("forwards a trimmed title to setConversationTitle", async () => { + const calls: { conversationId: string; title: string }[] = []; + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle(conversationId, title) { + calls.push({ conversationId, title }); + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1", title: " spaced " }), + }); + + expect(res.status).toBe(200); + expect(calls).toEqual([{ conversationId: "conv1", title: "spaced" }]); + }); + + it("does not call setConversationTitle when title is omitted", async () => { + let setTitleCalled = false; + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle() { + setTitleCalled = true; + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1" }), + }); + + expect(res.status).toBe(200); + expect(setTitleCalled).toBe(false); + }); + + it("does not call setConversationTitle for a whitespace-only title", async () => { + let setTitleCalled = false; + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle() { + setTitleCalled = true; + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1", title: " " }), + }); + + expect(res.status).toBe(200); + expect(setTitleCalled).toBe(false); + }); + + it("returns 400 when title is not a string", async () => { + let setTitleCalled = false; + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle() { + setTitleCalled = true; + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1", title: 42 }), + }); + + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toContain("title"); + expect(setTitleCalled).toBe(false); + }); + + it("proceeds with the turn even if setConversationTitle throws", async () => { + const store: ConversationStore = { + ...createFakeConversationStore(), + async setConversationTitle() { + throw new Error("store unavailable"); + }, + }; + const app = createApp({ + conversationStore: store, + orchestrator: createFakeOrchestrator([ + { type: "done", conversationId: "conv1", turnId: "t1", reason: "stop" }, + ]), + credentialStore: createFakeCredentialStore([]), + }); + + const res = await app.request("/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message: "hi", conversationId: "conv1", title: "My Task" }), + }); + + expect(res.status).toBe(200); + const text = await res.text(); + expect(text.trim().split("\n")).toHaveLength(1); + }); }); describe("POST /chat/warm", () => { @@ -2069,6 +2227,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..03b6ec2 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, @@ -456,6 +457,7 @@ export function createApp(opts: CreateServerOptions): Hono { reasoningEffort, workspaceId, images, + title, } = result; log.info("chat: request accepted", { conversationId, @@ -467,6 +469,22 @@ export function createApp(opts: CreateServerOptions): Hono { imageCount: images?.length ?? 0, }); + // Persist an explicit title BEFORE the turn starts so the tab shows it + // immediately (and before `--open` signals the frontend to open it). The + // store creates the conversation meta if none exists yet; a subsequent + // append preserves a non-"Untitled" title. A title-set failure is logged + // but never blocks the turn — the title is a nicety, the answer is not. + if (title !== undefined) { + try { + await opts.conversationStore.setConversationTitle(conversationId, title); + log.info("chat: title set", { conversationId }); + } catch (err) { + log.warn("chat: title set failure", { + error: err instanceof Error ? err.message : String(err), + }); + } + } + const events: AgentEvent[] = []; let controllerRef: ReadableStreamDefaultController<Uint8Array> | undefined; let streamClosed = false; @@ -796,6 +814,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-http/src/logic.test.ts b/packages/transport-http/src/logic.test.ts index 67632f3..271ee96 100644 --- a/packages/transport-http/src/logic.test.ts +++ b/packages/transport-http/src/logic.test.ts @@ -183,6 +183,56 @@ describe("parseChatBody", () => { } }); + // ── title ──────────────────────────────────────────────────────────────── + + it("extracts title when present", () => { + const result = parseChatBody({ message: "hi", title: "My Task" }, fakeId); + expect(isParseError(result)).toBe(false); + if (!isParseError(result)) { + expect(result.title).toBe("My Task"); + } + }); + + it("trims title whitespace", () => { + const result = parseChatBody({ message: "hi", title: " spaced title " }, fakeId); + expect(isParseError(result)).toBe(false); + if (!isParseError(result)) { + expect(result.title).toBe("spaced title"); + } + }); + + it("omits title when absent (backward compatible)", () => { + const result = parseChatBody({ message: "hi" }, fakeId); + expect(isParseError(result)).toBe(false); + if (!isParseError(result)) { + expect(result.title).toBeUndefined(); + } + }); + + it("omits title when whitespace-only (treated as absent)", () => { + const result = parseChatBody({ message: "hi", title: " " }, fakeId); + expect(isParseError(result)).toBe(false); + if (!isParseError(result)) { + expect(result.title).toBeUndefined(); + } + }); + + it("omits title when empty string (treated as absent)", () => { + const result = parseChatBody({ message: "hi", title: "" }, fakeId); + expect(isParseError(result)).toBe(false); + if (!isParseError(result)) { + expect(result.title).toBeUndefined(); + } + }); + + it("returns error when title is not a string", () => { + const result = parseChatBody({ message: "hi", title: 42 }, fakeId); + expect(isParseError(result)).toBe(true); + if (isParseError(result)) { + expect(result.error).toContain("title"); + } + }); + // ── images ────────────────────────────────────────────────────────────── it("parses images array with data URLs", () => { diff --git a/packages/transport-http/src/logic.ts b/packages/transport-http/src/logic.ts index c97f320..5cf96cc 100644 --- a/packages/transport-http/src/logic.ts +++ b/packages/transport-http/src/logic.ts @@ -56,6 +56,14 @@ export interface ChatCommand { readonly reasoningEffort?: ReasoningEffort; readonly workspaceId?: string; /** + * A human-readable title for the conversation tab, set at creation time. + * Parsed from the `ChatRequest.title` field; trimmed server-side. A + * whitespace-only value is treated as absent (omitted) so the auto-derived + * title applies. Forwarded to the `/chat` route which persists it via the + * conversation store's `setConversationTitle` before the turn starts. + */ + readonly title?: string; + /** * Images attached to this turn (data URLs or http URLs). Parsed from the * `ChatRequest.images` field; forwarded to the orchestrator which converts * them to `image` chunks on the user message. Each entry must have a non-empty @@ -128,6 +136,18 @@ export function parseChatBody(body: unknown, generateId: () => string): ParseRes (result as { workspaceId?: string }).workspaceId = obj.workspaceId; } + if (obj.title !== undefined) { + if (typeof obj.title !== "string") { + return { error: "Field 'title' must be a string" }; + } + const title = obj.title.trim(); + // A whitespace-only title is treated as absent so the auto-derived title + // applies (mirrors omitting the field) — never persist an empty title. + if (title.length > 0) { + (result as { title?: string }).title = title; + } + } + if (obj.images !== undefined) { if (!Array.isArray(obj.images)) { return { error: "Field 'images' must be an array" }; 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( diff --git a/packages/vision-handoff/src/service.ts b/packages/vision-handoff/src/service.ts index 01245df..397d81a 100644 --- a/packages/vision-handoff/src/service.ts +++ b/packages/vision-handoff/src/service.ts @@ -354,7 +354,9 @@ export function createVisionHandoffService(deps: VisionHandoffDeps): VisionHando }; const stream = vision.provider.stream([userMessage], [], { model: vision.model, - systemPrompt: "You are a vision assistant. Describe images faithfully and thoroughly.", + systemPrompt: + "You are a vision assistant. Describe images faithfully and thoroughly. " + + "Do not use any tools — just use your vision to see the image and describe it directly.", }); const description = (await collectTextFromStream(stream)).trim(); const text = @@ -657,7 +659,10 @@ export function createVisionHandoffService(deps: VisionHandoffDeps): VisionHando modelName: vision.modelName, ...(opts.cwd !== undefined ? { cwd: opts.cwd } : {}), systemPrompt: - "You are a vision assistant. A developer who cannot see images is asking you specific questions about an image they attached. Answer their question precisely and thoroughly.", + "You are a vision assistant. A developer who cannot see images is asking you specific " + + "questions about an image they attached. Answer their question precisely and thoroughly. " + + "Do not use any tools unless specifically asked to — just use your vision to see the " + + "image and describe it directly.", onEvent: (event: AgentEvent) => { if (event.type === "text-delta") { responseText += event.delta; |