docs: 异步子代理设计文档

记录 v8 设计方案的三个核心不变量、wait 模式、取消机制与崩溃恢复策略

docs/ASYNC_SUBAGENT_DESIGN.md

@@ -0,0 +1,588 @@
+# 异步子 Agent 方案设计
+
+> 版本: v8 (最终版)
+> 日期: 2026-08-12
+> 状态: 设计定稿
+
+## 一、核心思路
+
+子 agent 执行与父 agent 流程解耦，通过**双通道**实现异步通信：
+
+- **子代理结果**走独立队列 `sub_done_q`（直达 wait，不绕 SQLite → bus → reload）
+- **用户消息**走现有 `bus`（保持现有机制，零 channel 适配器改动）
+
+wait 工具通过 **释放锁 + select! 等待 + 重新获取锁** 实现真等待，bus 消息能在 wait 期间通过 `is_waiting` 分流注入 history 并唤醒 wait。
+
+### 核心机制
+
+```
+sub_done_q（独立队列）  ← 后台子代理完成时注入 → wait select! 直达消费
+bus（现有机制）         ← ws.rs 用户消息注入    → process_one 分流注入 + wakeup 唤醒 wait
+```
+
+## 二、关键设计决策
+
+| 决策点 | 选择 | 理由 |
+|---|---|---|
+| 子 agent 执行方式 | 异步（tokio::spawn） | 消除最大阻塞源 |
+| 子代理结果通道 | sub_done_q 独立队列 | 直达 wait，不绕 SQLite → bus → reload |
+| 用户消息通道 | 复用 bus | 现有机制，channel 零改动 |
+| wait 语义 | 释放锁 + select! + 重获取锁 | 真等待，无死锁，无 TOCTOU |
+| is_waiting 判断 | 持锁后判断 | 原子操作，无 TOCTOU |
+| 完成回调内容 | 当前结果 + 未完成子代理 id 列表 | LLM 据此决策下一步 |
+| task tool_result | running + 引导提示（只提 wait） | LLM 知道应调 wait |
+| 嵌套层 | 仅顶层异步 | 避免复杂度爆炸 |
+| 模式1（中断） | 不做 | 复杂度高，收益低 |
+
+## 三、数据结构
+
+### 1. SessionHistory 扩展（核心）
+
+```rust
+// src/gateway/session_history.rs
+pub(crate) struct SessionHistory {
+    // 现有字段
+    topic_histories: HashMap<String, Vec<ChatMessage>>,
+    chat_topic_ids: HashMap<String, String>,
+    compression_in_flight: HashSet<String>,
+    topic_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
+    conversations: Arc<dyn ConversationRepository>,
+    skill_events: Arc<dyn SkillEventRepository>,
+
+    // 新增：per-topic 子代理完成队列
+    sub_done_queues: HashMap<String, mpsc::Sender<SubagentResult>>,
+    // 新增：per-topic wait 唤醒信号
+    wait_wakeups: HashMap<String, Arc<tokio::sync::Notify>>,
+    // 新增：per-topic 等待状态
+    waiting_flags: HashMap<String, bool>,
+}
+```
+
+**生命周期与 topic_serial_locks 完全一致**：
+- 创建：`topic_serial_lock(topic_id)` 时同步创建
+- 驱逐：`evict_inactive_if_needed` 时同步移除
+- 清理：topic 删除时同步清理
+
+### 2. SubagentResult 结构
+
+```rust
+struct SubagentResult {
+    task_id: String,
+    status: SubagentStatus,        // completed/failed/timeout/cancelled
+    output: String,                // 当前子代理的输出（与 task 工具返回格式一致）
+    pending_task_ids: Vec<String>, // 未完成的子代理 id 列表
+}
+```
+
+**关键设计**：每个完成回调都带上 `pending_task_ids`，LLM 据此判断全局进度。
+
+### 3. pending_subagents 表（新增）
+
+```sql
+CREATE TABLE pending_subagents (
+    task_id TEXT PRIMARY KEY,
+    parent_session_id TEXT NOT NULL,
+    parent_topic_id TEXT NOT NULL,
+    parent_chat_id TEXT NOT NULL,
+    parent_channel TEXT NOT NULL,
+    def_name TEXT,
+    spawned_at INTEGER NOT NULL,
+    status TEXT NOT NULL DEFAULT 'running'  -- running/completed/failed/interrupted/cancelled
+);
+```
+
+## 四、执行流程
+
+### 阶段 ① 发起 task
+
+```
+用户消息 → bus → process_one → lock(topic) → agent.process()
+  → LLM 调用 task 工具
+  → TaskTool::execute():
+      ① INSERT pending_subagents (status='running')
+      ② tokio::spawn 子 agent（后台执行）
+      ③ 立即返回 tool_result:
+         "running, task_id=t1。
+          调用 wait_for_subagents(timeout_secs) 等待子代理完成并获取结果。"
+  → LLM 看到 running + 引导，自主决策下一步
+```
+
+**LLM 的 3 种可能行为**：
+- 调用 wait_for_subagents（主动等待）— 主路径
+- 调用其他工具（继续工作）— 罕见
+- 生成最终回答（不调 wait）— 后端兜底
+
+### 阶段 ② 后台子 agent 执行
+
+```
+tokio::spawn 独立运行:
+  → SubAgentEmitter 实时推消息流给前端
+  → Semaphore 限流（默认 8 并发）
+  → 执行完成或失败或超时
+```
+
+### 阶段 ③ 子 agent 完成回调（直达 sub_done_q）
+
+```
+后台 task 结束:
+  ① 查询 pending_subagents WHERE parent_topic_id=? AND status='running'
+     → 得到未完成列表 ["t2", "t3"]
+  ② sub_done_q.send(SubagentResult{
+       task_id: "t1",
+       status: "completed",
+       output: "{与 task 工具一致的输出格式}",
+       pending_task_ids: ["t2", "t3"]
+     })
+  ③ UPDATE pending_subagents SET status='completed' WHERE task_id='t1'
+```
+
+**关键**：子代理结果**不写 SQLite、不入 bus**，直接通过 sub_done_q 传递给 wait。
+
+**顺序保证**：查询未完成 → send → UPDATE
+
+### 阶段 ④ wait 工具（释放锁 + select! + 重获取锁）
+
+```
+LLM 调用 wait_for_subagents(timeout_secs):
+  ① 查询 pending_subagents
+     → 无 running → 返回 "无需等待"，继续循环
+     → 有 running → 进入等待
+
+  ② 释放锁，标记等待
+     drop(lock_guard)
+     session.set_waiting(topic_id, true)
+
+  ③ select! 等待（不持锁）
+     select! {
+       result = sub_done_q.recv() => Event::Subagent(result),
+       _ = wakeup.notified()       => Event::UserMessage,
+       _ = timeout(60s)            => Event::Timeout,
+     }
+
+  ④ 重新获取锁，清除标记（顺序重要：先获取锁，再清除）
+     let lock_guard = lock(topic_id).await
+     session.set_waiting(topic_id, false)
+
+  ⑤ 返回 tool_result
+     Subagent(r) => "子代理 {task_id} 完成: {output}。未完成: {pending}"
+     UserMessage => "有新用户消息到达"
+     Timeout     => "等待超时。未完成子代理: {pending}"
+```
+
+### 阶段 ⑤ process_one 分流（wait 期间注入用户消息）
+
+```rust
+async fn process_one(&self, inbound: InboundMessage) {
+    let guard = lock(topic_id).await;
+    
+    if session.is_waiting(topic_id) {
+        // wait 期间：只注入消息到 history，不启动 process
+        session.inject_to_history(inbound).await;  // INSERT + 更新内存 history
+        drop(guard);
+        session.wakeup(topic_id);  // 唤醒 wait
+        return;
+    }
+    
+    // 正常处理（topic 空闲或 agent 已退出）
+    agent.process().await;
+}
+```
+
+### 阶段 ⑥ process 退出兜底
+
+```
+process 退出前:
+  if waited=true:
+    → 不发 ExecutionCompleted（等 bus 消息触发新轮）
+  else if pending_subagents 有 running:
+    → 不发 ExecutionCompleted（LLM 没调 wait 但有 pending）
+  else:
+    → 发 ExecutionCompleted
+```
+
+## 五、wait 工具的完整设计
+
+### 工具定义
+
+```rust
+// 工具名：wait_for_subagents
+// 参数：timeout_secs（可选，默认 60s）
+// 描述：等待子代理完成或用户消息。用于有 pending 子代理时进入等待状态。
+
+wait_for_subagents(timeout_secs: Option<u64>) -> ToolResult {
+    let timeout = Duration::from_secs(timeout_secs.unwrap_or(60));
+    let pending = query_pending_task_ids(topic_id);
+    
+    if pending.is_empty() {
+        return ToolResult {
+            output: "无 pending 子代理，无需等待".to_string(),
+            metadata: { wait_marker: false }
+        };
+    }
+    
+    // 释放锁，标记等待
+    drop(lock_guard);
+    session.set_waiting(topic_id, true);
+    
+    // 等待（不持锁）
+    let event = select! {
+        result = sub_done_q.recv() => Event::Subagent(result),
+        _ = wakeup.notified() => Event::UserMessage,
+        _ = sleep(timeout) => Event::Timeout,
+    };
+    
+    // 重新获取锁，清除标记（顺序重要）
+    let lock_guard = lock(topic_id).await;
+    session.set_waiting(topic_id, false);
+    
+    // 先 drain sub_done_q 积压（处理多子代理同时完成）
+    match event {
+        Event::Subagent(result) => {
+            format_subagent_result(result)
+        }
+        Event::UserMessage => {
+            "有新用户消息到达".to_string()
+        }
+        Event::Timeout => {
+            let pending = query_pending_task_ids(topic_id);
+            format!("等待超时。未完成子代理: {:?}", pending)
+        }
+    }
+}
+```
+
+### wait 的三种返回场景
+
+| 事件 | 返回内容 | LLM 行为 |
+|---|---|---|
+| 子代理完成 | "子代理 t1 完成: {output}。未完成: [t2, t3]" | 处理结果，看 pending 决定继续 wait 或综合 |
+| 用户消息到达 | "有新用户消息到达" | LLM 下一轮看到注入的 user 消息 |
+| 超时 | "等待超时。未完成子代理: [t1, t2]" | 决定继续等或放弃 |
+
+## 六、两个通道的职责（核心设计）
+
+### sub_done_q：wait 专属消费
+
+```
+后台子代理完成 → sub_done_q.send(result)  ← 不获取锁，直达 wait
+                    ↓
+wait select! → 立即收到 → 返回给 LLM
+```
+
+**不经过 SQLite、不经过 bus**，直接队列传递。
+
+### bus：用户消息，wait 期间分流注入
+
+```
+用户消息 → bus.publish_inbound(UserMessage)
+  → process_one → lock(topic)
+      ├─ is_waiting=true → 注入 history + wakeup 唤醒 wait
+      └─ is_waiting=false → 正常启动 process
+```
+
+**wait 期间**：用户消息持锁注入 history，wakeup 唤醒 wait。
+**非 wait 期间**：正常走 process_one。
+
+## 七、LLM 看到的信息流
+
+| 时机 | LLM 看到的内容 |
+|---|---|
+| task 调用后 | `tool_result("running, task_id=t1。调用 wait_for_subagents 等待...")` |
+| wait 调用后 | `tool_result("已进入等待，2 个子代理运行中: [t1,t2]")` |
+| t1 完成（wait 返回） | `tool_result("子代理 t1 完成: {output}。未完成: [t2]")` |
+| 用户消息（wait 返回） | `tool_result("有新用户消息到达")` + 下轮 `user("用户消息")` |
+| t2 完成（wait 返回） | `tool_result("子代理 t2 完成: {output}。未完成: []")` |
+
+LLM 通过 `pending_task_ids` 能判断：
+- `[]` 空列表 → 全部完成，综合回答
+- `["t3"]` 非空 → 还有未完成的，继续调 wait
+
+## 八、消息的完整路径
+
+### 子代理结果
+
+```
+后台 task 完成 → sub_done_q.send(SubagentResult)
+  → wait select! 立即收到 → 返回 tool_result 给 LLM
+```
+**直达，不绕路**。
+
+### 用户消息
+
+```
+所有 channel 消息 → bus.publish_inbound(UserMessage)
+  → process_one → lock(topic)
+      ├─ is_waiting=true → 注入 history + wakeup 唤醒 wait
+      │                   → wait 返回 "有新用户消息到达"
+      │                   → LLM 下一轮看到注入的 user 消息
+      └─ is_waiting=false → 正常启动 process
+```
+
+**两条路径职责清晰，互不干扰**。
+
+## 九、三个边界问题的解法
+
+### 边界 1：wakeup vs timeout 竞态
+
+```
+timeout 和 wakeup 同时触发 → select! 随机选一个
+如果 timeout 赢：
+  - 用户消息已注入 history（process_one 在锁内完成注入）
+  - wait 返回 "超时"
+  - LLM 下一轮 reload history 会看到用户消息
+  - 不会丢失
+```
+**解法**：无需特殊处理，消息已持久化，reload 能读到。
+
+### 边界 2：is_waiting 清除时机
+
+```
+错误顺序：清除 is_waiting → 获取锁
+  → 中间窗口 process_one 看到 is_waiting=false → 启动 process → 死锁
+
+正确顺序：获取锁 → 清除 is_waiting
+  → 持锁后才清除，process_one 在锁外看到 is_waiting=true → 走注入路径
+```
+**解法**：`let guard = lock().await; set_waiting(false);` 顺序保证。
+
+### 边界 3：多条用户消息
+
+```
+wait 释放锁后，3 条用户消息依次到达：
+  msg1 → lock → 注入 → drop → wakeup
+  msg2 → lock → 注入 → drop → wakeup
+  msg3 → lock → 注入 → drop → wakeup
+
+wakeup 是 Notify，3 次 notify 只存储 1 个
+wait 被唤醒 1 次，但 history 已有 3 条消息
+wait 返回 "有新用户消息到达"
+LLM 下一轮看到 3 条 user 消息
+```
+**解法**：Notify 合并是正确行为，history 完整。
+
+## 十、时序场景表现
+
+### 场景 1：多子代理并发完成
+
+```
+① task(t1, t2, t3) → running
+② wait → 释放锁 → select!
+③ t1, t2, t3 几乎同时完成 → sub_done_q: [t1, t2, t3]
+④ wait select 收到 t1 → 返回 "t1 完成。未完成: [t2, t3]"
+⑤ LLM 处理 t1 → 再调 wait
+   wait 先 try_recv → 立即拿到 t2 → 返回 "t2 完成。未完成: [t3]"
+⑥ LLM 处理 t2 → 再调 wait
+   wait 先 try_recv → 立即拿到 t3 → 返回 "t3 完成。未完成: []"
+⑦ LLM 综合 t1+t2+t3 回答
+```
+**3 轮 wait，但无延迟（try_recv 立即返回积压）**。
+
+### 场景 2：wait 期间用户发消息
+
+```
+① task(t1) → running → wait → 释放锁 → select!
+② 用户发消息 → bus → process_one → lock（wait 已释放，获取成功）
+   → is_waiting=true → 注入 history + wakeup
+③ wait 被 wakeup 唤醒 → 重新获取锁 → 返回 "有新用户消息到达"
+④ LLM 看到 [wait_result, user("用户消息")] → 处理用户消息
+```
+**用户消息即时响应，wait 被唤醒**。
+
+### 场景 3：子代理完成 + 用户消息同时到达
+
+```
+① task(t1) → running → wait → 释放锁 → select!
+② t1 完成 → sub_done_q
+   用户发消息 → bus → process_one → 注入 + wakeup
+③ select! 随机选一个：
+   - 选 sub_done_q → 返回 "t1 完成" → LLM 下一轮看到 user 消息
+   - 选 wakeup → 返回 "有新用户消息" → LLM 下一轮看到 t1 结果（try_recv）
+```
+**两种情况都不丢失消息**。
+
+### 场景 4：父 agent 生成最终回答（不调 wait）
+
+```
+① task(t1) → running
+② LLM 生成最终回答（无 tool_calls）→ process 退出
+③ 退出前检查: pending 有 running → 不发 ExecutionCompleted
+④ t1 完成 → sub_done_q.send（无人消费，留在队列）
+   → 触发新轮 process_one（通过 bus 或定期检查）
+⑤ 新轮 LLM 看到 [自己之前的回答] → try_recv sub_done_q → 处理 t1 结果
+```
+
+**注意**：此场景需要额外机制触发新轮 process（sub_done_q 有残留时）。可通过：
+- 后台 task 完成后同时 publish_inbound(bus) 作为触发信号
+- 或 wait 退出兜底检查 sub_done_q
+
+### 场景 5：wait 超时
+
+```
+① task(t1) → running → wait(60s) → 释放锁 → select!
+② 60s 内无事件 → timeout 触发
+③ wait 重新获取锁 → 返回 "等待超时。未完成: [t1]"
+④ LLM 决定继续等（再调 wait）或放弃
+```
+
+## 十一、关键机制
+
+### 1. 队列与状态生命周期管理
+
+```rust
+// session_history.rs
+
+/// 获取或创建该 topic 的 sub_done_q sender
+pub(crate) fn sub_done_queue(&mut self, topic_id: &str) -> mpsc::Sender<SubagentResult> {
+    self.sub_done_queues
+        .entry(topic_id.to_string())
+        .or_insert_with(|| {
+            let (tx, _rx) = mpsc::channel(32);
+            tx
+        })
+        .clone()
+}
+
+/// 获取或创建该 topic 的 wakeup
+pub(crate) fn wait_wakeup(&mut self, topic_id: &str) -> Arc<Notify> {
+    self.wait_wakeups
+        .entry(topic_id.to_string())
+        .or_insert_with(|| Arc::new(Notify::new()))
+        .clone()
+}
+
+/// 设置等待状态
+pub(crate) fn set_waiting(&mut self, topic_id: &str, waiting: bool) {
+    self.waiting_flags.insert(topic_id.to_string(), waiting);
+}
+
+/// 检查等待状态
+pub(crate) fn is_waiting(&self, topic_id: &str) -> bool {
+    self.waiting_flags.get(topic_id).copied().unwrap_or(false)
+}
+
+// evict_inactive_if_needed 中同步清理
+fn evict_inactive_if_needed(&mut self) {
+    // ... 现有逻辑
+    if let Some(tid) = to_evict.cloned() {
+        self.topic_histories.remove(&tid);
+        self.topic_serial_locks.remove(&tid);
+        self.sub_done_queues.remove(&tid);      // 新增
+        self.wait_wakeups.remove(&tid);          // 新增
+        self.waiting_flags.remove(&tid);         // 新增
+    }
+}
+```
+
+### 2. pending_subagents 状态管理
+
+| 状态 | 含义 |
+|---|---|
+| running | 后台执行中 |
+| completed | 正常完成 |
+| failed | 执行失败 |
+| interrupted | 进程崩溃，启动扫描标记 |
+| cancelled | 用户取消 |
+
+### 3. 崩溃恢复
+
+**启动扫描**：
+```sql
+UPDATE pending_subagents SET status='interrupted' WHERE status='running';
+```
+
+**history 加载时**：
+- 遇到 `tool_result("running, task_id=...")` → 查 pending_subagents
+- status=interrupted → 替换为"子代理因重启中断，请决定是否重新发起"
+
+**两步操作之间崩溃**（查询未完成 → send → UPDATE）：
+- 崩溃在任何点：启动扫描标记 interrupted，history 加载时替换占位
+- 最坏情况：sub_done_q 消息丢失（wait 未消费），但 pending 表状态可恢复
+
+### 4. 取消传播
+
+- `task_id → CancellationToken` 映射
+- 用户取消 → 遍历 pending → 触发 cancel_token
+- 子代理退出 → sub_done_q.send(SubagentResult{status:"cancelled"})
+
+### 5. 去重
+
+- DefaultSubAgentRuntime 维护 `completed_tasks: HashSet<task_id>`
+- 入队前检查，已入队的不再入
+
+### 6. topic_serial_lock 一致性
+
+- 用户消息走 bus，用 topic_id 作 lock_key
+- 子代理结果走 sub_done_q，不需要 lock（直达 wait）
+- wait 释放锁后，bus 的 process_one 能获取锁注入消息
+
+## 十二、改动范围
+
+| 文件 | 改动 |
+|---|---|
+| `session_history.rs` | 新增 sub_done_q + wait_wakeup + waiting_flags（per-topic）+ 生命周期 |
+| `runtime.rs` | spawn 异步化 + 完成回调（含 pending_task_ids）注入 sub_done_q |
+| `task_tool.rs`（新） | TaskTool 返回 running + 引导提示 + spawn |
+| `wait_tool.rs`（新） | wait 释放锁 + select! + 重获取锁 + 返回 pending 信息 |
+| `agent_loop.rs` | 传递 lock_guard 给 wait 工具（支持释放/重获取） |
+| `processor.rs` | process_one 分流：is_waiting 检查 + 注入 + wakeup |
+| `db.rs` | pending_subagents 表 + 启动扫描 |
+| `history.rs` | running 占位替换 |
+| `cancel.rs` | 取消传播 |
+
+**预估总改动：~800 行**
+
+## 十三、与之前方案的对比
+
+| 维度 | v6（bus + break） | v8（双通道 + wait 释放锁） |
+|---|---|---|
+| 子代理结果通道 | bus（绕 SQLite → reload） | **sub_done_q 直达** ✅ |
+| 用户消息通道 | bus | bus |
+| wait 语义 | break 退出 | **select! 真等待** ✅ |
+| wait 期间用户消息 | 等锁（agent 退出后） | **即时注入 + wakeup** ✅ |
+| is_waiting 判断 | 不需要 | 需要（持锁后判断，无 TOCTOU） |
+| TOCTOU 风险 | 无 | **无**（持锁后判断） |
+| 死锁风险 | 无 | **无**（wait 释放锁） |
+| LLM 调用次数 | 2（batch）/ N+2 | **2-N** |
+| process 退出 | 每次都退出 reload | wait 不退出，连续推理 |
+| 上下文连贯 | 每轮 reload | **同 process 内连续** ✅ |
+| 实现复杂度 | 低 | 中（~20 行新增 + 边界解法） |
+| channel 适配器改动 | 零 | **零** ✅ |
+
+## 十四、不做的事
+
+- ❌ 模式1（中断插入）
+- ❌ user_msg_q（用户消息走 bus）
+- ❌ 嵌套层异步
+- ❌ 子代理结果走 bus（改回 sub_done_q）
+- ❌ wait 不持锁（改为释放锁 + select + 重获取）
+
+## 十五、方案定位总结
+
+| 机制 | 定位 | 通道 |
+|---|---|---|
+| wait 释放锁 + select! | **主路径**（真等待，子代理直达） | sub_done_q |
+| process_one 分流 + wakeup | **用户消息路径**（注入 + 唤醒） | bus |
+| process 退出兜底（pending 检查） | **安全网**（LLM 不调 wait 时） | - |
+
+三机制分工清晰：
+- wait select! 是核心（子代理结果直达 + 用户消息 wakeup）
+- process_one 分流是用户消息路径（wait 期间注入 + 唤醒）
+- 退出兜底是安全网（防止 LLM 不调 wait 时丢消息）
+
+## 十六、配置说明
+
+```json
+{
+  "subagent": {
+    "max_concurrent": 8,
+    "spawn_timeout_secs": 300,
+    "wait_default_timeout_secs": 60
+  }
+}
+```
+
+| 配置项 | 默认值 | 说明 |
+|---|---|---|
+| max_concurrent | 8 | 并发子 agent 数量（Semaphore） |
+| spawn_timeout_secs | 300 | 子 agent 执行超时 |
+| wait_default_timeout_secs | 60 | wait 工具默认超时 |