fix(ai): require exact SQL confirmation for agent writes

crates/dbx-core/src/agent_events.rs

@@ -8,6 +8,12 @@ pub enum AgentEvent {
     TurnStart { turn: u32 },
     /// Text delta from the LLM response.
     TextDelta { delta: String },
+    /// Request a localized, exact-SQL confirmation after an unconfirmed write
+    /// tool call. The desktop client owns the user-facing wording.
+    WriteSqlConfirmationRequired { sql: String },
+    /// Tell the desktop client to show localized manual-review guidance after
+    /// an AI write attempt against a production target.
+    ProductionWriteBlocked { sql: String },
     /// Reasoning/thinking delta (for models that support it).
     ReasoningDelta { delta: String },
     /// The LLM wants to call a tool.

crates/dbx-core/src/agent_loop.rs

@@ -248,8 +248,9 @@ pub async fn run_agent_loop(
         )
         .await;
     }
+    let mut sql_permissions = agent_ctx.sql_permissions.clone();
     let tools = if is_agent_mode {
-        agent_tools::all_tools(agent_ctx.db_type, agent_ctx.sql_permissions.clone())
+        agent_tools::all_tools(agent_ctx.db_type, sql_permissions.clone())
     } else {
         agent_tools::read_only_tools(agent_ctx.db_type)
     };
@@ -469,6 +470,33 @@ pub async fn run_agent_loop(
             break;
         }
 
+        // Some models call execute_query for a write despite the prompt requiring
+        // a proposal first. Stop before dispatch so DBX can return an exact SQL
+        // proposal for non-production targets, or a non-confirmable production
+        // block rather than an impossible confirmation loop.
+        if let Some(sql) = unconfirmed_write_sql(&collected_tool_calls, agent_ctx.db_type, &sql_permissions) {
+            let targets_production = {
+                let configs = agent_ctx.state.configs.read().await;
+                configs.get(&agent_ctx.connection_id).is_some_and(|config| {
+                    crate::production_safety::targets_production_database(config, &agent_ctx.database, sql)
+                })
+            };
+            let response = write_attempt_response(sql, targets_production);
+            on_event(match &response {
+                WriteAttemptResponse::ConfirmationRequired { sql } => {
+                    AgentEvent::WriteSqlConfirmationRequired { sql: sql.clone() }
+                }
+                WriteAttemptResponse::ProductionBlocked { sql } => {
+                    AgentEvent::ProductionWriteBlocked { sql: sql.clone() }
+                }
+            });
+            // The frontend localizes this semantic event before persisting it in
+            // chat history. Keep a non-user-facing result for API consumers.
+            final_text = response.result_text();
+            loop_exit = LoopExit::Completed;
+            break;
+        }
+
         // Execute each tool call
         // Emit all ToolCallStart events first
         for tc in &collected_tool_calls {
@@ -485,7 +513,7 @@ pub async fn run_agent_loop(
         let db2 = agent_ctx.database.clone();
         let schema2 = agent_ctx.schema.clone();
         let db_type = agent_ctx.db_type;
-        let sql_permissions = agent_ctx.sql_permissions.clone();
+        let parallel_sql_permissions = sql_permissions.clone();
 
         // Split by index into parallel and sequential groups using tool metadata
         let tool_parallel_map: std::collections::HashMap<&str, bool> =
@@ -510,7 +538,7 @@ pub async fn run_agent_loop(
                     let conn = conn2.clone();
                     let db = db2.clone();
                     let schema = schema2.clone();
-                    let perms = sql_permissions.clone();
+                    let perms = parallel_sql_permissions.clone();
                     async move {
                         agent_tools::execute_tool(&tc, &state, &conn, &db, schema.as_deref(), &db_type, perms).await
                     }
@@ -522,6 +550,7 @@ pub async fn run_agent_loop(
         let mut sequential_results = Vec::with_capacity(sequential_indices.len());
         for &i in &sequential_indices {
             let tc = make_tc(&collected_tool_calls[i]);
+            let execution_permissions = sequential_tool_permissions(&tc, db_type, &mut sql_permissions);
             sequential_results.push(
                 agent_tools::execute_tool(
                     &tc,
@@ -530,7 +559,7 @@ pub async fn run_agent_loop(
                     &db2,
                     schema2.as_deref(),
                     &db_type,
-                    sql_permissions.clone(),
+                    execution_permissions,
                 )
                 .await,
             );
@@ -1184,6 +1213,72 @@ async fn maybe_compact(
     CompactResult::Compacted
 }
 
+/// Semantic write-attempt outcomes. Keep user-visible language in the desktop
+/// i18n catalog rather than hard-coding English into a backend event.
+enum WriteAttemptResponse {
+    ConfirmationRequired { sql: String },
+    ProductionBlocked { sql: String },
+}
+
+impl WriteAttemptResponse {
+    fn result_text(&self) -> String {
+        match self {
+            Self::ConfirmationRequired { sql } | Self::ProductionBlocked { sql } => sql.clone(),
+        }
+    }
+}
+
+fn write_attempt_response(sql: &str, targets_production: bool) -> WriteAttemptResponse {
+    let sql = sql.trim().to_string();
+    if targets_production {
+        WriteAttemptResponse::ProductionBlocked { sql }
+    } else {
+        WriteAttemptResponse::ConfirmationRequired { sql }
+    }
+}
+
+fn sequential_tool_permissions(
+    tool_call: &ToolCall,
+    db_type: DatabaseType,
+    permissions: &mut agent_tools::AgentSqlPermissions,
+) -> agent_tools::AgentSqlPermissions {
+    if tool_call.name == "execute_query" {
+        if let Some(sql) = tool_call.arguments.get("sql").and_then(|value| value.as_str()) {
+            return agent_tools::take_sql_permissions_for_execution(sql, db_type, permissions);
+        }
+    }
+    permissions.clone()
+}
+
+/// Returns a deterministic confirmation proposal when the current turn contains
+/// an unconfirmed write/DDL tool call. The caller must stop before dispatching
+/// any tools so the database never sees the attempted SQL.
+fn unconfirmed_write_sql<'a>(
+    tool_calls: &'a [ToolCall],
+    db_type: DatabaseType,
+    permissions: &agent_tools::AgentSqlPermissions,
+) -> Option<&'a str> {
+    // Mongo execute_query is a read-only shell-command tool; mutating commands
+    // are rejected by the tool itself, so a confirmation proposal could never
+    // be executed. And because Mongo commands are not SQL, the risk classifier
+    // would flag even reads (db.collection.find(...)) as writes, turning every
+    // Mongo agent query into an impossible confirmation. Preserve the previous
+    // behavior: Mongo writes keep failing with a clear read-only error.
+    if db_type == DatabaseType::MongoDb {
+        return None;
+    }
+    tool_calls.iter().find_map(|tool_call| {
+        if tool_call.name != "execute_query" {
+            return None;
+        }
+        let sql = tool_call.arguments.get("sql").and_then(|value| value.as_str())?;
+        agent_tools::write_requires_confirmation(sql, db_type, permissions)
+            .ok()
+            .filter(|required| *required)
+            .map(|_| sql)
+    })
+}
+
 fn tool_result_for_followup_context(tool_name: &str, content: &str) -> String {
     let result = compact_tool_result_for_context(tool_name, content);
     format!(
@@ -1440,6 +1535,160 @@ mod tests {
         }
     }
 
+    #[test]
+    fn unconfirmed_write_attempt_requests_confirmation_with_exact_sql() {
+        let sql = "CREATE TABLE users (id INT);";
+        let response = write_attempt_response(sql, false);
+
+        assert!(
+            matches!(response, WriteAttemptResponse::ConfirmationRequired { sql: response_sql } if response_sql == sql)
+        );
+    }
+
+    #[test]
+    fn production_write_attempt_is_not_an_executable_confirmation() {
+        let response = write_attempt_response("DELETE FROM users WHERE id = 7;", true);
+
+        assert!(
+            matches!(response, WriteAttemptResponse::ProductionBlocked { sql } if sql == "DELETE FROM users WHERE id = 7;")
+        );
+    }
+
+    #[test]
+    fn direct_unconfirmed_ddl_tool_call_becomes_a_confirmation_proposal() {
+        let tool_calls = vec![ToolCall {
+            id: "create-table".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "CREATE TABLE users (id INT);" }),
+            provider_payload: None,
+        }];
+
+        let sql =
+            unconfirmed_write_sql(&tool_calls, DatabaseType::Postgres, &agent_tools::AgentSqlPermissions::default())
+                .expect("unconfirmed DDL must be intercepted before tool dispatch");
+        assert!(
+            matches!(write_attempt_response(sql, false), WriteAttemptResponse::ConfirmationRequired { sql } if sql == "CREATE TABLE users (id INT);")
+        );
+    }
+
+    #[test]
+    fn confirmed_or_read_only_tool_calls_do_not_become_confirmation_proposals() {
+        let create = ToolCall {
+            id: "create-table".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "CREATE TABLE users (id INT);" }),
+            provider_payload: None,
+        };
+        let select = ToolCall {
+            id: "select-users".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "SELECT * FROM users" }),
+            provider_payload: None,
+        };
+        let confirmed =
+            agent_tools::confirmed_write_sql_permissions(false, true, Some("CREATE TABLE users (id INT);".to_string()));
+
+        assert!(unconfirmed_write_sql(&[create], DatabaseType::Postgres, &confirmed).is_none());
+        assert!(unconfirmed_write_sql(&[select], DatabaseType::Postgres, &agent_tools::AgentSqlPermissions::default())
+            .is_none());
+    }
+
+    #[test]
+    fn duplicate_confirmed_writes_in_one_batch_receive_one_grant() {
+        let sql = "DELETE FROM users WHERE id = 7;";
+        let call = ToolCall {
+            id: "delete-user".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": sql }),
+            provider_payload: None,
+        };
+        let mut permissions = agent_tools::confirmed_write_sql_permissions(false, true, Some(sql.to_string()));
+
+        let first = sequential_tool_permissions(&call, DatabaseType::Postgres, &mut permissions);
+        let second = sequential_tool_permissions(&call, DatabaseType::Postgres, &mut permissions);
+
+        assert!(first.allow_writes);
+        assert_eq!(first.confirmed_write_sql.as_deref(), Some(sql));
+        assert!(!second.allow_writes);
+        assert_eq!(second.confirmed_write_sql, None);
+        assert_eq!(permissions, agent_tools::AgentSqlPermissions::default());
+    }
+
+    #[test]
+    fn confirmed_write_grant_stays_consumed_across_turns() {
+        let sql = "DELETE FROM users WHERE id = 7;";
+        let call = ToolCall {
+            id: "delete-user".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": sql }),
+            provider_payload: None,
+        };
+        let mut permissions = agent_tools::confirmed_write_sql_permissions(false, true, Some(sql.to_string()));
+
+        let _ = sequential_tool_permissions(&call, DatabaseType::Postgres, &mut permissions);
+
+        assert_eq!(unconfirmed_write_sql(std::slice::from_ref(&call), DatabaseType::Postgres, &permissions), Some(sql));
+    }
+
+    #[test]
+    fn read_before_confirmed_write_does_not_consume_the_grant() {
+        let confirmed_sql = "DELETE FROM users WHERE id = 7;";
+        let read = ToolCall {
+            id: "read-user".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "SELECT * FROM users WHERE id = 7" }),
+            provider_payload: None,
+        };
+        let write = ToolCall {
+            id: "delete-user".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": confirmed_sql }),
+            provider_payload: None,
+        };
+        let mut permissions =
+            agent_tools::confirmed_write_sql_permissions(false, true, Some(confirmed_sql.to_string()));
+
+        let read_permissions = sequential_tool_permissions(&read, DatabaseType::Postgres, &mut permissions);
+        let write_permissions = sequential_tool_permissions(&write, DatabaseType::Postgres, &mut permissions);
+
+        assert!(read_permissions.allow_writes);
+        assert!(write_permissions.allow_writes);
+        assert_eq!(write_permissions.confirmed_write_sql.as_deref(), Some(confirmed_sql));
+        assert_eq!(permissions, agent_tools::AgentSqlPermissions::default());
+    }
+
+    #[test]
+    fn mongo_agent_commands_are_never_intercepted_as_write_confirmations() {
+        // Mongo execute_query is read-only at the tool level, and its commands are
+        // not SQL — the risk classifier would flag even reads as writes. Neither a
+        // read nor a mutating command may become an impossible confirmation.
+        let read = ToolCall {
+            id: "mongo-read".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "db.collection.find({})" }),
+            provider_payload: None,
+        };
+        let write = ToolCall {
+            id: "mongo-write".to_string(),
+            name: "execute_query".to_string(),
+            arguments: json!({ "sql": "db.collection.insertOne({ name: \"x\" })" }),
+            provider_payload: None,
+        };
+        let permissions = agent_tools::AgentSqlPermissions::default();
+        // Without the Mongo exclusion this read would be flagged as an unconfirmed
+        // write (Mongo commands are not SQL), producing a confirmation for every
+        // agent query. Assert the misclassification so the exclusion's necessity
+        // stays explicit.
+        assert!(agent_tools::write_requires_confirmation(
+            "db.collection.find({})",
+            DatabaseType::MongoDb,
+            &permissions
+        )
+        .unwrap());
+        assert!(unconfirmed_write_sql(&[read], DatabaseType::MongoDb, &permissions).is_none());
+        assert!(unconfirmed_write_sql(&[write], DatabaseType::MongoDb, &permissions).is_none());
+    }
+
     #[test]
     fn generate_contract_rejects_schema_summary_without_sql() {
         let contract = generate_contract("帮我生成统计 2026年1月2日新注册会员数量的 sql", "ask");

crates/dbx-core/src/agent_tools.rs

@@ -152,6 +152,44 @@ fn sql_risk_allowed(risk: SqlRisk, permissions: AgentSqlPermissions) -> bool {
     }
 }
 
+/// Returns true when an Agent attempted a write or DDL call before DBX has a
+/// user-confirmed SQL binding for the current run. The caller must turn that
+/// attempt into a confirmation proposal instead of sending it to the database.
+pub fn write_requires_confirmation(
+    sql: &str,
+    db_type: DatabaseType,
+    permissions: &AgentSqlPermissions,
+) -> Result<bool, String> {
+    if permissions.allow_writes || permissions.allow_dangerous || permissions.confirmed_write_sql.is_some() {
+        return Ok(false);
+    }
+    let risk = crate::sql_risk::classify_sql_risk_for_database(sql, db_type)?;
+    Ok(matches!(risk, SqlRisk::Write | SqlRisk::Ddl))
+}
+
+/// Snapshot the permissions for one execute_query call and consume an exact
+/// confirmed write/DDL grant before dispatch. Later calls in the same agent run
+/// receive the cleared permissions and must request a new confirmation.
+pub(crate) fn take_sql_permissions_for_execution(
+    sql: &str,
+    db_type: DatabaseType,
+    permissions: &mut AgentSqlPermissions,
+) -> AgentSqlPermissions {
+    let execution_permissions = permissions.clone();
+    let consumes_confirmation = match crate::sql_risk::classify_sql_risk_for_database(sql, db_type) {
+        Ok(risk @ (SqlRisk::Write | SqlRisk::Ddl)) => {
+            sql_risk_allowed(risk, execution_permissions.clone())
+                && sql_matches_confirmed_write(sql, &execution_permissions.confirmed_write_sql)
+                && execution_permissions.confirmed_write_sql.is_some()
+        }
+        _ => false,
+    };
+    if consumes_confirmation {
+        *permissions = AgentSqlPermissions::default();
+    }
+    execution_permissions
+}
+
 /// Returns true for vector database types (Qdrant, Milvus, Weaviate, ChromaDb).
 /// If modifying this, also update VECTOR_DB_TYPES in apps/desktop/src/lib/ai.ts.
 pub fn is_vector_db(db_type: DatabaseType) -> bool {
@@ -328,7 +366,7 @@ fn execute_query_tool(sql_permissions: AgentSqlPermissions) -> ToolDefinition {
     } else if sql_permissions.allow_writes {
         "Execute SQL after the user explicitly confirmed this operation. Read queries and non-DDL writes are allowed for this run."
     } else {
-        "Execute a read-only SQL query and return results (max 50 rows). Only SELECT, WITH, SHOW, DESCRIBE, EXPLAIN statements are allowed. Write operations (INSERT/UPDATE/DELETE/DDL) are blocked."
+        "Execute a read-only SQL query and return results (max 50 rows). This run cannot execute writes or DDL because no specific SQL has been confirmed yet; this does not mean the database itself is read-only. When the user requests a write, first propose the exact SQL in one ```sql code block and ask for confirmation. After confirmation, DBX starts a new run that can execute only that exact SQL. Only SELECT, WITH, SHOW, DESCRIBE, EXPLAIN statements may be executed in this run."
     };
     ToolDefinition {
         name: "execute_query",
@@ -712,12 +750,26 @@ async fn execute_execute_query(
     // Classify SQL risk using the concrete database dialect.
     let risk = crate::sql_risk::classify_sql_risk_for_database(sql, *db_type)?;
     let connection_config = state.configs.read().await.get(connection_id).cloned();
-    if let Some(config) = connection_config {
-        if risk != SqlRisk::ReadOnly && crate::production_safety::targets_production_database(&config, database, sql) {
-            return Err("Blocked: AI agents cannot execute writes or DDL on a production database. Return the SQL for the user to review and execute manually in DBX.".to_string());
-        }
+    let targets_production = connection_config.as_ref().is_some_and(|config| {
+        risk != SqlRisk::ReadOnly && crate::production_safety::targets_production_database(config, database, sql)
+    });
+    let risk_allowed = sql_risk_allowed(risk, sql_permissions.clone());
+    let confirmed_sql_matches =
+        risk == SqlRisk::ReadOnly || sql_matches_confirmed_write(sql, &sql_permissions.confirmed_write_sql);
+
+    // Keep permission diagnostics useful without logging SQL, connection details, or result data.
+    log::debug!(
+        "[agent:execute_query:permission] risk={risk:?} targets_production={targets_production} allow_writes={} allow_dangerous={} has_confirmed_sql={} confirmed_sql_matches={confirmed_sql_matches} will_execute={}",
+        sql_permissions.allow_writes,
+        sql_permissions.allow_dangerous,
+        sql_permissions.confirmed_write_sql.is_some(),
+        !targets_production && risk_allowed && confirmed_sql_matches,
+    );
+
+    if targets_production {
+        return Err("Blocked: AI agents cannot execute writes or DDL on a production database. Return the SQL for the user to review and execute manually in DBX.".to_string());
     }
-    if !sql_risk_allowed(risk, sql_permissions.clone()) {
+    if !risk_allowed {
         if risk == SqlRisk::Transaction {
             return Err("Blocked: transaction control statements are not available to the AI agent.".to_string());
         }
@@ -729,7 +781,7 @@ async fn execute_execute_query(
 
     // When the user confirmed a specific write SQL, the agent must
     // execute only that SQL — not an arbitrary different statement.
-    if risk != SqlRisk::ReadOnly && !sql_matches_confirmed_write(sql, &sql_permissions.confirmed_write_sql) {
+    if !confirmed_sql_matches {
         let confirmed = sql_permissions.confirmed_write_sql.as_deref().unwrap_or("");
         return Err(format!(
             "Blocked: the executed SQL does not match the user-confirmed SQL.\n\
@@ -1438,6 +1490,52 @@ for line in sys.stdin:
         ));
     }
 
+    #[test]
+    fn unconfirmed_sql_tool_describes_the_confirmation_flow() {
+        let execute_query = all_tools(DatabaseType::Postgres, AgentSqlPermissions::default())
+            .into_iter()
+            .find(|tool| tool.name == "execute_query")
+            .expect("PostgreSQL Agent mode should expose execute_query");
+
+        assert!(execute_query.description.contains("no specific SQL has been confirmed yet"));
+        assert!(execute_query.description.contains("does not mean the database itself is read-only"));
+        assert!(execute_query.description.contains("propose the exact SQL in one ```sql code block"));
+        assert!(!sql_risk_allowed(SqlRisk::Write, AgentSqlPermissions::default()));
+        assert!(!sql_risk_allowed(SqlRisk::Ddl, AgentSqlPermissions::default()));
+    }
+
+    #[test]
+    fn write_confirmation_is_required_only_for_unconfirmed_write_or_ddl() {
+        let permissions = AgentSqlPermissions::default();
+        assert!(write_requires_confirmation("INSERT INTO users (id) VALUES (1)", DatabaseType::Postgres, &permissions)
+            .unwrap());
+        assert!(write_requires_confirmation("CREATE TABLE users (id INT)", DatabaseType::Mysql, &permissions).unwrap());
+        assert!(!write_requires_confirmation("SELECT * FROM users", DatabaseType::Postgres, &permissions).unwrap());
+
+        let confirmed = confirmed_write_sql_permissions(false, true, Some("CREATE TABLE users (id INT)".to_string()));
+        assert!(
+            !write_requires_confirmation("CREATE TABLE users (id INT)", DatabaseType::Postgres, &confirmed).unwrap()
+        );
+    }
+
+    #[test]
+    fn confirmed_postgres_article_tags_insert_is_write_enabled_for_nonproduction_agent() {
+        let sql = "INSERT INTO \"public\".\"article_tags\" (article_id, tag_id)\nVALUES (1, 4), (2, 4), (3, 8), (4, 5), (4, 6);";
+        let permissions = confirmed_write_sql_permissions(false, true, Some(sql.to_string()));
+
+        assert!(permissions.allow_writes);
+        assert!(permissions.allow_dangerous);
+        assert!(sql_risk_allowed(SqlRisk::Write, permissions.clone()));
+        assert!(sql_matches_confirmed_write(sql, &permissions.confirmed_write_sql));
+
+        let execute_query = all_tools(DatabaseType::Postgres, permissions)
+            .into_iter()
+            .find(|tool| tool.name == "execute_query")
+            .expect("PostgreSQL Agent mode should expose execute_query");
+        assert!(execute_query.description.contains("writes"));
+        assert!(!execute_query.description.contains("Write operations (INSERT/UPDATE/DELETE/DDL) are blocked"));
+    }
+
     #[test]
     fn oracle_agent_tools_include_explain_query() {
         let tools = all_tools(DatabaseType::Oracle, AgentSqlPermissions::default());