Merge pull request #481 from buster-so/dallin/bus-1328-ability-to-kill-a-chat-while-running-stop-button

Add '@buster/ai' dependency and enhance chat cancellation logic

.github/workflows/database-migrations.yml

@@ -3,6 +3,10 @@ name: Database Migrations
 on:
   push:
     branches: [main, staging]
+    paths:
+      - 'packages/database/drizzle/**'
+      - 'packages/database/drizzle.config.ts'
+      - '.github/workflows/database-migrations.yml'
 
 jobs:
   migrate:

apps/server/package.json

@@ -9,7 +9,8 @@
   "scripts": {
     "prebuild": "bun run scripts/validate-env.js && pnpm run typecheck",
     "build": "tsup",
-    "dev": "tsup --watch",
+    "dev": "bun --watch src/index.ts",
+    "dev:build": "tsup --watch",
     "lint": "biome check",
     "start": "bun dist/index.js",
     "test": "vitest run",
@@ -19,6 +20,7 @@
   },
   "dependencies": {
     "@buster/access-controls": "workspace:*",
+    "@buster/ai": "workspace:*",
     "@buster/database": "workspace:*",
     "@buster/server-shared": "workspace:*",
     "@buster/slack": "workspace:*",
@@ -29,6 +31,7 @@
     "@supabase/supabase-js": "catalog:",
     "@trigger.dev/sdk": "catalog:",
     "drizzle-orm": "catalog:",
+    "ai": "catalog:",
     "hono": "catalog:",
     "hono-pino": "^0.9.1",
     "pino": "^9.7.0",

apps/server/src/api/v2/chats/cancel-chat.ts

@@ -0,0 +1,311 @@
+import { canUserAccessChatCached } from '@buster/access-controls';
+import {
+  type ToolCallContent,
+  type ToolResultContent,
+  isToolCallContent,
+  isToolResultContent,
+} from '@buster/ai/utils/database/types';
+import type { User } from '@buster/database';
+import { and, eq, isNotNull, updateMessageFields } from '@buster/database';
+import { db, messages } from '@buster/database';
+import type {
+  ChatMessageReasoningMessage,
+  ChatMessageResponseMessage,
+} from '@buster/server-shared/chats';
+import { runs } from '@trigger.dev/sdk';
+import type { CoreMessage } from 'ai';
+import { errorResponse } from '../../../utils/response';
+
+/**
+ * Cancel a chat and clean up any incomplete messages
+ *
+ * Strategy:
+ * 1. Cancel the trigger runs
+ * 2. Fetch fresh data and clean up messages
+ * 3. Mark messages as completed with proper cleanup
+ */
+export async function cancelChatHandler(chatId: string, user: User): Promise<void> {
+  const userHasAccessToChat = await canUserAccessChatCached({
+    userId: user.id,
+    chatId,
+  });
+
+  if (!userHasAccessToChat) {
+    throw errorResponse('You do not have access to this chat', 403);
+  }
+
+  // First, query just for IDs and trigger run IDs
+  const messagesToCancel = await db
+    .select({
+      id: messages.id,
+      triggerRunId: messages.triggerRunId,
+    })
+    .from(messages)
+    .where(
+      and(
+        eq(messages.chatId, chatId),
+        eq(messages.isCompleted, false),
+        isNotNull(messages.triggerRunId)
+      )
+    );
+
+  // Type narrow to ensure triggerRunId is not null
+  const incompleteTriggerMessages = messagesToCancel.filter(
+    (result): result is { id: string; triggerRunId: string } => result.triggerRunId !== null
+  );
+
+  // Cancel all trigger runs first
+  const cancellationPromises = incompleteTriggerMessages.map(async (message) => {
+    try {
+      await runs.cancel(message.triggerRunId);
+      console.info(`Cancelled trigger run ${message.triggerRunId} for message ${message.id}`);
+    } catch (error) {
+      console.error(`Failed to cancel trigger run ${message.triggerRunId}:`, error);
+      // Continue with cleanup even if cancellation fails
+    }
+  });
+
+  // Wait for all cancellations to complete
+  await Promise.allSettled(cancellationPromises);
+
+  await new Promise((resolve) => setTimeout(resolve, 500));
+
+  // Now fetch the latest message data and clean up each message
+  const cleanupPromises = incompleteTriggerMessages.map(async (message) => {
+    // Fetch the latest message data
+    const [latestMessageData] = await db
+      .select({
+        rawLlmMessages: messages.rawLlmMessages,
+        reasoning: messages.reasoning,
+        responseMessages: messages.responseMessages,
+      })
+      .from(messages)
+      .where(eq(messages.id, message.id));
+
+    if (latestMessageData) {
+      await cleanUpMessage(
+        message.id,
+        latestMessageData.rawLlmMessages,
+        latestMessageData.reasoning,
+        latestMessageData.responseMessages
+      );
+    }
+  });
+
+  // Wait for all cleanups to complete
+  await Promise.allSettled(cleanupPromises);
+}
+/**
+ * Find tool calls without corresponding tool results
+ */
+function findIncompleteToolCalls(messages: CoreMessage[]): ToolCallContent[] {
+  const toolCalls = new Map<string, ToolCallContent>();
+  const toolResults = new Set<string>();
+
+  // First pass: collect all tool calls and tool results
+  for (const message of messages) {
+    if (message.role === 'assistant' && Array.isArray(message.content)) {
+      for (const content of message.content) {
+        if (isToolCallContent(content)) {
+          toolCalls.set(content.toolCallId, content);
+        }
+      }
+    } else if (message.role === 'tool' && Array.isArray(message.content)) {
+      for (const content of message.content) {
+        if (isToolResultContent(content)) {
+          toolResults.add(content.toolCallId);
+        }
+      }
+    }
+  }
+
+  // Second pass: find tool calls without results
+  const incompleteToolCalls: ToolCallContent[] = [];
+  for (const [toolCallId, toolCall] of toolCalls) {
+    if (!toolResults.has(toolCallId)) {
+      incompleteToolCalls.push(toolCall);
+    }
+  }
+
+  return incompleteToolCalls;
+}
+
+/**
+ * Create tool result messages for incomplete tool calls
+ */
+function createCancellationToolResults(incompleteToolCalls: ToolCallContent[]): CoreMessage[] {
+  if (incompleteToolCalls.length === 0) {
+    return [];
+  }
+
+  const toolResultMessages: CoreMessage[] = [];
+
+  for (const toolCall of incompleteToolCalls) {
+    const toolResult: ToolResultContent = {
+      type: 'tool-result',
+      toolCallId: toolCall.toolCallId,
+      toolName: toolCall.toolName,
+      result: {
+        error: true,
+        message: 'The user ended the chat',
+      },
+    };
+
+    toolResultMessages.push({
+      role: 'tool',
+      content: [toolResult],
+    });
+  }
+
+  return toolResultMessages;
+}
+
+/**
+ * Clean up messages by adding tool results for incomplete tool calls
+ */
+function cleanUpRawLlmMessages(messages: CoreMessage[]): CoreMessage[] {
+  const incompleteToolCalls = findIncompleteToolCalls(messages);
+
+  if (incompleteToolCalls.length === 0) {
+    return messages;
+  }
+
+  // Create tool result messages for incomplete tool calls
+  const toolResultMessages = createCancellationToolResults(incompleteToolCalls);
+
+  // Append tool results to the messages
+  return [...messages, ...toolResultMessages];
+}
+
+/**
+ * Ensure reasoning messages are marked as completed
+ */
+function ensureReasoningMessagesCompleted(
+  reasoning: ChatMessageReasoningMessage[]
+): ChatMessageReasoningMessage[] {
+  console.info('Ensuring reasoning messages are completed:', {
+    totalMessages: reasoning.length,
+    loadingMessages: reasoning.filter(
+      (msg) => msg && typeof msg === 'object' && 'status' in msg && msg.status === 'loading'
+    ).length,
+  });
+
+  return reasoning.map((msg, index) => {
+    if (msg && typeof msg === 'object' && 'status' in msg && msg.status === 'loading') {
+      console.info(`Marking reasoning message ${index} as completed:`, {
+        id: 'id' in msg ? msg.id : 'unknown',
+        title: 'title' in msg ? msg.title : 'unknown',
+        previousStatus: msg.status,
+      });
+      return {
+        ...msg,
+        status: 'completed' as const,
+      };
+    }
+    return msg;
+  });
+}
+
+/**
+ * Clean up and finalize all message fields for a cancelled chat
+ */
+interface CleanedMessageFields {
+  rawLlmMessages: CoreMessage[];
+  reasoning: ChatMessageReasoningMessage[];
+  responseMessages: ChatMessageResponseMessage[];
+}
+
+function cleanUpMessageFields(
+  rawLlmMessages: CoreMessage[],
+  reasoning: ChatMessageReasoningMessage[],
+  responseMessages: ChatMessageResponseMessage[]
+): CleanedMessageFields {
+  // Clean up raw LLM messages by adding tool results for incomplete tool calls
+  const cleanedRawMessages = cleanUpRawLlmMessages(rawLlmMessages);
+
+  // Ensure all reasoning messages are marked as completed
+  const completedReasoning = ensureReasoningMessagesCompleted(reasoning);
+
+  return {
+    rawLlmMessages: cleanedRawMessages,
+    reasoning: completedReasoning,
+    responseMessages: responseMessages,
+  };
+}
+
+async function cleanUpMessage(
+  messageId: string,
+  rawLlmMessages: unknown,
+  reasoning: unknown,
+  responseMessages: unknown
+): Promise<void> {
+  try {
+    // Parse and validate the message fields
+    const currentRawMessages = Array.isArray(rawLlmMessages)
+      ? (rawLlmMessages as CoreMessage[])
+      : [];
+    const currentReasoning = Array.isArray(reasoning)
+      ? (reasoning as ChatMessageReasoningMessage[])
+      : [];
+
+    // Handle responseMessages which could be an array or object
+    let currentResponseMessages: ChatMessageResponseMessage[] = [];
+    if (Array.isArray(responseMessages)) {
+      currentResponseMessages = responseMessages as ChatMessageResponseMessage[];
+    } else if (responseMessages && typeof responseMessages === 'object') {
+      // Convert object to array if it has values
+      const values = Object.values(responseMessages);
+      if (values.length > 0 && values.every((v) => v && typeof v === 'object')) {
+        currentResponseMessages = values as ChatMessageResponseMessage[];
+      }
+    }
+
+    console.info(`Cleaning up message ${messageId}:`, {
+      rawMessagesCount: currentRawMessages.length,
+      reasoningCount: currentReasoning.length,
+      responseMessagesCount: currentResponseMessages.length,
+      responseMessagesType: Array.isArray(responseMessages) ? 'array' : typeof responseMessages,
+    });
+
+    // Clean up all message fields
+    const cleanedFields = cleanUpMessageFields(
+      currentRawMessages,
+      currentReasoning,
+      currentResponseMessages
+    );
+
+    // Determine the final reasoning message based on whether we were in response phase
+    const hasResponseMessages = currentResponseMessages.length > 0;
+    const finalReasoningMessage = hasResponseMessages
+      ? 'Stopped during final response'
+      : 'Stopped reasoning';
+
+    // Log the cleaned reasoning to debug
+    console.info('Cleaned reasoning before save:', {
+      reasoningCount: cleanedFields.reasoning.length,
+      loadingCount: cleanedFields.reasoning.filter(
+        (r) => r && typeof r === 'object' && 'status' in r && r.status === 'loading'
+      ).length,
+      lastReasoningMessage: cleanedFields.reasoning[cleanedFields.reasoning.length - 1],
+    });
+
+    // Ensure the reasoning array is properly serializable
+    const serializableReasoning = JSON.parse(JSON.stringify(cleanedFields.reasoning));
+
+    // Update the message in the database
+    await updateMessageFields(messageId, {
+      rawLlmMessages: cleanedFields.rawLlmMessages,
+      reasoning: serializableReasoning,
+      responseMessages: cleanedFields.responseMessages,
+      finalReasoningMessage: finalReasoningMessage,
+      isCompleted: true,
+    });
+
+    console.info(
+      `Successfully cleaned up message ${messageId} with finalReasoningMessage: ${finalReasoningMessage}`
+    );
+  } catch (error) {
+    console.error(`Failed to clean up message ${messageId}:`, error);
+    // Don't throw - we want to continue processing other messages
+  }
+}

apps/server/src/api/v2/chats/handler.ts

@@ -88,6 +88,12 @@ export async function createChatHandler(
           throw new Error('Trigger service returned invalid handle');
         }
 
+        // Update the message with the trigger run ID
+        const { updateMessage } = await import('@buster/database');
+        await updateMessage(messageId, {
+          triggerRunId: taskHandle.id,
+        });
+
         // Task was successfully queued - background analysis will proceed
       } catch (triggerError) {
         console.error('Failed to trigger analyst agent task:', triggerError);

apps/server/src/api/v2/chats/index.ts

@@ -3,6 +3,7 @@ import {
   ChatError,
   type ChatWithMessages,
   ChatWithMessagesSchema,
+  CancelChatParamsSchema,
 } from '@buster/server-shared/chats';
 import { zValidator } from '@hono/zod-validator';
 import { Hono } from 'hono';
@@ -11,6 +12,7 @@ import '../../../types/hono.types'; //I added this to fix intermitent type error
 import { HTTPException } from 'hono/http-exception';
 import { z } from 'zod';
 import { createChatHandler } from './handler';
+import { cancelChatHandler } from './cancel-chat';
 
 const app = new Hono()
   // Apply authentication middleware
@@ -48,6 +50,14 @@ const app = new Hono()
       });
     }
   )
+  // DELETE /chats/:chat_id/cancel - Cancel a chat and its running triggers
+  .delete('/:chat_id/cancel', zValidator('param', CancelChatParamsSchema), async (c) => {
+    const params = c.req.valid('param');
+    const user = c.get('busterUser');
+    
+    await cancelChatHandler(params.chat_id, user);
+    return c.json({ success: true, message: 'Chat cancelled successfully' });
+  })
   .onError((e, c) => {
     if (e instanceof ChatError) {
       // we need to use this syntax instead of HTTPException because hono bubbles up 500 errors

packages/data-source/tests/integration/adapters/snowflake.test.ts

@@ -523,7 +523,7 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle connection drops gracefully',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // Simulate network interruption by destroying connection
       // @ts-expect-error - Testing private property
       const conn = adapter.connection;
@@ -534,10 +534,10 @@ describe('SnowflakeAdapter Integration', () => {
           });
         });
       }
-      
+
       // Next query should fail
       await expect(adapter.query('SELECT 1')).rejects.toThrow();
-      
+
       // But adapter should be able to reinitialize
       await adapter.initialize(credentials);
       const result = await adapter.query('SELECT 1 as test');
@@ -550,7 +550,7 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle query cancellation',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // Start a long-running query and cancel it
       const longQuery = adapter.query(
         `SELECT COUNT(*) FROM SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.LINEITEM 
@@ -559,9 +559,9 @@ describe('SnowflakeAdapter Integration', () => {
         undefined,
         100 // Very short timeout to force cancellation
       );
-      
+
       await expect(longQuery).rejects.toThrow(/timeout/i);
-      
+
       // Should be able to run another query immediately
       const result = await adapter.query('SELECT 1 as test');
       expect(result.rows[0].TEST).toBe(1);
@@ -573,10 +573,10 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle very long strings in queries',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // Test with a very long string (1MB)
       const veryLongString = 'x'.repeat(1000000);
-      
+
       // This should work but might be slow
       const result = await adapter.query(
         `SELECT LENGTH('${veryLongString}') as str_length`,
@@ -584,7 +584,7 @@ describe('SnowflakeAdapter Integration', () => {
         undefined,
         60000 // 60 second timeout for large string
       );
-      
+
       expect(result.rows[0].STR_LENGTH).toBe(1000000);
     },
     120000 // 2 minute timeout for this test
@@ -594,7 +594,7 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle Unicode and special characters',
     async () => {
       await adapter.initialize(credentials);
-      
+
       const result = await adapter.query(`
         SELECT 
           '🎉emoji🎊' as emoji_text,
@@ -603,7 +603,7 @@ describe('SnowflakeAdapter Integration', () => {
           'Special: <>&"\\/' as special_chars,
           'Line' || CHR(10) || 'Break' as line_break
       `);
-      
+
       expect(result.rows[0].EMOJI_TEXT).toBe('🎉emoji🎊');
       expect(result.rows[0].CHINESE_TEXT).toBe('Chinese: 你好');
       expect(result.rows[0].ARABIC_TEXT).toBe('Arabic: مرحبا');
@@ -617,14 +617,14 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle extremely large result sets with maxRows',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // Query that would return millions of rows
       const result = await adapter.query(
         'SELECT * FROM SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.LINEITEM',
         undefined,
         1000 // Limit to 1000 rows
       );
-      
+
       expect(result.rows.length).toBe(1000);
       expect(result.hasMoreRows).toBe(true);
       expect(result.rowCount).toBe(1000);
@@ -636,19 +636,19 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle rapid connection cycling',
     async () => {
       const results = [];
-      
+
       // Rapidly create, use, and close connections
       for (let i = 0; i < 5; i++) {
         const tempAdapter = new SnowflakeAdapter();
         await tempAdapter.initialize(credentials);
-        
+
         const result = await tempAdapter.query(`SELECT ${i} as cycle_num`);
         results.push(result.rows[0].CYCLE_NUM);
-        
+
         await tempAdapter.close();
         // No delay - test rapid cycling
       }
-      
+
       expect(results).toEqual([0, 1, 2, 3, 4]);
     },
     60000 // 1 minute for rapid cycling
@@ -658,14 +658,14 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle warehouse suspension gracefully',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // First query to ensure warehouse is running
       await adapter.query('SELECT 1');
-      
+
       // Note: In production, warehouse might auto-suspend
       // This test simulates querying after potential suspension
-      await new Promise(resolve => setTimeout(resolve, 5000)); // 5 second delay
-      
+      await new Promise((resolve) => setTimeout(resolve, 5000)); // 5 second delay
+
       // Should still work (Snowflake should auto-resume)
       const result = await adapter.query('SELECT 2 as test');
       expect(result.rows[0].TEST).toBe(2);
@@ -677,7 +677,7 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle malformed SQL gracefully',
     async () => {
       await adapter.initialize(credentials);
-      
+
       const malformedQueries = [
         'SELECT * FROM', // Incomplete
         'SELCT * FROM table', // Typo
@@ -685,11 +685,11 @@ describe('SnowflakeAdapter Integration', () => {
         'SELECT * FROM "non.existent.schema"."table"', // Invalid schema
         'SELECT 1; DROP TABLE test;', // Multiple statements
       ];
-      
+
       for (const query of malformedQueries) {
         await expect(adapter.query(query)).rejects.toThrow();
       }
-      
+
       // Should still be able to run valid queries
       const result = await adapter.query('SELECT 1 as test');
       expect(result.rows[0].TEST).toBe(1);
@@ -702,24 +702,25 @@ describe('SnowflakeAdapter Integration', () => {
     async () => {
       const adapters: SnowflakeAdapter[] = [];
       const promises: Promise<any>[] = [];
-      
+
       // Create many adapters without closing them (simulating pool exhaustion)
       for (let i = 0; i < 20; i++) {
         const tempAdapter = new SnowflakeAdapter();
         adapters.push(tempAdapter);
-        
-        const promise = tempAdapter.initialize(credentials)
+
+        const promise = tempAdapter
+          .initialize(credentials)
           .then(() => tempAdapter.query(`SELECT ${i} as num`));
-        
+
         promises.push(promise);
       }
-      
+
       // All should complete successfully
       const results = await Promise.all(promises);
       expect(results).toHaveLength(20);
-      
+
       // Cleanup
-      await Promise.all(adapters.map(a => a.close()));
+      await Promise.all(adapters.map((a) => a.close()));
     },
     120000 // 2 minutes for many connections
   );
@@ -728,17 +729,15 @@ describe('SnowflakeAdapter Integration', () => {
     'should maintain connection integrity under load',
     async () => {
       await adapter.initialize(credentials);
-      
+
       // Run many queries in parallel on same adapter
       const queryPromises = [];
       for (let i = 0; i < 50; i++) {
-        queryPromises.push(
-          adapter.query(`SELECT ${i} as num, CURRENT_TIMESTAMP() as ts`)
-        );
+        queryPromises.push(adapter.query(`SELECT ${i} as num, CURRENT_TIMESTAMP() as ts`));
       }
-      
+
       const results = await Promise.all(queryPromises);
-      
+
       // Verify all queries succeeded and returned correct data
       expect(results).toHaveLength(50);
       results.forEach((result, index) => {
@@ -753,14 +752,14 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle binary data correctly',
     async () => {
       await adapter.initialize(credentials);
-      
+
       const result = await adapter.query(`
         SELECT 
           TO_BINARY('48656C6C6F', 'HEX') as hex_binary,
           TO_BINARY('SGVsbG8=', 'BASE64') as base64_binary,
           BASE64_ENCODE(TO_BINARY('48656C6C6F', 'HEX')) as encoded_text
       `);
-      
+
       expect(result.rows[0].HEX_BINARY).toBeTruthy();
       expect(result.rows[0].BASE64_BINARY).toBeTruthy();
       expect(result.rows[0].ENCODED_TEXT).toBe('SGVsbG8=');
@@ -772,7 +771,7 @@ describe('SnowflakeAdapter Integration', () => {
     'should handle timezone-aware timestamps',
     async () => {
       await adapter.initialize(credentials);
-      
+
       const result = await adapter.query(`
         SELECT 
           CONVERT_TIMEZONE('UTC', 'America/New_York', '2024-01-01 12:00:00'::TIMESTAMP_NTZ) as ny_time,
@@ -780,7 +779,7 @@ describe('SnowflakeAdapter Integration', () => {
           CURRENT_TIMESTAMP() as current_ts,
           SYSDATE() as sys_date
       `);
-      
+
       expect(result.rows[0].NY_TIME).toBeTruthy();
       expect(result.rows[0].TOKYO_TIME).toBeTruthy();
       expect(result.rows[0].CURRENT_TS).toBeTruthy();

packages/database/drizzle/0078_adorable_layla_miller.sql

@@ -0,0 +1,2 @@
+-- Add trigger_run_id column to messages table
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS trigger_run_id text;

packages/database/drizzle/meta/_journal.json

@@ -547,6 +547,13 @@
       "when": 1752150366202,
       "tag": "0077_short_marvex",
       "breakpoints": true
+    },
+    {
+      "idx": 78,
+      "version": "7",
+      "when": 1752184722835,
+      "tag": "0078_adorable_layla_miller",
+      "breakpoints": true
     }
   ]
 }

packages/database/src/helpers/messages.ts

@@ -110,6 +110,7 @@ export async function updateMessageFields(
     reasoning?: unknown;
     rawLlmMessages?: unknown;
     finalReasoningMessage?: string;
+    isCompleted?: boolean;
   }
 ): Promise<{ success: boolean }> {
   try {
@@ -125,6 +126,7 @@ export async function updateMessageFields(
       reasoning?: unknown;
       rawLlmMessages?: unknown;
       finalReasoningMessage?: string;
+      isCompleted?: boolean;
     } = {
       updatedAt: new Date().toISOString(),
     };
@@ -143,6 +145,10 @@ export async function updateMessageFields(
       updateData.finalReasoningMessage = fields.finalReasoningMessage;
     }
 
+    if ('isCompleted' in fields) {
+      updateData.isCompleted = fields.isCompleted;
+    }
+
     await db
       .update(messages)
       .set(updateData)

packages/database/src/schema.ts

@@ -828,6 +828,7 @@ export const messages = pgTable(
     feedback: text(),
     isCompleted: boolean('is_completed').default(false).notNull(),
     postProcessingMessage: jsonb('post_processing_message'),
+    triggerRunId: text('trigger_run_id'),
   },
   (table) => [
     index('messages_chat_id_idx').using('btree', table.chatId.asc().nullsLast().op('uuid_ops')),

packages/server-shared/src/chats/chat.types.ts

@@ -60,9 +60,15 @@ export const ChatCreateHandlerRequestSchema = z.object({
   asset_type: AssetType.optional(),
 });
 
+// Cancel chat params schema
+export const CancelChatParamsSchema = z.object({
+  chat_id: z.string().uuid(),
+});
+
 // Infer types from schemas
 export type AssetPermissionRole = z.infer<typeof AssetPermissionRoleSchema>;
 export type BusterShareIndividual = z.infer<typeof BusterShareIndividualSchema>;
 export type ChatWithMessages = z.infer<typeof ChatWithMessagesSchema>;
 export type ChatCreateRequest = z.infer<typeof ChatCreateRequestSchema>;
 export type ChatCreateHandlerRequest = z.infer<typeof ChatCreateHandlerRequestSchema>;
+export type CancelChatParams = z.infer<typeof CancelChatParamsSchema>;

pnpm-lock.yaml

@@ -105,6 +105,9 @@ importers:
       '@buster/access-controls':
         specifier: workspace:*
         version: link:../../packages/access-controls
+      '@buster/ai':
+        specifier: workspace:*
+        version: link:../../packages/ai
       '@buster/database':
         specifier: workspace:*
         version: link:../../packages/database
@@ -132,6 +135,9 @@ importers:
       '@trigger.dev/sdk':
         specifier: 'catalog:'
         version: 4.0.0-v4-beta.22(ai@4.3.16(react@18.3.1)(zod@3.25.75))(zod@3.25.75)
+      ai:
+        specifier: 'catalog:'
+        version: 4.3.16(react@18.3.1)(zod@3.25.75)
       drizzle-orm:
         specifier: 'catalog:'
         version: 0.44.2(@opentelemetry/api@1.9.0)(@types/pg@8.15.4)(mysql2@3.14.1)(pg@8.16.3)(postgres@3.4.7)