diff --git a/packages/control-plane/src/routes/session-child-spawn.ts b/packages/control-plane/src/routes/session-child-spawn.ts index 1258edc527..3c2548b708 100644 --- a/packages/control-plane/src/routes/session-child-spawn.ts +++ b/packages/control-plane/src/routes/session-child-spawn.ts @@ -294,6 +294,7 @@ async function handleSpawnChild( authorId: spawnContext.promptAuthor.userId, canonicalUserId: spawnContext.promptAuthor.canonicalUserId ?? undefined, source: "agent", + cancellableByUser: false, } satisfies EnqueuePromptRequest; promptResponse = await ctx.sessionRuntime.fetch(childId, SessionInternalPaths.prompt, { diff --git a/packages/control-plane/src/routes/session-prompt.ts b/packages/control-plane/src/routes/session-prompt.ts index 5916dfeea0..ec7b2520cf 100644 --- a/packages/control-plane/src/routes/session-prompt.ts +++ b/packages/control-plane/src/routes/session-prompt.ts @@ -135,6 +135,7 @@ async function handleSessionPrompt( authorId, canonicalUserId, source: body.source || "web", + cancellableByUser: ctx.principal?.kind === "user", model: body.model, reasoningEffort: body.reasoningEffort, attachments, diff --git a/packages/control-plane/src/scheduler/scheduler.ts b/packages/control-plane/src/scheduler/scheduler.ts index 3ec8d45b77..bed96d2591 100644 --- a/packages/control-plane/src/scheduler/scheduler.ts +++ b/packages/control-plane/src/scheduler/scheduler.ts @@ -1605,7 +1605,7 @@ export class Scheduler { const promptResponse = await stub.fetch("http://internal/internal/prompt", { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify(body), + body: JSON.stringify({ ...body, cancellableByUser: false }), }); if (!promptResponse.ok) { diff --git a/packages/control-plane/src/session/enqueue-prompt-contract.ts b/packages/control-plane/src/session/enqueue-prompt-contract.ts index 77e065a0ef..c2829855d0 100644 --- a/packages/control-plane/src/session/enqueue-prompt-contract.ts +++ b/packages/control-plane/src/session/enqueue-prompt-contract.ts @@ -13,6 +13,7 @@ export const enqueuePromptRequestSchema = z authorId: z.string(), canonicalUserId: z.string().nullable().optional(), source: messageSourceSchema, + cancellableByUser: z.boolean().default(false), model: z.string().optional(), reasoningEffort: z.string().optional(), attachments: sessionAttachmentReferencesSchema.optional(), diff --git a/packages/control-plane/src/session/http/handlers/child-sessions.handler.test.ts b/packages/control-plane/src/session/http/handlers/child-sessions.handler.test.ts index 1bebbcf757..5f45f04c95 100644 --- a/packages/control-plane/src/session/http/handlers/child-sessions.handler.test.ts +++ b/packages/control-plane/src/session/http/handlers/child-sessions.handler.test.ts @@ -136,6 +136,7 @@ describe("ChildSessionsHandler", () => { authorId: "owner-1", canonicalUserId: "canonical-1", source: "agent", + cancellableByUser: false, scmEnrichment: { userId: null, login: null, diff --git a/packages/control-plane/src/session/http/handlers/child-sessions.handler.ts b/packages/control-plane/src/session/http/handlers/child-sessions.handler.ts index 45fe58d269..b4d63c97ea 100644 --- a/packages/control-plane/src/session/http/handlers/child-sessions.handler.ts +++ b/packages/control-plane/src/session/http/handlers/child-sessions.handler.ts @@ -150,6 +150,7 @@ export class ChildSessionsHandler { authorId: parsed.data.author.userId, canonicalUserId: parsed.data.author.canonicalUserId ?? undefined, source: "agent", + cancellableByUser: false, scmEnrichment: { userId: parsed.data.author.scmUserId, login: parsed.data.author.scmLogin, diff --git a/packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts b/packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts index 2abe3725b3..110bf00794 100644 --- a/packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts +++ b/packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts @@ -105,6 +105,7 @@ function createMessage(overrides: Partial = {}): MessageRow { reasoning_effort: null, attachments: null, callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, diff --git a/packages/control-plane/src/session/http/handlers/messages.handler.test.ts b/packages/control-plane/src/session/http/handlers/messages.handler.test.ts index fe057f0ec8..9fe3ec1978 100644 --- a/packages/control-plane/src/session/http/handlers/messages.handler.test.ts +++ b/packages/control-plane/src/session/http/handlers/messages.handler.test.ts @@ -55,6 +55,7 @@ describe("MessagesHandler", () => { content: "hello", authorId: "user-1", source: "web", + cancellableByUser: false, }); }); @@ -94,7 +95,10 @@ describe("MessagesHandler", () => { ); expect(response.status).toBe(200); - expect(messageService.enqueuePrompt).toHaveBeenCalledWith(body); + expect(messageService.enqueuePrompt).toHaveBeenCalledWith({ + ...body, + cancellableByUser: false, + }); }); it("returns 400 for malformed prompt bodies", async () => { diff --git a/packages/control-plane/src/session/message-queue.test.ts b/packages/control-plane/src/session/message-queue.test.ts index c732ce7639..e009b7e440 100644 --- a/packages/control-plane/src/session/message-queue.test.ts +++ b/packages/control-plane/src/session/message-queue.test.ts @@ -80,6 +80,7 @@ function createMessage(overrides: Partial = {}): MessageRow { reasoning_effort: null, attachments: null, callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, @@ -427,12 +428,15 @@ describe("SessionMessageQueue", () => { }); expect(h.repository.cancelPendingMessage).toHaveBeenCalledWith("msg-1"); + expect(h.broadcast).toHaveBeenCalledWith({ type: "prompt_queue_updated", promptQueue: [] }); expect(h.wsManager.send).toHaveBeenCalledWith(ws, { type: "prompt_cancelled", clientRequestId: "request-1", messageId: "msg-1", }); - expect(h.broadcast).toHaveBeenCalledWith({ type: "prompt_queue_updated", promptQueue: [] }); + expect(h.broadcast.mock.invocationCallOrder[0]).toBeLessThan( + h.wsManager.send.mock.invocationCallOrder[0] + ); expect(h.sessionStatus.reconcileAfterQueueRemoval).toHaveBeenCalledOnce(); }); @@ -451,7 +455,10 @@ describe("SessionMessageQueue", () => { message: "This prompt is no longer pending and cannot be removed", clientRequestId: "request-1", }); - expect(h.broadcast).not.toHaveBeenCalled(); + expect(h.broadcast).toHaveBeenCalledWith({ type: "prompt_queue_updated", promptQueue: [] }); + expect(h.broadcast.mock.invocationCallOrder[0]).toBeLessThan( + h.wsManager.send.mock.invocationCallOrder[0] + ); }); it("reconciles session status after removing a prompt", async () => { @@ -1514,6 +1521,7 @@ describe("SessionMessageQueue", () => { content: "Continue", authorId: "user-1", source: "agent", + cancellableByUser: false, }) ).rejects.toMatchObject({ sessionStatus: status }); @@ -1547,6 +1555,7 @@ describe("SessionMessageQueue", () => { content: "Fix bug", authorId: "github:1001", source: "github", + cancellableByUser: false, scmEnrichment: { userId: "1001", login: "octocat", @@ -1569,6 +1578,7 @@ describe("SessionMessageQueue", () => { content: "Fix bug", authorId: "github:1001", source: "github", + cancellableByUser: false, }); expect(h.participantService.create).toHaveBeenCalledWith("github:1001", "github:1001"); @@ -1581,6 +1591,7 @@ describe("SessionMessageQueue", () => { content: "Fix bug", authorId: "github:1001", source: "github", + cancellableByUser: false, scmEnrichment: { userId: "1001", login: "octocat", @@ -1610,6 +1621,7 @@ describe("SessionMessageQueue", () => { content: "Fix bug", authorId: "github:1001", source: "github", + cancellableByUser: false, }); expect(h.repository.updateParticipantCoalesce).not.toHaveBeenCalled(); diff --git a/packages/control-plane/src/session/message-queue.ts b/packages/control-plane/src/session/message-queue.ts index df854e5c2b..031ff816be 100644 --- a/packages/control-plane/src/session/message-queue.ts +++ b/packages/control-plane/src/session/message-queue.ts @@ -65,6 +65,7 @@ interface EnqueuePromptCoreData { userId: string; content: string; source: MessageSource; + cancellableByUser: boolean; model?: string; reasoningEffort?: string; attachments?: SessionAttachmentReference[]; @@ -194,6 +195,7 @@ export class SessionMessageQueue { authorId: participant.id, content: command.prompt, source: "github", + cancellableByUser: false, status: "pending", createdAt: now, }, @@ -257,6 +259,7 @@ export class SessionMessageQueue { userId: client.userId, content: data.content, source: "web", + cancellableByUser: true, model: data.model, reasoningEffort: data.reasoningEffort, attachments: data.attachments, @@ -328,7 +331,9 @@ export class SessionMessageQueue { ws: WebSocket, data: { messageId: string; clientRequestId: string } ): Promise { - if (!this.messageRepository.cancelPendingMessage(data.messageId)) { + const cancelled = this.messageRepository.cancelPendingMessage(data.messageId); + this.broadcastPromptQueue(); + if (!cancelled) { this.wsManager.send(ws, { type: "error", code: "PROMPT_NOT_CANCELLABLE", @@ -343,7 +348,6 @@ export class SessionMessageQueue { clientRequestId: data.clientRequestId, messageId: data.messageId, }); - this.broadcastPromptQueue(); this.log.info("prompt.cancelled", { event: "prompt.cancelled", message_id: data.messageId, @@ -740,6 +744,7 @@ export class SessionMessageQueue { userId: data.authorId, content: data.content, source: data.source, + cancellableByUser: data.cancellableByUser, model: data.model, reasoningEffort: data.reasoningEffort, attachments: data.attachments, @@ -822,6 +827,7 @@ export class SessionMessageQueue { authorId: data.participant.id, content: data.content, source: data.source, + cancellableByUser: data.cancellableByUser, model: messageModel, reasoningEffort: messageReasoningEffort, attachments: attachments ? JSON.stringify(attachments) : null, diff --git a/packages/control-plane/src/session/message-repository.test.ts b/packages/control-plane/src/session/message-repository.test.ts index 7ccca04831..6417ace1dc 100644 --- a/packages/control-plane/src/session/message-repository.test.ts +++ b/packages/control-plane/src/session/message-repository.test.ts @@ -125,10 +125,31 @@ describe("MessageRepository", () => { it("projects unfinished messages into the prompt queue", () => { vi.spyOn(repository, "listUnfinishedMessages").mockReturnValue([ - { id: "msg-1", content: "Continue", status: "pending" } as never, + { + id: "msg-1", + content: "Continue", + status: "pending", + source: "linear", + callback_context: "{}", + cancellable_by_user: 1, + } as never, + { + id: "msg-2", + content: "Reply in Linear", + status: "pending", + source: "web", + callback_context: null, + cancellable_by_user: 0, + } as never, ]); expect(repository.listPromptQueue()).toEqual([ - { messageId: "msg-1", content: "Continue", status: "pending" }, + { messageId: "msg-1", content: "Continue", status: "pending", cancellable: true }, + { + messageId: "msg-2", + content: "Reply in Linear", + status: "pending", + cancellable: false, + }, ]); }); @@ -138,6 +159,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Hello", source: "web", + cancellableByUser: true, model: "claude-sonnet-4", attachments: "[]", callbackContext: '{"channel":"C123"}', @@ -154,6 +176,7 @@ describe("MessageRepository", () => { null, "[]", '{"channel":"C123"}', + 1, null, null, null, @@ -176,6 +199,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -199,6 +223,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -223,6 +248,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -247,6 +273,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -271,6 +298,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -300,6 +328,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Fix feedback", source: "github", + cancellableByUser: false, status: "pending", createdAt: 2000, }, @@ -326,6 +355,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Look", source: "web", + cancellableByUser: true, status: "pending", createdAt: 1, }, @@ -345,6 +375,7 @@ describe("MessageRepository", () => { authorId: "p-1", content: "Look", source: "web", + cancellableByUser: true, status: "pending", createdAt: 1, }, @@ -354,23 +385,20 @@ describe("MessageRepository", () => { expect(mock.calls).toHaveLength(1); }); - it("atomically releases attachments and cancels a pending web message", () => { - mock.setData(`SELECT status, source, callback_context FROM messages WHERE id = ?`, [ - { status: "pending", source: "web", callback_context: null }, + it("atomically cancels a user-cancellable pending message and releases attachments", () => { + mock.setMatchingData(/DELETE FROM messages[\s\S]*cancellable_by_user = 1[\s\S]*RETURNING id/, [ + { id: "msg-1" }, ]); - mock.setRowsWritten(1); expect(repository.cancelPendingMessage("msg-1")).toBe(true); expect(transactionSyncCalls).toBe(1); + expect(mock.calls[0].query).toContain("DELETE FROM messages"); expect(mock.calls[1].query).toContain("UPDATE attachments SET message_id = NULL"); - expect(mock.calls[2].query).toContain("DELETE FROM messages"); }); - it("rejects cancellation for messages that may need callbacks", () => { - mock.setData(`SELECT status, source, callback_context FROM messages WHERE id = ?`, [ - { status: "pending", source: "linear", callback_context: null }, - ]); + it("rejects cancellation without canonical user ownership", () => { expect(repository.cancelPendingMessage("msg-1")).toBe(false); expect(mock.calls).toHaveLength(1); + expect(mock.calls[0].query).toContain("cancellable_by_user = 1"); }); it("atomically starts processing and creates the canonical user event", () => { diff --git a/packages/control-plane/src/session/message-repository.ts b/packages/control-plane/src/session/message-repository.ts index 8dc4b2f68f..57199e4136 100644 --- a/packages/control-plane/src/session/message-repository.ts +++ b/packages/control-plane/src/session/message-repository.ts @@ -29,6 +29,7 @@ export interface CreateMessageData { reasoningEffort?: string | null; attachments?: string | null; callbackContext?: string | null; + cancellableByUser: boolean; clientRequestId?: string | null; requestFingerprint?: string | null; autofixFeedbackKey?: string | null; @@ -227,37 +228,22 @@ export class MessageRepository { messageId: message.id, content: message.content, status: message.status as "pending" | "processing", + cancellable: message.status === "pending" && message.cancellable_by_user === 1, })); } cancelPendingMessage(messageId: string): boolean { return this.transactionSync(() => { - const result = this.sql.exec( - `SELECT status, source, callback_context FROM messages WHERE id = ?`, + const deleted = this.sql.exec( + `DELETE FROM messages + WHERE id = ? AND status = 'pending' AND cancellable_by_user = 1 + RETURNING id`, messageId ); - const message = ( - result.toArray() as Array<{ - status: MessageStatus; - source: string; - callback_context: string | null; - }> - )[0]; - if ( - message?.status !== "pending" || - message.source !== "web" || - message.callback_context !== null - ) { - return false; - } + if (deleted.toArray().length !== 1) return false; this.attachments.releaseForMessage(messageId); - const deleted = this.sql.exec( - `DELETE FROM messages WHERE id = ? AND status = 'pending'`, - messageId - ); - deleted.toArray(); - return deleted.rowsWritten === 1; + return true; }); } @@ -279,9 +265,9 @@ export class MessageRepository { this.sql.exec( `INSERT INTO messages ( id, author_id, content, source, model, reasoning_effort, attachments, - callback_context, client_request_id, request_fingerprint, autofix_feedback_key, + callback_context, cancellable_by_user, client_request_id, request_fingerprint, autofix_feedback_key, autofix_pr_key, origin_context, status, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, data.id, data.authorId, data.content, @@ -290,6 +276,7 @@ export class MessageRepository { data.reasoningEffort ?? null, data.attachments ?? null, data.callbackContext ?? null, + data.cancellableByUser ? 1 : 0, data.clientRequestId ?? null, data.requestFingerprint ?? null, data.autofixFeedbackKey ?? null, diff --git a/packages/control-plane/src/session/schema.test.ts b/packages/control-plane/src/session/schema.test.ts index 3625154b1d..00a0094430 100644 --- a/packages/control-plane/src/session/schema.test.ts +++ b/packages/control-plane/src/session/schema.test.ts @@ -416,6 +416,33 @@ describe("applyMigrations", () => { } }); + it("adds canonical prompt cancellation ownership and backfills eligible legacy messages", () => { + expect(SCHEMA_SQL).toContain("cancellable_by_user INTEGER NOT NULL DEFAULT 0"); + const migration = MIGRATIONS.find((entry) => entry.id === 47); + const db = new DatabaseSync(":memory:"); + const sql = createDatabaseSql(db); + try { + db.exec(`CREATE TABLE messages ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + callback_context TEXT + )`); + db.exec(`INSERT INTO messages VALUES + ('web', 'web', NULL), + ('linear', 'linear', NULL), + ('callback', 'web', '{}')`); + (migration!.run as (sql: SqlStorage) => void)(sql); + + expect(db.prepare("SELECT id, cancellable_by_user FROM messages ORDER BY id").all()).toEqual([ + { id: "callback", cancellable_by_user: 0 }, + { id: "linear", cancellable_by_user: 0 }, + { id: "web", cancellable_by_user: 1 }, + ]); + } finally { + db.close(); + } + }); + it("initializes a legacy messages table before creating indexes for new columns", () => { expect(SCHEMA_SQL).not.toMatch(/\bCREATE (?:UNIQUE )?INDEX\b/); @@ -453,6 +480,7 @@ describe("applyMigrations", () => { expect.objectContaining({ name: "client_request_id", type: "TEXT" }), expect.objectContaining({ name: "request_fingerprint", type: "TEXT" }), expect.objectContaining({ name: "stop_confirmation_deadline", type: "INTEGER" }), + expect.objectContaining({ name: "cancellable_by_user", type: "INTEGER" }), ]) ); expect( diff --git a/packages/control-plane/src/session/schema.ts b/packages/control-plane/src/session/schema.ts index 8ee706494f..229cf2060a 100644 --- a/packages/control-plane/src/session/schema.ts +++ b/packages/control-plane/src/session/schema.ts @@ -114,6 +114,7 @@ CREATE TABLE IF NOT EXISTS messages ( reasoning_effort TEXT, -- Per-message reasoning effort override attachments TEXT, -- JSON array callback_context TEXT, -- JSON callback context for Slack follow-up notifications + cancellable_by_user INTEGER NOT NULL DEFAULT 0, -- Set only by authenticated user ingress client_request_id TEXT, -- Web-client idempotency key request_fingerprint TEXT, -- Participant-scoped canonical request hash autofix_feedback_key TEXT, -- Stable provider feedback identity for idempotency @@ -630,6 +631,18 @@ export const MIGRATIONS: readonly SchemaMigration[] = [ ); }, }, + { + id: 47, + description: "Add canonical prompt cancellation ownership", + run: (sql) => { + runMigration( + sql, + `ALTER TABLE messages ADD COLUMN cancellable_by_user INTEGER NOT NULL DEFAULT 0` + ); + sql.exec(`UPDATE messages SET cancellable_by_user = 1 + WHERE source = 'web' AND callback_context IS NULL`); + }, + }, ]; /** diff --git a/packages/control-plane/src/session/services/message.service.test.ts b/packages/control-plane/src/session/services/message.service.test.ts index 8ec38db74f..a9a6d5a12f 100644 --- a/packages/control-plane/src/session/services/message.service.test.ts +++ b/packages/control-plane/src/session/services/message.service.test.ts @@ -55,6 +55,7 @@ describe("MessageService", () => { content: "hello", authorId: "user-1", source: "web", + cancellableByUser: true, }); expect(result).toEqual({ messageId: "msg-1", status: "queued" }); @@ -62,6 +63,7 @@ describe("MessageService", () => { content: "hello", authorId: "user-1", source: "web", + cancellableByUser: true, }); }); @@ -193,6 +195,7 @@ describe("MessageService", () => { }, ]), callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, @@ -214,6 +217,7 @@ describe("MessageService", () => { reasoning_effort: null, attachments: "invalid-json", callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, @@ -235,6 +239,7 @@ describe("MessageService", () => { reasoning_effort: null, attachments: null, callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, @@ -282,6 +287,7 @@ describe("MessageService", () => { reasoning_effort: null, attachments: "[]", callback_context: null, + cancellable_by_user: 1, client_request_id: null, request_fingerprint: null, autofix_feedback_key: null, diff --git a/packages/control-plane/src/session/types.ts b/packages/control-plane/src/session/types.ts index 626b003953..65cfd95cd5 100644 --- a/packages/control-plane/src/session/types.ts +++ b/packages/control-plane/src/session/types.ts @@ -103,6 +103,7 @@ export interface MessageRow { reasoning_effort: string | null; // Reasoning effort for per-message override attachments: string | null; // JSON callback_context: string | null; // JSON: { channel, threadTs, repoFullName, model } + cancellable_by_user: number; client_request_id: string | null; request_fingerprint: string | null; autofix_feedback_key: string | null; diff --git a/packages/control-plane/test/integration/helpers.ts b/packages/control-plane/test/integration/helpers.ts index 3f25f015c8..882c7be0bd 100644 --- a/packages/control-plane/test/integration/helpers.ts +++ b/packages/control-plane/test/integration/helpers.ts @@ -321,18 +321,22 @@ export async function seedMessage( status: string; createdAt: number; startedAt?: number; + cancellableByUser?: boolean; } ): Promise { await runInSessionDO(stub, (instance: SessionDO, state) => { state.storage.sql.exec( - "INSERT INTO messages (id, author_id, content, source, status, created_at, started_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + `INSERT INTO messages + (id, author_id, content, source, status, created_at, started_at, cancellable_by_user) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, msg.id, msg.authorId, msg.content, msg.source, msg.status, msg.createdAt, - msg.startedAt ?? null + msg.startedAt ?? null, + msg.cancellableByUser ? 1 : 0 ); }); } diff --git a/packages/control-plane/test/integration/prompt-enqueue.test.ts b/packages/control-plane/test/integration/prompt-enqueue.test.ts index 314ae5d147..7886580c97 100644 --- a/packages/control-plane/test/integration/prompt-enqueue.test.ts +++ b/packages/control-plane/test/integration/prompt-enqueue.test.ts @@ -6,6 +6,7 @@ import { openSandboxWs, queryDO, seedSandboxAuth, + serviceFetch, } from "./helpers"; const SANDBOX_TOKEN = "prompt-order-sandbox-token"; @@ -52,6 +53,25 @@ describe("POST /internal/prompt", () => { expect(["pending", "processing"]).toContain(messages[0].status); }); + it("derives user cancellation ownership independently of caller-supplied source", async () => { + const { stub, sessionName } = await initSession(); + + const response = await serviceFetch(`https://cp.test/sessions/${sessionName}/prompt`, { + method: "POST", + body: JSON.stringify({ content: "Browser prompt", source: "linear" }), + }); + + expect(response.status).toBe(200); + const { messageId } = await response.json<{ messageId: string }>(); + expect( + await queryDO<{ source: string; cancellable_by_user: number }>( + stub, + "SELECT source, cancellable_by_user FROM messages WHERE id = ?", + messageId + ) + ).toEqual([{ source: "linear", cancellable_by_user: 1 }]); + }); + it("persists queued prompts in FIFO order", async () => { const { stub } = await initSession(); const enqueue = async (content: string) => { diff --git a/packages/control-plane/test/integration/service-auth.test.ts b/packages/control-plane/test/integration/service-auth.test.ts index c1146490e1..5a44af0864 100644 --- a/packages/control-plane/test/integration/service-auth.test.ts +++ b/packages/control-plane/test/integration/service-auth.test.ts @@ -11,6 +11,7 @@ import { generateInternalToken } from "@open-inspect/shared/auth"; import { GlobalSecretsStore } from "../../src/db/global-secrets"; import { UserStore } from "../../src/db/user-store"; import { cleanD1Tables } from "./cleanup"; +import { queryDO } from "./helpers"; import { insertCanonicalUser } from "./identity-seed-helpers"; const SERVICE_SECRET: Record = { @@ -311,10 +312,51 @@ describe("sig1 service-credential authentication", () => { method: "POST", url: `https://test.local/sessions/${createdBody.sessionId}/prompt`, actor: "slack:U0002", - body: JSON.stringify({ content: "Cross-session prompt" }), + body: JSON.stringify({ content: "Integration source prompt", source: "linear" }), }); expect(collaborator.status).toBe(200); + const callbackPrompt = await signedFetch({ + service: "slack-bot", + method: "POST", + url: `https://test.local/sessions/${createdBody.sessionId}/prompt`, + actor: "slack:U0002", + body: JSON.stringify({ + content: "Integration callback prompt", + source: "web", + callbackContext: { + source: "slack", + channel: "C1", + threadTs: "1.0", + repoFullName: "acme/repo", + model: "anthropic/claude-haiku-4-5", + }, + }), + }); + expect(callbackPrompt.status).toBe(200); + + const stub = env.SESSION.get(env.SESSION.idFromName(createdBody.sessionId)); + expect( + await queryDO<{ content: string; source: string; cancellable_by_user: number }>( + stub, + `SELECT content, source, cancellable_by_user FROM messages + WHERE content IN (?, ?) ORDER BY content`, + "Integration source prompt", + "Integration callback prompt" + ) + ).toEqual([ + { + content: "Integration callback prompt", + source: "web", + cancellable_by_user: 0, + }, + { + content: "Integration source prompt", + source: "linear", + cancellable_by_user: 0, + }, + ]); + const deniedByServiceCeiling = await signedFetch({ service: "slack-bot", method: "DELETE", diff --git a/packages/control-plane/test/integration/websocket-client.test.ts b/packages/control-plane/test/integration/websocket-client.test.ts index 01576b5ec7..736d8395b8 100644 --- a/packages/control-plane/test/integration/websocket-client.test.ts +++ b/packages/control-plane/test/integration/websocket-client.test.ts @@ -634,7 +634,7 @@ describe("Client WebSocket (via SELF.fetch)", () => { promptQueue: Array>; }; expect(subscribed.promptQueue).toEqual([ - expect.objectContaining({ content: "Only once", status: "pending" }), + expect.objectContaining({ content: "Only once", status: "pending", cancellable: true }), ]); expect(subscribed.promptQueue[0]).not.toHaveProperty("model"); expect(subscribed.promptQueue[0]).not.toHaveProperty("reasoningEffort"); @@ -753,6 +753,7 @@ describe("Client WebSocket (via SELF.fetch)", () => { source: "web", status: "pending", createdAt: now + index, + cancellableByUser: true, }); } await queryDO( @@ -905,8 +906,11 @@ describe("Client WebSocket (via SELF.fetch)", () => { ws.close(); }); - it("does not allow web clients to cancel integration-owned prompts", async () => { - const name = `ws-client-cancel-integration-${Date.now()}`; + it.each([ + { ownership: "integration source", source: "linear", callbackContext: null }, + { ownership: "callback context", source: "web", callbackContext: { channel: "C1" } }, + ])("does not allow web clients to cancel prompts owned by $ownership", async (testCase) => { + const name = `ws-client-cancel-integration-${crypto.randomUUID()}`; const { stub } = await initNamedSession(name); const [{ id: participantId }] = await queryDO<{ id: string }>( stub, @@ -916,22 +920,24 @@ describe("Client WebSocket (via SELF.fetch)", () => { id: "message-linear", authorId: participantId, content: "Reply in Linear", - source: "linear", + source: testCase.source, status: "pending", createdAt: Date.now(), }); - await queryDO( - stub, - "UPDATE messages SET source = 'web', callback_context = ? WHERE id = ?", - JSON.stringify({ channel: "C1", threadTs: "1.0" }), - "message-linear" - ); + if (testCase.callbackContext) { + await queryDO( + stub, + "UPDATE messages SET callback_context = ? WHERE id = ?", + JSON.stringify(testCase.callbackContext), + "message-linear" + ); + } const { ws, messages } = await openClientWs(name, { subscribe: true }); const subscribed = messages.find((message) => message.type === "subscribed") as { promptQueue: Array<{ messageId: string }>; }; expect(subscribed.promptQueue).toContainEqual( - expect.objectContaining({ messageId: "message-linear" }) + expect.objectContaining({ messageId: "message-linear", cancellable: false }) ); const clientRequestId = crypto.randomUUID(); const rejected = collectMessages(ws, { diff --git a/packages/shared/src/types/server-messages.test.ts b/packages/shared/src/types/server-messages.test.ts index 61f426c974..4b0c5d5a29 100644 --- a/packages/shared/src/types/server-messages.test.ts +++ b/packages/shared/src/types/server-messages.test.ts @@ -167,11 +167,13 @@ describe("session view contracts", () => { messageId: "message-running", content: "Run this", status: "processing", + cancellable: false, }, { messageId: "message-pending", content: "Then this", status: "pending", + cancellable: true, }, ]; @@ -188,6 +190,18 @@ describe("session view contracts", () => { ).toEqual(promptQueue); }); + it("treats queue items from older servers as non-cancellable", () => { + const message = serverMessageSchema.parse({ + type: "prompt_queue_updated", + promptQueue: [{ messageId: "message-1", content: "Wait", status: "pending" }], + }); + + expect(message).toMatchObject({ + type: "prompt_queue_updated", + promptQueue: [{ cancellable: false }], + }); + }); + it("echoes prompt request correlation", () => { expect( serverMessageSchema.parse({ diff --git a/packages/shared/src/types/server-messages.ts b/packages/shared/src/types/server-messages.ts index 432c62356c..638beb79ed 100644 --- a/packages/shared/src/types/server-messages.ts +++ b/packages/shared/src/types/server-messages.ts @@ -11,6 +11,7 @@ export const promptQueueItemSchema = z.object({ messageId: z.string(), content: z.string(), status: z.enum(["pending", "processing"]), + cancellable: z.boolean().default(false), }); export type PromptQueueItem = z.infer; diff --git a/packages/web/src/app/(app)/(sidebar)/session/[id]/page.tsx b/packages/web/src/app/(app)/(sidebar)/session/[id]/page.tsx index 91192a80a6..3ce3194b15 100644 --- a/packages/web/src/app/(app)/(sidebar)/session/[id]/page.tsx +++ b/packages/web/src/app/(app)/(sidebar)/session/[id]/page.tsx @@ -157,7 +157,7 @@ export default function SessionPage() { if (!capabilities.lifecycle) return; if (cancellingPromptIdsRef.current.has(messageId)) return; const queuedPrompt = promptQueue.find((item) => item.messageId === messageId); - if (!queuedPrompt || queuedPrompt.status !== "pending") return; + if (!queuedPrompt?.cancellable) return; cancellingPromptIdsRef.current.add(messageId); setCancellingPromptIds(new Set(cancellingPromptIdsRef.current)); diff --git a/packages/web/src/components/queued-prompt-stack.test.tsx b/packages/web/src/components/queued-prompt-stack.test.tsx index 199f283ef1..ec3ecbfbae 100644 --- a/packages/web/src/components/queued-prompt-stack.test.tsx +++ b/packages/web/src/components/queued-prompt-stack.test.tsx @@ -22,7 +22,9 @@ describe("QueuedPromptStack", () => { it("shows queued prompts without removal controls in read-only mode", () => { render( { onRemove={vi.fn()} capabilities={FULL_CAPABILITIES} promptQueue={[ - { messageId: "running", content: "Already running", status: "processing" }, - { messageId: "next", content: "Run next", status: "pending" }, - { messageId: "later", content: "Run after that", status: "pending" }, + { + messageId: "running", + content: "Already running", + status: "processing", + cancellable: false, + }, + { messageId: "next", content: "Run next", status: "pending", cancellable: true }, + { messageId: "later", content: "Run after that", status: "pending", cancellable: true }, ]} /> ); @@ -60,7 +67,14 @@ describe("QueuedPromptStack", () => { cancellingPromptIds={new Set()} onRemove={vi.fn()} capabilities={FULL_CAPABILITIES} - promptQueue={[{ messageId: "running", content: "Already running", status: "processing" }]} + promptQueue={[ + { + messageId: "running", + content: "Already running", + status: "processing", + cancellable: false, + }, + ]} /> ); @@ -71,7 +85,9 @@ describe("QueuedPromptStack", () => { const onRemove = vi.fn(); render( { const onRemove = vi.fn(); render( { expect(onRemove).toHaveBeenCalledWith("next"); }); - it("offers removal for pending prompts whose eligibility is server-owned", () => { + it("does not offer removal for a server-owned prompt", () => { const onRemove = vi.fn(); render( { messageId: "linear-prompt", content: "Reply in Linear", status: "pending", + cancellable: false, }, ]} cancellingPromptIds={new Set()} @@ -116,7 +135,10 @@ describe("QueuedPromptStack", () => { /> ); - fireEvent.click(screen.getByRole("button", { name: "Remove queued prompt: Reply in Linear" })); - expect(onRemove).toHaveBeenCalledWith("linear-prompt"); + expect(screen.getByText("Reply in Linear")).toBeInTheDocument(); + expect( + screen.queryByRole("button", { name: "Remove queued prompt: Reply in Linear" }) + ).not.toBeInTheDocument(); + expect(onRemove).not.toHaveBeenCalled(); }); }); diff --git a/packages/web/src/components/queued-prompt-stack.tsx b/packages/web/src/components/queued-prompt-stack.tsx index 60e4a7fd63..fcffe57de1 100644 --- a/packages/web/src/components/queued-prompt-stack.tsx +++ b/packages/web/src/components/queued-prompt-stack.tsx @@ -31,7 +31,7 @@ export function QueuedPromptStack({

{prompt.content}

- {capabilities.lifecycle && ( + {capabilities.lifecycle && prompt.cancellable && (