feat(agent-composer): add message identity and queue cancel APIs

apps/web/src/lib/cloud-agent-next/cloud-agent-client.ts

@@ -367,6 +367,12 @@ export type InterruptResult = {
   processesFound: boolean;
 };
 
+/** Input for canceling one queued (not yet accepted) message. */
+export type CancelQueuedMessageInput = {
+  sessionId: string;
+  messageId: string;
+};
+
 export type AnswerQuestionInput = {
   sessionId: string;
   questionId: string;
@@ -521,6 +527,9 @@ type CloudAgentNextTRPCClient = {
   interruptSession: {
     mutate: (input: { sessionId: string }) => Promise<InterruptResult>;
   };
+  cancelQueuedMessage: {
+    mutate: (input: CancelQueuedMessageInput) => Promise<{ dropped: boolean }>;
+  };
   getSession: {
     query: (input: GetSessionInput) => Promise<GetSessionOutput>;
   };
@@ -683,6 +692,26 @@ export class CloudAgentNextClient {
     }
   }
 
+  /**
+   * Cancel one queued (not yet accepted) message by id without interrupting the
+   * active run. Returns whether a pending message was dropped.
+   */
+  async cancelQueuedMessage(sessionId: string, messageId: string): Promise<{ dropped: boolean }> {
+    try {
+      return await this.client.cancelQueuedMessage.mutate({ sessionId, messageId });
+    } catch (error) {
+      console.error(`Error canceling queued message ${messageId} in session ${sessionId}:`, error);
+      captureException(error, {
+        tags: {
+          source: 'cloud-agent-next-client',
+          endpoint: 'cancelQueuedMessage',
+        },
+        extra: { sessionId, messageId },
+      });
+      throw error;
+    }
+  }
+
   /**
    * Get session state from cloud-agent DO.
    */

apps/web/src/routers/cloud-agent-next-router.test.ts

@@ -54,10 +54,14 @@ const mockGenerateCloudAgentAttachmentDownloadUrl = jest.fn<
 
 const mockGetSession = jest.fn<(cloudAgentSessionId: string) => Promise<{ model?: string }>>();
 
+const mockCancelQueuedMessage =
+  jest.fn<(input: { sessionId: string; messageId: string }) => Promise<{ dropped: boolean }>>();
+
 const mockCreateCloudAgentNextClient = jest.fn(() => ({
   prepareSession: mockPrepareSession,
   sendMessage: mockSendMessage,
   getSession: mockGetSession,
+  cancelQueuedMessage: mockCancelQueuedMessage,
 }));
 
 const mockCreateCloudAgentNextClientForModel = jest.fn(
@@ -177,6 +181,7 @@ let createCaller: (ctx: { user: User }) => {
     contentLength: number;
   }) => Promise<unknown>;
   getAttachmentDownloadUrl: (input: { messageUuid: string; filename: string }) => Promise<unknown>;
+  cancelQueuedMessage: (input: { sessionId: string; messageId: string }) => Promise<unknown>;
   checkEligibility: () => Promise<{
     balance: number;
     minBalance: number;
@@ -413,6 +418,30 @@ describe('cloudAgentNextRouter attachment forwarding', () => {
   });
 });
 
+describe('cloudAgentNextRouter.cancelQueuedMessage', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockVerifyUserOwnsSessionV2ByCloudAgentId.mockResolvedValue({
+      kiloSessionId: 'ses_12345678901234567890123456',
+    });
+    mockCancelQueuedMessage.mockResolvedValue({ dropped: true });
+  });
+
+  it('denies canceling a queued message on a session the user does not own', async () => {
+    mockVerifyUserOwnsSessionV2ByCloudAgentId.mockResolvedValueOnce(null);
+    const caller = createCaller({ user: { id: 'user-1', is_admin: false } as User });
+
+    await expect(
+      caller.cancelQueuedMessage({
+        sessionId: 'agent_123',
+        messageId: 'msg_123456789abc123456789ABCDE',
+      })
+    ).rejects.toThrow('Session not found or access denied');
+
+    expect(mockCancelQueuedMessage).not.toHaveBeenCalled();
+  });
+});
+
 describe('cloudAgentNextRouter helper procedures', () => {
   beforeEach(() => {
     jest.clearAllMocks();

apps/web/src/routers/cloud-agent-next-router.ts

@@ -23,6 +23,7 @@ import {
   baseInitiateSessionNextOutputSchema,
   baseSendMessageNextSchema,
   baseInterruptSessionNextSchema,
+  baseCancelQueuedMessageNextSchema,
   baseGetSessionNextSchema,
   baseGetSessionNextOutputSchema,
   baseAnswerQuestionNextSchema,
@@ -460,6 +461,22 @@ export const cloudAgentNextRouter = createTRPCRouter({
       return await client.interruptSession(input.sessionId);
     }),
 
+  /**
+   * Cancel one queued (not yet accepted) message by id. Never interrupts the
+   * active run; a missing id or the accepted current message returns
+   * `{ dropped: false }`.
+   */
+  cancelQueuedMessage: baseProcedure
+    .input(baseCancelQueuedMessageNextSchema)
+    .output(z.object({ dropped: z.boolean() }))
+    .mutation(async ({ ctx, input }) => {
+      await assertUserOwnsSession(ctx.user.id, input.sessionId);
+      const authToken = generateCloudAgentToken(ctx.user);
+      const client = createCloudAgentNextClient(authToken);
+
+      return await client.cancelQueuedMessage(input.sessionId, input.messageId);
+    }),
+
   answerQuestion: baseProcedure
     .input(baseAnswerQuestionNextSchema)
     .output(z.object({ success: z.boolean() }))

apps/web/src/routers/cloud-agent-next-schemas.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from '@jest/globals';
 import {
   basePrepareSessionNextSchema,
+  baseCancelQueuedMessageNextSchema,
   cloudAgentGetAttachmentDownloadUrlSchema,
   cloudAgentGetAttachmentUploadUrlSchema,
   cloudAgentRelaxedAttachmentFilenameSchema,
@@ -264,3 +265,25 @@ describe('basePrepareSessionNextSchema cloneFromKiloSessionId union', () => {
     expect(result.success).toBe(true);
   });
 });
+
+describe('baseCancelQueuedMessageNextSchema', () => {
+  const VALID_MESSAGE_ID = 'msg_123456789abc123456789ABCDE';
+
+  it('accepts a session id with a message id', () => {
+    expect(
+      baseCancelQueuedMessageNextSchema.safeParse({
+        sessionId: 'agent_123',
+        messageId: VALID_MESSAGE_ID,
+      }).success
+    ).toBe(true);
+  });
+
+  it('requires both sessionId and messageId', () => {
+    expect(baseCancelQueuedMessageNextSchema.safeParse({ sessionId: 'agent_123' }).success).toBe(
+      false
+    );
+    expect(
+      baseCancelQueuedMessageNextSchema.safeParse({ messageId: VALID_MESSAGE_ID }).success
+    ).toBe(false);
+  });
+});

apps/web/src/routers/cloud-agent-next-schemas.ts

@@ -517,6 +517,12 @@ export const baseInterruptSessionNextSchema = z.object({
   sessionId: z.string(),
 });
 
+// Schema for canceling one queued (not yet accepted) message by id.
+export const baseCancelQueuedMessageNextSchema = z.object({
+  sessionId: z.string(),
+  messageId: messageIdNextSchema,
+});
+
 // Schema for getting session state
 export const baseGetSessionNextSchema = z.object({
   cloudAgentSessionId: z.string(),

apps/web/src/routers/organizations/organization-cloud-agent-next-router.test.ts

@@ -58,10 +58,14 @@ const mockGenerateCloudAgentAttachmentUploadUrl = jest.fn<
 
 const mockGetSession = jest.fn<(cloudAgentSessionId: string) => Promise<{ model?: string }>>();
 
+const mockCancelQueuedMessage =
+  jest.fn<(input: { sessionId: string; messageId: string }) => Promise<{ dropped: boolean }>>();
+
 const mockCreateCloudAgentNextClient = jest.fn(() => ({
   prepareSession: mockPrepareSession,
   sendMessage: mockSendMessage,
   getSession: mockGetSession,
+  cancelQueuedMessage: mockCancelQueuedMessage,
 }));
 
 const mockCreateCloudAgentNextClientForModel = jest.fn(
@@ -218,6 +222,11 @@ let createCaller: (ctx: { user: User }) => {
     contentType: 'text/markdown';
     contentLength: number;
   }) => Promise<unknown>;
+  cancelQueuedMessage: (input: {
+    organizationId: string;
+    sessionId: string;
+    messageId: string;
+  }) => Promise<unknown>;
   listBitbucketRepositories: (input: {
     organizationId: string;
     forceRefresh?: boolean;
@@ -472,6 +481,32 @@ describe('organizationCloudAgentNextRouter attachment forwarding', () => {
   });
 });
 
+describe('organizationCloudAgentNextRouter.cancelQueuedMessage', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockEnsureOrganizationAccess.mockImplementation(() => undefined);
+    mockVerifyOrgOwnsSessionV2ByCloudAgentId.mockResolvedValue({
+      kiloSessionId: 'ses_12345678901234567890123456',
+    });
+    mockCancelQueuedMessage.mockResolvedValue({ dropped: true });
+  });
+
+  it('denies canceling a queued message on a session outside the organization', async () => {
+    mockVerifyOrgOwnsSessionV2ByCloudAgentId.mockResolvedValueOnce(null);
+    const caller = createCaller({ user: { id: 'user-1', is_admin: false } as User });
+
+    await expect(
+      caller.cancelQueuedMessage({
+        organizationId: ORGANIZATION_ID,
+        sessionId: 'agent_123',
+        messageId: 'msg_123456789abc123456789ABCDE',
+      })
+    ).rejects.toThrow('Organization does not own this session');
+
+    expect(mockCancelQueuedMessage).not.toHaveBeenCalled();
+  });
+});
+
 describe('organizationCloudAgentNextRouter helper procedures', () => {
   beforeEach(() => {
     jest.clearAllMocks();

apps/web/src/routers/organizations/organization-cloud-agent-next-router.ts

@@ -31,6 +31,7 @@ import {
   baseInitiateSessionNextOutputSchema,
   baseSendMessageNextSchema,
   baseInterruptSessionNextSchema,
+  baseCancelQueuedMessageNextSchema,
   baseGetSessionNextSchema,
   baseGetSessionNextOutputSchema,
   baseAnswerQuestionNextSchema,
@@ -141,6 +142,10 @@ const InterruptSessionInput = baseInterruptSessionNextSchema.extend({
   organizationId: z.uuid(),
 });
 
+const CancelQueuedMessageInput = baseCancelQueuedMessageNextSchema.extend({
+  organizationId: z.uuid(),
+});
+
 const ImageUploadUrlInput = cloudAgentGetImageUploadUrlSchema.extend({
   organizationId: z.uuid(),
 });
@@ -606,6 +611,26 @@ export const organizationCloudAgentNextRouter = createTRPCRouter({
       return await client.interruptSession(input.sessionId);
     }),
 
+  /**
+   * Cancel one queued (not yet accepted) message by id. Never interrupts the
+   * active run; a missing id or the accepted current message returns
+   * `{ dropped: false }`.
+   */
+  cancelQueuedMessage: organizationMemberMutationProcedure
+    .input(CancelQueuedMessageInput)
+    .output(z.object({ dropped: z.boolean() }))
+    .mutation(async ({ ctx, input }) => {
+      await assertOrganizationOwnsSession({
+        organizationId: input.organizationId,
+        userId: ctx.user.id,
+        cloudAgentSessionId: input.sessionId,
+      });
+      const authToken = generateCloudAgentToken(ctx.user);
+      const client = createCloudAgentNextClient(authToken);
+
+      return await client.cancelQueuedMessage(input.sessionId, input.messageId);
+    }),
+
   answerQuestion: organizationMemberMutationProcedure
     .input(AnswerQuestionInput)
     .output(z.object({ success: z.boolean() }))

packages/cloud-agent-sdk/src/cli-live-transport.test.ts

@@ -1062,6 +1062,74 @@ describe('CliLiveTransport unified user web connection', () => {
     transport.destroy();
   });
 
+  it('includes messageID on send_message when the client assigns a message id', async () => {
+    const connection = createConnection();
+    jest
+      .mocked(connection.sendCommand)
+      .mockImplementation((_sessionId, command) =>
+        Promise.resolve(command === 'list_models' ? WIRE_CATALOG : { ok: true })
+      );
+    const { transport } = createTransportWithSinks({ connection });
+
+    transport.connect();
+    emitOwner(connection);
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    jest.mocked(connection.sendCommand).mockClear();
+
+    await transport.send?.({
+      payload: { type: 'prompt', prompt: 'hello' },
+      messageId: 'msg-queued-1',
+    });
+
+    expect(connection.sendCommand).toHaveBeenCalledWith(
+      KILO_SESSION_ID,
+      'send_message',
+      {
+        sessionID: KILO_SESSION_ID,
+        parts: [{ type: 'text', text: 'hello' }],
+        messageID: 'msg-queued-1',
+      },
+      'owner'
+    );
+    transport.destroy();
+  });
+
+  it('relays drop_queued_message with messageID without sending interrupt', async () => {
+    const connection = createConnection();
+    jest
+      .mocked(connection.sendCommand)
+      .mockImplementation((_sessionId, command) =>
+        Promise.resolve(command === 'list_models' ? WIRE_CATALOG : { ok: true })
+      );
+    const { transport } = createTransportWithSinks({ connection });
+
+    transport.connect();
+    emitOwner(connection);
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    jest.mocked(connection.sendCommand).mockClear();
+
+    await transport.dropQueuedMessage?.('msg-drop-1');
+
+    expect(connection.sendCommand).toHaveBeenCalledTimes(1);
+    expect(connection.sendCommand).toHaveBeenCalledWith(
+      KILO_SESSION_ID,
+      'drop_queued_message',
+      { protocolVersion: 1, messageID: 'msg-drop-1' },
+      'owner'
+    );
+    expect(connection.sendCommand).not.toHaveBeenCalledWith(
+      KILO_SESSION_ID,
+      'interrupt',
+      expect.anything(),
+      'owner'
+    );
+    transport.destroy();
+  });
+
   it.each([
     ['Kilo', { providerID: 'kilo', modelID: 'anthropic/claude-sonnet-4' }],
     ['non-Kilo', { providerID: 'anthropic', modelID: 'claude-sonnet-4' }],

packages/cloud-agent-sdk/src/cli-live-transport.ts

@@ -990,6 +990,10 @@ function createCliLiveTransport(config: CliLiveTransportConfig): TransportFactor
         return sendCommand('send_message', {
           sessionID: config.kiloSessionId,
           parts,
+          // Old form is `send_message` without `messageID`; include it once the
+          // client assigns an id so the CLI can correlate the queued turn.
+          // Remove the omission when every client sends it.
+          ...(input.messageId ? { messageID: input.messageId } : {}),
           ...(payload.mode ? { agent: payload.mode } : {}),
           ...(remoteModel.kind === 'none'
             ? {}
@@ -1000,6 +1004,8 @@ function createCliLiveTransport(config: CliLiveTransportConfig): TransportFactor
         });
       },
       interrupt: () => sendCommand('interrupt', {}),
+      dropQueuedMessage: messageId =>
+        sendCommand('drop_queued_message', { protocolVersion: 1, messageID: messageId }),
       answer: payload =>
         sendCommand('question_reply', {
           requestID: payload.requestId,