fix: 修复 daemon 架构中 3 个 blocking 正确性 bug

1. call_capability 元工具不可用
   - execute_tool() 中缺失 call_capability 处理分支
   - 添加解包 {name, prompt} → 提取目标工具并派发的逻辑

2. tool_call_id 传空字符串
   - Tool Consumer 中 tool_call_id 被 _ 忽略,ToolResult 传空串
   - 改为使用 LLM 原始 tool_call_id 传递

3. 高风险工具审批无限循环
   - Tool Consumer 无条件触发审批,approved_tool 从未被读取
   - ToolCall 新增 approved 字段,已批准的工具跳过审批
   - ToolExecutor 根据 approved_tool 设置标记

4. Session TTL 时间戳不更新(should-fix)
   - ToolResult 回喂时更新 last_access,防止长对话被误清理

5. 其他清理
   - load_memories 魔数字符串比较改为 starts_with
   - _sock_path 添加文档说明保留用于兼容
   - MessageKind::ToolCall 新增 approved: bool 字段
This commit is contained in:
2026-06-09 22:10:39 +08:00
parent 8ed608f01b
commit 713cbb7cc3
5 changed files with 121 additions and 11 deletions
+16
View File
@@ -0,0 +1,16 @@
# CodeGraph data files
# These are local to each machine and should not be committed
# Database
*.db
*.db-wal
*.db-shm
# Cache
cache/
# Logs
*.log
# Hook markers
.dirty
+6
View File
@@ -0,0 +1,6 @@
{
"pid": 2145387,
"version": "0.9.9",
"socketPath": "/home/xiao/project/iAs/.codegraph/daemon.sock",
"startedAt": 1781008946109
}
File diff suppressed because one or more lines are too long
+67 -11
View File
@@ -82,6 +82,8 @@ struct DaemonCtx {
}
/// Daemon 入口
///
/// `sock_path` 保留用于兼容旧版 daemon 命令的 CLI 接口,实际已不再使用。
pub async fn run(
_sock_path: String,
database: &Option<Arc<Database>>,
@@ -352,22 +354,30 @@ async fn llm_consumer_loop(
let uid = user_id.clone();
let corr = correlation_id;
let session = conv.session();
let approved_tool = approved_tool.clone();
let executor: ToolExecutor = Arc::new(move |name: &str, args_json: &str| {
// 所有工具统一通过消息队列异步执行
let eq = eq.clone();
let uid = uid.clone();
let corr = corr;
let session = session.clone();
let n = name.to_string();
let args = args_json.to_string();
let is_approved = approved_tool.as_ref().map_or(false, |t| t == &n);
Box::pin(async move {
let ch = ChannelId::wechat(&uid);
let sid = { session.lock().await.session_id };
eq.enqueue(PipelineMessage::tool_call(
ch, corr, sid, &Uuid::new_v4().to_string(), &n, &args, None,
)).await;
let tool_call_id = Uuid::new_v4().to_string();
if is_approved {
eq.enqueue(PipelineMessage::approved_tool_call(
ch, corr, sid, &tool_call_id, &n, &args, None,
)).await;
} else {
eq.enqueue(PipelineMessage::tool_call(
ch, corr, sid, &tool_call_id, &n, &args, None,
)).await;
}
Ok(format!("{}{}", ASYNC_MARKER, n))
})
});
@@ -385,9 +395,10 @@ async fn llm_consumer_loop(
// 从缓存恢复会话,注入工具结果并恢复 LLM 循环
let should_continue = {
let map = sessions.lock().await;
if let Some((conv, _)) = map.get(&correlation_id) {
let mut map = sessions.lock().await;
if let Some((conv, last_access)) = map.get_mut(&correlation_id) {
conv.session().lock().await.add_tool_result(&tool_call_id, &tool_name, &result);
*last_access = Instant::now(); // 更新访问时间,防止 TTL 误清理
true
} else {
warn!("会话已过期: corr={}", correlation_id);
@@ -534,11 +545,11 @@ async fn tool_consumer_loop(
let correlation_id = msg.correlation_id;
match msg.kind {
MessageKind::ToolCall { session_id, tool_call_id: _, tool_name, arguments, context_token } => {
MessageKind::ToolCall { session_id, tool_call_id, tool_name, arguments, context_token, approved } => {
info!("Tool Consumer: user={} tool={}", user_id, tool_name);
// 高风险工具 → 审批
if crate::tools::builtin::BuiltinRegistry::is_high_risk(&tool_name) {
// 高风险工具 → 审批(除非已被用户批准)
if !approved && crate::tools::builtin::BuiltinRegistry::is_high_risk(&tool_name) {
info!("高风险工具 {} 需要审批", tool_name);
ctx.approval_ctx.lock().await.insert(
user_id.clone(),
@@ -569,7 +580,7 @@ async fn tool_consumer_loop(
let output = execute_tool(&tool_name, &arguments, ctx, &user_id).await;
enqueue.enqueue(PipelineMessage::tool_result(
ChannelId::wechat(&user_id), correlation_id, session_id, "", &tool_name, &output, context_token,
ChannelId::wechat(&user_id), correlation_id, session_id, &tool_call_id, &tool_name, &output, context_token,
)).await;
}
MessageKind::ApprovalRequest { .. } => {
@@ -612,6 +623,50 @@ async fn execute_tool(name: &str, arguments: &str, ctx: &DaemonCtx, user_id: &st
let specs = crate::tools::builtin::BuiltinRegistry::specs();
return crate::tools::build_capability_guide(&specs);
}
"call_capability" => {
// 解包 {name, prompt} → 提取目标工具名和参数,直接派发
let cp: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default();
let target_name = cp.get("name").and_then(|v| v.as_str()).unwrap_or("").to_string();
if target_name.is_empty() {
return "请提供 name 参数".to_string();
}
let target_args = serde_json::to_string(&crate::tools::unpack_call_params(&cp))
.unwrap_or_default();
// 重新派发到 execute_tool(通过顶层 match 分支避免 async 递归)
// 上下文工具
match target_name.as_str() {
"read_memories" => {
let mems = load_memories(&ctx.memory_store, user_id).await;
if mems.is_empty() { return "暂无长期记忆".to_string(); }
return mems.iter().enumerate()
.map(|(i, v)| format!("{}. {}", i + 1, v)).collect::<Vec<_>>().join("\n");
}
"write_memory" => {
let params: serde_json::Value = serde_json::from_str(&target_args).unwrap_or_default();
let content = params.get("content").and_then(|v| v.as_str()).unwrap_or("");
if content.is_empty() { return "请提供 content 参数".to_string(); }
ctx.memory_store.write_for(user_id, content).await;
return format!("已添加长期记忆: {}", content);
}
"read_summaries" => {
let sums = load_summaries(&ctx.db, user_id).await;
if sums.is_empty() { return "暂无历史摘要".to_string(); }
return sums.iter().enumerate()
.map(|(i, s)| format!("{}. {}", i + 1, s))
.collect::<Vec<_>>().join("\n---\n");
}
"query_capabilities" => {
let specs = crate::tools::builtin::BuiltinRegistry::specs();
return crate::tools::build_capability_guide(&specs);
}
_ => {}
}
// 内置工具注册表
match crate::tools::builtin::BuiltinRegistry::execute(&target_name, &target_args).await {
Some(r) => return r.output,
None => return format!("未知工具: {}", target_name),
}
}
_ => {}
}
@@ -685,7 +740,8 @@ async fn load_history(db: &Option<Arc<Database>>, user_id: &str) -> Vec<HistoryE
async fn load_memories(memory_store: &MemoryStore, user_id: &str) -> Vec<String> {
let text = memory_store.read_for(user_id).await;
if text == "暂无长期记忆" { vec![] }
// read_for() 返回 "暂无长期记忆" 表示无记忆,否则返回 "1. ...\n2. ..." 格式
if text.starts_with("暂无") { vec![] }
else { text.lines().map(|l| l.trim().to_string()).filter(|l| !l.is_empty()).collect() }
}
+27
View File
@@ -60,6 +60,8 @@ pub enum MessageKind {
tool_name: String,
arguments: String,
context_token: Option<String>,
/// 该工具调用是否已被用户批准(跳过审批流程)
approved: bool,
},
/// 工具执行结果 → 回喂 LLM Consumer
ToolResult {
@@ -157,6 +159,31 @@ impl PipelineMessage {
tool_name: tool_name.into(),
arguments: arguments.into(),
context_token,
approved: false,
},
}
}
/// 快速构造一条已批准的工具调用(跳过审批)
pub fn approved_tool_call(
channel: ChannelId,
correlation_id: Uuid,
session_id: Uuid,
tool_call_id: impl Into<String>,
tool_name: impl Into<String>,
arguments: impl Into<String>,
context_token: Option<String>,
) -> Self {
Self {
channel,
correlation_id,
kind: MessageKind::ToolCall {
session_id,
tool_call_id: tool_call_id.into(),
tool_name: tool_name.into(),
arguments: arguments.into(),
context_token,
approved: true,
},
}
}