fix: buffer HTML chunks to avoid flashing, flush on closing tags

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;

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., </div>, </h1>, </p>, </ul>, </li>)
+                // 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("</div>")
+                    || html_buffer.contains("</h1>")
+                    || html_buffer.contains("</h2>")
+                    || html_buffer.contains("</p>")
+                    || html_buffer.contains("</ul>")
+                    || html_buffer.contains("</ol>")
+                    || html_buffer.contains("</li>")
+                    || html_buffer.contains("</section>")
+                    || html_buffer.contains("</header>")
+                    || html_buffer.contains("</footer>");
 
-                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<crate::core::shared::models::Suggestion> = 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.

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

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

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

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

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

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

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<String>,
+    pub last_modified: Option<DateTime<Utc>>,
+    pub file_size: Option<i64>,
+    pub indexed: bool,
+    pub fail_count: i32,
+    pub last_failed_at: Option<DateTime<Utc>>,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+#[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<String>,
+    pub last_modified: Option<DateTime<Utc>>,
+    pub file_size: Option<i64>,
+    pub indexed: bool,
+}
+
+#[derive(AsChangeset, Debug)]
+#[diesel(table_name = drive_files)]
+pub struct DriveFileUpdate {
+    pub etag: Option<String>,
+    pub last_modified: Option<DateTime<Utc>>,
+    pub file_size: Option<i64>,
+    pub indexed: Option<bool>,
+    pub fail_count: Option<i32>,
+    pub last_failed_at: Option<DateTime<Utc>>,
+    pub updated_at: DateTime<Utc>,
+}
+
+diesel::table! {
+    drive_files (id) {
+        id -> Uuid,
+        bot_id -> Uuid,
+        file_path -> Text,
+        file_type -> Varchar,
+        etag -> Nullable<Text>,
+        last_modified -> Nullable<Timestamptz>,
+        file_size -> Nullable<Int8>,
+        indexed -> Bool,
+        fail_count -> Int4,
+        last_failed_at -> Nullable<Timestamptz>,
+        created_at -> Timestamptz,
+        updated_at -> Timestamptz,
+    }
+}

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

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

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

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::*;
-

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

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