From f05f09b63648a4d729f0ca7313485cb59d789482 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Thu, 13 Aug 2026 22:13:00 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E5=BC=82=E6=AD=A5=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E8=AE=BE=E8=AE=A1=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 记录 v8 设计方案的三个核心不变量、wait 模式、取消机制与崩溃恢复策略 --- docs/ASYNC_SUBAGENT_DESIGN.md | 588 ++++++++++++++++++++++++++++++++++ 1 file changed, 588 insertions(+) create mode 100644 docs/ASYNC_SUBAGENT_DESIGN.md diff --git a/docs/ASYNC_SUBAGENT_DESIGN.md b/docs/ASYNC_SUBAGENT_DESIGN.md new file mode 100644 index 0000000..d27758e --- /dev/null +++ b/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>, + chat_topic_ids: HashMap, + compression_in_flight: HashSet, + topic_serial_locks: HashMap>>, + conversations: Arc, + skill_events: Arc, + + // 新增：per-topic 子代理完成队列 + sub_done_queues: HashMap>, + // 新增：per-topic wait 唤醒信号 + wait_wakeups: HashMap>, + // 新增：per-topic 等待状态 + waiting_flags: HashMap, +} +``` + +**生命周期与 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, // 未完成的子代理 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) -> 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 { + 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 { + 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` +- 入队前检查，已入队的不再入 + +### 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 工具默认超时 |