修复: gen_stream判重+重试分类+状态机最短路径+既有测试修复

crates/df-storage/src/crud/conversation_repo.rs

@@ -319,6 +319,38 @@ impl AiToolExecutionRepo {
         .map_err(storage_err)?
     }
 
+    /// 按 (conversation_id, tool_name, arguments) 查最新一条审计记录(重试防护用)。
+    ///
+    /// 语义:同一会话内同一工具同一参数的重试判重。区别于 `find_by_tool_call_id`
+    /// (裸 id 匹配,弱模型 provider 的 tool_call_id 每轮重排,裸 id 判重会误杀跨轮合法调用,
+    /// 实证 2026-08-11)。
+    pub async fn find_by_conv_tool_args(
+        &self,
+        conversation_id: &str,
+        tool_name: &str,
+        arguments: &str,
+    ) -> Result<Option<AiToolExecutionRecord>> {
+        let conn = self.conn.clone();
+        let cid = conversation_id.to_owned();
+        let tname = tool_name.to_owned();
+        let args = arguments.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let mut stmt = guard
+                .prepare(
+                    "SELECT * FROM ai_tool_executions WHERE conversation_id = ?1 AND tool_name = ?2 AND arguments = ?3 ORDER BY requested_at DESC LIMIT 1",
+                )
+                .map_err(storage_err)?;
+            let row = stmt
+                .query_row(params![cid, tname, args], |row| ai_tool_execution_from_row(row))
+                .optional()
+                .map_err(storage_err)?;
+            Ok(row)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
     /// 列出所有 status=pending 的审计行(启动重建 pending_approvals 用)
     ///
     /// 专用 SELECT(非 query 宏——后者硬编码 ORDER BY created_at,而本表无该列)。
@@ -428,9 +460,9 @@ impl AiToolExecutionRepo {
                 params_vec.push(Box::new(r.clone()));
             }
             if let Some(k) = &kw {
-                let escaped = k.replace('%', "\\%").replace('_', "\\_");
+                let escaped = k.replace('|', "||").replace('%', "|%").replace('_', "|_");
                 let pat = format!("%{escaped}%");
-                where_clauses.push(format!("tool_name LIKE ?{} ESCAPE '\\'", params_vec.len() + 1));
+                where_clauses.push(format!("tool_name LIKE ?{} ESCAPE '|'", params_vec.len() + 1));
                 params_vec.push(Box::new(pat));
             }
 
@@ -491,9 +523,9 @@ impl AiToolExecutionRepo {
                 params_vec.push(Box::new(r.clone()));
             }
             if let Some(k) = &kw {
-                let escaped = k.replace('%', "\\%").replace('_', "\\_");
+                let escaped = k.replace('|', "||").replace('%', "|%").replace('_', "|_");
                 let pat = format!("%{escaped}%");
-                where_clauses.push(format!("tool_name LIKE ?{} ESCAPE '\\'", params_vec.len() + 1));
+                where_clauses.push(format!("tool_name LIKE ?{} ESCAPE '|'", params_vec.len() + 1));
                 params_vec.push(Box::new(pat));
             }
 
@@ -844,4 +876,71 @@ mod tests {
         assert_eq!(legacy.model_configs.len(), 2, "老字符串数组应转 2 个默认 ModelConfig");
         assert_eq!(legacy.model_configs[0].model_id, "glm-4-flash");
     }
+
+    // ---------- find_by_conv_tool_args(重试判重,SC-260811-P0-1) ----------
+
+    /// 构造一条审计记录(参数自洽,满足 NOT NULL 约束)。
+    fn audit_rec(
+        id: &str, conv: &str, tc_id: &str, tool: &str, args: &str, status: &str,
+    ) -> AiToolExecutionRecord {
+        AiToolExecutionRecord {
+            id: id.into(),
+            conversation_id: Some(conv.into()),
+            message_id: None,
+            tool_call_id: tc_id.into(),
+            tool_name: tool.into(),
+            arguments: args.into(),
+            result: None,
+            status: status.into(),
+            risk_level: "low".into(),
+            requested_at: "0".into(),
+            executed_at: None,
+            decided_by: None,
+        }
+    }
+
+    /// P0-1 核心语义:弱模型跨轮复用同 tool_call_id(gen_stream_0),但工具/参数不同 →
+    /// find_by_conv_tool_args 不得误判为重试(旧 find_by_tool_call_id 会误命中)。
+    #[tokio::test]
+    async fn find_by_conv_tool_args_ignores_same_id_different_tool() {
+        let db = Database::open_in_memory().await.expect("open_in_memory");
+        let repo = AiToolExecutionRepo::new(&db);
+        // 会话开头已执行过 list_tasks(gen_stream_0),之后同会话又出现 gen_stream_0 的 advance_task
+        repo.insert(audit_rec("a1", "conv1", "gen_stream_0", "list_tasks", "{\"limit\":50}", "completed")).await.expect("insert");
+        repo.insert(audit_rec("a2", "conv1", "gen_stream_0", "advance_task", "{\"id\":\"t1\",\"target_status\":\"blocked\"}", "completed")).await.expect("insert");
+
+        // 查 advance_task 同三元组 → 命中(a2)
+        let got = repo.find_by_conv_tool_args("conv1", "advance_task", "{\"id\":\"t1\",\"target_status\":\"blocked\"}").await.expect("query");
+        assert!(got.is_some(), "同 (conv,tool,args) 应命中");
+        assert_eq!(got.unwrap().id, "a2");
+
+        // 查 list_tasks 同三元组 → 命中(a1),互不串扰
+        let got2 = repo.find_by_conv_tool_args("conv1", "list_tasks", "{\"limit\":50}").await.expect("query");
+        assert!(got2.is_some());
+        assert_eq!(got2.unwrap().id, "a1");
+
+        // 跨会话同 tool_call_id 不命中(不同 conv)
+        let got3 = repo.find_by_conv_tool_args("conv2", "advance_task", "{\"id\":\"t1\",\"target_status\":\"blocked\"}").await.expect("query");
+        assert!(got3.is_none(), "不同会话同参数不应命中");
+    }
+
+    /// 同 (conv,tool,args) 且已落定(completed) → 判重返回记录(调用方 retry_count≥1);
+    /// pending 视为"尚未落定"(首次挂起审批)不判重。
+    #[tokio::test]
+    async fn find_by_conv_tool_args_pending_not_retry() {
+        let db = Database::open_in_memory().await.expect("open_in_memory");
+        let repo = AiToolExecutionRepo::new(&db);
+        repo.insert(audit_rec("a1", "conv1", "call_x", "write_file", "{\"path\":\"a.go\"}", "pending")).await.expect("insert");
+        // pending 记录不视为重试
+        let got = repo.find_by_conv_tool_args("conv1", "write_file", "{\"path\":\"a.go\"}").await.expect("query");
+        assert!(got.is_some(), "pending 记录也应能查到(状态判定在调用方 detect_retry_count 做,此处只负责按三元组定位)");
+        assert_eq!(got.unwrap().status, "pending");
+
+        // 同三元组 latest(有 completed 应返回最新一条 requested_at DESC)
+        let mut rec2 = audit_rec("a2", "conv1", "call_x", "write_file", "{\"path\":\"a.go\"}", "completed");
+        rec2.requested_at = "1".into(); // 晚于 a1 的 "0",保证 DESC 序可辨
+        repo.insert(rec2).await.expect("insert");
+        let got2 = repo.find_by_conv_tool_args("conv1", "write_file", "{\"path\":\"a.go\"}").await.expect("query");
+        assert_eq!(got2.unwrap().id, "a2", "同三元组应返回最新落定记录");
+    }
 }