diff options
49 files changed, 4013 insertions, 263 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/notes/heartbeat-setting-handoff.md b/notes/heartbeat-setting-handoff.md new file mode 100644 index 0000000..53382aa --- /dev/null +++ b/notes/heartbeat-setting-handoff.md @@ -0,0 +1,130 @@ +# FE handoff — heartbeat `inactiveOnly` (skip while workspace is busy) + +Courier this to `../frontend` (cross-repo contract change; `lsp references` does not span +repos — ORCHESTRATOR §7). All changes are ADDITIVE — nothing existing breaks. The new +field has a default, so a frontend that never sends it gets the new behavior automatically. + +## What shipped (backend) + +A new per-workspace heartbeat setting, **`inactiveOnly`**: when on (the default), the +heartbeat SKIPS a fire whenever the configured workspace has any active agents — i.e. a +conversation whose persisted status is `"active"` (driving a turn) or `"queued"` (waiting on +the message queue). The heartbeat stays quiet while the user is actively working and only +fires when the workspace is idle. + +The spawned heartbeat conversation lives in the DEDICATED heartbeat workspace (not the +configured one), so a heartbeat run never counts as an "active agent" of the configured +workspace — there is no self-block. + +## The setting + +```ts +// @dispatch/transport-contract (HeartbeatConfig) +interface HeartbeatConfig { + enabled: boolean; + inactiveOnly: boolean; // ← NEW. default: true + systemPrompt: string; + taskPrompt: string; + intervalMinutes: number; + model: string; + reasoningEffort: ReasoningEffort | null; +} +``` + +- **Name:** `inactiveOnly` +- **Type:** `boolean` +- **Default:** `true` (the heartbeat is quiet-by-default while the workspace is busy) +- **Semantics:** + - `true` → before each scheduled fire, the backend checks the configured workspace for + active agents. If any exist, the fire is **silently skipped** (no run is recorded, no + `HeartbeatRun` created). The scheduler re-arms and tries again at the next + `intervalMinutes` interval. + - `false` → the heartbeat fires unconditionally on every interval, regardless of + workspace activity (the pre-existing behavior). +- **What counts as an "active agent":** any conversation in the configured workspace whose + persisted `ConversationStatus` is `"active"` or `"queued"`. The orchestrator sets + `"active"` when a turn starts and `"idle"` when it settles, so this is the live + busy/idle signal. +- **Self-block safety:** heartbeat-spawned conversations are filed in the `heartbeat` + workspace, so an in-flight heartbeat run does NOT keep the configured workspace "busy". + +## API endpoints + +The setting rides on the existing heartbeat config endpoints — no new routes. + +### Read + +`GET /workspaces/:id/heartbeat` → `HeartbeatConfig` (200) + +The response now includes `inactiveOnly`. For a workspace that never configured a +heartbeat, the full default config is returned (`enabled: false`, `inactiveOnly: true`, +`intervalMinutes: 30`, …). + +### Update + +`PUT /workspaces/:id/heartbeat` with a partial `UpdateHeartbeatRequest` body → `HeartbeatConfig` (200) + +```ts +// @dispatch/transport-contract (UpdateHeartbeatRequest) +interface UpdateHeartbeatRequest { + enabled?: boolean; + inactiveOnly?: boolean; // ← NEW. optional; absent = unchanged + systemPrompt?: string; + taskPrompt?: string; + intervalMinutes?: number; + model?: string; + reasoningEffort?: ReasoningEffort | null; +} +``` + +All fields optional (a partial update); only provided fields are applied. Absent +`inactiveOnly` leaves it unchanged (distinct from sending `false`). + +**Validation:** `inactiveOnly` must be a JSON `boolean` when present. A non-boolean +(e.g. `"yes"`, `1`) → HTTP 400 `{ error: "Field 'inactiveOnly' must be a boolean" }`. +The service is NOT called on validation failure. + +**Example:** + +```http +PUT /workspaces/proj/heartbeat +Content-Type: application/json + +{ "inactiveOnly": false } +``` + +```json +200 OK +{ + "enabled": false, + "inactiveOnly": false, + "systemPrompt": "", + "taskPrompt": "", + "intervalMinutes": 30, + "model": "", + "reasoningEffort": null +} +``` + +## Frontend implementation notes + +- Render a **checkbox** (default checked) in the workspace heartbeat settings UI, labelled + something like "Only run when the workspace is idle" / "Pause while agents are active". +- The checkbox reflects and edits `inactiveOnly` on the heartbeat config. +- On toggle, send `PUT /workspaces/:id/heartbeat` with `{ inactiveOnly: <bool> }` (a partial + update — no need to send the rest of the config). +- A skipped fire produces **no `HeartbeatRun`**, so the runs list (`GET + /workspaces/:id/heartbeat/runs`) and the next-run countdown (`GET + /workspaces/:id/heartbeat/next-run`) behave exactly as on any other idle interval: the + next-run timestamp advances to `now + intervalMinutes`. No new event type is emitted on a + skip (it is silent); the backend logs it at `info` level only. +- Legacy workspaces: a config persisted by an older backend (no `inactiveOnly` field) is + read back as `inactiveOnly: true` (the default) — the feature is ON by default for + everyone, including pre-existing workspaces. + +## Versions / deps + +The new field is added to `HeartbeatConfig` / `UpdateHeartbeatRequest` in +`@dispatch/transport-contract`. It is a purely additive field with a default, so no +version bump of the pinned `file:` dependency is strictly required to keep working — but +bump it when convenient to pick up the typed field. 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/heartbeat/src/config-store.test.ts b/packages/heartbeat/src/config-store.test.ts index e77d10a..9a6772d 100644 --- a/packages/heartbeat/src/config-store.test.ts +++ b/packages/heartbeat/src/config-store.test.ts @@ -77,6 +77,21 @@ describe("applyConfigUpdate (pure)", () => { const untouched = applyConfigUpdate(withEffort, { enabled: true }); expect(untouched.reasoningEffort).toBe("low"); }); + + it("defaults inactiveOnly to true (the heartbeat is quiet by default while the workspace is busy)", () => { + expect(DEFAULT_HEARTBEAT_CONFIG.inactiveOnly).toBe(true); + }); + + it("applies an inactiveOnly update", () => { + const next = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { inactiveOnly: false }); + expect(next.inactiveOnly).toBe(false); + }); + + it("leaves inactiveOnly unchanged when absent from the update", () => { + const off = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { inactiveOnly: false }); + const untouched = applyConfigUpdate(off, { enabled: true }); + expect(untouched.inactiveOnly).toBe(false); + }); }); describe("createHeartbeatConfigStore", () => { @@ -106,6 +121,28 @@ describe("createHeartbeatConfigStore", () => { expect(next.taskPrompt).toBe("c"); }); + it("round-trips inactiveOnly through persistence", async () => { + const storage = createMemoryStorage(); + const store = createHeartbeatConfigStore(storage); + const next = await store.update("ws-1", { inactiveOnly: false }); + expect(next.inactiveOnly).toBe(false); + const store2 = createHeartbeatConfigStore(storage); + expect((await store2.get("ws-1")).inactiveOnly).toBe(false); + }); + + it("defaults inactiveOnly to true for legacy configs persisted before the field existed", async () => { + const storage = createMemoryStorage(); + // Simulate a config written by an older build that had no inactiveOnly + // field (a pre-inactive-only heartbeat config). + await storage.set("config:ws-1", JSON.stringify({ enabled: true, intervalMinutes: 5 })); + const store = createHeartbeatConfigStore(storage); + const config = await store.get("ws-1"); + expect(config.enabled).toBe(true); + expect(config.intervalMinutes).toBe(5); + // The missing field defaults ON (feature on by default) — never `undefined`. + expect(config.inactiveOnly).toBe(true); + }); + it("lists persisted workspace ids", async () => { const store = createHeartbeatConfigStore(createMemoryStorage()); await store.update("ws-b", { enabled: true }); diff --git a/packages/heartbeat/src/config-store.ts b/packages/heartbeat/src/config-store.ts index f06c9e8..52aca3e 100644 --- a/packages/heartbeat/src/config-store.ts +++ b/packages/heartbeat/src/config-store.ts @@ -7,6 +7,7 @@ import type { HeartbeatConfig, UpdateHeartbeatRequest } from "@dispatch/transpor */ export const DEFAULT_HEARTBEAT_CONFIG: HeartbeatConfig = { enabled: false, + inactiveOnly: true, systemPrompt: "", taskPrompt: "", intervalMinutes: 30, @@ -38,6 +39,7 @@ export function applyConfigUpdate( ): HeartbeatConfig { const next: HeartbeatConfig = { enabled: update.enabled !== undefined ? update.enabled : current.enabled, + inactiveOnly: update.inactiveOnly !== undefined ? update.inactiveOnly : current.inactiveOnly, systemPrompt: update.systemPrompt !== undefined ? update.systemPrompt : current.systemPrompt, taskPrompt: update.taskPrompt !== undefined ? update.taskPrompt : current.taskPrompt, intervalMinutes: @@ -72,6 +74,7 @@ export function createHeartbeatConfigStore(storage: StorageNamespace): Heartbeat const parsed = JSON.parse(raw) as Partial<HeartbeatConfig>; return { enabled: typeof parsed.enabled === "boolean" ? parsed.enabled : false, + inactiveOnly: typeof parsed.inactiveOnly === "boolean" ? parsed.inactiveOnly : true, systemPrompt: typeof parsed.systemPrompt === "string" ? parsed.systemPrompt : "", taskPrompt: typeof parsed.taskPrompt === "string" ? parsed.taskPrompt : "", intervalMinutes: diff --git a/packages/heartbeat/src/extension.ts b/packages/heartbeat/src/extension.ts index f245b7e..b044619 100644 --- a/packages/heartbeat/src/extension.ts +++ b/packages/heartbeat/src/extension.ts @@ -90,6 +90,19 @@ export const extension: Extension = { // Lazy (resolved at fire time, mirroring resolvePrompt). getWorkspaceCwd: async (wsId) => (await conversationStore.getWorkspace(wsId))?.defaultCwd ?? null, + // inactiveOnly: the configured workspace is "busy" while any of its + // conversations are driving or queued for a turn. The orchestrator + // sets persisted status "active" on turn start and "idle" on settle, + // so a single store read (filtered to the configured workspaceId) is + // the live active-agent check. The heartbeat's own spawned conversation + // lives in the DEDICATED heartbeat workspace, so it never self-blocks. + hasActiveAgents: async (wsId) => { + const active = await conversationStore.listConversations({ + workspaceId: wsId, + status: ["active", "queued"], + }); + return active.length > 0; + }, }); // Reconcile stale runs + arm enabled workspaces on boot. diff --git a/packages/heartbeat/src/heartbeat.test.ts b/packages/heartbeat/src/heartbeat.test.ts index ff1ec10..a6f7ebb 100644 --- a/packages/heartbeat/src/heartbeat.test.ts +++ b/packages/heartbeat/src/heartbeat.test.ts @@ -157,6 +157,7 @@ function createService(opts: { ) => Promise<string>; readonly getGlobalSystemPrompt?: () => Promise<string>; readonly getWorkspaceCwd?: (workspaceId: string) => Promise<string | null>; + readonly hasActiveAgents?: (workspaceId: string) => Promise<boolean>; }) { const fake = createFakeTimers(); let id = 0; @@ -171,6 +172,7 @@ function createService(opts: { ? { getGlobalSystemPrompt: opts.getGlobalSystemPrompt } : {}), ...(opts.getWorkspaceCwd !== undefined ? { getWorkspaceCwd: opts.getWorkspaceCwd } : {}), + ...(opts.hasActiveAgents !== undefined ? { hasActiveAgents: opts.hasActiveAgents } : {}), }); return { svc, advance: fake.advance, storage }; } @@ -181,6 +183,131 @@ describe("createHeartbeatService", () => { const cfg = await svc.getConfig("ws-1"); expect(cfg.enabled).toBe(false); expect(cfg.intervalMinutes).toBe(30); + // inactiveOnly defaults ON (the heartbeat is quiet by default while the + // workspace is busy). + expect(cfg.inactiveOnly).toBe(true); + }); + + describe("inactiveOnly (skip fire while the workspace has active agents)", () => { + it("skips the fire (records no run) when inactiveOnly is true and the workspace has active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(true), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // interval elapses → fire + await flush(); // let the async fire() reach the active-agent check + return + + expect(orch.pending).toHaveLength(0); // no turn started + expect(await svc.listRuns("ws-1")).toHaveLength(0); // no run recorded + }); + + it("still fires when inactiveOnly is true but the workspace has NO active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(false), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("fires unconditionally when inactiveOnly is false, even with active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + // Active agents reported, but the setting is OFF → must not block. + hasActiveAgents: () => Promise.resolve(true), + }); + await svc.updateConfig("ws-1", { + enabled: true, + inactiveOnly: false, + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("re-arms and fires on the next interval after a skipped fire (the scheduler keeps ticking)", async () => { + const orch = createFakeOrchestrator(); + let busy = true; // workspace busy on the first fire, free on the next + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(busy), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // first fire — skipped (busy) + await flush(); + expect(orch.pending).toHaveLength(0); + + busy = false; // workspace goes idle + advance(60_000); // next interval → fire again + await flush(); + expect(orch.pending).toHaveLength(1); // now it fires + orch.pending[0]!.resolve(); + await flush(); + }); + + it("does not consult hasActiveAgents when inactiveOnly is false (degrades off cleanly)", async () => { + const orch = createFakeOrchestrator(); + let consulted = false; + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => { + consulted = true; + return Promise.resolve(true); + }, + }); + await svc.updateConfig("ws-1", { + enabled: true, + inactiveOnly: false, + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(consulted).toBe(false); // never asked — setting is off + expect(orch.pending).toHaveLength(1); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("checks active agents per configured workspace (only the configured workspace is consulted)", async () => { + const orch = createFakeOrchestrator(); + const busyWorkspaces = new Set<string>(["ws-busy"]); + const { svc, advance } = createService({ + orch, + hasActiveAgents: (wsId) => Promise.resolve(busyWorkspaces.has(wsId)), + }); + // ws-free is idle, ws-busy has active agents. + await svc.updateConfig("ws-free", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + await svc.updateConfig("ws-busy", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // both fire + await flush(); + // ws-free fired (idle), ws-busy skipped (busy). + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-free"))[0]?.status).toBe("running"); + expect(await svc.listRuns("ws-busy")).toHaveLength(0); + orch.pending[0]!.resolve(); + await flush(); + }); }); it("arming an enabled config does not fire immediately (waits for the interval)", () => { diff --git a/packages/heartbeat/src/heartbeat.ts b/packages/heartbeat/src/heartbeat.ts index 12e70e3..e04c538 100644 --- a/packages/heartbeat/src/heartbeat.ts +++ b/packages/heartbeat/src/heartbeat.ts @@ -116,6 +116,19 @@ export interface HeartbeatServiceDeps { * (against `conversationStore.getWorkspace`); tests inject a fake. */ readonly getWorkspaceCwd?: (workspaceId: string) => Promise<string | null>; + /** + * Whether the configured workspace currently has any ACTIVE agents — + * conversations driving (or queued for) a turn. When `config.inactiveOnly` + * is `true`, the heartbeat SKIPS a fire while this returns `true` (the + * workspace is busy). The extension wires the real check against + * `conversationStore.listConversations({ workspaceId, status: ["active", + * "queued"] })` (the configured workspace's persisted statuses — the + * orchestrator sets `"active"` on turn start, `"idle"` on settle); tests + * inject a fake. When omitted, `false` (no active agents → never skip) so + * the inactive-only feature degrades off cleanly — the heartbeat fires + * unconditionally, matching the pre-inactive-only behavior. + */ + readonly hasActiveAgents?: (workspaceId: string) => Promise<boolean>; } interface ActiveRun { @@ -143,6 +156,10 @@ export function createHeartbeatService(deps: HeartbeatServiceDeps): HeartbeatSer // conversation's workspace). The extension wires the real getter so the // turn pins to the CONFIGURED workspace's defaultCwd; tests inject a fake. const getWorkspaceCwd = deps.getWorkspaceCwd ?? (() => Promise.resolve(null)); + // Default: no active agents (never skip) — the inactive-only feature degrades + // off cleanly. The extension wires the real check (against the conversation + // store's persisted statuses); tests inject a fake. + const hasActiveAgents = deps.hasActiveAgents ?? (() => Promise.resolve(false)); // runId → active-run tracking (in-memory; the durable record lives in the // run store). Used to (a) map a stop request to its conversation, and @@ -160,6 +177,18 @@ export function createHeartbeatService(deps: HeartbeatServiceDeps): HeartbeatSer // Race: disabled/disarmed between the timer firing and now. if (!config.enabled) return; + // inactiveOnly: skip this fire when the configured workspace has active + // agents (a conversation driving or queued for a turn). The fire is + // silently skipped — no run is recorded — and the scheduler re-arms to + // try again at the next interval. The spawned heartbeat conversation lives + // in the DEDICATED heartbeat workspace, so it never self-blocks (a prior + // in-flight heartbeat run is NOT an active agent of the configured + // workspace). Disabled (inactiveOnly === false) fires unconditionally. + if (config.inactiveOnly && (await hasActiveAgents(workspaceId))) { + logger?.info("heartbeat: fire skipped — workspace has active agents", { workspaceId }); + return; + } + const conversationId = generateId(); const runId = generateId(); const triggeredAt = new Date(timers.now()).toISOString(); diff --git a/packages/kernel/src/contracts/runtime.ts b/packages/kernel/src/contracts/runtime.ts index 71d2211..deae126 100644 --- a/packages/kernel/src/contracts/runtime.ts +++ b/packages/kernel/src/contracts/runtime.ts @@ -121,13 +121,19 @@ export interface RunTurnInput { * results. When omitted or returning an empty array, no injection happens * (the runtime is unchanged). * - * Injected (not ambient) so the kernel stays pure: it owns no queue and - * names no feature — it just calls the callback and appends what it gets. - * Only invoked when a step PRODUCED tool calls (the tool-result boundary); - * a step that ends without tool calls does not drain (the caller decides - * what to do with any pending messages after the turn ends). + * May return a Promise (the runtime `await`s it): the shell uses this to + * PERSIST the injected messages to the store as part of the same critical + * section as the injection, so they are never lost (a fire-and-forget + * persist would race with the next step's `onStepComplete` append and + * collide on the store's seq counter). A sync return is still supported + * (backward-compatible). Injected (not ambient) so the kernel stays pure: + * it owns no queue and names no feature — it just calls the callback, + * awaits it, and appends what it gets. Only invoked when a step PRODUCED + * tool calls (the tool-result boundary); a step that ends without tool + * calls does not drain (the caller decides what to do with any pending + * messages after the turn ends). */ - readonly drainSteering?: () => readonly ChatMessage[]; + readonly drainSteering?: () => readonly ChatMessage[] | Promise<readonly ChatMessage[]>; /** * Optional. Called by the runtime after each step's messages are finalized @@ -142,6 +148,33 @@ export interface RunTurnInput { readonly onStepComplete?: (messages: readonly ChatMessage[]) => Promise<void> | void; /** + * Optional. Called by the runtime at each tool-result boundary — after a + * step that produced tool calls has been finalized (`onStepComplete` has run + * and any `drainSteering` messages have been appended), before the next step + * begins — giving the caller a chance to REPLACE the running message + * history. The caller returns a new message list which the runtime adopts + * as its working history for all subsequent steps, or `undefined`/an empty + * array to keep the history unchanged. + * + * The runtime receives the step's token `usage` (so the caller can check a + * threshold against the model's context window) and the current `messages` + * (the full prompt the next step would otherwise see). Generic and + * feature-agnostic: the runtime calls it and adopts whatever message list it + * returns — it names no feature, owns no threshold policy, and performs no + * I/O, mirroring `drainSteering`. The shell uses this to compact the history + * in-flight when the context window nears capacity, so a long-running turn + * (e.g. left overnight) does not run out of context mid-turn: it summarizes + * the old history and continues with the summary + recent messages. Only + * invoked when a step PRODUCED tool calls (there is a "next step" to compact + * before); a step that ends without tool calls ends the turn, so there is no + * boundary to compact at. Injected (not ambient) so the kernel stays pure. + */ + readonly onStepBoundary?: (ctx: { + readonly stepUsage: Usage; + readonly messages: readonly ChatMessage[]; + }) => Promise<readonly ChatMessage[] | undefined> | (readonly ChatMessage[] | undefined); + + /** * Optional injected retry strategy for retryable provider errors (e.g. HTTP * 429 / 5xx "overloaded"). When omitted, a retryable error ends the step * exactly as before (backward-compatible). When provided, the runtime wraps diff --git a/packages/kernel/src/runtime/run-turn.test.ts b/packages/kernel/src/runtime/run-turn.test.ts index 8d20975..452e162 100644 --- a/packages/kernel/src/runtime/run-turn.test.ts +++ b/packages/kernel/src/runtime/run-turn.test.ts @@ -2853,6 +2853,53 @@ describe("runTurn", () => { expect(drainCallCount).toBe(2); }); + it("async drainSteering (returns a Promise) is awaited — its messages reach the next step (the shell persists before returning)", async () => { + // The shell's drainSteering is async so it can persist the injected + // messages before returning. The kernel must `await` it (a sync call + // would get a Promise, not the array). This test pins that contract. + let drainCallCount = 0; + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: "async steer!" }], + }; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + // Async drainSteering — resolves on a microtask, like a real persist. + drainSteering: () => + new Promise((resolve) => { + drainCallCount++; + // Defer the resolve so the kernel MUST await to get the array. + queueMicrotask(() => resolve([steeringMessage])); + }), + }); + + expect(drainCallCount).toBe(1); + const secondStepMessages = capturedMessages[1] ?? []; + // The async-returned steering message was awaited and appended AFTER the + // tool result, before the next step — proving the kernel awaited it. + expect(secondStepMessages).toHaveLength(4); + expect(secondStepMessages[3]).toEqual(steeringMessage); + }); + it("MAX_STEPS=0 (unlimited): turn runs past the old 50-step limit and drains at every tool-result boundary until the model stops naturally", async () => { let drainCallCount = 0; // 100 tool-call steps (past the old MAX_STEPS=50) + 1 text-only step @@ -2899,6 +2946,335 @@ describe("runTurn", () => { }); }); + // ── onStepBoundary (in-flight history replacement) ────────────────────── + // + // PURE tests: a capturing provider + a fake `onStepBoundary` that returns a + // replacement message list (or void). The kernel must adopt the returned + // list as its working history for subsequent steps. ZERO mocks of + // `@dispatch/*` — the hook is injected like `drainSteering`. + + describe("onStepBoundary", () => { + it("returns a replacement → next step's provider input is the replaced history (old messages dropped)", async () => { + const compactedHistory: ChatMessage[] = [ + { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] }, + { role: "user", chunks: [{ type: "text", text: "recent" }] }, + ]; + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 999, outputTokens: 1 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + // Sanity: the kernel passes the full running history at the boundary + // (user, assistant tool-call, tool result) — 3 messages. + expect(messages.length).toBe(3); + return compactedHistory; + }, + }); + + expect(boundaryCallCount).toBe(1); + expect(capturedMessages).toHaveLength(2); + // The second step must see ONLY the replaced history — the original + // 3 messages are gone, replaced by [SUMMARY, recent]. + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(2); + expect(secondStepMessages[0]).toEqual(compactedHistory[0]); + expect(secondStepMessages[1]).toEqual(compactedHistory[1]); + }); + + it("returns undefined → history unchanged (byte-identical to omitting the hook)", async () => { + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => { + boundaryCallCount++; + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(1); + const secondStepMessages = capturedMessages[1] ?? []; + // user, assistant(tool-call), tool-result — unchanged. + expect(secondStepMessages).toHaveLength(3); + expect(secondStepMessages[0]?.role).toBe("user"); + expect(secondStepMessages[1]?.role).toBe("assistant"); + expect(secondStepMessages[2]?.role).toBe("tool"); + }); + + it("returns an empty array → treated as no replacement (history unchanged)", async () => { + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => [], + }); + + const secondStepMessages = capturedMessages[1] ?? []; + // Empty replacement is ignored — history unchanged. + expect(secondStepMessages).toHaveLength(3); + }); + + it("NOT called when a step has no tool calls (text-only turn) — there is no next step", async () => { + let boundaryCallCount = 0; + const provider = createFakeProvider([ + [ + { type: "text-delta", delta: "hello" }, + { type: "finish", reason: "stop" }, + ], + ]); + + await runTurn({ + provider, + messages: [userMessage], + tools: [], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => { + boundaryCallCount++; + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(0); + }); + + it("multiple tool-call steps → called once per tool-call step, AFTER drainSteering (steering present in messages)", async () => { + let boundaryCallCount = 0; + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: "steer!" }], + }; + let sawSteeringAtBoundary = false; + + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + drainSteering: () => [steeringMessage], + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + // drainSteering runs BEFORE onStepBoundary, so the steering message + // must already be present in the boundary's `messages`. + if (messages.some((m) => m === steeringMessage)) { + sawSteeringAtBoundary = true; + } + return undefined; + }, + }); + + // Steps 0 and 1 produced tool calls → boundary once each. + // Step 2 (text-only) ends the turn → no boundary. Total = 2. + expect(boundaryCallCount).toBe(2); + expect(sawSteeringAtBoundary).toBe(true); + expect(capturedMessages).toHaveLength(3); + }); + + it("receives the step's usage as stepUsage", async () => { + const seenUsages: Array<{ inputTokens: number; outputTokens: number }> = []; + const { provider } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 4242, outputTokens: 17 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: ({ stepUsage }) => { + seenUsages.push({ + inputTokens: stepUsage.inputTokens, + outputTokens: stepUsage.outputTokens, + }); + return undefined; + }, + }); + + expect(seenUsages).toEqual([{ inputTokens: 4242, outputTokens: 17 }]); + }); + + it("omitted → strict no-op (byte-identical to before)", async () => { + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + const result = await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + // onStepBoundary omitted — must be a strict no-op. + }); + + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(3); + expect(result.messages).toHaveLength(3); + }); + + it("replacement adopted across multiple subsequent steps (history stays compacted)", async () => { + const compactedHistory: ChatMessage[] = [ + { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] }, + ]; + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + // Compact only at the FIRST boundary; subsequent boundaries return void. + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + if (boundaryCallCount === 1) { + // Pre-replacement: user + assistant(tool-call) + tool result = 3. + expect(messages.length).toBe(3); + return compactedHistory; + } + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(2); + // After the FIRST boundary compacts to [SUMMARY], step 1's provider call + // (captured[1]) sees ONLY [SUMMARY] — the original user message is GONE, + // proving the replacement took effect before the next step. + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(1); + expect(secondStepMessages[0]).toEqual(compactedHistory[0]); + // Step 1 then produced tc2 + a tool result (pushed onto messages), so + // step 2's provider call (captured[2]) sees [SUMMARY, assistant(tc2), + // tool-result] = 3 — the compaction PERSISTED across step 1→2 (the + // SUMMARY is still heading the history; the original user message + // never came back). + const thirdStepMessages = capturedMessages[2] ?? []; + expect(thirdStepMessages).toHaveLength(3); + expect(thirdStepMessages[0]).toEqual(compactedHistory[0]); + expect(thirdStepMessages[1]?.role).toBe("assistant"); + expect(thirdStepMessages[2]?.role).toBe("tool"); + }); + }); + // ── Retry with backoff ────────────────────────────────────────────────── // // PURE tests: a fake `sleep` (records calls, resolves instantly, can abort diff --git a/packages/kernel/src/runtime/run-turn.ts b/packages/kernel/src/runtime/run-turn.ts index 3460033..8f68865 100644 --- a/packages/kernel/src/runtime/run-turn.ts +++ b/packages/kernel/src/runtime/run-turn.ts @@ -720,11 +720,34 @@ export async function runTurn(input: RunTurnInput): Promise<RunTurnResult> { // and append them after the tool results, before the next call. // The kernel owns no queue and names no feature — it just calls // the callback and appends. Emits nothing (caller emits the - // `steering` AgentEvent in its own wrapper). - const steering = input.drainSteering?.() ?? []; + // `steering` AgentEvent in its own wrapper). The callback MAY + // return a Promise (the shell persists the injected messages + // before returning) — `await` handles both sync and async. + const steering = (await input.drainSteering?.()) ?? []; for (const msg of steering) { messages.push(msg); } + + // In-flight history replacement boundary: give the caller a chance + // to swap the running message history before the next step (e.g. to + // compact it when the context window nears capacity). The kernel names + // no feature and performs no I/O — it calls the callback and adopts + // whatever message list it returns, mirroring `drainSteering`. When the + // caller returns a list, the runtime replaces its working history with + // it (in place); `undefined`/empty leaves it unchanged. Only reached + // when there IS a next step (tool calls were produced). + if (input.onStepBoundary !== undefined) { + const replacement = await input.onStepBoundary({ + stepUsage: stepResult.usage, + messages, + }); + if (replacement !== undefined && replacement.length > 0) { + messages.length = 0; + for (const msg of replacement) { + messages.push(msg); + } + } + } } } } finally { diff --git a/packages/mcp/src/client.test.ts b/packages/mcp/src/client.test.ts index ca8a11b..4542a36 100644 --- a/packages/mcp/src/client.test.ts +++ b/packages/mcp/src/client.test.ts @@ -203,4 +203,70 @@ describe("McpClient", () => { content: [{ type: "text", text: "too late" }], }); }); + + /** A connection whose initialize never responds (simulates a framing- + * incompatible server like chrome-devtools-mcp under Content-Length framing): + * the pending JSON-RPC request would hang forever without a timeout/abort. */ + function makeHangingConnection(): Connection { + const never = new Promise<unknown>(() => {}); + return { + send: () => never, + notify: () => {}, + onNotification: () => {}, + close: () => {}, + pid: 1, + }; + } + + it("initialize raises McpTimeoutError when the server never responds", async () => { + const { McpTimeoutError } = await import("./timeout.js"); + const client = new McpClient({ connection: makeHangingConnection() }); + await expect(client.initialize(undefined, 20)).rejects.toBeInstanceOf(McpTimeoutError); + expect(client.getState()).toBe("error"); + }); + + it("initialize is abortable: an aborting signal rejects immediately", async () => { + const client = new McpClient({ connection: makeHangingConnection() }); + const controller = new AbortController(); + const p = client.initialize(controller.signal, 50_000); + controller.abort(); + await expect(p).rejects.toThrow("Aborted"); + expect(client.getState()).toBe("error"); + }); + + it("initialize rejects immediately when the signal is already aborted", async () => { + const client = new McpClient({ connection: makeHangingConnection() }); + const controller = new AbortController(); + controller.abort(); + await expect(client.initialize(controller.signal, 50_000)).rejects.toThrow("Aborted"); + }); + + it("listTools raises McpTimeoutError when the server never responds", async () => { + const { McpTimeoutError } = await import("./timeout.js"); + // Reach "connected" with a fast (auto-responding) connection, then swap in + // a hanging connection for the tools/list call. + const fast = makeMockConnection(); + const client = new McpClient({ connection: fast }); + await client.initialize(); + + const hanging = makeHangingConnection(); + // Swap the connection so tools/list hangs. + (client as unknown as { connection: Connection }).connection = hanging; + + await expect(client.listTools(undefined, 20)).rejects.toBeInstanceOf(McpTimeoutError); + }); + + it("listTools is abortable", async () => { + const fast = makeMockConnection(); + const client = new McpClient({ connection: fast }); + await client.initialize(); + + const hanging = makeHangingConnection(); + (client as unknown as { connection: Connection }).connection = hanging; + + const controller = new AbortController(); + const p = client.listTools(controller.signal, 50_000); + controller.abort(); + await expect(p).rejects.toThrow("Aborted"); + }); }); diff --git a/packages/mcp/src/client.ts b/packages/mcp/src/client.ts index d5d06fc..6a69c00 100644 --- a/packages/mcp/src/client.ts +++ b/packages/mcp/src/client.ts @@ -3,8 +3,15 @@ * * Manages a single MCP server connection: initialize handshake, * tool discovery, tool invocation, and list_changed notifications. + * + * Every awaited handshake/list operation is bounded by `withTimeout` (a default + * timeout) and an optional `AbortSignal`, so a misbehaving or framing- + * incompatible server can never hang the caller (the per-turn tools filter) + * indefinitely. `callTool` was already abort-aware; `initialize`/`listTools` + * now are too. */ +import { MCP_DEFAULT_TIMEOUT_MS, withTimeout } from "./timeout.js"; import type { Connection } from "./transport.js"; import type { McpCallResult, @@ -47,14 +54,27 @@ export class McpClient { this.toolsChangedHandler = handler; } - async initialize(): Promise<McpInitializeResult> { + /** + * Perform the MCP `initialize` handshake. Bounded by `timeoutMs` (default + * {@link MCP_DEFAULT_TIMEOUT_MS}) and the optional `signal` (the turn's abort + * signal) so a server that never responds cannot hang the caller forever. + */ + async initialize( + signal?: AbortSignal, + timeoutMs: number = MCP_DEFAULT_TIMEOUT_MS, + ): Promise<McpInitializeResult> { this.state = "connecting"; try { - const result = (await this.connection.send("initialize", { - protocolVersion: "2025-11-25", - capabilities: {}, - clientInfo: { name: "dispatch", version: "0.0.0" }, - })) as McpInitializeResult; + const result = (await withTimeout( + this.connection.send("initialize", { + protocolVersion: "2025-11-25", + capabilities: {}, + clientInfo: { name: "dispatch", version: "0.0.0" }, + }), + "initialize", + timeoutMs, + signal, + )) as McpInitializeResult; this.capabilities = result.capabilities; this.connection.notify("notifications/initialized", {}); @@ -73,11 +93,23 @@ export class McpClient { } } - async listTools(): Promise<readonly McpToolInfo[]> { + /** + * List the server's tools. Bounded by `timeoutMs` (default + * {@link MCP_DEFAULT_TIMEOUT_MS}) and the optional `signal`. + */ + async listTools( + signal?: AbortSignal, + timeoutMs: number = MCP_DEFAULT_TIMEOUT_MS, + ): Promise<readonly McpToolInfo[]> { if (this.state !== "connected") { throw new Error("Client not connected"); } - const result = (await this.connection.send("tools/list")) as McpListToolsResult; + const result = (await withTimeout( + this.connection.send("tools/list"), + "tools/list", + timeoutMs, + signal, + )) as McpListToolsResult; this.tools = result.tools; return this.tools; } diff --git a/packages/mcp/src/extension.test.ts b/packages/mcp/src/extension.test.ts index b937c2b..9e029d2 100644 --- a/packages/mcp/src/extension.test.ts +++ b/packages/mcp/src/extension.test.ts @@ -81,6 +81,9 @@ describe("filterMcpTools (pure)", () => { interface FakeServer { tools: McpToolInfo[]; failInitialize: boolean; + /** When true, the spawn never responds to `initialize` (a hanging / + * framing-incompatible server) — used to exercise timeout/abort paths. */ + hangInitialize: boolean; emitListChanged: () => void; } @@ -113,7 +116,11 @@ function makeFakeSpawn(server: FakeServer): SpawnProcess { const id = parsed.id ?? 0; const method = parsed.method; if (method === "initialize") { - if (server.failInitialize) { + if (server.hangInitialize) { + // Never respond — simulates a framing-incompatible server + // (e.g. chrome-devtools-mcp under the old Content-Length + // framing). The connect must be bounded by timeout/abort. + } else if (server.failInitialize) { emit( encode( JSON.stringify({ @@ -246,7 +253,12 @@ const assembly = (tools: ToolContract[], cwd = "/proj"): ToolAssembly => ({ const flush = () => new Promise((r) => setTimeout(r, 0)); function makeServer(initialTools: McpToolInfo[]): FakeServer { - return { tools: [...initialTools], failInitialize: false, emitListChanged: () => {} }; + return { + tools: [...initialTools], + failInitialize: false, + hangInitialize: false, + emitListChanged: () => {}, + }; } function makeExt(server: FakeServer, configJson: string): Extension { @@ -360,4 +372,95 @@ describe("mcp extension lifecycle", () => { const after = await service.status("/proj"); expect(after[0].state).toBe("disconnected"); }); + + // ------------------------------------------------------------------------- + // Bug 2 + Bug 3: a misbehaving/hanging server must not hang a turn, and the + // turn's AbortSignal (assembly.signal) must interrupt a stuck connect. + // ------------------------------------------------------------------------- + + it("degrades gracefully (no MCP tools) when the turn's signal is already aborted", async () => { + const server = makeServer([tool("create_object")]); + const ext = makeExt(server, dispatchConfig({ freecad: { command: "fake" } })); + const { host, getFilter } = makeFakeHost(); + ext.activate(host); + const filter = requireFilter(getFilter); + + const controller = new AbortController(); + controller.abort(); + + const base = assembly([]); + // The filter must NOT hang on the (never-needed) connect: an aborted turn + // signal propagates to initialize, which rejects immediately. + const result = await filter({ + tools: base.tools, + cwd: base.cwd, + conversationId: base.conversationId, + signal: controller.signal, + }); + + expect(result.tools).toEqual([]); + ext.deactivate?.(); + }); + + it("the turn's signal aborts a hanging server connect (POST /stop interrupts)", async () => { + // hangInitialize: the spawn never responds to initialize (a framing- + // incompatible / misbehaving server). Without abort propagation this + // would hang the filter until MCP_CONNECT_TIMEOUT_MS; with propagation + // the abort breaks it immediately. + const server = makeServer([tool("create_object")]); + server.hangInitialize = true; + const ext = makeExt(server, dispatchConfig({ chrome: { command: "fake" } })); + const { host, getFilter } = makeFakeHost(); + ext.activate(host); + const filter = requireFilter(getFilter); + + const controller = new AbortController(); + const base = assembly([]); + const resultPromise = filter({ + tools: base.tools, + cwd: base.cwd, + conversationId: base.conversationId, + signal: controller.signal, + }); + + // Let the filter progress into the hanging initialize (withTimeout has + // its abort listener armed), THEN abort — a true mid-flight cancel + // simulating POST /conversations/:id/stop. Without signal propagation + // this would hang ~30s (the connect backstop) and time out the test. + await flush(); + controller.abort(); + + const result = await resultPromise; + // Degraded: no MCP tools surfaced, and the filter resolved (did not hang). + expect(result.tools).toEqual([]); + ext.deactivate?.(); + }); + + it("non-MCP tools pass through unchanged when an MCP connect fails", async () => { + // failInitialize: the server rejects initialize (a fast failure, not a + // hang) so the connect degrades promptly without waiting on a backstop. + const server = makeServer([tool("create_object")]); + server.failInitialize = true; + const ext = makeExt(server, dispatchConfig({ chrome: { command: "fake" } })); + const { host, getFilter } = makeFakeHost(); + ext.activate(host); + const filter = requireFilter(getFilter); + + const stubNonMcp: ToolContract = { + name: "run_shell", + description: "kept", + parameters: { type: "object" }, + execute: async () => ({ content: "" }), + }; + + const result = await filter({ + tools: [stubNonMcp], + cwd: "/proj", + conversationId: "c", + }); + + // Non-MCP tool survives; the failed MCP server contributed no tools. + expect(result.tools.map((t) => t.name)).toEqual(["run_shell"]); + ext.deactivate?.(); + }); }); diff --git a/packages/mcp/src/extension.ts b/packages/mcp/src/extension.ts index d12d420..bb951ee 100644 --- a/packages/mcp/src/extension.ts +++ b/packages/mcp/src/extension.ts @@ -18,6 +18,7 @@ import { resolveServers } from "./config.js"; import type { Logger } from "./manager.js"; import { McpManager } from "./manager.js"; import { adaptTool, namespace } from "./registry.js"; +import { MCP_CONNECT_TIMEOUT_MS } from "./timeout.js"; import type { SpawnedProcess, SpawnProcess } from "./transport.js"; import { createStdioTransport } from "./transport.js"; import type { McpServerStatus, McpService, ResolvedMcpServer } from "./types.js"; @@ -111,13 +112,19 @@ export function makeMcpExtension(deps: McpExtensionDeps): Extension { } } - async function connectAndRegister(server: ResolvedMcpServer, cwd: string): Promise<void> { - const client = await manager.ensureConnected(server, cwd); + async function connectAndRegister( + server: ResolvedMcpServer, + cwd: string, + signal?: AbortSignal, + ): Promise<void> { + const client = await manager.ensureConnected(server, cwd, signal); registerToolsFromClient(server.id, client); // Wire list_changed → re-list → re-register. onToolsChanged replaces // the handler; ensureConnected returns the same cached client so this - // is idempotent across turns. + // is idempotent across turns. The async re-list runs LATER (not during + // this filter), so it is NOT bound to the filter's signal (already + // done) — it relies on listTools()'s own default timeout instead. client.onToolsChanged(async () => { try { await client.listTools(); @@ -133,18 +140,44 @@ export function makeMcpExtension(deps: McpExtensionDeps): Extension { // Resolve config + ensure servers connected, then drop tools whose // server is not connected. Lazy-spawn happens here (first turn). + // + // The whole connect phase is wrapped in a per-filter AbortController + // that fires on EITHER (a) the turn's signal (`assembly.signal`, so + // POST /conversations/:id/stop interrupts a stuck connect immediately) + // OR (b) a timeout (`MCP_CONNECT_TIMEOUT_MS`, so a misbehaving / + // framing-incompatible server cannot hang the turn forever). On abort + // we degrade gracefully: skip MCP tools for this turn rather than block. host.addFilter(toolsFilter, async (assembly: ToolAssembly): Promise<ToolAssembly> => { const cwd = assembly.cwd ?? deps.getCwd(); const dispatchMcpJson = await deps.readFile(joinPath(cwd, ".dispatch", "mcp.json")); const opencodeJson = await deps.readFile(joinPath(cwd, "opencode.json")); const { servers } = resolveServers({ dispatchMcpJson, opencodeJson }); - for (const server of servers) { - try { - await connectAndRegister(server, cwd); - } catch { - // Connection failure — the manager tracks broken state. + const controller = new AbortController(); + const parentSignal = assembly.signal; + const onParentAbort = (): void => controller.abort(); + if (parentSignal !== undefined) { + if (parentSignal.aborted) { + controller.abort(); + } else { + parentSignal.addEventListener("abort", onParentAbort, { once: true }); + } + } + const timer = setTimeout(() => controller.abort(), MCP_CONNECT_TIMEOUT_MS); + + try { + for (const server of servers) { + try { + await connectAndRegister(server, cwd, controller.signal); + } catch { + // Connection failure / timeout / aborted — the manager tracks + // broken state; we keep going (or abort cascades) below. + } + if (controller.signal.aborted) break; } + } finally { + clearTimeout(timer); + parentSignal?.removeEventListener("abort", onParentAbort); } const statuses = manager.status(servers); diff --git a/packages/mcp/src/framing.test.ts b/packages/mcp/src/framing.test.ts index 3b207a3..9832c74 100644 --- a/packages/mcp/src/framing.test.ts +++ b/packages/mcp/src/framing.test.ts @@ -1,84 +1,74 @@ import { describe, expect, it } from "vitest"; import { encode, FrameDecoder } from "./framing.js"; +/** Build a legacy Content-Length frame (for exercising the decoder's CL path). */ +function contentLengthFrame(body: string): Uint8Array { + const bodyBytes = new TextEncoder().encode(body); + return new TextEncoder().encode(`Content-Length: ${bodyBytes.length}\r\n\r\n${body}`); +} + describe("encode", () => { - it("produces correct Content-Length header", () => { + it("produces newline-delimited JSON (current MCP spec framing)", () => { const msg = '{"jsonrpc":"2.0","id":1,"method":"initialize"}'; const encoded = encode(msg); const text = new TextDecoder().decode(encoded); - expect(text).toBe(`Content-Length: ${new TextEncoder().encode(msg).length}\r\n\r\n${msg}`); + expect(text).toBe(`${msg}\n`); }); - it("handles empty message", () => { + it("appends a trailing newline to an empty message", () => { const encoded = encode(""); const text = new TextDecoder().decode(encoded); - expect(text).toBe("Content-Length: 0\r\n\r\n"); + expect(text).toBe("\n"); }); }); -describe("FrameDecoder", () => { - it("reassembles a complete message from one chunk", () => { +describe("FrameDecoder — newline-delimited JSON", () => { + it("decodes a single newline-delimited message", () => { + const msg = '{"jsonrpc":"2.0","id":1}'; + const decoder = new FrameDecoder(); + expect(decoder.decode(encode(msg))).toEqual([msg]); + }); + + it("decodes a CRLF-terminated message (trailing \\r tolerated)", () => { const msg = '{"jsonrpc":"2.0","id":1}'; - const encoded = encode(msg); const decoder = new FrameDecoder(); - const messages = decoder.decode(encoded); - expect(messages).toEqual([msg]); + const framed = new TextEncoder().encode(`${msg}\r\n`); + expect(decoder.decode(framed)).toEqual([msg]); }); - it("handles split across chunks", () => { + it("handles a split across chunks", () => { const msg = '{"jsonrpc":"2.0","id":1,"method":"initialize"}'; const encoded = encode(msg); const mid = Math.floor(encoded.length / 2); - const chunk1 = encoded.slice(0, mid); - const chunk2 = encoded.slice(mid); const decoder = new FrameDecoder(); - const result1 = decoder.decode(chunk1); - expect(result1).toEqual([]); - - const result2 = decoder.decode(chunk2); - expect(result2).toEqual([msg]); + expect(decoder.decode(encoded.slice(0, mid))).toEqual([]); + expect(decoder.decode(encoded.slice(mid))).toEqual([msg]); }); it("handles two messages in one chunk", () => { const msg1 = '{"jsonrpc":"2.0","id":1}'; const msg2 = '{"jsonrpc":"2.0","id":2}'; - const encoded1 = encode(msg1); - const encoded2 = encode(msg2); - const combined = new Uint8Array(encoded1.length + encoded2.length); - combined.set(encoded1); - combined.set(encoded2, encoded1.length); + const combined = new Uint8Array(encode(msg1).length + encode(msg2).length); + combined.set(encode(msg1)); + combined.set(encode(msg2), encode(msg1).length); const decoder = new FrameDecoder(); - const messages = decoder.decode(combined); - expect(messages).toEqual([msg1, msg2]); + expect(decoder.decode(combined)).toEqual([msg1, msg2]); }); - it("rejects negative Content-Length by skipping header", () => { - const header = "Content-Length: -5\r\n\r\n"; - const encoded = new TextEncoder().encode(`${header}extra`); - const decoder = new FrameDecoder(); - const messages = decoder.decode(encoded); - // Negative length does not match the digit capture, so the header is skipped. - expect(messages).toEqual([]); - }); - - it("rejects zero Content-Length", () => { - const encoded = encode(""); + it("skips blank lines between messages", () => { + const msg = '{"jsonrpc":"2.0","id":1}'; + const framed = new TextEncoder().encode(`\n\n${msg}\n\n`); const decoder = new FrameDecoder(); - const messages = decoder.decode(encoded); - expect(messages).toEqual([""]); + expect(decoder.decode(framed)).toEqual([msg]); }); - it("reassembles multi-byte UTF-8 content (byte-length, not char-length)", () => { - // "héllo" — é is two UTF-8 bytes; Content-Length counts bytes. + it("reassembles multi-byte UTF-8 content (byte-aware, not char-aware)", () => { const msg = '{"text":"héllo 🚀"}'; - const encoded = encode(msg); expect(new TextEncoder().encode(msg).length).toBeGreaterThan(msg.length); - const decoder = new FrameDecoder(); - const messages = decoder.decode(encoded); - expect(messages).toEqual([msg]); + expect(decoder.decode(encode(msg))).toEqual([msg]); }); it("reassembles multi-byte content split across a chunk boundary", () => { @@ -89,4 +79,85 @@ describe("FrameDecoder", () => { expect(decoder.decode(encoded.slice(0, mid))).toEqual([]); expect(decoder.decode(encoded.slice(mid))).toEqual([msg]); }); + + it("does not split a JSON string containing an escaped \\n (no raw newline)", () => { + // JSON escapes newlines inside strings as the two chars `\` + `n`; a raw + // 0x0a only ever appears as a message separator. So a JSON body carrying + // an embedded newline literal survives intact. + const msg = '{"text":"line1\\nline2"}'; + const decoder = new FrameDecoder(); + expect(decoder.decode(encode(msg))).toEqual([msg]); + }); +}); + +describe("FrameDecoder — legacy Content-Length framing (auto-detected)", () => { + it("decodes a Content-Length-framed message", () => { + const msg = '{"jsonrpc":"2.0","id":1}'; + const decoder = new FrameDecoder(); + expect(decoder.decode(contentLengthFrame(msg))).toEqual([msg]); + }); + + it("reassembles a Content-Length frame split across chunks", () => { + const msg = '{"jsonrpc":"2.0","id":1,"method":"initialize"}'; + const encoded = contentLengthFrame(msg); + const mid = Math.floor(encoded.length / 2); + const decoder = new FrameDecoder(); + expect(decoder.decode(encoded.slice(0, mid))).toEqual([]); + expect(decoder.decode(encoded.slice(mid))).toEqual([msg]); + }); + + it("decodes two Content-Length frames in one chunk", () => { + const msg1 = '{"jsonrpc":"2.0","id":1}'; + const msg2 = '{"jsonrpc":"2.0","id":2}'; + const combined = new Uint8Array( + contentLengthFrame(msg1).length + contentLengthFrame(msg2).length, + ); + combined.set(contentLengthFrame(msg1)); + combined.set(contentLengthFrame(msg2), contentLengthFrame(msg1).length); + + const decoder = new FrameDecoder(); + expect(decoder.decode(combined)).toEqual([msg1, msg2]); + }); + + it("rejects negative Content-Length by skipping header", () => { + const encoded = new TextEncoder().encode("Content-Length: -5\r\n\r\nextra"); + const decoder = new FrameDecoder(); + expect(decoder.decode(encoded)).toEqual([]); + }); + + it("accepts zero Content-Length as an empty message", () => { + const decoder = new FrameDecoder(); + expect(decoder.decode(contentLengthFrame(""))).toEqual([""]); + }); + + it("reassembles multi-byte UTF-8 via Content-Length (byte count)", () => { + const msg = '{"text":"héllo 🚀"}'; + const decoder = new FrameDecoder(); + expect(decoder.decode(contentLengthFrame(msg))).toEqual([msg]); + }); + + it("does not mis-read a partial 'Content-Length' prefix as newline-delimited", () => { + // A buffer that is a partial prefix of "Content-Length:" must WAIT for more + // bytes rather than being split on a (nonexistent) newline. + const decoder = new FrameDecoder(); + const partial = new TextEncoder().encode("Content-Len"); + expect(decoder.decode(partial)).toEqual([]); + const rest = new TextEncoder().encode("gth: 3\r\n\r\nabc"); + expect(decoder.decode(rest)).toEqual(["abc"]); + }); +}); + +describe("FrameDecoder — mixed framings", () => { + it("decodes a Content-Length frame followed by a newline-delimited message", () => { + const clMsg = '{"jsonrpc":"2.0","id":1}'; + const nlMsg = '{"jsonrpc":"2.0","id":2}'; + const cl = contentLengthFrame(clMsg); + const nl = encode(nlMsg); + const combined = new Uint8Array(cl.length + nl.length); + combined.set(cl); + combined.set(nl, cl.length); + + const decoder = new FrameDecoder(); + expect(decoder.decode(combined)).toEqual([clMsg, nlMsg]); + }); }); diff --git a/packages/mcp/src/framing.ts b/packages/mcp/src/framing.ts index 5f818e2..d30b871 100644 --- a/packages/mcp/src/framing.ts +++ b/packages/mcp/src/framing.ts @@ -1,32 +1,40 @@ /** - * Content-Length framing for MCP stdio transport. + * MCP stdio framing. * - * Each JSON-RPC message is framed as: - * Content-Length: <byte-length>\r\n\r\n<JSON bytes> + * The MCP spec (revision 2025-03-26 and later, including 2025-11-25) frames + * JSON-RPC messages over stdio as **newline-delimited JSON**: each message is + * a single JSON document followed by a `\n` (or `\r\n`). This replaced the + * older LSP-style `Content-Length: N\r\n\r\n<JSON>` framing that the protocol + * inherited originally. Modern servers (e.g. chrome-devtools-mcp) speak ONLY + * newline-delimited JSON — they emit `<JSON>\n` on stdout and do not respond to + * `Content-Length`-framed input at all. * - * Same framing as LSP — the MCP spec inherited this from the LSP base protocol. + * Outgoing messages are therefore encoded as newline-delimited JSON (the + * current spec default), so we can talk to modern servers. The decoder + * **auto-detects** the incoming framing per message — it accepts BOTH + * newline-delimited JSON and legacy `Content-Length` frames — so we still + * interoperate with servers that respond with the old framing. * * PURE: no I/O. Operates on bytes (Uint8Array) so multi-byte UTF-8 content is * handled correctly — `Content-Length` is a *byte* count, not a character count. */ +const HEADER_PREFIX = "Content-Length:"; const HEADER_SEP = "\r\n\r\n"; const CONTENT_LENGTH_RE = /Content-Length:\s*(\d+)/i; -const SEP_BYTES = new TextEncoder().encode(HEADER_SEP); +const encoder = new TextEncoder(); +const SEP_BYTES = encoder.encode(HEADER_SEP); + +const CR = 0x0d; +const LF = 0x0a; /** - * Encode a JSON string into a single Content-Length-framed message. - * Returns the full frame (header + blank line + body) as bytes. + * Encode a JSON string as a single newline-delimited frame (the current MCP + * spec stdio framing): `<JSON>\n`. This is what we send to MCP servers. */ export function encode(msg: string): Uint8Array { - const body = new TextEncoder().encode(msg); - const header = `Content-Length: ${body.length}\r\n\r\n`; - const frame = new TextEncoder().encode(header); - const result = new Uint8Array(frame.length + body.length); - result.set(frame); - result.set(body, frame.length); - return result; + return encoder.encode(`${msg}\n`); } /** Find the first occurrence of `needle` in `haystack` at or after `from`. -1 if absent. */ @@ -46,9 +54,48 @@ function indexOfBytes(haystack: Uint8Array, needle: Uint8Array, from: number): n return -1; } +/** Find the first occurrence of a single byte at or after `from`. -1 if absent. */ +function indexOfByte(haystack: Uint8Array, needle: number, from: number): number { + for (let i = from; i < haystack.length; i++) { + if (haystack[i] === needle) return i; + } + return -1; +} + +/** Lowercase an ASCII byte (A-Z → a-z); leave everything else unchanged. */ +function toLowerByte(b: number): number { + return b >= 0x41 && b <= 0x5a ? b + 0x20 : b; +} + +function isLineTerminator(b: number | undefined): boolean { + return b === CR || b === LF; +} + +/** + * How does `buf` relate to the `Content-Length:` header prefix (case-insensitive)? + * - `HEADER_PREFIX.length` → `buf` starts with the full `Content-Length:` prefix. + * - a positive number `< HEADER_PREFIX.length` → `buf` is a (possibly partial) + * prefix of `Content-Length:` — ambiguous, the caller must wait for more bytes. + * - `-1` → `buf` definitively does NOT start with `Content-Length:` (not CL framing). + */ +function contentLengthPrefixLength(buf: Uint8Array): number { + const n = Math.min(buf.length, HEADER_PREFIX.length); + for (let i = 0; i < n; i++) { + const a = buf[i]; + if (a === undefined) return -1; // unreachable: i < n <= buf.length + if (toLowerByte(a) !== toLowerByte(HEADER_PREFIX.charCodeAt(i))) return -1; + } + return n; +} + /** * Feed raw bytes into the decoder. Returns all complete JSON messages that can * be extracted from the accumulated buffer. Buffers partial frames across calls. + * + * Auto-detects framing per message: a `Content-Length:`-prefixed buffer is parsed + * as a Content-Length frame (legacy/LSP-style); anything else is parsed as + * newline-delimited JSON (current MCP spec). Both framings may be mixed in a + * single stream. */ export class FrameDecoder { private buf: Uint8Array = new Uint8Array(0); @@ -56,45 +103,90 @@ export class FrameDecoder { decode(chunk: Uint8Array): string[] { // Append the incoming chunk to the internal byte buffer. - const next = new Uint8Array(this.buf.length + chunk.length); - next.set(this.buf); - next.set(chunk, this.buf.length); - this.buf = next; + if (chunk.length > 0) { + const next = new Uint8Array(this.buf.length + chunk.length); + next.set(this.buf); + next.set(chunk, this.buf.length); + this.buf = next; + } const messages: string[] = []; while (true) { - const sepIdx = indexOfBytes(this.buf, SEP_BYTES, 0); - if (sepIdx === -1) break; - - // The header block is everything before the separator; parse - // Content-Length from it (ASCII, so decoding the slice is safe). - const headerText = this.decoder.decode(this.buf.subarray(0, sepIdx)); - const match = CONTENT_LENGTH_RE.exec(headerText); - const bodyStart = sepIdx + SEP_BYTES.length; - - if (!match?.[1]) { - // No usable Content-Length — drop this header and continue scanning. - this.buf = this.buf.subarray(bodyStart); - continue; - } + // 1. Skip leading CR/LF whitespace between frames. + let i = 0; + while (i < this.buf.length && isLineTerminator(this.buf[i])) i++; + if (i > 0) this.buf = this.buf.subarray(i); + if (this.buf.length === 0) break; - const length = Number.parseInt(match[1], 10); - if (length < 0) { - this.buf = this.buf.subarray(bodyStart); + // 2. Detect framing. + const prefix = contentLengthPrefixLength(this.buf); + if (prefix >= 0 && prefix < HEADER_PREFIX.length) { + // Buffer is a (possibly partial) prefix of "Content-Length:" — ambiguous; + // wait for more bytes before deciding this is (or isn't) a CL frame. + break; + } + if (prefix === HEADER_PREFIX.length) { + // Content-Length framing. + const result = this.tryParseContentLength(); + if (result === "incomplete") break; + if (result !== "skip") messages.push(result); continue; } - if (this.buf.length - bodyStart < length) { - // Body not fully received yet; wait for more bytes. - break; - } + // 3. Newline-delimited JSON: one message per line, terminated by `\n` + // (a trailing `\r` before the `\n` is tolerated). A raw newline byte + // can only ever appear as a message separator — JSON escapes newlines + // inside strings as the two characters `\n`, never a literal 0x0a — so + // splitting on `\n` bytes is safe for valid JSON. + const nl = indexOfByte(this.buf, LF, 0); + if (nl === -1) break; // incomplete line — wait for more bytes - // Decode exactly `length` body bytes (preserves multi-byte UTF-8). - messages.push(this.decoder.decode(this.buf.subarray(bodyStart, bodyStart + length))); - this.buf = this.buf.subarray(bodyStart + length); + let end = nl; + if (end > 0 && this.buf[end - 1] === CR) end--; + const text = this.decoder.decode(this.buf.subarray(0, end)); + this.buf = this.buf.subarray(nl + 1); + if (text.length > 0) messages.push(text); } return messages; } + + /** + * Parse one Content-Length frame from the front of `this.buf`. Returns: + * - the decoded body string (possibly `""` for a zero-length body), + * - `"incomplete"` if the header or body hasn't fully arrived (caller waits), + * - `"skip"` if the header was consumed but carried no usable Content-Length + * (caller continues scanning without emitting a message). + */ + private tryParseContentLength(): string | "incomplete" | "skip" { + const sepIdx = indexOfBytes(this.buf, SEP_BYTES, 0); + if (sepIdx === -1) return "incomplete"; // header not fully received yet + + const headerText = this.decoder.decode(this.buf.subarray(0, sepIdx)); + const match = CONTENT_LENGTH_RE.exec(headerText); + const bodyStart = sepIdx + SEP_BYTES.length; + + if (!match?.[1]) { + // No usable Content-Length — drop this header and continue scanning. + this.buf = this.buf.subarray(bodyStart); + return "skip"; + } + + const length = Number.parseInt(match[1], 10); + if (length < 0) { + this.buf = this.buf.subarray(bodyStart); + return "skip"; + } + + if (this.buf.length - bodyStart < length) { + // Body not fully received yet; wait for more bytes. + return "incomplete"; + } + + // Decode exactly `length` body bytes (preserves multi-byte UTF-8). + const body = this.decoder.decode(this.buf.subarray(bodyStart, bodyStart + length)); + this.buf = this.buf.subarray(bodyStart + length); + return body; + } } diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index 6902244..6d186d7 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -11,6 +11,12 @@ export { encode, FrameDecoder } from "./framing.js"; export { type Logger, McpManager, type McpManagerDeps } from "./manager.js"; export { adaptTool, flattenContent, namespace } from "./registry.js"; export { + MCP_CONNECT_TIMEOUT_MS, + MCP_DEFAULT_TIMEOUT_MS, + McpTimeoutError, + withTimeout, +} from "./timeout.js"; +export { type Connection, createStdioTransport, type SpawnedProcess, diff --git a/packages/mcp/src/manager.test.ts b/packages/mcp/src/manager.test.ts index 7cc53b8..d6179b7 100644 --- a/packages/mcp/src/manager.test.ts +++ b/packages/mcp/src/manager.test.ts @@ -213,4 +213,29 @@ describe("McpManager", () => { expect(manager.getClient("nonexistent")).toBeUndefined(); }); + + it("forwards the abort signal: a hanging initialize is interrupted", async () => { + // A connection whose initialize never resolves (a misbehaving / + // framing-incompatible server). Without a signal this hangs forever. + const hangingConn: Connection = { + send: () => new Promise(() => {}), + notify: () => {}, + onNotification: () => {}, + close: () => {}, + pid: 300, + }; + const manager = makeManager(() => { + return { connection: hangingConn, promise: Promise.resolve() }; + }); + + const controller = new AbortController(); + const connectPromise = manager.ensureConnected(testServer, "/tmp", controller.signal); + + // Abort mid-connect — the signal must reach initialize() and break it. + controller.abort(); + + await expect(connectPromise).rejects.toThrow(); + // The server is recorded as broken (not silently wedged). + expect(manager.status([testServer])[0].state).toBe("error"); + }); }); diff --git a/packages/mcp/src/manager.ts b/packages/mcp/src/manager.ts index 08ad389..1d719c6 100644 --- a/packages/mcp/src/manager.ts +++ b/packages/mcp/src/manager.ts @@ -104,7 +104,19 @@ export class McpManager { return results; } - async ensureConnected(server: ResolvedMcpServer, cwd: string): Promise<McpClient> { + /** + * Ensure a client for `server` is connected, lazily spawning + handshaking on + * first access. The optional `signal` (the turn's abort signal) is forwarded + * into the `initialize`/`listTools` handshake so `POST /conversations/:id/stop` + * can interrupt a stuck connect; the operations are independently bounded by + * their own default timeout, so a misbehaving server cannot hang a turn even + * when no signal is supplied. + */ + async ensureConnected( + server: ResolvedMcpServer, + cwd: string, + signal?: AbortSignal, + ): Promise<McpClient> { const existing = this.clients.get(server.id); if (existing && existing.client.getState() === "connected") { return existing.client; @@ -119,7 +131,7 @@ export class McpManager { this.broken.delete(server.id); } - await this.spawnClient(server, cwd); + await this.spawnClient(server, cwd, signal); const entry = this.clients.get(server.id); if (!entry) { throw new Error(`Failed to spawn MCP client for ${server.id}`); @@ -131,11 +143,15 @@ export class McpManager { return entry.client; } - private async spawnClient(server: ResolvedMcpServer, cwd: string): Promise<void> { + private async spawnClient( + server: ResolvedMcpServer, + cwd: string, + signal?: AbortSignal, + ): Promise<void> { const existingSpawn = this.spawning.get(server.id); if (existingSpawn) return existingSpawn; - const spawnPromise = this.doSpawn(server, cwd); + const spawnPromise = this.doSpawn(server, cwd, signal); this.spawning.set(server.id, spawnPromise); try { @@ -145,7 +161,11 @@ export class McpManager { } } - private async doSpawn(server: ResolvedMcpServer, cwd: string): Promise<void> { + private async doSpawn( + server: ResolvedMcpServer, + cwd: string, + signal?: AbortSignal, + ): Promise<void> { const { connection, promise } = this.connectionFactory(server, cwd); const client = new McpClient({ connection }); @@ -153,7 +173,7 @@ export class McpManager { const entry: ClientEntry = { client, server, - promise: this.initClient(client, server, promise), + promise: this.initClient(client, server, promise, signal), }; this.clients.set(server.id, entry); @@ -172,10 +192,11 @@ export class McpManager { client: McpClient, server: ResolvedMcpServer, _transportPromise: Promise<void>, + signal?: AbortSignal, ): Promise<void> { try { - await client.initialize(); - await client.listTools(); + await client.initialize(signal); + await client.listTools(signal); this.deps.logger?.info("MCP server connected", { serverId: server.id, toolCount: String(client.getTools().length), diff --git a/packages/mcp/src/rpc.test.ts b/packages/mcp/src/rpc.test.ts index dc295af..34b6af4 100644 --- a/packages/mcp/src/rpc.test.ts +++ b/packages/mcp/src/rpc.test.ts @@ -87,11 +87,10 @@ describe("JsonRpcClient", () => { client.request("c"); expect(written.length).toBe(3); - // Extract JSON body from Content-Length framed messages + // Outgoing messages are newline-delimited JSON (current MCP spec framing). const parse = (bytes: Uint8Array): { id: number } => { const text = new TextDecoder().decode(bytes); - const bodyStart = text.indexOf("\r\n\r\n") + 4; - return JSON.parse(text.slice(bodyStart)) as { id: number }; + return JSON.parse(text.replace(/\r?\n$/, "")) as { id: number }; }; const msg1 = parse(written[0]); const msg2 = parse(written[1]); diff --git a/packages/mcp/src/timeout.test.ts b/packages/mcp/src/timeout.test.ts new file mode 100644 index 0000000..38961d0 --- /dev/null +++ b/packages/mcp/src/timeout.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it } from "vitest"; +import { McpTimeoutError, withTimeout } from "./timeout.js"; + +/** + * `withTimeout` uses a single `setTimeout` (the only edge). Tests drive it + * deterministically via `AbortController` for the abort path, and use real + * (tiny) timers for the timeout path — vitest's async scheduler is fast enough + * at single-digit ms that this is stable. + */ + +describe("withTimeout", () => { + it("resolves with the underlying value when the promise settles first", async () => { + const result = await withTimeout(Promise.resolve("ok"), "initialize", 1000); + expect(result).toBe("ok"); + }); + + it("rejects with the underlying error when the promise rejects first", async () => { + await expect( + withTimeout(Promise.reject(new Error("boom")), "initialize", 1000), + ).rejects.toThrow("boom"); + }); + + it("rejects with McpTimeoutError when the timeout fires first", async () => { + const never = new Promise<string>(() => {}); // never settles + const p = withTimeout(never, "initialize", 20); + await expect(p).rejects.toBeInstanceOf(McpTimeoutError); + await expect(p).rejects.toMatchObject({ method: "initialize", timeoutMs: 20 }); + }); + + it("cleans up the timer after the promise settles (no leak)", async () => { + // If the timer were not cleared, vitest would report an unhandled timer; + // resolving quickly should leave nothing pending. + const result = await withTimeout(Promise.resolve(1), "tools/list", 50_000); + expect(result).toBe(1); + }); + + it("rejects with Error('Aborted') when the signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(); + await expect( + withTimeout(new Promise(() => {}), "initialize", 50_000, controller.signal), + ).rejects.toThrow("Aborted"); + }); + + it("rejects with Error('Aborted') when the signal aborts mid-flight", async () => { + const controller = new AbortController(); + const never = new Promise<string>(() => {}); + const p = withTimeout(never, "initialize", 50_000, controller.signal); + controller.abort(); + await expect(p).rejects.toThrow("Aborted"); + }); + + it("abort beats timeout (immediate cancellation)", async () => { + const controller = new AbortController(); + const never = new Promise<string>(() => {}); + const p = withTimeout(never, "initialize", 50_000, controller.signal); + controller.abort(); + await expect(p).rejects.toThrow("Aborted"); + }); + + it("passes through when timeout is disabled (0) and no signal", async () => { + const result = await withTimeout(Promise.resolve("passthrough"), "initialize", 0); + expect(result).toBe("passthrough"); + }); + + it("passes through when timeout is Infinity and no signal", async () => { + const result = await withTimeout(Promise.resolve(42), "tools/list", Number.POSITIVE_INFINITY); + expect(result).toBe(42); + }); + + it("still honors the signal when timeout is disabled", async () => { + const controller = new AbortController(); + const p = withTimeout(new Promise<string>(() => {}), "initialize", 0, controller.signal); + controller.abort(); + await expect(p).rejects.toThrow("Aborted"); + }); + + it("removes the abort listener after the promise settles", async () => { + const controller = new AbortController(); + const addSpy = controller.signal.addEventListener.bind(controller.signal); + let added = 0; + const spiedAdd = (...args: unknown[]) => { + added++; + return (addSpy as (...a: unknown[]) => void)(...args); + }; + controller.signal.addEventListener = spiedAdd as typeof controller.signal.addEventListener; + + await withTimeout(Promise.resolve("ok"), "initialize", 50_000, controller.signal); + // The resolve path cleans up: the listener was added then removed. The + // important assertion is that resolution works and no abort fires after. + expect(added).toBe(1); + // Aborting after settlement must NOT reject the already-settled promise. + controller.abort(); + }); +}); + +describe("McpTimeoutError", () => { + it("carries method + timeoutMs and a descriptive message", () => { + const err = new McpTimeoutError("initialize", 5000); + expect(err.method).toBe("initialize"); + expect(err.timeoutMs).toBe(5000); + expect(err.message).toBe("MCP initialize timed out after 5000ms"); + expect(err.name).toBe("McpTimeoutError"); + }); +}); diff --git a/packages/mcp/src/timeout.ts b/packages/mcp/src/timeout.ts new file mode 100644 index 0000000..294bfc5 --- /dev/null +++ b/packages/mcp/src/timeout.ts @@ -0,0 +1,114 @@ +/** + * Timeout + abort helper for MCP operations. + * + * A single misbehaving or framing-incompatible MCP server must never be able to + * hang an agent turn indefinitely: the JSON-RPC `initialize` / `tools/list` + * requests are awaited in-band during the per-turn tools filter, so a server + * that never responds would block the whole turn forever. `withTimeout` bounds + * any such awaited operation by BOTH a timeout (always) and an optional + * `AbortSignal` (so the turn's stop can interrupt an in-flight connect). + * + * Edge effect: uses `setTimeout` (the clock is the only I/O). The timer + the + * signal listener are always cleaned up on settlement, so a resolved operation + * never leaks a pending timer. The underlying promise ALWAYS has a handler + * attached (even when abort/timeout wins first), so it can never surface as an + * unhandled rejection. Mocking the OUTERMOST edge (real clock) is fine; tests + * drive this via `AbortController` (deterministic) rather than the timer. + */ + +/** Default per-operation timeout for MCP handshake/tool-list requests (ms). */ +export const MCP_DEFAULT_TIMEOUT_MS = 30_000; + +/** + * Backstop timeout bounding the ENTIRE per-turn MCP connect phase (spawn + + * initialize + listTools across all configured servers), applied by the tools + * filter. A misbehaving or framing-incompatible server cannot hang a turn + * longer than this; on expiry the filter degrades gracefully (skips MCP tools + * for that turn) instead of blocking the turn. Generous enough to absorb a + * legitimate slow server startup (e.g. a browser-launching MCP server). + */ +export const MCP_CONNECT_TIMEOUT_MS = 30_000; + +/** + * Raised when an MCP operation does not settle within its timeout. Distinct + * from a plain `Error` so callers (the manager's broken-state tracking, tests) + * can tell a timeout/incompatibility apart from a server-reported RPC error. + */ +export class McpTimeoutError extends Error { + readonly method: string; + readonly timeoutMs: number; + constructor(method: string, timeoutMs: number) { + super(`MCP ${method} timed out after ${timeoutMs}ms`); + this.name = "McpTimeoutError"; + this.method = method; + this.timeoutMs = timeoutMs; + } +} + +/** + * Race `promise` against a timeout (always) and an optional `AbortSignal`. + * Resolves/rejects with `promise`'s outcome if it settles first; rejects with + * `McpTimeoutError` on timeout, or `Error("Aborted")` if `signal` aborts first. + * + * @param method JSON-RPC method name (for the timeout message). + * @param timeoutMs Milliseconds before a timeout is raised. Pass `0` or + * `Infinity` to disable the timeout (only the `signal` then bounds the call). + * @param signal Optional abort signal — typically the turn's signal, so + * `POST /conversations/:id/stop` can interrupt a stuck connect immediately. + */ +export function withTimeout<T>( + promise: Promise<T>, + method: string, + timeoutMs: number, + signal?: AbortSignal, +): Promise<T> { + // No timeout and no signal → pass straight through (nothing to race). + const hasTimeout = timeoutMs > 0 && Number.isFinite(timeoutMs); + if (!hasTimeout && signal === undefined) { + return promise; + } + + return new Promise<T>((resolve, reject) => { + let settled = false; + + const finish = (action: () => void): void => { + if (settled) return; + settled = true; + cleanup(); + action(); + }; + + let timer: ReturnType<typeof setTimeout> | undefined; + const onAbort = (): void => { + finish(() => reject(new Error("Aborted"))); + }; + + const cleanup = (): void => { + if (timer !== undefined) clearTimeout(timer); + if (signal !== undefined) signal.removeEventListener("abort", onAbort); + }; + + if (hasTimeout) { + timer = setTimeout( + () => finish(() => reject(new McpTimeoutError(method, timeoutMs))), + timeoutMs, + ); + } + if (signal !== undefined) { + if (signal.aborted) { + // Already aborted: abort wins immediately. The `.then` below still + // attaches a handler so the underlying promise never rejects unhandled. + finish(() => reject(new Error("Aborted"))); + } else { + signal.addEventListener("abort", onAbort, { once: true }); + } + } + + // Always attach handlers so the underlying promise is never unhandled — + // even when abort/timeout already won (settled), this is a no-op. + promise.then( + (value) => finish(() => resolve(value)), + (err: unknown) => finish(() => reject(err)), + ); + }); +} diff --git a/packages/mcp/src/transport.test.ts b/packages/mcp/src/transport.test.ts index e828340..8c69ad2 100644 --- a/packages/mcp/src/transport.test.ts +++ b/packages/mcp/src/transport.test.ts @@ -62,7 +62,7 @@ describe("createStdioTransport", () => { connection.close(); }); - it("connection sends framed messages via stdin", () => { + it("connection sends newline-delimited messages via stdin (current MCP spec)", () => { const pair = makePipe(); const spawn: SpawnProcess = () => pair.process; @@ -73,8 +73,33 @@ describe("createStdioTransport", () => { const writes = pair.writtenToStdin(); expect(writes.length).toBe(1); const text = new TextDecoder().decode(writes[0]); - expect(text).toContain("Content-Length:"); + // Outgoing framing is newline-delimited JSON (not Content-Length). + expect(text).not.toContain("Content-Length:"); expect(text).toContain('"method":"test/method"'); + expect(text.endsWith("\n")).toBe(true); + connection.close(); + }); + + it("decodes a newline-delimited server response (auto-detect)", async () => { + const pair = makePipe(); + const spawn: SpawnProcess = () => pair.process; + + const { connection } = createStdioTransport({ spawn, command: ["test"] }, "/tmp"); + + const resultPromise = connection.send("tools/list"); + + // Server responds with newline-delimited JSON (e.g. chrome-devtools-mcp). + const response = `${JSON.stringify({ + jsonrpc: "2.0", + id: 1, + result: { tools: [{ name: "t", description: "d", inputSchema: { type: "object" } }] }, + })}\n`; + pair.emitStdout(new TextEncoder().encode(response)); + + const result = await resultPromise; + expect(result).toEqual({ + tools: [{ name: "t", description: "d", inputSchema: { type: "object" } }], + }); connection.close(); }); 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.test.ts b/packages/session-orchestrator/src/orchestrator.test.ts index 400ca0b..c4be03c 100644 --- a/packages/session-orchestrator/src/orchestrator.test.ts +++ b/packages/session-orchestrator/src/orchestrator.test.ts @@ -19,11 +19,14 @@ import type { TurnMetrics, } from "@dispatch/kernel"; import { createLogger, runTurn } from "@dispatch/kernel"; +import { createMessageQueueService } from "@dispatch/message-queue"; import type { SystemPromptService } from "@dispatch/system-prompt"; import { describe, expect, it } from "vitest"; import { + type ConversationCompactedPayload, type ConversationOpenedPayload, type ConversationStatusChangedPayload, + conversationCompacted, createCompactionService, createSessionOrchestrator, createWarmService, @@ -49,6 +52,7 @@ function createInMemoryStore(): ConversationStore & { const effortData = new Map<string, ReasoningEffort>(); const modelData = new Map<string, string>(); const workspaceIdData = new Map<string, string>(); + const compactPercentData = new Map<string, number>(); // Track conversations that have a meta row. In the real store, append, // setWorkspaceId, setConversationStatus, setConversationTitle, and // setCompactedFrom all create a minimal meta row on first contact. @@ -158,10 +162,12 @@ function createInMemoryStore(): ConversationStore & { knownConversations.add(conversationId); data.set(conversationId, [...messages]); }, - async getCompactPercent() { - return null; + async getCompactPercent(conversationId) { + return compactPercentData.get(conversationId) ?? null; + }, + async setCompactPercent(conversationId, percent) { + compactPercentData.set(conversationId, percent); }, - async setCompactPercent() {}, async forkHistory(_sourceId, targetId) { knownConversations.add(targetId); }, @@ -3873,6 +3879,194 @@ describe("system prompt: regular turn flow", () => { }); }); +describe("title (summon-title): deferred until after workspace initialization", () => { + // Regression: an earlier implementation set the title in the HTTP /chat + // route BEFORE the turn started, which pre-created the conversation meta + // and made the orchestrator's `meta === null` newness check falsely report + // an EXISTING conversation — so ensureWorkspace / setWorkspaceId / the + // first-turn system-prompt construct were ALL skipped. The fix defers the + // title set into workspaceSetupPromise, AFTER the newness check + workspace + // assignment, so a titled new conversation is still initialized correctly. + + /** Wrap the in-memory store to record the ORDER of init-relevant calls. */ + function createCallRecordingStore() { + const base = createInMemoryStore(); + const calls: string[] = []; + const titleCalls: { conversationId: string; title: string }[] = []; + return { + store: { + ...base, + async getConversationMeta(conversationId: string) { + calls.push(`getMeta:${conversationId}`); + return base.getConversationMeta(conversationId); + }, + async ensureWorkspace(id: string) { + calls.push(`ensureWorkspace:${id}`); + return base.ensureWorkspace(id); + }, + async setWorkspaceId(conversationId: string, workspaceId: string) { + calls.push(`setWorkspaceId:${workspaceId}`); + await base.setWorkspaceId(conversationId, workspaceId); + }, + async setConversationTitle(conversationId: string, title: string) { + calls.push(`setTitle:${title}`); + titleCalls.push({ conversationId, title }); + await base.setConversationTitle(conversationId, title); + }, + } as ConversationStore, + calls, + titleCalls, + }; + } + + it("titled new conversation: workspace assigned, system prompt constructed, title set", async () => { + const { store, calls, titleCalls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + const constructCalls: string[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + resolveSystemPrompt: () => + createFakeSystemPromptService(async (conversationId) => { + constructCalls.push(conversationId); + return "CONSTRUCTED_PROMPT"; + }), + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-new", + text: "hi", + onEvent: () => {}, + title: "My Task", + workspaceId: "my-workspace", + }); + + // The bug: workspace init was skipped. It must NOT be. + expect(calls).toContain("ensureWorkspace:my-workspace"); + expect(calls).toContain("setWorkspaceId:my-workspace"); + // First-turn system prompt construct runs (proves isNewConversation was + // true — the newness check was not fooled by a pre-created meta). + expect(constructCalls).toEqual(["conv-title-new"]); + // The title is persisted. + expect(titleCalls).toEqual([{ conversationId: "conv-title-new", title: "My Task" }]); + }); + + it("title is set AFTER the newness check + workspace assignment (order)", async () => { + const { store, calls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-order", + text: "hi", + onEvent: () => {}, + title: "Ordered", + }); + + const getMetaIdx = calls.findIndex((c) => c.startsWith("getMeta:")); + const ensureIdx = calls.findIndex((c) => c.startsWith("ensureWorkspace:")); + const setWsIdx = calls.findIndex((c) => c.startsWith("setWorkspaceId:")); + const setTitleIdx = calls.findIndex((c) => c.startsWith("setTitle:")); + expect(getMetaIdx).toBeGreaterThanOrEqual(0); + expect(ensureIdx).toBeGreaterThan(getMetaIdx); + expect(setWsIdx).toBeGreaterThan(ensureIdx); + expect(setTitleIdx).toBeGreaterThan(setWsIdx); + }); + + it("no title: setConversationTitle is not called", async () => { + const { store, calls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-no-title", + text: "hi", + onEvent: () => {}, + }); + + expect(calls.some((c) => c.startsWith("setTitle:"))).toBe(false); + }); + + it("existing conversation with a title: workspace NOT re-assigned, title still set", async () => { + const { store, calls, titleCalls } = createCallRecordingStore(); + // Seed an existing conversation (meta non-null, workspace already set). + await store.setWorkspaceId("conv-title-existing", "prior-workspace"); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-existing", + text: "hi", + onEvent: () => {}, + title: "Renamed", + }); + + // Existing conversation: workspace init must not run again. + expect(calls.some((c) => c.startsWith("ensureWorkspace:"))).toBe(false); + // But the title is still applied (rename on an existing conversation). + expect(titleCalls).toEqual([{ conversationId: "conv-title-existing", title: "Renamed" }]); + }); + + it("turn still completes if setConversationTitle throws", async () => { + const base = createInMemoryStore(); + const store: ConversationStore = { + ...base, + async setConversationTitle() { + throw new Error("title store unavailable"); + }, + }; + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captured, captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-throws", + text: "hi", + onEvent: () => {}, + title: "Resilient", + }); + + // The turn ran despite the title-set failure. + expect(captured).toHaveLength(1); + }); +}); + describe("system prompt: compaction flow", () => { function seedHistory( store: ReturnType<typeof createInMemoryStore>, @@ -3985,6 +4179,484 @@ describe("system prompt: compaction flow", () => { }); }); +describe("in-flight compaction", () => { + // Seeds a conversation with `count` alternating user/assistant text messages + // so the history is long enough to compact (> DEFAULT_KEEP_LAST_N = 10). + function seedHistory( + store: ReturnType<typeof createInMemoryStore>, + conversationId: string, + count: number, + ): void { + const messages: ChatMessage[] = []; + for (let i = 0; i < count; i++) { + messages.push({ + role: i % 2 === 0 ? "user" : "assistant", + chunks: [{ type: "text", text: `seed message ${i}` }], + }); + } + store.data.set(conversationId, messages); + } + + // A provider whose `stream` serves a SCRIPT of per-call event lists, in + // order. Captures the messages passed to each call so a test can assert what + // the model saw at each step (incl. after in-flight compaction replaced it). + function createScriptedCapturingProvider(script: ProviderEvent[][]): { + provider: ProviderContract; + capturedMessages: ChatMessage[][]; + } { + const capturedMessages: ChatMessage[][] = []; + let callIndex = 0; + const provider: ProviderContract = { + id: "fake", + stream(messages) { + capturedMessages.push([...messages]); + const events = script[callIndex] ?? []; + callIndex++; + return (async function* () { + for (const event of events) { + yield event; + } + })(); + }, + }; + return { provider, capturedMessages }; + } + + function echoTool(): ToolContract { + return { + name: "echo", + description: "echo", + parameters: { type: "object" }, + execute: async () => ({ content: "echoed" }), + }; + } + + it("triggers when a step's usage exceeds the threshold: history is compacted mid-turn and the prompt continues with the summary", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-inflight", 15); // > keepLastN(10) → compactable + + // contextWindow 1000, default percent 85 → threshold 850. + // Step 0 emits a tool call + usage(inputTokens 900) → 910 > 850 → trigger. + // Then the compaction summary call, then step 1 ends the turn. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 900, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + // Compaction summary call (performCompaction): + [ + { type: "text-delta", delta: "COMPACTED SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + // Step 1 (post-compaction) — the turn CONTINUES: + [ + { type: "text-delta", delta: "all done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + const { events, onEvent } = collectEvents(); + + await orchestrator.handleMessage({ + conversationId: "conv-inflight", + text: "keep working overnight", + onEvent, + modelName: "test/model", + }); + + // 1) The conversationCompacted event fired mid-turn. + expect(compactedEvents).toHaveLength(1); + expect(compactedEvents[0]?.conversationId).toBe("conv-inflight"); + expect(compactedEvents[0]?.messagesSummarized).toBeGreaterThan(0); + expect(compactedEvents[0]?.messagesKept).toBe(10); + + // 2) The store history was replaced: it now begins with the system summary + // message, and the OLDEST seed messages are gone (summarized). The most + // recent messages are retained (keepLastN = 10), so some later seed + // messages may survive — that is correct. + const stored = store.data.get("conv-inflight") ?? []; + expect(stored.length).toBeGreaterThan(0); + expect(stored[0]?.role).toBe("system"); + expect(stored[0]?.chunks[0]).toMatchObject({ type: "text" }); + const firstText = (stored[0]?.chunks[0] as { text: string } | undefined)?.text ?? ""; + expect(firstText).toContain("COMPACTED SUMMARY"); + // The earliest seed messages were summarized away (not retained). + expect( + stored.some((m) => m.chunks.some((c) => c.type === "text" && c.text === "seed message 0")), + ).toBe(false); + + // 3) The turn CONTINUED after compaction: 3 provider calls happened + // (step 0, compaction summary, step 1) and the final assistant text + // was produced + persisted. + expect(capturedMessages).toHaveLength(3); + const step1Messages = capturedMessages[2] ?? []; + // Step 1 saw the COMPACTED history: it must start with the summary + // system message, NOT the original seed/user prefix. + expect(step1Messages[0]?.role).toBe("system"); + + const turnSealed = events.some((e) => e.type === "turn-sealed"); + expect(turnSealed).toBe(true); + }); + + it("does NOT trigger when the step usage is below the threshold (history unchanged, no event)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-below", 15); + + // contextWindow 1000 → threshold 850. Step usage 100 < 850 → no trigger. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 100, outputTokens: 5 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-below", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + // No compaction event, and only 2 provider calls (no summary call). + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(2); + // The seed messages are still the start of history (uncompacted). + const stored = store.data.get("conv-below") ?? []; + expect(stored[0]?.chunks[0]).toMatchObject({ type: "text", text: "seed message 0" }); + }); + + it("does NOT trigger when auto-compact is disabled (compact percent = 0)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-disabled", 15); + await store.setCompactPercent("conv-disabled", 0); + + // Usage would exceed the default threshold, but percent=0 disables it. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 950, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-disabled", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(2); + }); + + it("does NOT trigger on a text-only turn (no tool calls → no next step → no boundary)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-textonly", 15); + + // Single text-only step with high usage — but no tool calls → the turn + // ends → there is no step boundary to compact at (post-seal handles it). + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "text-delta", delta: "final answer" }, + { type: "usage", usage: { inputTokens: 950, outputTokens: 10 } }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-textonly", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + // No in-flight compaction (only 1 provider call — the turn ended). + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(1); + }); + + it("the manual compaction SERVICE still refuses while a conversation is generating, but in-flight compaction runs anyway", async () => { + // This documents the two-path design: compact() (the service) guards on + // activeConversations and refuses mid-turn; the in-flight path bypasses + // that guard (it IS the mid-turn path) using performCompaction directly. + const store = createInMemoryStore(); + seedHistory(store, "conv-twopath", 15); + + const { provider } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 900, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + const activeConversations = new Set<string>(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + // The compaction SERVICE shares the orchestrator's activeConversations set; + // build it against the SAME set so the guard reflects reality. + const compactionService = createCompactionService( + { + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn, + emit: () => {}, + }, + activeConversations, + ); + + // Drive a turn that triggers in-flight compaction. We can't easily inspect + // activeConversations mid-turn, so we assert the observable contract: + // in-flight compaction produced an event (it ran WHILE active), and the + // store was compacted. + await orchestrator.handleMessage({ + conversationId: "conv-twopath", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + expect(compactedEvents).toHaveLength(1); + const stored = store.data.get("conv-twopath") ?? []; + expect(stored[0]?.role).toBe("system"); + + // After the turn settles (idle), the manual service CAN compact (no longer + // active) — and it succeeds (history is compactable again only if long + // enough; here it is short post-compaction, so it reports too-short, which + // proves the service path is reachable and its guard is the ONLY reason it + // would have refused mid-turn). + const manual = await compactionService.compact("conv-twopath"); + // Post-compaction the history is short (summary + ~10 + turn tail) → the + // service reports an error (too short / threshold), NOT "generating". + expect("error" in manual).toBe(true); + if ("error" in manual) { + expect(manual.error).not.toBe("conversation is generating"); + } + }); + + it("steering messages are persisted and survive in-flight compaction — the store and the LLM's context stay aligned (Bug A + B)", async () => { + // Regression test for the two critical bugs: + // A) drainSteering injected steering into the kernel's in-memory messages + // but never persisted it → the user could never see it, and + // compaction (loading the store) scrubbed it. + // B) compaction sliced the store and the kernel's messages independently; + // the unpersisted steering offset the slices → DB and LLM dropped + // DIFFERENT messages (structural divergence). + // Fix: drainSteering persists (awaited); compaction uses the kernel's LIVE + // messages array, so the store write and the kernel's replacement use the + // SAME recent slice → aligned, and the steering is retained. + const store = createInMemoryStore(); + seedHistory(store, "conv-align", 15); // > keepLastN(10) → compactable + const queue = createMessageQueueService({ + id: () => `q-${Math.random().toString(36).slice(2, 8)}`, + now: () => 1000, + notify: () => {}, + }); + queue.enqueue("conv-align", "STEER MID-TURN"); // drained at step 0's boundary + + // contextWindow 1000, percent 85 → threshold 850. Step 0 usage 900 → fire. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 900, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + // Compaction summary call: + [ + { type: "text-delta", delta: "ALIGN SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + // Step 1 (post-compaction) — the turn continues: + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + resolveQueue: () => queue, + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-align", + text: "keep working overnight", + onEvent: () => {}, + modelName: "test/model", + }); + + // Bug A: the steering message was PERSISTED to the store (the user CAN see + // it). It is either retained in the kept recent slice or captured in the + // summary; either way it must be present in the store, not lost. + expect(compactedEvents).toHaveLength(1); + const stored = store.data.get("conv-align") ?? []; + // The compacted history begins with the summary system message. + expect(stored[0]?.role).toBe("system"); + expect((stored[0]?.chunks[0] as { text?: string } | undefined)?.text).toContain( + "ALIGN SUMMARY", + ); + // The steering message survived in the kept recent slice (it was the most + // recent user message before compaction, so it is within keepLastN=10). + const storedSteering = stored.find( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(storedSteering).toBeDefined(); + + // Bug B: the store and the LLM's context are ALIGNED. The provider's + // post-compaction call (captured[2]) is the kernel's working history AFTER + // compaction replaced it. The store was written with the SAME compacted + // history, then step 1's assistant output was appended on top. So the + // kernel's view (captured[2]) must be an exact PREFIX of the store — same + // summary, same recent slice, same steering at the same index. (Before the + // fix, the store dropped the steering while the kernel kept it, so the two + // diverged structurally.) + const step1Messages = capturedMessages[2] ?? []; + expect(step1Messages[0]?.role).toBe("system"); // summary heads both + const kernelSteering = step1Messages.find( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(kernelSteering).toBeDefined(); + // The kernel's post-compaction history is an exact PREFIX of the store + // (the store then has step 1's appended assistant output after it). Same + // length, same roles in order, same steering index => structural alignment. + expect(stored.length).toBeGreaterThanOrEqual(step1Messages.length); + const storePrefix = stored.slice(0, step1Messages.length); + expect(storePrefix).toHaveLength(step1Messages.length); + for (let i = 0; i < step1Messages.length; i++) { + expect(storePrefix[i]?.role).toBe(step1Messages[i]?.role); + } + const storeSteerIdx = storePrefix.findIndex( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + const kernelSteerIdx = step1Messages.findIndex( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(kernelSteerIdx).toBe(storeSteerIdx); + expect(kernelSteerIdx).toBeGreaterThanOrEqual(0); + }); +}); + describe("per-turn memory telemetry", () => { function capturingLogger(): { logger: Logger; records: LogRecord[] } { let id = 0; diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts index aaf418a..a2e141a 100644 --- a/packages/session-orchestrator/src/orchestrator.ts +++ b/packages/session-orchestrator/src/orchestrator.ts @@ -131,6 +131,16 @@ export interface StartTurnInput { * system-prompt service when loaded). */ readonly systemPrompt?: string; + /** + * A human-readable title for the conversation tab. When provided, it is + * persisted via `setConversationTitle` AFTER the new-conversation workspace + * setup resolves (so the `meta === null` newness detection still fires and + * `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + * construction are NOT skipped) and BEFORE the first message append (so the + * append's auto-title does not overwrite it). Omit to keep the auto-derived + * title. The caller is responsible for trimming/validation. + */ + readonly title?: string; } export type StartTurnResult = @@ -170,6 +180,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 +355,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; /** @@ -370,6 +406,8 @@ export interface SessionOrchestrator { systemPrompt?: string; /** Images attached to this turn — see {@link StartTurnInput.images}. */ images?: readonly ImageInput[]; + /** Conversation tab title — see {@link StartTurnInput.title}. */ + title?: string; }): Promise<void>; } @@ -548,6 +586,7 @@ export function createSessionOrchestrator( workspaceId: string, systemPromptOverride: string | undefined, images: readonly ImageInput[] | undefined, + title: string | undefined, ): void { const turnId = generateTurnId(); const promptStartedAt = deps.now?.() ?? Date.now(); @@ -567,14 +606,34 @@ export function createSessionOrchestrator( // The newness flag is also reused to decide whether to construct // (first turn) or get (subsequent turn) the system prompt — see the // providerOpts assembly below. + // + // An explicit `title` (e.g. the CLI `--title` flag) is persisted HERE, + // AFTER the workspace setup resolves — deliberately NOT before the turn. + // Setting it earlier (e.g. in the HTTP route) would pre-create the meta + // row, make `meta !== null`, and fool this newness check into skipping + // `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + // construction. By deferring it to here, the title lands after the + // workspace is assigned but BEFORE the first message append (so the + // append's auto-title sees a non-"Untitled" title and preserves it). const workspaceSetupPromise = (async (): Promise<boolean> => { const meta = await deps.conversationStore.getConversationMeta(conversationId); if (meta === null) { await deps.conversationStore.ensureWorkspace(workspaceId); await deps.conversationStore.setWorkspaceId(conversationId, workspaceId); - return true; } - return false; + if (title !== undefined) { + // Best-effort: a title-set failure must NOT break the turn (the + // workspace setup above already succeeded). Log and continue — the + // append's auto-derived title applies instead. + try { + await deps.conversationStore.setConversationTitle(conversationId, title); + } catch (err) { + deps.logger?.child({ conversationId }).warn("orchestrator: title set failure", { + error: err instanceof Error ? err.message : String(err), + }); + } + } + return meta === null; })(); // ALWAYS resolve the effective cwd through getEffectiveCwd, passing the @@ -760,6 +819,11 @@ export function createSessionOrchestrator( conversationId, ...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}), ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), + // Thread the turn's abort signal into the filter chain so a filter + // awaiting slow I/O (the MCP tools filter connecting to MCP servers) + // can be interrupted by POST /conversations/:id/stop instead of + // blocking the turn until its own timeout fires. + signal: controller.signal, }); const dispatch = deps.resolveDispatch?.() ?? defaultDispatchPolicy(); const turnLogger = deps.logger?.child({ conversationId, turnId }); @@ -844,17 +908,33 @@ export function createSessionOrchestrator( const drainSteering = queue === undefined ? undefined - : (): readonly ChatMessage[] => { + : async (): Promise<readonly ChatMessage[]> => { const queued = queue.drain(conversationId); if (queued.length === 0) return []; const steerText = queued.map((q) => q.text).join("\n\n"); + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: steerText }], + }; + // Persist the injected steering message to the store as part + // of the SAME critical section as the injection, so it is + // never lost. Without this, the message would live only in + // the kernel's in-memory messages array (never persisted), + // so a user could never see it — and in-flight compaction + // (which loads the store) would scrub it. A fire-and-forget + // append would race with the next step's `onStepComplete` + // append and collide on the store's seq counter, so we + // `await` it (the kernel awaits drainSteering). Errors + // propagate (a DB failure ends the turn, matching + // `onStepComplete`'s behavior). + await deps.conversationStore.append(conversationId, [steeringMessage]); emitToHub(conversationId, { type: "steering", conversationId, turnId, text: steerText, }); - return [{ role: "user", chunks: [{ type: "text", text: steerText }] }]; + return [steeringMessage]; }; // Vision handoff: transform the message list for the provider. When the @@ -899,6 +979,81 @@ export function createSessionOrchestrator( ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), ...(deps.now !== undefined ? { now: deps.now } : {}), ...(drainSteering !== undefined ? { drainSteering } : {}), + // In-flight compaction: at every tool-result boundary the kernel + // calls this with the step's usage + the running messages. When the + // context size exceeds the compact-percent threshold (percent of the + // model's context window), the old history is summarized and + // replaced with [summary, ...recent] — mid-turn, without stopping — + // so a long-running turn (e.g. left overnight) does not run out of + // context. This is DISTINCT from the post-seal auto-compact below: + // that one runs AFTER the turn ends (preparing the next turn) and + // refuses while the conversation is active; this one runs DURING the + // turn (saving the running turn) and uses the live step usage (not + // persisted metrics, which are only written at turn end). When the + // threshold is not exceeded, compaction is disabled (percent 0), or + // the model's context window is unknown, it returns void and the + // kernel keeps its history unchanged (a strict no-op). + onStepBoundary: async ({ stepUsage, messages }) => { + const stored = await deps.conversationStore.getCompactPercent(conversationId); + const percent = stored ?? DEFAULT_COMPACT_PERCENT; + if (percent <= 0) return; // auto-compact disabled + // contextSize mirrors the persisted definition: this step's + // inputTokens + outputTokens (the prompt the NEXT step would + // inherit, grown by this step's output). + const contextSize = stepUsage.inputTokens + stepUsage.outputTokens; + if (effectiveModelName === undefined || deps.resolveModelInfo === undefined) return; + const info = await deps.resolveModelInfo(effectiveModelName); + if (info?.contextWindow === undefined) return; + const threshold = Math.floor(info.contextWindow * (percent / 100)); + if (contextSize < threshold) return; // threshold not exceeded + + const keepLastN = DEFAULT_KEEP_LAST_N; + turnLogger?.info("compaction:in-flight", { + conversationId, + turnId, + contextSize, + threshold, + percent, + }); + const outcome = await performCompaction( + { + conversationStore: deps.conversationStore, + resolveProvider: deps.resolveProvider, + ...(deps.resolveModel !== undefined ? { resolveModel: deps.resolveModel } : {}), + ...(deps.resolveSystemPrompt !== undefined + ? { resolveSystemPrompt: deps.resolveSystemPrompt } + : {}), + ...(deps.resolveConcurrencyLimiter !== undefined + ? { resolveConcurrencyLimiter: deps.resolveConcurrencyLimiter } + : {}), + ...(deps.logger !== undefined ? { logger: deps.logger } : {}), + ...(deps.now !== undefined ? { now: deps.now } : {}), + // emit is required by performCompaction; fall back to a no-op + // when the orchestrator was constructed without one (tests). + emit: deps.emit ?? noopEmit, + }, + conversationId, + // Pass the kernel's LIVE messages array (not a store reload): + // it includes mid-turn steering messages (now persisted by + // drainSteering) and is the authoritative prompt state. Using it + // for the split keeps the store write and the kernel's + // replacement aligned (same recent slice) — no DB↔LLM divergence. + { keepLastN, modelName: effectiveModelName, messages }, + ); + if ("error" in outcome) { + turnLogger?.warn("compaction:in-flight:skipped", { + conversationId, + turnId, + error: outcome.error, + }); + return; // too short / empty summary / unknown model → no replacement + } + // Return the compacted history the kernel should adopt. This is + // EXACTLY what performCompaction wrote to the store + // ([summary, ...recent-from-live]), so the kernel's working + // history and the store stay byte-aligned. + return outcome.compactedMessages; + }, }; // Persist the user message at turn start so it has a seq @@ -1021,6 +1176,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { if (activeTurns.has(conversationId)) { return { started: false, reason: "already-active" }; @@ -1035,6 +1191,7 @@ export function createSessionOrchestrator( workspaceId ?? "default", systemPrompt, images, + title, ); const turn = activeTurns.get(conversationId); const turnId = turn !== undefined ? turn.turnId : ""; @@ -1060,6 +1217,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) { @@ -1140,6 +1310,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { const turnInput: StartTurnInput = { conversationId, @@ -1151,6 +1322,7 @@ export function createSessionOrchestrator( ...(workspaceId !== undefined ? { workspaceId } : {}), ...(systemPrompt !== undefined ? { systemPrompt } : {}), ...(images !== undefined ? { images } : {}), + ...(title !== undefined ? { title } : {}), }; const result = orchestrator.startTurn(turnInput); if (!result.started) { @@ -1296,6 +1468,16 @@ export function createWarmService( const DEFAULT_KEEP_LAST_N = 10; const DEFAULT_COMPACT_PERCENT = 85; +/** + * No-op emit used as a fallback when the orchestrator is constructed without an + * `emit` (some tests). `performCompaction` requires a non-optional `emit` (it + * emits `conversationCompacted`); the in-flight path degrades to emitting + * nothing rather than skipping compaction entirely. Generic-typed so it + * satisfies `PerformCompactionDeps["emit"]` for any hook payload type. + */ +const noopEmit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void = + () => {}; + const COMPACTION_SYSTEM_PROMPT = "You are a conversation summarizer. Summarize the following conversation concord concisely but comprehensively. " + "Focus on key decisions, context, file paths, and any unresolved questions. " + @@ -1317,6 +1499,222 @@ function formatMessagesForSummary(messages: readonly ChatMessage[]): string { .join("\n\n"); } +/** + * Deps for {@link performCompaction} — the subset of `SessionOrchestratorDeps` + * needed to summarize old history, fork it to an archive, and replace it with + * a summary + recent messages. Structural so both the compaction SERVICE + * (`compact`, manual + post-seal auto) and the IN-FLIGHT compaction path (the + * turn loop's `onStepBoundary`) can call the same shared core without duplicating + * the summarization/fork/replace/emit logic. The active-conversation guard and + * the threshold check are the CALLERS' policy (they differ between the two + * paths) and are NOT performed here. + */ +interface PerformCompactionDeps { + readonly conversationStore: ConversationStore; + readonly resolveProvider: () => ProviderContract; + readonly resolveModel?: ( + modelName: string, + ) => { provider: ProviderContract; model: string } | undefined; + readonly resolveSystemPrompt?: () => SystemPromptService | undefined; + readonly resolveConcurrencyLimiter?: () => ConcurrencyLimiter | undefined; + readonly logger?: Logger; + readonly now?: () => number; + readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; +} + +/** Result of a successful {@link performCompaction}. */ +interface PerformCompactionResult { + readonly summary: string; + readonly newConversationId: string; + readonly messagesSummarized: number; + readonly messagesKept: number; + /** + * The system-role summary message that heads the compacted history + * (`[summaryMessage, ...recentKept]`). Returned so the in-flight caller can + * build the kernel's replacement history with the SAME summary object. + */ + readonly summaryMessage: ChatMessage; + /** + * The full compacted history `[summaryMessage, ...recentKept]` exactly as + * written to the store. The in-flight caller returns this to the kernel so + * the kernel's working history and the store stay byte-aligned (the same + * `recentKept` slice — taken from the caller-supplied live `messages` — is + * used for BOTH the store write and this return value). + */ + readonly compactedMessages: readonly ChatMessage[]; +} + +/** + * The shared compaction core: summarize the oldest `history.length - + * keepLastN` messages via a provider stream, fork the full pre-compaction + * history to an archive (non-destructive), and replace the live history with + * `[summaryMessage, ...recentKept]`. Emits `conversationCompacted`. Returns the + * result (incl. the `summaryMessage` + the `compactedMessages`) or an error. + * + * History source: when `opts.messages` is provided (the in-flight path), it is + * used as the authoritative history — this is the kernel's LIVE messages array, + * which includes mid-turn steering messages (and the vision-transformed + * provider view) that a store reload could miss (the steering persist may not + * have completed, or — before this fix — was never done at all). Using the live + * array keeps the store write and the kernel's replacement aligned (same + * `recentKept` slice), avoiding the DB↔LLM structural divergence where + * independent slices dropped different messages. When `opts.messages` is + * omitted (the post-seal/manual `compact()` path — the turn has ended, so the + * store is stable), the history is loaded from the store. + * + * Performs NO active-conversation guard and NO threshold check — those are the + * callers' policy. No-ops (returns an error) when the conversation is too + * short to compact (≤ keepLastN messages) or the summary is empty. + */ +async function performCompaction( + deps: PerformCompactionDeps, + conversationId: string, + opts: { + readonly keepLastN?: number; + readonly modelName?: string; + /** The kernel's live messages array (in-flight path). Omit to load the store (post-seal/manual). */ + readonly messages?: readonly ChatMessage[]; + }, +): Promise<PerformCompactionResult | { readonly error: string }> { + // Use the caller-supplied live messages (in-flight) or load the store + // (post-seal/manual — the store is stable once the turn has ended). + const history = opts.messages ?? (await deps.conversationStore.load(conversationId)); + const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; + + if (history.length <= keepLastN) { + return { error: "conversation too short to compact" }; + } + + // Split: old messages to summarize + recent messages to keep. + const toSummarize = history.slice(0, history.length - keepLastN); + const toKeep = history.slice(history.length - keepLastN); + + // Resolve provider + let provider: ProviderContract; + let modelOverride: string | undefined; + if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { + const resolved = deps.resolveModel(opts.modelName); + if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; + provider = resolved.provider; + modelOverride = resolved.model; + } else { + provider = deps.resolveProvider(); + } + + // Wrap with concurrency limiting (same as the main turn path). + const compactionLimiter = deps.resolveConcurrencyLimiter?.(); + if (compactionLimiter !== undefined) { + const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + provider = wrapProviderWithConcurrency( + provider, + compactionLimiter, + conversationId, + compactionWorkspaceId, + deps.now?.() ?? Date.now(), + ); + } + + // Build the summarization request: system prompt + conversation text + instruction + const conversationText = formatMessagesForSummary(toSummarize); + const summaryRequest: ChatMessage = { + role: "user", + chunks: [ + { + type: "text", + text: `Please summarize the following conversation:\n\n${conversationText}`, + }, + ], + }; + + const providerOpts: ProviderStreamOptions = { + maxTokens: 2000, + ...(modelOverride !== undefined ? { model: modelOverride } : {}), + ...(deps.logger !== undefined + ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } + : {}), + }; + + // Reconstruct the system prompt on compaction (fresh variable + // resolution — files/cwd/time may have changed since construction). + // The construct call also persists the result for future turns. When + // the system-prompt service is unavailable, fall back to the + // compaction-only system prompt (current behavior, no regression). + const systemPromptService = deps.resolveSystemPrompt?.(); + let compactionSystemPrompt: string; + if (systemPromptService !== undefined) { + const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); + const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); + const constructed = await systemPromptService.construct(conversationId, cwd, { + ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), + workspaceId, + ...(computerId !== null ? { computerId } : {}), + }); + compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; + } else { + compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; + } + + // Call the provider and accumulate the summary + let summary = ""; + for await (const event of provider.stream([summaryRequest], [], { + ...providerOpts, + systemPrompt: compactionSystemPrompt, + })) { + if ((event as ProviderEvent).type === "text-delta") { + summary += (event as { delta: string }).delta; + } else if ((event as ProviderEvent).type === "error") { + return { error: (event as { message: string }).message }; + } + } + + if (summary.trim().length === 0) { + return { error: "model produced empty summary" }; + } + + // Non-destructive: fork the full pre-compaction history to a new + // archive conversation. The original conversation keeps its ID + // (so messaging between agents still works) and gets the compacted + // content. The archive inherits the original's compactedFrom, + // creating a chain: A → Y → X → ... + const archiveId = crypto.randomUUID(); + await deps.conversationStore.forkHistory(conversationId, archiveId); + + // Replace history: [system: summary] + the recent kept messages. `toKeep` + // is sliced from the caller-supplied live `messages` (in-flight) — the SAME + // slice returned below as `compactedMessages` — so the store and the kernel's + // working history stay byte-aligned (same messages kept/dropped). + const summaryMessage: ChatMessage = { + role: "system", + chunks: [ + { + type: "text", + text: `The following is a summary of the previous conversation:\n\n${summary}`, + }, + ], + }; + + const compactedMessages: readonly ChatMessage[] = [summaryMessage, ...toKeep]; + await deps.conversationStore.replaceHistory(conversationId, compactedMessages); + await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + + deps.emit(conversationCompacted, { + conversationId, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + }); + + return { + summary, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + summaryMessage, + compactedMessages, + }; +} + export function createCompactionService( deps: SessionOrchestratorDeps & { readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; @@ -1329,14 +1727,9 @@ export function createCompactionService( return { error: "conversation is generating" }; } - const history = await deps.conversationStore.load(conversationId); - const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; - - if (history.length <= keepLastN) { - return { error: "conversation too short to compact" }; - } - // Auto mode: check if contextSize exceeds percent of contextWindow. + // The threshold check is the caller's policy (uses persisted turn + // metrics) and is NOT performed by the shared `performCompaction` core. if (opts?.auto === true) { const stored = await deps.conversationStore.getCompactPercent(conversationId); const percent = stored ?? DEFAULT_COMPACT_PERCENT; @@ -1360,129 +1753,22 @@ export function createCompactionService( if (contextSize < threshold) return { error: "threshold not exceeded" }; } - // Split: old messages to summarize + recent messages to keep. - const toSummarize = history.slice(0, history.length - keepLastN); - const toKeep = history.slice(history.length - keepLastN); - - // Resolve provider - let provider: ProviderContract; - let modelOverride: string | undefined; - if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { - const resolved = deps.resolveModel(opts.modelName); - if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; - provider = resolved.provider; - modelOverride = resolved.model; - } else { - provider = deps.resolveProvider(); - } - - // Wrap with concurrency limiting (same as the main turn path). - const compactionLimiter = deps.resolveConcurrencyLimiter?.(); - if (compactionLimiter !== undefined) { - const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - provider = wrapProviderWithConcurrency( - provider, - compactionLimiter, - conversationId, - compactionWorkspaceId, - deps.now?.() ?? Date.now(), - ); - } - - // Build the summarization request: system prompt + conversation text + instruction - const conversationText = formatMessagesForSummary(toSummarize); - const summaryRequest: ChatMessage = { - role: "user", - chunks: [ - { - type: "text", - text: `Please summarize the following conversation:\n\n${conversationText}`, - }, - ], - }; - - const providerOpts: ProviderStreamOptions = { - maxTokens: 2000, - ...(modelOverride !== undefined ? { model: modelOverride } : {}), - ...(deps.logger !== undefined - ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } - : {}), - }; - - // Reconstruct the system prompt on compaction (fresh variable - // resolution — files/cwd/time may have changed since construction). - // The construct call also persists the result for future turns. When - // the system-prompt service is unavailable, fall back to the - // compaction-only system prompt (current behavior, no regression). - const systemPromptService = deps.resolveSystemPrompt?.(); - let compactionSystemPrompt: string; - if (systemPromptService !== undefined) { - const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); - const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); - const constructed = await systemPromptService.construct(conversationId, cwd, { - ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), - workspaceId, - ...(computerId !== null ? { computerId } : {}), - }); - compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; - } else { - compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; - } - - // Call the provider and accumulate the summary - let summary = ""; - for await (const event of provider.stream([summaryRequest], [], { - ...providerOpts, - systemPrompt: compactionSystemPrompt, - })) { - if ((event as ProviderEvent).type === "text-delta") { - summary += (event as { delta: string }).delta; - } else if ((event as ProviderEvent).type === "error") { - return { error: (event as { message: string }).message }; - } - } - - if (summary.trim().length === 0) { - return { error: "model produced empty summary" }; - } - - // Non-destructive: fork the full pre-compaction history to a new - // archive conversation. The original conversation keeps its ID - // (so messaging between agents still works) and gets the compacted - // content. The archive inherits the original's compactedFrom, - // creating a chain: A → Y → X → ... - const archiveId = crypto.randomUUID(); - await deps.conversationStore.forkHistory(conversationId, archiveId); - - // Replace history: [system: summary] + recent messages - const summaryMessage: ChatMessage = { - role: "system", - chunks: [ - { - type: "text", - text: `The following is a summary of the previous conversation:\n\n${summary}`, - }, - ], - }; - - await deps.conversationStore.replaceHistory(conversationId, [summaryMessage, ...toKeep]); - await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + // Shared summarize + fork + replace + emit core (no active guard, no + // threshold — those are this caller's policy above). The length check + // ("conversation too short to compact") lives inside the core. + const outcome = await performCompaction(deps, conversationId, { + ...(opts?.keepLastN !== undefined ? { keepLastN: opts.keepLastN } : {}), + ...(opts?.modelName !== undefined ? { modelName: opts.modelName } : {}), + }); + if ("error" in outcome) return { error: outcome.error }; + const { summary, newConversationId, messagesSummarized, messagesKept } = outcome; const result: CompactionResult = { summary, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, + newConversationId, + messagesSummarized, + messagesKept, }; - - deps.emit(conversationCompacted, { - conversationId, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, - }); - return result; }, }; diff --git a/packages/session-orchestrator/src/queue.test.ts b/packages/session-orchestrator/src/queue.test.ts index a09a441..e5746c3 100644 --- a/packages/session-orchestrator/src/queue.test.ts +++ b/packages/session-orchestrator/src/queue.test.ts @@ -217,7 +217,9 @@ function createDrainingCaptureRunTurn(): { captured.push(input); if (input.drainSteering !== undefined) { drainCalled = true; - const drained = input.drainSteering(); + // The kernel awaits drainSteering (it may return a Promise that + // persists the injected messages); the fake mirrors that. + const drained = await input.drainSteering(); drainedMessages.push(...drained); } return { @@ -332,6 +334,18 @@ describe("drainSteering", () => { expect(steering?.conversationId).toBe("conv-drain"); expect(steering?.text).toBe("first\n\nsecond"); expect(steering?.turnId).toMatch(/^turn-/); + + // The steering message was PERSISTED to the store (not just injected into + // the kernel's in-memory array). Without this, a user could never see the + // steering message, and in-flight compaction (which uses the live messages) + // would be the only thing keeping it — but only if it fired. + const stored = store.data.get("conv-drain") ?? []; + const storedSteering = stored.find( + (m) => + m.role === "user" && + m.chunks.some((c) => c.type === "text" && c.text === "first\n\nsecond"), + ); + expect(storedSteering).toBeDefined(); }); it("drainSteering on an empty queue returns [] and emits nothing", async () => { @@ -593,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/session-orchestrator/src/tools-filter.ts b/packages/session-orchestrator/src/tools-filter.ts index 28e82bf..22bc5cd 100644 --- a/packages/session-orchestrator/src/tools-filter.ts +++ b/packages/session-orchestrator/src/tools-filter.ts @@ -15,6 +15,15 @@ export interface ToolAssembly { readonly computerId?: string; /** The conversation this turn belongs to. */ readonly conversationId: string; + /** + * The turn's abort signal, threaded through the filter chain so a filter that + * awaits slow I/O (e.g. the MCP tools filter connecting to MCP servers) can be + * interrupted by `POST /conversations/:id/stop` instead of blocking the turn + * until its own timeout fires. Optional: omitted by paths that have no turn + * controller (e.g. the cache-warm probe), in which case filters fall back to + * their own timeouts. Filters that return a fresh assembly MUST preserve it. + */ + readonly signal?: AbortSignal; } /** Filter chain run once per turn to transform the tool set before it reaches runTurn. */ @@ -55,5 +64,6 @@ export function filterRemoteIncompatibleTools(assembly: ToolAssembly): ToolAssem ...(assembly.cwd !== undefined ? { cwd: assembly.cwd } : {}), ...(assembly.computerId !== undefined ? { computerId: assembly.computerId } : {}), conversationId: assembly.conversationId, + ...(assembly.signal !== undefined ? { signal: assembly.signal } : {}), }; } 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..32b03d3 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 — persisted at creation + * time, after the new-conversation workspace setup resolves (so workspace + * assignment and first-turn system-prompt construction are not skipped) and + * before the first message append (so the append's auto-derived title does + * not overwrite it). The tab shows it 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 the + * turn (no second round-trip from the client). + */ + 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 @@ -968,10 +1025,26 @@ export interface TestComputerResponse { * level. The scheduler resets its timer after each run completes (not a fixed * wall-clock schedule); on backend restart it resumes scheduling for enabled * heartbeats. + * + * `inactiveOnly` (default `true`) gates each fire on the configured workspace + * having NO active agents (no conversation with status `"active"` or `"queued"`): + * the heartbeat stays quiet while the user is actively working, and only fires + * when the workspace is idle. Set `false` to fire unconditionally. */ export interface HeartbeatConfig { /** Whether the heartbeat loop is active for this workspace. */ readonly enabled: boolean; + /** + * When `true` (the default), the heartbeat SKIPS a fire when the configured + * workspace has any active agents — conversations whose persisted status is + * `"active"` (driving a turn) or `"queued"` (waiting on the message queue). + * The fire is silently skipped (no run is recorded); the scheduler re-arms + * and tries again at the next interval. When `false`, the heartbeat fires + * unconditionally regardless of workspace activity. The spawned heartbeat + * conversation lives in the DEDICATED heartbeat workspace, so it never + * counts as an "active agent" of the configured workspace (no self-block). + */ + readonly inactiveOnly: boolean; /** Custom system prompt for the heartbeat AI (empty = no system prompt). */ readonly systemPrompt: string; /** Task prompt sent as the first user message when the heartbeat fires. */ @@ -993,6 +1066,7 @@ export interface HeartbeatConfig { */ export interface UpdateHeartbeatRequest { readonly enabled?: boolean; + readonly inactiveOnly?: boolean; readonly systemPrompt?: string; readonly taskPrompt?: string; readonly intervalMinutes?: number; diff --git a/packages/transport-http/src/app.test.ts b/packages/transport-http/src/app.test.ts index 03f1959..557fb44 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 }; }, @@ -539,6 +549,35 @@ function createFakeHeartbeatService(nextRunAt: string | null): HeartbeatService }; } +/** + * A HeartbeatService fake that CAPTURES the updateConfig call (workspaceId + + * partial update) and returns a config echoing the captured update on top of + * the defaults — for asserting the PUT /workspaces/:id/heartbeat route forwards + * validated fields to the service. + */ +function createCapturingHeartbeatService(): HeartbeatService & { + readonly captured: { workspaceId: string; update: Record<string, unknown> }[]; +} { + const captured: { workspaceId: string; update: Record<string, unknown> }[] = []; + const svc: HeartbeatService = { + getConfig: async () => DEFAULT_HEARTBEAT_CONFIG, + async updateConfig(workspaceId, update) { + captured.push({ workspaceId, update: update as Record<string, unknown> }); + return { ...DEFAULT_HEARTBEAT_CONFIG, ...update }; + }, + listRuns: async () => [], + stopRun: async () => ({ ok: true }), + startAll: async () => {}, + stopAll: () => {}, + nextRunAt: async () => null, + }; + return Object.assign(svc, { + get captured() { + return captured; + }, + }); +} + const noopLogger = createFakeLogger(); describe("GET /health", () => { @@ -789,6 +828,124 @@ describe("POST /chat", () => { expect(cap.received?.modelName).toBeUndefined(); expect(cap.received?.cwd).toBeUndefined(); }); + + it("forwards the title to the orchestrator", async () => { + const cap = createCapturingOrchestrator(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: cap, + 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(cap.received).toBeDefined(); + expect(cap.received?.title).toBe("My Task"); + }); + + it("forwards a trimmed title to the orchestrator", async () => { + const cap = createCapturingOrchestrator(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: cap, + 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(cap.received?.title).toBe("spaced"); + }); + + it("does not forward a title when omitted", async () => { + const cap = createCapturingOrchestrator(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: cap, + 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(cap.received?.title).toBeUndefined(); + }); + + it("does not forward a title for a whitespace-only title", async () => { + const cap = createCapturingOrchestrator(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: cap, + 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(cap.received?.title).toBeUndefined(); + }); + + it("returns 400 when title is not a string", async () => { + const app = createApp({ + conversationStore: createFakeConversationStore(), + 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"); + }); + + it("does not call setConversationTitle itself (the orchestrator owns it)", 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: "My Task" }), + }); + + expect(res.status).toBe(200); + // The route must NOT pre-create the meta — that would bypass the + // orchestrator's new-conversation workspace/system-prompt init. The + // orchestrator sets the title after workspace setup instead. + expect(setTitleCalled).toBe(false); + }); }); describe("POST /chat/warm", () => { @@ -2069,6 +2226,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({ @@ -4541,3 +4834,64 @@ describe("GET /workspaces/:id/heartbeat/next-run", () => { expect(body.nextRunAt).toBeNull(); }); }); + +describe("PUT /workspaces/:id/heartbeat", () => { + it("forwards inactiveOnly to the service and echoes it in the response", async () => { + const hb = createCapturingHeartbeatService(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + heartbeatService: hb, + logger: noopLogger, + }); + const res = await app.request("/workspaces/ws-1/heartbeat", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ inactiveOnly: false }), + }); + expect(res.status).toBe(200); + const body = (await res.json()) as { inactiveOnly: boolean }; + expect(body.inactiveOnly).toBe(false); + expect(hb.captured).toHaveLength(1); + expect(hb.captured[0]?.workspaceId).toBe("ws-1"); + expect(hb.captured[0]?.update.inactiveOnly).toBe(false); + }); + + it("rejects a non-boolean inactiveOnly with 400", async () => { + const hb = createCapturingHeartbeatService(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + heartbeatService: hb, + logger: noopLogger, + }); + const res = await app.request("/workspaces/ws-1/heartbeat", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ inactiveOnly: "yes" }), + }); + expect(res.status).toBe(400); + // The service was NOT called (validation happened first). + expect(hb.captured).toHaveLength(0); + }); + + it("omits inactiveOnly from the forwarded update when absent (leaves it unchanged)", async () => { + const hb = createCapturingHeartbeatService(); + const app = createApp({ + conversationStore: createFakeConversationStore(), + orchestrator: createFakeOrchestrator([]), + credentialStore: createFakeCredentialStore([]), + heartbeatService: hb, + logger: noopLogger, + }); + const res = await app.request("/workspaces/ws-1/heartbeat", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ enabled: true }), + }); + expect(res.status).toBe(200); + expect(hb.captured[0]?.update.inactiveOnly).toBeUndefined(); + }); +}); diff --git a/packages/transport-http/src/app.ts b/packages/transport-http/src/app.ts index 656be9d..32a92f1 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, @@ -516,6 +518,7 @@ export function createApp(opts: CreateServerOptions): Hono { ...(reasoningEffort !== undefined ? { reasoningEffort } : {}), ...(workspaceId !== undefined ? { workspaceId } : {}), ...(images !== undefined ? { images } : {}), + ...(title !== undefined ? { title } : {}), }; opts.orchestrator @@ -796,6 +799,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 { @@ -1673,6 +1698,16 @@ export function createApp(opts: CreateServerOptions): Hono { update.enabled = obj.enabled; } + // inactiveOnly: when true (the default), the heartbeat skips a fire while + // the configured workspace has active agents. A boolean; absent leaves it + // unchanged. + if (obj.inactiveOnly !== undefined) { + if (typeof obj.inactiveOnly !== "boolean") { + return c.json({ error: "Field 'inactiveOnly' must be a boolean" }, 400); + } + update.inactiveOnly = obj.inactiveOnly; + } + if (obj.systemPrompt !== undefined) { if (typeof obj.systemPrompt !== "string") { return c.json({ error: "Field 'systemPrompt' must be a string" }, 400); 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..c703049 100644 --- a/packages/transport-http/src/logic.ts +++ b/packages/transport-http/src/logic.ts @@ -56,6 +56,16 @@ 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 orchestrator, which persists it via the + * conversation store's `setConversationTitle` AFTER the new-conversation + * workspace setup (so workspace assignment / first-turn system-prompt + * construction are not skipped) and before the first message append. + */ + 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 +138,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( |