diff --git a/migrations/6.3.1-01-drive-files/down.sql b/migrations/6.3.1-01-drive-files/down.sql new file mode 100644 index 00000000..23f379f7 --- /dev/null +++ b/migrations/6.3.1-01-drive-files/down.sql @@ -0,0 +1,6 @@ +-- ============================================ +-- Drive Files State Table - Rollback +-- Version: 6.3.1 +-- ============================================ + +DROP TABLE IF EXISTS drive_files; diff --git a/src/core/bot/mod.rs b/src/core/bot/mod.rs index 6e08b60a..6055b252 100644 --- a/src/core/bot/mod.rs +++ b/src/core/bot/mod.rs @@ -833,6 +833,7 @@ impl BotOrchestrator { let mut in_analysis = false; let mut tool_call_buffer = String::new(); // Accumulate potential tool call JSON chunks let mut accumulating_tool_call = false; // Track if we're currently accumulating a tool call + let mut html_buffer = String::new(); // Buffer for HTML content let _handler = llm_models::get_handler(&model); trace!("Using model handler for {}", model); @@ -1149,25 +1150,47 @@ impl BotOrchestrator { if !in_analysis { full_response.push_str(&chunk); + html_buffer.push_str(&chunk); - let response = BotResponse { - bot_id: message.bot_id.clone(), - user_id: message.user_id.clone(), - session_id: message.session_id.clone(), - channel: message.channel.clone(), - content: chunk.clone(), - message_type: MessageType::BOT_RESPONSE, - stream_token: None, - is_complete: false, - suggestions: Vec::new(), - context_name: None, - context_length: 0, - context_max_length: 0, - }; + // Check if we should flush the buffer: + // 1. HTML tag pair completed (e.g., , ,

, , ) + // 2. Buffer is large enough (> 500 chars) + // 3. This is the last chunk (is_complete will be true next iteration) + let should_flush = html_buffer.len() > 500 + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains("

") + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains("") + || html_buffer.contains(""); - if response_tx.send(response).await.is_err() { - warn!("Response channel closed"); - break; + if should_flush { + let content_to_send = html_buffer.clone(); + html_buffer.clear(); + + let response = BotResponse { + bot_id: message.bot_id.clone(), + user_id: message.user_id.clone(), + session_id: message.session_id.clone(), + channel: message.channel.clone(), + content: content_to_send, + message_type: MessageType::BOT_RESPONSE, + stream_token: None, + is_complete: false, + suggestions: Vec::new(), + context_name: None, + context_length: 0, + context_max_length: 0, + }; + + if response_tx.send(response).await.is_err() { + warn!("Response channel closed"); + break; + } } } } @@ -1208,6 +1231,27 @@ impl BotOrchestrator { #[cfg(not(feature = "chat"))] let suggestions: Vec = Vec::new(); + // Flush any remaining HTML buffer before sending final response + if !html_buffer.is_empty() { + trace!("Flushing remaining {} chars in HTML buffer", html_buffer.len()); + let final_chunk = BotResponse { + bot_id: message.bot_id.clone(), + user_id: message.user_id.clone(), + session_id: message.session_id.clone(), + channel: message.channel.clone(), + content: html_buffer.clone(), + message_type: MessageType::BOT_RESPONSE, + stream_token: None, + is_complete: false, + suggestions: Vec::new(), + context_name: None, + context_length: 0, + context_max_length: 0, + }; + let _ = response_tx.send(final_chunk).await; + html_buffer.clear(); + } + // Content was already sent as streaming chunks. // Sending full_response again would duplicate it (especially for WhatsApp which accumulates buffer). // The final response is just a signal that streaming is complete - it should not contain content. diff --git a/src/core/shared/schema/analytics.rs b/src/core/shared/schema/analytics.rs index 6706b263..498e0fd1 100644 --- a/src/core/shared/schema/analytics.rs +++ b/src/core/shared/schema/analytics.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { dashboards (id) { diff --git a/src/core/shared/schema/attendant.rs b/src/core/shared/schema/attendant.rs index 14e15040..a6cd9caf 100644 --- a/src/core/shared/schema/attendant.rs +++ b/src/core/shared/schema/attendant.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { attendant_queues (id) { diff --git a/src/core/shared/schema/billing.rs b/src/core/shared/schema/billing.rs index 2904d88a..c1847ca1 100644 --- a/src/core/shared/schema/billing.rs +++ b/src/core/shared/schema/billing.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { billing_invoices (id) { diff --git a/src/core/shared/schema/calendar.rs b/src/core/shared/schema/calendar.rs index 921220f2..c57a1d0b 100644 --- a/src/core/shared/schema/calendar.rs +++ b/src/core/shared/schema/calendar.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { calendars (id) { diff --git a/src/core/shared/schema/canvas.rs b/src/core/shared/schema/canvas.rs index e6cc5498..f1e7c696 100644 --- a/src/core/shared/schema/canvas.rs +++ b/src/core/shared/schema/canvas.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { canvases (id) { diff --git a/src/core/shared/schema/compliance.rs b/src/core/shared/schema/compliance.rs index 5d8fb798..5c33645d 100644 --- a/src/core/shared/schema/compliance.rs +++ b/src/core/shared/schema/compliance.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { legal_documents (id) { diff --git a/src/core/shared/schema/drive.rs b/src/core/shared/schema/drive.rs new file mode 100644 index 00000000..fc06b425 --- /dev/null +++ b/src/core/shared/schema/drive.rs @@ -0,0 +1,61 @@ +use chrono::{DateTime, Utc}; +use diesel::prelude::*; +use uuid::Uuid; + +#[derive(Queryable, Insertable, AsChangeset, Debug, Clone)] +#[diesel(table_name = drive_files)] +pub struct DriveFile { + pub id: Uuid, + pub bot_id: Uuid, + pub file_path: String, + pub file_type: String, + pub etag: Option, + pub last_modified: Option>, + pub file_size: Option, + pub indexed: bool, + pub fail_count: i32, + pub last_failed_at: Option>, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +#[derive(Insertable, Debug)] +#[diesel(table_name = drive_files)] +pub struct NewDriveFile { + pub bot_id: Uuid, + pub file_path: String, + pub file_type: String, + pub etag: Option, + pub last_modified: Option>, + pub file_size: Option, + pub indexed: bool, +} + +#[derive(AsChangeset, Debug)] +#[diesel(table_name = drive_files)] +pub struct DriveFileUpdate { + pub etag: Option, + pub last_modified: Option>, + pub file_size: Option, + pub indexed: Option, + pub fail_count: Option, + pub last_failed_at: Option>, + pub updated_at: DateTime, +} + +diesel::table! { + drive_files (id) { + id -> Uuid, + bot_id -> Uuid, + file_path -> Text, + file_type -> Varchar, + etag -> Nullable, + last_modified -> Nullable, + file_size -> Nullable, + indexed -> Bool, + fail_count -> Int4, + last_failed_at -> Nullable, + created_at -> Timestamptz, + updated_at -> Timestamptz, + } +} diff --git a/src/core/shared/schema/goals.rs b/src/core/shared/schema/goals.rs index b94780f6..70f31900 100644 --- a/src/core/shared/schema/goals.rs +++ b/src/core/shared/schema/goals.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { okr_objectives (id) { diff --git a/src/core/shared/schema/kb.rs b/src/core/shared/schema/kb.rs index eca8da7d..67fde463 100644 --- a/src/core/shared/schema/kb.rs +++ b/src/core/shared/schema/kb.rs @@ -30,7 +30,4 @@ diesel::joinable!(kb_group_associations -> kb_collections (kb_id)); diesel::joinable!(kb_group_associations -> rbac_groups (group_id)); diesel::joinable!(kb_group_associations -> users (granted_by)); -diesel::allow_tables_to_appear_in_same_query!( - kb_collections, - kb_group_associations, -); +diesel::allow_tables_to_appear_in_same_query!(kb_collections, kb_group_associations,); diff --git a/src/core/shared/schema/meet.rs b/src/core/shared/schema/meet.rs index ed2f606d..2637ba27 100644 --- a/src/core/shared/schema/meet.rs +++ b/src/core/shared/schema/meet.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { meeting_rooms (id) { diff --git a/src/core/shared/schema/mod.rs b/src/core/shared/schema/mod.rs index e1882b53..611dc624 100644 --- a/src/core/shared/schema/mod.rs +++ b/src/core/shared/schema/mod.rs @@ -89,7 +89,10 @@ pub mod project; #[cfg(feature = "dashboards")] pub mod dashboards; +// Drive (always available - used by DriveMonitor) +pub mod drive; +pub use self::drive::*; + // Email integration (always available) pub mod email_integration; pub use self::email_integration::*; - diff --git a/src/core/shared/schema/social.rs b/src/core/shared/schema/social.rs index 179a2705..f58c0b90 100644 --- a/src/core/shared/schema/social.rs +++ b/src/core/shared/schema/social.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { social_communities (id) { diff --git a/src/core/shared/schema/workspaces.rs b/src/core/shared/schema/workspaces.rs index 074462f2..1b7eec12 100644 --- a/src/core/shared/schema/workspaces.rs +++ b/src/core/shared/schema/workspaces.rs @@ -1,4 +1,4 @@ -use crate::core::shared::schema::core::{organizations, bots}; +use crate::core::shared::schema::core::{bots, organizations}; diesel::table! { workspaces (id) {