Fix queued prompt cancellation state synchronization

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, {

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,

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) {

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(),

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,

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,

packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts

@@ -105,6 +105,7 @@ function createMessage(overrides: Partial<MessageRow> = {}): MessageRow {
     reasoning_effort: null,
     attachments: null,
     callback_context: null,
+    cancellable_by_user: 1,
     client_request_id: null,
     request_fingerprint: null,
     autofix_feedback_key: null,

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 () => {

packages/control-plane/src/session/message-queue.test.ts

@@ -80,6 +80,7 @@ function createMessage(overrides: Partial<MessageRow> = {}): 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();

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<void> {
-    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,

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", () => {