# 异步子 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 工具默认超时 |