feat(ai): run agent sessions as background tasks

crates/dbx-core/src/agent_tools.rs

@@ -1,4 +1,5 @@
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex as StdMutex, OnceLock, Weak};
 
 use serde_json::json;
 
@@ -36,6 +37,32 @@ const MAX_QUERY_CELL_CHAR_LIMIT: usize = 4_000;
 /// Bounds explicit sliding-window scans without changing the underlying database request.
 const MAX_QUERY_CELL_CHAR_OFFSET: usize = 1_000_000;
 
+fn connection_tool_lock(connection_id: &str) -> Arc<tokio::sync::Mutex<()>> {
+    static LOCKS: OnceLock<StdMutex<HashMap<String, Weak<tokio::sync::Mutex<()>>>>> = OnceLock::new();
+    let locks = LOCKS.get_or_init(|| StdMutex::new(HashMap::new()));
+    let mut registry = locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    registry.retain(|_, lock| lock.strong_count() > 0);
+    if let Some(lock) = registry.get(connection_id).and_then(Weak::upgrade) {
+        return lock;
+    }
+    let lock = Arc::new(tokio::sync::Mutex::new(()));
+    registry.insert(connection_id.to_string(), Arc::downgrade(&lock));
+    lock
+}
+
+fn tool_uses_database(tool_name: &str) -> bool {
+    matches!(
+        tool_name,
+        "list_tables"
+            | "get_columns"
+            | "execute_query"
+            | "get_sample_data"
+            | "list_collections"
+            | "browse_collection"
+            | "explain_query"
+    )
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct QueryCellWindow {
     offset: usize,
@@ -506,6 +533,15 @@ pub async fn execute_tool(
     db_type: &DatabaseType,
     sql_permissions: AgentSqlPermissions,
 ) -> ToolResult {
+    // Agent loops for different conversations may share a physical connection.
+    // Lock only the database tool future; model generation and non-DB tools stay
+    // concurrent. Dropping a cancelled future releases either the waiter or the
+    // acquired guard automatically.
+    let _connection_guard = if tool_uses_database(&tool_call.name) {
+        Some(connection_tool_lock(connection_id).lock_owned().await)
+    } else {
+        None
+    };
     let result = match tool_call.name.as_str() {
         "list_tables" => execute_list_tables(tool_call, state, connection_id, database, default_schema, db_type).await,
         "get_columns" => execute_get_columns(tool_call, state, connection_id, database, default_schema, db_type).await,
@@ -1243,6 +1279,24 @@ async fn resolve_chroma_collection_uuid(
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[tokio::test]
+    async fn database_tool_locks_are_connection_scoped_and_cancel_safe() {
+        let first = connection_tool_lock("agent-tool-lock-a").lock_owned().await;
+        let same_connection = connection_tool_lock("agent-tool-lock-a");
+        let other_connection = connection_tool_lock("agent-tool-lock-b");
+
+        assert!(same_connection.try_lock().is_err(), "same connection must serialize tool calls");
+        assert!(other_connection.try_lock().is_ok(), "different connections must remain concurrent");
+
+        let waiter = tokio::spawn(async move { same_connection.lock_owned().await });
+        tokio::task::yield_now().await;
+        waiter.abort();
+        let _ = waiter.await;
+        drop(first);
+
+        assert!(connection_tool_lock("agent-tool-lock-a").try_lock().is_ok());
+    }
     #[cfg(unix)]
     use crate::connection::PoolKind;
     #[cfg(unix)]

crates/dbx-core/src/ai.rs

@@ -558,6 +558,131 @@ pub struct AiConversation {
     pub connection_name: String,
     pub database: String,
     pub messages: Vec<AiChatMessage>,
+    /// One editable "send later" input saved while an active run occupies the
+    /// conversation (parent PRD §5). Persisted with the conversation so it
+    /// survives a restart and is restored when the conversation reopens.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub queued_input: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum AiRunStatus {
+    Preparing,
+    Queued,
+    Running,
+    AwaitingWriteConfirmation,
+    Completed,
+    Failed,
+    Cancelled,
+    Interrupted,
+    /// A normal-send FIFO item recovered after a process restart. The user's
+    /// input was never actually submitted, so the run is surfaced as an
+    /// editable, unsent pending draft — never as a failure/interruption.
+    /// (Parent PRD §7: "恢复为其会话中可编辑、未发送的待发输入".)
+    PendingRecoverable,
+}
+
+impl AiRunStatus {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Preparing => "preparing",
+            Self::Queued => "queued",
+            Self::Running => "running",
+            Self::AwaitingWriteConfirmation => "awaiting_write_confirmation",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+            Self::Cancelled => "cancelled",
+            Self::Interrupted => "interrupted",
+            Self::PendingRecoverable => "pending_recoverable",
+        }
+    }
+
+    pub fn parse(value: &str) -> Result<Self, String> {
+        match value {
+            "preparing" => Ok(Self::Preparing),
+            "queued" => Ok(Self::Queued),
+            "running" => Ok(Self::Running),
+            "awaiting_write_confirmation" => Ok(Self::AwaitingWriteConfirmation),
+            "completed" => Ok(Self::Completed),
+            "failed" => Ok(Self::Failed),
+            "cancelled" => Ok(Self::Cancelled),
+            "interrupted" => Ok(Self::Interrupted),
+            "pending_recoverable" => Ok(Self::PendingRecoverable),
+            other => Err(format!("Unknown AI run status: {other}")),
+        }
+    }
+
+    pub fn protects_conversation(self) -> bool {
+        matches!(
+            self,
+            Self::Preparing | Self::Queued | Self::Running | Self::AwaitingWriteConfirmation | Self::PendingRecoverable
+        )
+    }
+}
+
+/// Why a run occupies the global FIFO queue. Normal sends and accepted
+/// write-confirmation resumes recover differently after a restart (parent
+/// PRD §3 line 58), so the category must be persisted.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum AiRunFifoCategory {
+    /// A user-sent message waiting for a global concurrency slot.
+    NormalSend,
+    /// An accepted write confirmation whose resume segment is queued.
+    WriteConfirmationResume,
+}
+
+impl AiRunFifoCategory {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::NormalSend => "normal_send",
+            Self::WriteConfirmationResume => "write_confirmation_resume",
+        }
+    }
+
+    pub fn parse(value: &str) -> Result<Self, String> {
+        match value {
+            "normal_send" => Ok(Self::NormalSend),
+            "write_confirmation_resume" => Ok(Self::WriteConfirmationResume),
+            other => Err(format!("Unknown AI run FIFO category: {other}")),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AiRun {
+    pub run_id: String,
+    pub conversation_id: String,
+    #[serde(default)]
+    pub session_ids: Vec<String>,
+    pub status: AiRunStatus,
+    #[serde(default)]
+    pub connection_id: String,
+    #[serde(default)]
+    pub database: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub schema: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pending_confirmation: Option<serde_json::Value>,
+    /// Which queue category this run occupied. Persisted so restart recovery
+    /// can distinguish a normal send (→ editable pending input) from an
+    /// accepted write-confirmation resume (→ back to awaiting confirmation).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub fifo_category: Option<AiRunFifoCategory>,
+    /// The user's input text for a normal-send FIFO item, recovered after
+    /// restart as an editable, unsent pending draft.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pending_input: Option<String>,
+    /// Highest event `seq` ever assigned to this run, across all its sessions
+    /// (parent PRD §8). Strictly increasing from 1; the run manager persists it
+    /// to drive event dedup, the unread baseline, and the "updates while you
+    /// were away" separator anchor.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_seq: Option<u64>,
     pub created_at: String,
     pub updated_at: String,
 }

crates/dbx-core/src/storage.rs

@@ -4,13 +4,16 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use log::warn;
 use rusqlite::{
-    params, params_from_iter, types::Value, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql,
+    params, params_from_iter, types::Value, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql, Transaction,
     TransactionBehavior,
 };
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::ai::{AiChatMessage, AiChatSelectionState, AiConfig, AiConfigItem, AiConversation, AiProvider};
+use crate::ai::{
+    AiChatMessage, AiChatSelectionState, AiConfig, AiConfigItem, AiConversation, AiProvider, AiRun, AiRunFifoCategory,
+    AiRunStatus,
+};
 use crate::connection_secrets::{
     MQ_AUTH_API_KEY_VALUE_KEY, MQ_AUTH_CLIENT_SECRET_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_SECRET_PREFIX,
     MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, MQ_TOKEN_SIGNING_SECRET_PREFIX, NACOS_AUTH_PASSWORD_KEY,
@@ -43,6 +46,7 @@ const USER_DATA_TABLES: &[&str] = &[
     "connection_secrets",
     "history",
     "ai_conversations",
+    "ai_runs",
     "mq_token_records",
     "saved_sql_folders",
     "saved_sql_files",
@@ -364,9 +368,27 @@ const SCHEMA_STATEMENTS: &[&str] = &[
         connection_name TEXT NOT NULL DEFAULT '',
         database TEXT NOT NULL DEFAULT '',
         messages_json TEXT NOT NULL DEFAULT '[]',
+        queued_input TEXT,
         created_at TEXT NOT NULL DEFAULT '',
         updated_at TEXT NOT NULL DEFAULT ''
     )",
+    "CREATE TABLE IF NOT EXISTS ai_runs (
+        run_id TEXT PRIMARY KEY,
+        conversation_id TEXT NOT NULL,
+        session_ids_json TEXT NOT NULL DEFAULT '[]',
+        status TEXT NOT NULL,
+        connection_id TEXT NOT NULL DEFAULT '',
+        database TEXT NOT NULL DEFAULT '',
+        schema_name TEXT,
+        pending_confirmation_json TEXT,
+        fifo_category TEXT,
+        pending_input TEXT,
+        max_seq INTEGER,
+        created_at TEXT NOT NULL DEFAULT '',
+        updated_at TEXT NOT NULL DEFAULT '',
+        FOREIGN KEY (conversation_id) REFERENCES ai_conversations(id) ON DELETE CASCADE
+    )",
+    "CREATE INDEX IF NOT EXISTS idx_ai_runs_conversation_status ON ai_runs(conversation_id, status)",
     "CREATE TABLE IF NOT EXISTS sidebar_layout (
         id INTEGER PRIMARY KEY CHECK (id = 1),
         layout_json TEXT NOT NULL
@@ -532,6 +554,8 @@ impl Storage {
             ensure_schema_cache_columns_sync(conn)?;
             ensure_ai_configs_columns_sync(conn)?;
             ensure_state_store_columns_sync(conn)?;
+            ensure_ai_conversations_columns_sync(conn)?;
+            ensure_ai_runs_columns_sync(conn)?;
             Ok(())
         })
     }
@@ -730,6 +754,24 @@ fn ensure_ai_configs_columns_sync(conn: &Connection) -> Result<(), String> {
 
     Ok(())
 }
+
+/// Adds the queued-input column to `ai_conversations` for databases created
+/// by earlier iterations of the uncommitted WIP, where the table predates it.
+fn ensure_ai_conversations_columns_sync(conn: &Connection) -> Result<(), String> {
+    const COLUMNS: &[(&str, &str)] = &[("queued_input", "TEXT")];
+
+    ensure_table_columns(conn, "ai_conversations", COLUMNS)
+}
+
+/// Adds the background-run recovery columns (`fifo_category`, `pending_input`,
+/// `max_seq`) to databases created by earlier iterations of the uncommitted
+/// WIP, where the `ai_runs` table predates these fields.
+fn ensure_ai_runs_columns_sync(conn: &Connection) -> Result<(), String> {
+    const COLUMNS: &[(&str, &str)] = &[("fifo_category", "TEXT"), ("pending_input", "TEXT"), ("max_seq", "INTEGER")];
+
+    ensure_table_columns(conn, "ai_runs", COLUMNS)
+}
+
 fn ensure_table_columns(conn: &Connection, table_name: &str, columns: &[(&str, &str)]) -> Result<(), String> {
     let mut stmt =
         conn.prepare(&format!("SELECT name FROM pragma_table_info('{table_name}')")).map_err(|e| e.to_string())?;
@@ -2136,34 +2178,109 @@ impl Storage {
 
 // AI Conversations
 
+// Terminal runs beyond this many per conversation are pruned on every AI save.
+// Only the newest terminal run per conversation is ever read at recovery (it
+// drives the history row's status badge after a restart); the older ones are
+// pure, unbounded storage + startup-load growth.
+const KEEP_TERMINAL_AI_RUNS_PER_CONVERSATION: i64 = 2;
+
+/// Caps each conversation's terminal run history. `save_ai_run` /
+/// `save_ai_run_state` persist every run unconditionally and `load_ai_runs`
+/// loads the whole table at startup, so without this cap repeated completed/
+/// failed/cancelled runs grow SQLite storage and recovery work forever. The
+/// frontend recovery loop dedups to the newest run per conversation, so keeping
+/// the newest few terminal runs preserves the row status badge exactly while
+/// bounding the table. Non-terminal statuses (preparing/queued/running/
+/// awaiting_write_confirmation/pending_recoverable) are never touched - they
+/// are the recovery payload.
+fn prune_terminal_ai_runs(tx: &Transaction<'_>) -> Result<(), String> {
+    tx.execute(
+        "DELETE FROM ai_runs
+         WHERE status IN ('completed', 'failed', 'cancelled', 'interrupted')
+           AND run_id NOT IN (
+               SELECT run_id FROM (
+                   SELECT run_id,
+                          ROW_NUMBER() OVER (
+                              PARTITION BY conversation_id
+                              ORDER BY updated_at DESC, run_id DESC
+                          ) AS rn
+                   FROM ai_runs
+                   WHERE status IN ('completed', 'failed', 'cancelled', 'interrupted')
+               ) WHERE rn <= ?1
+           )",
+        params![KEEP_TERMINAL_AI_RUNS_PER_CONVERSATION],
+    )
+    .map_err(|e| e.to_string())?;
+    Ok(())
+}
+
+fn prune_ai_conversations(tx: &Transaction<'_>) -> Result<(), String> {
+    // The 50-row limit is a soft cap: conversations with active or actionable
+    // runs remain reachable even when they exceed the cap. Only terminal,
+    // unprotected conversations compete for the remaining budget.
+    tx.execute(
+        "WITH protected AS (
+             SELECT DISTINCT conversation_id FROM ai_runs
+             WHERE status IN ('preparing', 'queued', 'running', 'awaiting_write_confirmation', 'pending_recoverable')
+         ), budget AS (
+             SELECT MAX(0, 50 - COUNT(*)) AS value FROM protected
+         ), keepers AS (
+             SELECT conversation_id AS id FROM protected
+             UNION
+             SELECT id FROM (
+                 SELECT id FROM ai_conversations
+                 WHERE id NOT IN (SELECT conversation_id FROM protected)
+                 ORDER BY updated_at DESC
+                 LIMIT (SELECT value FROM budget)
+             )
+         )
+         DELETE FROM ai_conversations WHERE id NOT IN (SELECT id FROM keepers)",
+        [],
+    )
+    .map_err(|e| e.to_string())?;
+    // Do not depend on a connection-wide foreign_keys pragma for cleanup.
+    tx.execute("DELETE FROM ai_runs WHERE conversation_id NOT IN (SELECT id FROM ai_conversations)", [])
+        .map_err(|e| e.to_string())?;
+    // Cap terminal run history for the conversations that survive the cap
+    // above (they are deliberately retained), so normal use cannot grow the
+    // ai_runs table without bound.
+    prune_terminal_ai_runs(tx)?;
+    Ok(())
+}
+
 impl Storage {
     pub async fn save_ai_conversation(&self, conv: &AiConversation) -> Result<(), String> {
         let conv = conv.clone();
         let messages_json = serde_json::to_string(&conv.messages).map_err(|e| e.to_string())?;
         self.with_conn(move |conn| {
-            conn.execute(
-                "INSERT OR REPLACE INTO ai_conversations \
-                 (id, title, connection_name, database, messages_json, created_at, updated_at) \
-                 VALUES (?, ?, ?, ?, ?, ?, ?)",
+            let tx = conn.transaction().map_err(|e| e.to_string())?;
+            tx.execute(
+                "INSERT INTO ai_conversations \
+                 (id, title, connection_name, database, messages_json, queued_input, created_at, updated_at) \
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?) \
+                 ON CONFLICT(id) DO UPDATE SET \
+                   title = excluded.title, \
+                   connection_name = excluded.connection_name, \
+                   database = excluded.database, \
+                   messages_json = excluded.messages_json, \
+                   queued_input = excluded.queued_input, \
+                   created_at = excluded.created_at, \
+                   updated_at = excluded.updated_at",
                 params![
                     conv.id,
                     conv.title,
                     conv.connection_name,
                     conv.database,
                     messages_json,
+                    conv.queued_input,
                     conv.created_at,
                     conv.updated_at
                 ],
             )
             .map_err(|e| e.to_string())?;
 
-            conn.execute(
-                "DELETE FROM ai_conversations WHERE id NOT IN \
-                 (SELECT id FROM ai_conversations ORDER BY updated_at DESC LIMIT 50)",
-                [],
-            )
-            .map_err(|e| e.to_string())?;
-            Ok(())
+            prune_ai_conversations(&tx)?;
+            tx.commit().map_err(|e| e.to_string())
         })
         .await
     }
@@ -2172,7 +2289,7 @@ impl Storage {
         self.with_conn(|conn| {
             let mut stmt = conn
                 .prepare(
-                    "SELECT id, title, connection_name, database, messages_json, created_at, updated_at \
+                    "SELECT id, title, connection_name, database, messages_json, queued_input, created_at, updated_at \
                      FROM ai_conversations ORDER BY updated_at DESC",
                 )
                 .map_err(|e| e.to_string())?;
@@ -2187,8 +2304,9 @@ impl Storage {
                         connection_name: row.get(2)?,
                         database: row.get(3)?,
                         messages,
-                        created_at: row.get(5)?,
-                        updated_at: row.get(6)?,
+                        queued_input: row.get(5)?,
+                        created_at: row.get(6)?,
+                        updated_at: row.get(7)?,
                     })
                 })
                 .map_err(|e| e.to_string())?;
@@ -2200,7 +2318,179 @@ impl Storage {
     pub async fn delete_ai_conversation(&self, id: &str) -> Result<(), String> {
         let id = id.to_string();
         self.with_conn(move |conn| {
-            conn.execute("DELETE FROM ai_conversations WHERE id = ?1", [id]).map(|_| ()).map_err(|e| e.to_string())
+            let tx = conn.transaction().map_err(|e| e.to_string())?;
+            tx.execute("DELETE FROM ai_runs WHERE conversation_id = ?1", [&id]).map_err(|e| e.to_string())?;
+            tx.execute("DELETE FROM ai_conversations WHERE id = ?1", [&id]).map_err(|e| e.to_string())?;
+            prune_ai_conversations(&tx)?;
+            tx.commit().map_err(|e| e.to_string())
+        })
+        .await
+    }
+
+    pub async fn save_ai_run(&self, run: &AiRun) -> Result<(), String> {
+        let run = run.clone();
+        let session_ids_json = serde_json::to_string(&run.session_ids).map_err(|e| e.to_string())?;
+        let pending_confirmation_json = run.pending_confirmation.map(|value| value.to_string());
+        let fifo_category = run.fifo_category.map(|c| c.as_str().to_string());
+        self.with_conn(move |conn| {
+            let tx = conn.transaction().map_err(|e| e.to_string())?;
+            tx.execute(
+                "INSERT OR REPLACE INTO ai_runs
+                 (run_id, conversation_id, session_ids_json, status, connection_id, database, schema_name,
+                  pending_confirmation_json, fifo_category, pending_input, max_seq, created_at, updated_at)
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                params![
+                    run.run_id,
+                    run.conversation_id,
+                    session_ids_json,
+                    run.status.as_str(),
+                    run.connection_id,
+                    run.database,
+                    run.schema,
+                    pending_confirmation_json,
+                    fifo_category,
+                    run.pending_input,
+                    run.max_seq,
+                    run.created_at,
+                    run.updated_at,
+                ],
+            )
+            .map_err(|e| e.to_string())?;
+            prune_ai_conversations(&tx)?;
+            tx.commit().map_err(|e| e.to_string())
+        })
+        .await
+    }
+
+    pub async fn save_ai_run_state(&self, conv: &AiConversation, run: &AiRun) -> Result<(), String> {
+        let conv = conv.clone();
+        let run = run.clone();
+        let messages_json = serde_json::to_string(&conv.messages).map_err(|e| e.to_string())?;
+        let session_ids_json = serde_json::to_string(&run.session_ids).map_err(|e| e.to_string())?;
+        let pending_confirmation_json = run.pending_confirmation.map(|value| value.to_string());
+        let fifo_category = run.fifo_category.map(|c| c.as_str().to_string());
+        self.with_conn(move |conn| {
+            let tx = conn.transaction().map_err(|e| e.to_string())?;
+            tx.execute(
+                "INSERT INTO ai_conversations
+                 (id, title, connection_name, database, messages_json, queued_input, created_at, updated_at)
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                 ON CONFLICT(id) DO UPDATE SET
+                   title = excluded.title,
+                   connection_name = excluded.connection_name,
+                   database = excluded.database,
+                   messages_json = excluded.messages_json,
+                   queued_input = excluded.queued_input,
+                   created_at = excluded.created_at,
+                   updated_at = excluded.updated_at",
+                params![
+                    conv.id,
+                    conv.title,
+                    conv.connection_name,
+                    conv.database,
+                    messages_json,
+                    conv.queued_input,
+                    conv.created_at,
+                    conv.updated_at
+                ],
+            )
+            .map_err(|e| e.to_string())?;
+            tx.execute(
+                "INSERT OR REPLACE INTO ai_runs
+                 (run_id, conversation_id, session_ids_json, status, connection_id, database, schema_name,
+                  pending_confirmation_json, fifo_category, pending_input, max_seq, created_at, updated_at)
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                params![
+                    run.run_id,
+                    run.conversation_id,
+                    session_ids_json,
+                    run.status.as_str(),
+                    run.connection_id,
+                    run.database,
+                    run.schema,
+                    pending_confirmation_json,
+                    fifo_category,
+                    run.pending_input,
+                    run.max_seq,
+                    run.created_at,
+                    run.updated_at,
+                ],
+            )
+            .map_err(|e| e.to_string())?;
+            prune_ai_conversations(&tx)?;
+            tx.commit().map_err(|e| e.to_string())
+        })
+        .await
+    }
+
+    pub async fn load_ai_runs(&self) -> Result<Vec<AiRun>, String> {
+        self.with_conn(|conn| {
+            let mut stmt = conn
+                .prepare(
+                    "SELECT run_id, conversation_id, session_ids_json, status, connection_id, database,
+                            schema_name, pending_confirmation_json, fifo_category, pending_input, max_seq, created_at, updated_at
+                     FROM ai_runs ORDER BY updated_at DESC",
+                )
+                .map_err(|e| e.to_string())?;
+            let rows = stmt
+                .query_map([], |row| {
+                    let session_ids_json: String = row.get(2)?;
+                    let status: String = row.get(3)?;
+                    let pending_confirmation_json: Option<String> = row.get(7)?;
+                    let fifo_category: Option<String> = row.get(8)?;
+                    let pending_input: Option<String> = row.get(9)?;
+                    Ok((
+                        row.get::<_, String>(0)?,
+                        row.get::<_, String>(1)?,
+                        session_ids_json,
+                        status,
+                        row.get::<_, String>(4)?,
+                        row.get::<_, String>(5)?,
+                        row.get::<_, Option<String>>(6)?,
+                        pending_confirmation_json,
+                        fifo_category,
+                        pending_input,
+                        row.get::<_, Option<u64>>(10)?,
+                        row.get::<_, String>(11)?,
+                        row.get::<_, String>(12)?,
+                    ))
+                })
+                .map_err(|e| e.to_string())?;
+            rows.map(|row| {
+                let (
+                    run_id,
+                    conversation_id,
+                    session_ids_json,
+                    status,
+                    connection_id,
+                    database,
+                    schema,
+                    pending_confirmation_json,
+                    fifo_category,
+                    pending_input,
+                    max_seq,
+                    created_at,
+                    updated_at,
+                ) = row.map_err(|e| e.to_string())?;
+                Ok(AiRun {
+                    run_id,
+                    conversation_id,
+                    session_ids: serde_json::from_str(&session_ids_json).map_err(|e| e.to_string())?,
+                    status: AiRunStatus::parse(&status)?,
+                    connection_id,
+                    database,
+                    schema,
+                    pending_confirmation: pending_confirmation_json
+                        .map(|json| serde_json::from_str(&json).map_err(|e| e.to_string()))
+                        .transpose()?,
+                    fifo_category: fifo_category.map(|value| AiRunFifoCategory::parse(&value)).transpose()?,
+                    pending_input,
+                    max_seq,
+                    created_at,
+                    updated_at,
+                })
+            })
+            .collect::<Result<Vec<_>, String>>()
         })
         .await
     }
@@ -4385,10 +4675,11 @@ fn map_from_sql_err(err: serde_json::Error) -> rusqlite::Error {
 mod tests {
     use super::{
         maybe_import_user_data_db, DataDbImportResult, DesktopIconTheme, DesktopSettings, McpGlobalPolicy,
-        McpGlobalPolicyState, Storage, MCP_GLOBAL_POLICY_KEY,
+        McpGlobalPolicyState, Storage, KEEP_TERMINAL_AI_RUNS_PER_CONVERSATION, MCP_GLOBAL_POLICY_KEY,
     };
     use crate::ai::{
-        AiActiveModelSelection, AiAssistantMode, AiChatSelectionState, AiEffortSelection, AiModelEffortPreference,
+        AiActiveModelSelection, AiAssistantMode, AiChatMessage, AiChatSelectionState, AiConversation,
+        AiEffortSelection, AiModelEffortPreference, AiRun, AiRunFifoCategory, AiRunStatus,
     };
     use crate::connection_secrets::NACOS_RNACOS_CONSOLE_PASSWORD_KEY;
     use crate::connection_secrets::{
@@ -4440,6 +4731,239 @@ mod tests {
         }
     }
 
+    fn ai_conversation(id: &str, updated_at: &str) -> AiConversation {
+        AiConversation {
+            id: id.to_string(),
+            title: id.to_string(),
+            connection_name: "local".to_string(),
+            database: "db".to_string(),
+            messages: vec![AiChatMessage {
+                role: "user".to_string(),
+                content: id.to_string(),
+                mentions: None,
+                reasoning: None,
+                kind: None,
+                covered_messages: None,
+            }],
+            queued_input: None,
+            created_at: updated_at.to_string(),
+            updated_at: updated_at.to_string(),
+        }
+    }
+
+    fn ai_run(id: &str, conversation_id: &str, status: AiRunStatus, updated_at: &str) -> AiRun {
+        AiRun {
+            run_id: id.to_string(),
+            conversation_id: conversation_id.to_string(),
+            session_ids: vec![],
+            status,
+            connection_id: "connection".to_string(),
+            database: "db".to_string(),
+            schema: None,
+            pending_confirmation: None,
+            fifo_category: None,
+            pending_input: None,
+            max_seq: None,
+            created_at: updated_at.to_string(),
+            updated_at: updated_at.to_string(),
+        }
+    }
+
+    #[tokio::test]
+    async fn ai_conversation_soft_cap_never_evicts_protected_runs() {
+        let path = temp_db_path("ai-conversation-soft-cap");
+        let storage = Storage::open(&path).await.unwrap();
+
+        let protected = ai_conversation("protected", "0000");
+        let protected_run = ai_run("protected-run", "protected", AiRunStatus::Running, "0000");
+        storage.save_ai_run_state(&protected, &protected_run).await.unwrap();
+        for index in 0..55 {
+            let timestamp = format!("{index:04}");
+            storage.save_ai_conversation(&ai_conversation(&format!("terminal-{index}"), &timestamp)).await.unwrap();
+        }
+
+        let conversations = storage.load_ai_conversations().await.unwrap();
+        assert_eq!(conversations.len(), 50);
+        assert!(conversations.iter().any(|conversation| conversation.id == "protected"));
+        assert!(!conversations.iter().any(|conversation| conversation.id == "terminal-0"));
+        assert!(conversations.iter().any(|conversation| conversation.id == "terminal-54"));
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn ai_conversation_soft_cap_allows_more_than_fifty_protected_runs() {
+        let path = temp_db_path("ai-conversation-protected-overflow");
+        let storage = Storage::open(&path).await.unwrap();
+
+        for index in 0..51 {
+            let id = format!("protected-{index}");
+            let timestamp = format!("{index:04}");
+            storage
+                .save_ai_run_state(
+                    &ai_conversation(&id, &timestamp),
+                    &ai_run(&format!("run-{index}"), &id, AiRunStatus::AwaitingWriteConfirmation, &timestamp),
+                )
+                .await
+                .unwrap();
+        }
+        storage.save_ai_conversation(&ai_conversation("terminal-extra", "9999")).await.unwrap();
+
+        let conversations = storage.load_ai_conversations().await.unwrap();
+        assert_eq!(conversations.len(), 51);
+        assert!(conversations.iter().all(|conversation| conversation.id.starts_with("protected-")));
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn ai_conversation_soft_cap_protects_pending_recoverable_runs() {
+        let path = temp_db_path("ai-conversation-pending-recoverable-protection");
+        let storage = Storage::open(&path).await.unwrap();
+
+        // A recovered pending-input run (PRD §7 line 93) must be protected like
+        // any other non-terminal run: its draft is not lost to pruning.
+        let protected = ai_conversation("recoverable", "0000");
+        let mut protected_run = ai_run("recoverable-run", "recoverable", AiRunStatus::PendingRecoverable, "0000");
+        protected_run.pending_input = Some("recover me".to_string());
+        storage.save_ai_run_state(&protected, &protected_run).await.unwrap();
+        for index in 0..55 {
+            let timestamp = format!("{index:04}");
+            storage.save_ai_conversation(&ai_conversation(&format!("terminal-{index}"), &timestamp)).await.unwrap();
+        }
+
+        let conversations = storage.load_ai_conversations().await.unwrap();
+        assert_eq!(conversations.len(), 50);
+        assert!(conversations.iter().any(|conversation| conversation.id == "recoverable"));
+        assert!(!conversations.iter().any(|conversation| conversation.id == "terminal-0"));
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn ai_run_roundtrips_fifo_category_and_pending_input() {
+        let path = temp_db_path("ai-run-fifo-category-roundtrip");
+        let storage = Storage::open(&path).await.unwrap();
+
+        let conversation = ai_conversation("fifo-conv", "0000");
+        let mut run = ai_run("fifo-run", "fifo-conv", AiRunStatus::Queued, "0000");
+        run.fifo_category = Some(AiRunFifoCategory::NormalSend);
+        run.pending_input = Some("select * from orders limit 5".to_string());
+        run.max_seq = Some(42);
+        storage.save_ai_run_state(&conversation, &run).await.unwrap();
+
+        let loaded = storage.load_ai_runs().await.unwrap();
+        assert_eq!(loaded.len(), 1);
+        assert_eq!(loaded[0].run_id, "fifo-run");
+        assert_eq!(loaded[0].status, AiRunStatus::Queued);
+        assert_eq!(loaded[0].fifo_category, Some(AiRunFifoCategory::NormalSend));
+        assert_eq!(loaded[0].pending_input.as_deref(), Some("select * from orders limit 5"));
+        assert_eq!(loaded[0].max_seq, Some(42));
+
+        // The write_confirmation_resume category survives too.
+        let mut resume = ai_run("resume-run", "fifo-conv", AiRunStatus::Queued, "0001");
+        resume.fifo_category = Some(AiRunFifoCategory::WriteConfirmationResume);
+        storage.save_ai_run(&resume).await.unwrap();
+        let loaded = storage.load_ai_runs().await.unwrap();
+        let resume = loaded.iter().find(|run| run.run_id == "resume-run").unwrap();
+        assert_eq!(resume.fifo_category, Some(AiRunFifoCategory::WriteConfirmationResume));
+        assert!(resume.pending_input.is_none());
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn saving_a_conversation_keeps_its_background_runs() {
+        let path = temp_db_path("ai-conversation-upsert-keeps-runs");
+        let storage = Storage::open(&path).await.unwrap();
+        // `ai_runs` declares an ON DELETE CASCADE relationship. Exercise the
+        // snapshot path with enforcement enabled so an accidental REPLACE
+        // (delete + insert) cannot silently erase an active run on restart.
+        storage
+            .with_conn(|conn| conn.execute_batch("PRAGMA foreign_keys = ON").map_err(|e| e.to_string()))
+            .await
+            .unwrap();
+
+        let mut conversation = ai_conversation("upsert-conv", "0000");
+        let run = ai_run("upsert-run", "upsert-conv", AiRunStatus::Running, "0000");
+        storage.save_ai_run_state(&conversation, &run).await.unwrap();
+
+        conversation.updated_at = "0001".to_string();
+        conversation.queued_input = Some("send later".to_string());
+        storage.save_ai_conversation(&conversation).await.unwrap();
+
+        let runs = storage.load_ai_runs().await.unwrap();
+        assert_eq!(runs.len(), 1);
+        assert_eq!(runs[0].run_id, "upsert-run");
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn terminal_ai_runs_are_capped_per_conversation_while_nonterminal_survive() {
+        // Reviewed finding (unbounded terminal run growth): save_ai_run /
+        // save_ai_run_state persist every terminal run and load_ai_runs loads
+        // the whole table at startup, but prune_ai_conversations only caps
+        // conversations and deliberately retains runs for the survivors - so
+        // repeated completed runs grew SQLite storage and recovery work
+        // forever. The storage layer now caps terminal history per
+        // conversation (keeping the newest few, which drive the row status
+        // badge after restart) and never touches recovery-relevant runs.
+        let path = temp_db_path("ai-terminal-runs-capped");
+        let storage = Storage::open(&path).await.unwrap();
+
+        // A non-terminal run must always survive - it is the recovery payload.
+        let conversation = ai_conversation("cap-conv", "0000");
+        storage
+            .save_ai_run_state(&conversation, &ai_run("active-run", "cap-conv", AiRunStatus::Running, "0000"))
+            .await
+            .unwrap();
+        // Repeated completed runs (normal use): older ones must be pruned.
+        for index in 0..5 {
+            let timestamp = format!("{index:04}");
+            storage
+                .save_ai_run(&ai_run(&format!("terminal-{index}"), "cap-conv", AiRunStatus::Completed, &timestamp))
+                .await
+                .unwrap();
+        }
+
+        let runs = storage.load_ai_runs().await.unwrap();
+        assert!(runs.iter().any(|run| run.run_id == "active-run"), "recovery-relevant run must survive");
+        let terminal: Vec<_> = runs.iter().filter(|run| run.status == AiRunStatus::Completed).collect();
+        assert_eq!(
+            terminal.len(),
+            KEEP_TERMINAL_AI_RUNS_PER_CONVERSATION as usize,
+            "only the newest terminal runs per conversation survive"
+        );
+        assert!(terminal.iter().any(|run| run.run_id == "terminal-4"), "the newest terminal run is retained");
+        assert!(!runs.iter().any(|run| run.run_id == "terminal-0"), "the oldest terminal runs are pruned");
+
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn ai_conversation_roundtrips_queued_input() {
+        let path = temp_db_path("ai-conversation-queued-input-roundtrip");
+        let storage = Storage::open(&path).await.unwrap();
+
+        let mut conversation = ai_conversation("queued-conv", "0000");
+        conversation.queued_input = Some("run this after the current task".to_string());
+        storage.save_ai_conversation(&conversation).await.unwrap();
+
+        let loaded = storage.load_ai_conversations().await.unwrap();
+        assert_eq!(loaded.len(), 1);
+        assert_eq!(loaded[0].id, "queued-conv");
+        assert_eq!(loaded[0].queued_input.as_deref(), Some("run this after the current task"));
+
+        // Overwriting clears a stale queued input.
+        conversation.queued_input = None;
+        storage.save_ai_conversation(&conversation).await.unwrap();
+        let loaded = storage.load_ai_conversations().await.unwrap();
+        assert!(loaded[0].queued_input.is_none());
+
+        let _ = std::fs::remove_file(path);
+    }
+
     #[tokio::test]
     async fn history_search_filters_connection_database_and_legacy_entries() {
         let path = temp_db_path("history-search-scope");

crates/dbx-web/src/routes/ai.rs

@@ -369,10 +369,17 @@ pub async fn ai_stream(
         })
         .await;
 
-        if let Err(_e) = result {
-            let error_chunk =
-                AiStreamChunk { session_id: sid.clone(), delta: String::new(), reasoning_delta: None, done: true };
-            let _ = tx.send(serde_json::to_string(&error_chunk).unwrap_or_default());
+        if let Err(error) = result {
+            // Preserve the existing POST/SSE lifecycle, but make failure an
+            // explicit terminal payload so Web cannot mistake it for a
+            // successful empty response.
+            let error_chunk = serde_json::json!({
+                "session_id": sid.clone(),
+                "delta": "",
+                "done": true,
+                "error": error,
+            });
+            let _ = tx.send(error_chunk.to_string());
         }
 
         dbx_core::ai::unregister_stream(&sid).await;