xt/agcore

Files

T

徐涛 28ca43ccb2 chore(docs): 将设计文档从 docs 移至 design 目录

将 note、pdd、prd、roadmap 四类文档分别归入 `design/` 下对应子目录中，并新增 `.gitkeep` 占位文件

2026-07-23 05:45:53 +08:00

37 KiB

Raw Permalink Blame History

Phase 18：Agent 角色热切换与子代理调度

背景与目标

问题空间

agcore 已完整交付 Phase 0-17，具备 SessionManager 会话生命周期管理、会话树（父子层次）、Checkpointer 检查点、流式输出、ContextSlot 上下文分区、MemoryStore 持久化、摘要自动生成等能力。当前 Session 在 create() 时绑定一个 Arc<dyn Agent>，此后无法变更角色；会话间调度仅通过 create_child() + submit_turn() 手动编排，缺乏内建的子代理派发机制。

Phase 18 要解决两个正交但关联的问题：

Agent 角色热切换：运行时替换 session 绑定的 Agent，保留上下文（slot 历史、turn_index、session_memory、cost_so_far）
子代理调度：在 SessionManager 上提供声明式的 dispatch / dispatch_stream / dispatch_all API，支持父子 session 间的 Memory 继承、bridge_keys 注入、并发控制、结构化回传

目标

提供 SessionManager::switch_agent(session_id, new_agent)，替换 Arc<dyn Agent>，全量保留 slot / turn_index / session_memory
提供 DispatchConfig / SubTaskResult / SubTaskStreamEvent 类型以及 dispatch / dispatch_stream / dispatch_all 三个核心方法
实现父转子三层级交互：父->子（Memory 快照继承 + bridge_keys）、子->父（SubTaskResult 结构化回传 + result_summary）、子<->子（shared namespace）
产出 4 个端到端示例：agent_switch_demo / sub_agent_dispatch_demo / bridge_keys_demo / dispatch_stream_demo

依赖与优先级

依赖：Phase 17（SessionManager + 会话树 + Checkpointer）[高]
优先级：P0
预估规模：约 720 行核心 + 210 行测试 + 450 行示例
审查修复：第 1 轮审查修复（Finalize 方案重写 + 4 个 🔴 阻塞 + 8 个 🟡 改进）

当前状态分析

现有架构中的关键接入点

接入点	位置	可用性	分析
`AgentSession.agent`	`src/agent/session.rs:53`	`pub` 字段	可直接替换，无需新增 setter [高]
`AgentSession.session_id`	`src/agent/session.rs:51`	`pub` 字段	子 session 创建后可读取 [高]
`AgentSession.session_memory`	`src/agent/session.rs:58`	`pub` 字段	dispatch 后父可读子 memory [高]
`SessionManager::create_child(parent_id, agent)`	`src/engine/session_manager.rs:202-239`	`pub async`	dispatch 可直接复用，bundle 继承避免重复构造 [高]
`SessionMemory::list_entries()`	`src/agent/session_memory.rs:81-110`	`pub async`	可获取全量条目用于父子继承 / bridge_keys 过滤 [高]
`SessionMemory::set_with_meta()`	`src/agent/session_memory.rs:61-79`	`pub async`	子 session 写入继承数据时保留原始 metadata [高]
`CostTracker`	`src/llm/cycle.rs`	derive `Clone`	SubTaskResult 可直接 clone usage [高]
`SessionMeta` 持久化	`src/engine/session_manager.rs:32-67`	`pub(crate)`	以 `session:{id}:meta` key 存到 MemoryStore；switch 后需更新 agent_name [高]
`SessionManager::save_session_meta()`	`src/engine/session_manager.rs:131-142`	`async fn` (非 pub)	switch_agent 需要类似的 meta 更新能力；考虑提取为 `pub(crate)` [高]
`SessionManager::destroy()`	`src/engine/session_manager.rs:495-512`	`pub async`	dispatch 失败时清理子 session 可直接复用 [高]
`SessionManager.sessions`	`src/engine/session_manager.rs:92`	`pub(crate)` RwLock	switch_agent 和 dispatch 的 get / replace 操作均依赖此字段 [高]
`EngineError` 枚举	`src/engine/error.rs:16-46`	`#[non_exhaustive]`	已有 6 个变体，需追加 DispatchFailed / SwitchFailed / SubAgentStreamError [中]
`futures-core` / `futures-util`	`Cargo.toml:17-18`	已引入	dispatch_stream 返回 `Pin<Box<dyn Stream>>` 所需依赖已就绪，无需新增 [高]

Agent trait 与 SessionManager 之间的关系

AgentSession {
    agent: Arc<dyn Agent>,         // 可替换
    session_memory: SessionMemory, // 可继承（clone backend）
    slots: HashMap<String, ContextSlot>,
    turn_index: u32,
    cost_so_far: CostTracker,
    // ... 其余内部字段
}

SessionManager {
    sessions: RwLock<HashMap<String, Arc<Mutex<AgentSession>>>>,
    checkpointer: Checkpointer,
    store: Arc<dyn MemoryStore>,
    config: SessionManagerConfig,
}

AgentSession.agent 是 pub 字段，这意味着 switch_agent 只需 get() → lock() → 替换 agent → 写回 meta。Route 明确、无架构阻力 [高]。

锁契约（需格外注意）

src/engine/session_manager.rs:4-12 记录了锁契约：不持有 RwLock 跨越 .await。所有 .await 点必须在 RwLock guard drop 之后。这意味着：

switch_agent：读锁 get() 返回 Arc<Mutex<AgentSession>> 后释放，然后 lock session 级别的 Mutex → 替换 agent → 释放 Mutex → save_session_meta（I/O）[高]
dispatch：读锁 get() 父 session → 释放 → create_child（内部写锁）→ lock 子 session → inherit memory → submit_turn → 释放 [高]
不会引入新的死锁风险 [高]

现有测试覆盖

SessionManager 已有 906 行（含 12 个测试），覆盖 create/get/destroy/create_child/replace/recover/children/parent/并发创建等场景：src/engine/session_manager.rs:527-906。Phase 18 新增测试不修改这些已有测试。

调研发现

1. bridge_keys 注入位置

问题：bridge_keys 本质是父 session 想注入到子 agent prompt 中的上下文数据。它应该放在哪里？

调研来源：

docs/note-opencode-subagent-dispatch.md — 明确反对修改 Agent 的 system_prompt() [高]
src/agent/agent.rs:21 — system_prompt() 返回 &str，无状态变更能力 [高]
src/agent/session_memory.rs — SessionMemory::set() 提供 key-value 写入，子 agent 可读 [高]

结论：bridge_keys 通过 SessionMemory 副本继承 + 过滤注入，不碰 system_prompt()。子 agent 通过 get_session_data(key) 读取桥接数据 [高]。

2. SessionMemory 继承策略

问题：子 agent 启动时，父 session 的 SessionMemory 如何传递？

方案 A —— 引用共享：父子共享同一 SessionMemory 实例（Arc clone 后端）。优点是零拷贝，缺点是父子隔离被破坏 [中]。

方案 B —— 快照副本：父调用 list_entries() 获取全量条目，子通过 set_with_meta() 写入自己的 namespace。优点是隔离性强，缺点是 O(n) 拷贝开销 [高]。

来源：

src/agent/session.rs:503-524 — to_snapshot() 已实现类似的 list_entries → HashMap 拍平 [高]
src/agent/session_memory.rs:61-79 — set_with_meta() 可保留原始 metadata [高]
docs/note-opencode-subagent-dispatch.md — SA 建议副本策略 [中]

结论：采用方案 B（快照副本），隔离性优先。inherit_session_memory 内部使用 list_entries() → 按 bridge_keys 过滤 → set_with_meta() 写入子 namespace [高]。

3. dispatch_all 部分成功语义

问题：当一批子代理中部分失败时，dispatch_all 应该整体失败还是返回部分成功的 Vec？

来源：docs/note-opencode-subagent-dispatch.md — PM 和 SA 一致认为应返回部分成功语义 [高]。

结论：返回 Vec<Result<SubTaskResult, EngineError>>。调用方可迭代检查每个结果，失败条目保留 checkpoint 以便审计 [高]。

4. dispatch_stream 生命周期

问题：dispatch_stream 需要返回一个流，流内部要做 submit_turn_stream + finalize_active_stream。session 所有权和生命周期如何管理？

来源：

src/engine/session_manager.rs:390-412 — submit_turn_stream 的锁模式：短持锁获取流后立即释放 [高]
src/agent/session.rs:384-445 — submit_turn_stream 自身不持有跨 await 的锁 [高]
docs/note-opencode-subagent-dispatch.md — SA 建议在 AgentSession 新增 finalize_active_stream() 内部方法 [中]

结论：dispatch_stream 使用 &Arc<Self> 签名 + tokio::spawn。内部流管道：create_child → inherit_memory → submit_turn_stream → mpsc channel 转发事件 → 流消费完毕后调用 finalize_active_stream()。session 通过 Arc<Mutex<AgentSession>> 在 spawned task 中持有 [高]。

5. Cargo.toml 依赖分析

来源：

Cargo.toml:17 — futures-util = "0.3" 已在依赖中 [高]
Cargo.toml:15 — tokio-stream = "0.1" 已在依赖中 [高]
Cargo.toml:16 — futures = "0.3" 已在依赖中 [高]

结论：dispatch_stream 所需的 StreamExt / ReceiverStream 所需的基础设施已全部就绪，无需新增任何依赖 [高]。

可选方案

方案 A：switch_agent 作为 AgentSession 方法 vs SessionManager 方法

维度	A1: AgentSession 方法	A2: SessionManager 方法
实现位置	`agent/session.rs`	`engine/switch.rs`
职责归属	session 实例级	管理器级
能否更新 SessionMeta	不能（无 store 引用）	能（有 store + checkpointer）
能否做自动 checkpoint	不能（无 checkpointer）	能
与 create_child / replace 对齐	不对齐（create 在 SM）	对齐（都在 SM）

来源：

src/agent/session.rs:49-71 — AgentSession 不持有 store / checkpointer 引用 [高]
src/engine/session_manager.rs:325-347 — replace() 是 SM 方法，涉及 meta 持久化 [高]
roadmap lines 838 — switch_agent(session_id, new_agent) 签名暗示 SM 方法 [中]

结论：采用 A2（SessionManager 方法）。AgentSession 没有 store 引用，无法更新 SessionMeta。独立文件 engine/switch.rs 作为 SessionManager 的 impl 块。

方案 B：bridge_keys 注入方式

维度	B1: SessionMemory 副本继承 + 过滤	B2: 修改 Agent trait
系统 prompt 侵入性	无	需新增 `set_bridge_data()` 方法
switch_agent 兼容性	天然兼容（与 agent 解耦）	switch 后需重新注入
实现复杂度	一个私有辅助函数	需改 Agent trait + 所有实现
测试增量	小（只测 `inherit_session_memory`）	大（需测所有 Agent impl）

来源：

src/agent/agent.rs:16-30 — Agent trait 当前仅 3 个方法，简洁 [高]
docs/note-opencode-subagent-dispatch.md — "bridge_keys 通过 slot 注入，不修改 Agent system_prompt" [高]

结论：采用 B1。隔离关注点：Agent 负责"角色"，SessionMemory 负责"桥接数据"。

方案 C：dispatch_stream 返回类型

维度	C1: `Pin<Box<dyn Stream<Item=SubTaskStreamEvent>+Send>>`	C2: 自定义 struct 包装
与现有 API 一致性	与 `submit_turn_stream` 一致 [高]	不一致
调用方灵活性	直接 `.next()` + StreamExt	需解包装
实现复杂度	直接返回 stream	需额外 struct + 方法
可组合性	高（可直接 map/filter/collect）	低

来源：

src/engine/session_manager.rs:390-412 — submit_turn_stream 返回 Pin<Box<dyn Stream<Item=StreamEvent>+Send>> [高]

结论：采用 C1。保持一致的模式，调用方可以 StreamExt::collect / map 等。

否决方案

方案	否决原因
switch_agent 做自动 checkpoint	与 auto_checkpoint 语义不一致（submit_turn 才触发），用户可手动 checkpoint。来源：`src/engine/session_manager.rs:357-384` auto_checkpoint 仅在 submit_turn/finalize_turn 触发
AgentSession 中的 `Agent` 用 `Box<dyn Agent>`	与现有 `Arc<dyn Agent>` 不一致，且 SessionSnapshot 不序列化 agent（`src/agent/session.rs:502`）。来源：`src/agent/session.rs:53`
dispatch_all 返回所有成功再返回	需要调用方等待全部完成才能拿到第一个结果。Rust 已有 `JoinSet` / `FuturesUnordered` 可选，但 v0.3 先保持简单
child_memory 加额外权限控制	子 session 是父创建的，父天然有 destroy / read 权限。来源：`docs/note-opencode-subagent-dispatch.md` PM 明确"不做额外权限控制"
子 session 失败时保留 checkpoint	`destroy()` 调 `checkpointer.delete_all`（`session_manager.rs:508`），不保留持久化残留。失败路径的调试信息通过 `tracing::error!` 日志记录

#	决策	结论	理由
D1	switch_agent 位置	SessionManager 方法，在 `engine/switch.rs`	需 store 更新 SessionMeta，AgentSession 无 store 引用
D2	bridge_keys 注入方式	SessionMemory 副本继承 + 过滤，不碰 system_prompt	概念正交，switch_agent 友好
D3	Memory 继承策略	快照副本（list_entries → set_with_meta）	父子隔离优先，O(n) 拷贝可接受
D4	dispatch_all 返回类型	`Vec<Result<SubTaskResult, EngineError>>`	部分成功语义，Rust-idiomatic
D5	dispatch_stream 返回类型	`Pin<Box<dyn Stream<Item=SubTaskStreamEvent>+Send>>`	与 `submit_turn_stream` 一致
D6	switch_agent 的 lock 策略	get() 读锁立即释放 → Mutex lock → 替换 → 释放 Mutex → I/O	严格遵循已有锁契约
D7	switch checkpoint	不自动 checkpoint	与 `auto_checkpoint` 语义一致（仅 submit_turn/finalize_turn 触发）
D8	dispatch 失败清理	destroy 子 session	不留僵尸 session
D9	dispatch 失败时 checkpoint	destroy 清理全部（含 checkpoint）	`destroy()` 内部调 `checkpointer.delete_all`，不留持久化垃圾
D10	dispatch_all 并发控制	`tokio::sync::Semaphore`	轻量、内建、语义清晰
D11	dispatch_all / dispatch_stream 签名	`self: &Arc<Self>`	满足 `tokio::spawn` `'static` 约束
D12	子 session 创建时 bundle	从父 session 的 `RuntimeBundle` clone	复用 `create_child` 已有逻辑

实施建议

阶段划分

共 9 个步骤，建议依次实施，不可并行。总预估时间由实现者在实施时评估。

Step 1 — `error.rs` 扩展

文件：src/engine/error.rs 内容：EngineError 追加 3 个变体

DispatchFailed(String) — 子代理调度通用失败
SwitchFailed(String) — 角色切换失败
SubAgentStreamError { child_id: String, detail: String } — 流式调度中的子代理错误

验证：cargo build 成功。注意：已存在的 #[non_exhaustive] 属性确保这不是 breaking change [高]。

Step 2 — `session.rs` 无变更

文件：src/agent/session.rs — 不修改现有 API。

审查发现：第一轮审查确认 finalize_active_stream() 假设不成立。submit_turn_stream（session.rs:384-445）返回 stream 后 LlmCycle 即被 drop（cycle.rs:647 通过 std::mem::take 移出消息），不存在"active stream 内部状态"可读取。

结论：改为在 dispatch_stream 的 spawn task 中从 StreamEvent 序列重建消息列表，直接调用已有的 finalize_turn(response, new_messages).await。详见 Step 7 第 3 项。

验证：不修改 session.rs，Step 7 实施前 cargo build 可通过。

Step 3 — `switch.rs`

文件：src/engine/switch.rs 内容：

impl SessionManager {
    /// 热切换指定 session 的 Agent 角色。
    ///
    /// - 保留 slot 历史 / turn_index / session_memory / cost_so_far
    /// - 自动更新 SessionMeta 中的 agent_name（保持原始 created_at / parent_id）
    /// - 不自动 checkpoint（与 `auto_checkpoint` 语义一致：仅 submit_turn 触发）
    /// - **注意**: 切换后新的 system_prompt 将与已有对话历史共存。
    ///   建议在切换后发送一条明确的上下文过渡提示
    ///   （如"你现在以新角色 X 的身份继续对话"）作为切换后的首条输入。
    /// - **安全提示**: `AgentSession.agent` 是 `pub` 字段可直接访问，
    ///   绕过 `switch_agent` 直接修改会导致 SessionMeta 中的 agent_name
    ///   与内存状态不一致，请始终使用此方法。
    pub async fn switch_agent(
        &self,
        session_id: &str,
        new_agent: Arc<dyn Agent>,
    ) -> Result<(), EngineError> {
        // 1. get session（RwLock 读锁，返回后释放）
        let session = self.get(session_id).await?;

        // 2. lock Mutex，替换 agent，读 name + turn_index
        let (agent_name, turn_index) = {
            let mut guard = session.lock().await;
            guard.agent = new_agent;
            (guard.agent.name().to_string(), guard.turn_index())
        }; // 释放 Mutex

        // 3. 读取原始 SessionMeta（用于保留 created_at / parent_id）
        let existing_meta = self
            .load_session_meta(session_id)
            .await?
            .ok_or_else(|| EngineError::SessionNotFound(session_id.to_string()))?;

        // 4. 构造新 meta 并持久化（I/O，无锁）
        let meta = SessionMeta {
            session_id: session_id.to_string(),
            agent_name,
            parent_id: existing_meta.parent_id,
            created_at: existing_meta.created_at,
            turn_count: turn_index,
        };
        self.save_session_meta(&meta).await?;

        tracing::info!(
            session_id = %session_id,
            agent_name = %meta.agent_name,
            previous_agent = %existing_meta.agent_name,
            "agent switched"
        );
        Ok(())
    }
}

测试（预计 4 个）：

基本切换：switch 后 agent.name() 返回新 name
上下文保留：turn_index / session_memory / slot 历史均不变
SessionMeta 持久化：load_session_meta 验证 agent_name 已更新
不存在的 session：返回 SessionNotFound

验证：cargo test --all-targets + clippy

Step 4 — `sub_agent.rs` 类型

文件：src/engine/sub_agent.rs 内容：3 个类型定义

/// 子代理调度配置。
#[derive(Debug, Clone)]
pub struct DispatchConfig {
    /// 最大并发数（dispatch_all 用）。默认 10。
    pub max_concurrency: usize,
    /// 是否继承父 SessionMemory。默认 true。
    pub inherit_session_memory: bool,
    /// 桥接 key 列表：
    /// - `None` = 不继承任何父 SessionMemory
    /// - `Some(vec![])` = 继承全部父 SessionMemory
    /// - `Some(keys)` = 仅继承指定的 keys
    /// 默认 `None`（零继承），显式选择加入。
    pub bridge_keys: Option<Vec<String>>,
    /// 子↔子共享 namespace。如果为 `Some(prefix)`，
    /// 子 agent 可通过 `session.get_session_data(key)` 访问
    /// `shared:{prefix}:{key}` 命名空间的数据。
    /// 默认 `Some(parent_session_id)`。
    pub shared_namespace: Option<String>,
}

impl Default for DispatchConfig {
    fn default() -> Self {
        Self {
            max_concurrency: 10,
            inherit_session_memory: true,
            bridge_keys: None,
            shared_namespace: None,
        }
    }
}

/// 子代理执行结果。
///
/// dispatch 成功后子 session **保留在 SessionManager 中**，调用方可
/// 通过 `sm.get(&result.child_id)` 获取子 session 引用，进而通过
/// `session_memory()` 读取子 SessionMemory（如 "result_summary"）。
#[derive(Debug)]
pub struct SubTaskResult {
    /// 子 session ID。可通过此 ID 在 SessionManager 中读取子 session。
    pub child_id: String,
    /// LLM 最终响应。
    pub response: MessageResponse,
    /// 本次调用的 token 用量。
    pub usage: CostTracker,
    /// 可选摘要（读取子 session_memory 中的 "result_summary"）。
    pub summary: Option<String>,
}

/// 流式子代理调度事件。
#[derive(Debug)]
pub enum SubTaskStreamEvent {
    /// 子 session 已创建（携带 child_id）。
    ChildCreated { child_id: String },
    /// LLM 流事件（透传）。
    Stream(StreamEvent),
    /// 执行完成（携带完整结果）。
    Completed(SubTaskResult),
}

SubTaskStreamEvent 需实现 Display 和 std::error::Error（Completed 和 ChildCreated 不触发错误路径，Display 仅用于调试日志）[中]。

验证：cargo build

Step 5 — `sub_agent.rs dispatch` 核心

文件：src/engine/sub_agent.rs 内容：

impl SessionManager {
    /// 私有辅助：从父 session memory 继承条目到子 session。
    ///
    /// **一致性模型**：捕获的是调用时刻的父 session_memory 快照。
    /// 即使在 `list_entries()` 返回后、`set_with_meta()` 写入前
    /// 父 session 被并发写入新数据，子 session 也**不会**看到这些
    /// 新数据（快照副本的内生特征）。[审查确认]
    async fn inherit_session_memory(
        &self,
        parent_id: &str,
        child_id: &str,
        config: &DispatchConfig,
    ) -> Result<(), EngineError> { /* ... */ }
 
    /// 派发一个子任务，返回结构化结果。
    pub async fn dispatch(
        &self,
        parent_id: &str,
        sub_agent: Arc<dyn Agent>,
        task: impl Into<String>,
        config: DispatchConfig,
    ) -> Result<SubTaskResult, EngineError> { /* ... */ }
}

dispatch 流程：

create_child(parent_id, sub_agent) → 获取 child_id
若 config.inherit_session_memory == true → inherit_session_memory(parent_id, child_id, config)
submit_turn(child_id, task) → 获取 response
读取 "result_summary"（可选）
返回 SubTaskResult（子 session 保留在 SessionManager 中，可通过 sm.get(&child_id) 读取 child_memory）
失败路径：let _ = self.destroy(&child_id).await; tracing::error!(...)（静默吞掉清理错误，原始 EngineError 优先；destroy 会清理子 session 的 SessionMeta + checkpoint 条目，不留僵尸）

测试（预计 5 个）：

基本调度：子 agent 返回预期响应
bridge_keys 过滤：仅指定的 key 被继承
memory 继承：父 set 的值子可读到
submit_turn 失败：错误传播 + 子 session 被销毁
无效 parent_id：返回 SessionNotFound

验证：cargo test --all-targets

Step 6 — `sub_agent.rs dispatch_all`

文件：src/engine/sub_agent.rs 内容：

impl SessionManager {
    /// 并行派发一批子任务。
    pub async fn dispatch_all(
        self: &Arc<Self>,
        parent_id: &str,
        tasks: Vec<(Arc<dyn Agent>, String)>,
        config: DispatchConfig,
    ) -> Vec<Result<SubTaskResult, EngineError>> { /* ... */ }
}

设计要点：

使用 tokio::sync::Semaphore 限制并发数（默认 config.max_concurrency）
Semaphore acquire 在 spawn 内：let permit = semaphore.clone().acquire_owned().await; — permit 所有权转移到 spawned task。避免 spawn N 个 task 时全量分配 Future 内存 [审查修复]
每个 task tokio::spawn + Arc<Self> clone
内部调用 dispatch 的同类逻辑（create_child → inherit → submit_turn）
indexed 收集：预分配 Vec<Option<Result<...>>> 按 tasks 索引填入，维持输入顺序。不使用排序（排序需等所有 child_id 生成后）[审查修复]
每个结果独立：Ok(SubTaskResult) 或 Err(EngineError)

测试（预计 4 个）：

并行 3 个全部成功
部分失败（MockProvider 对特定 task 返回错误）
Semaphore 上限验证（max_concurrency=1 时串行执行）
空 tasks 列表

验证：cargo test --all-targets

Step 7 — `sub_agent.rs dispatch_stream`

文件：src/engine/sub_agent.rs 内容：

impl SessionManager {
    pub async fn dispatch_stream(
        self: &Arc<Self>,
        parent_id: &str,
        sub_agent: Arc<dyn Agent>,
        task: impl Into<String>,
        config: DispatchConfig,
    ) -> Result<
        Pin<Box<dyn Stream<Item = SubTaskStreamEvent> + Send>>,
        EngineError,
    > { /* ... */ }
}

设计要点：

同步部分（lock 外）：create_child + inherit_memory
获取 Stream 后通过 tokio::sync::mpsc::unbounded_channel 转发事件（与 LLM stream 内部背压策略一致，避免有界 channel 的 sender 阻塞风险）[审查修复]
spawn task 持有 Arc<Mutex<AgentSession>> 消费 LLM stream

消息重建机制（替代已移除的 finalize_active_stream()）：[审查修复]

// 在 spawn task 中：
let mut new_messages: Vec<Message> = vec![Message::user_text(&task)];
let mut final_response: Option<MessageResponse> = None;

while let Some(event) = llm_stream.next().await {
    // 转发事件到输出 channel
    tx.send(SubTaskStreamEvent::Stream(event.clone()))?;
    // 从 ToolExecutionCompleted 构造 ToolResult 消息
    if let StreamEvent::ToolExecutionCompleted { tool_name, tool_call_id, input, output } = &event {
        new_messages.push(Message::tool_result(tool_call_id, tool_name, output));
    }
    // 捕获最终响应
    if let StreamEvent::MessageComplete(ref resp) = event {
        final_response = Some(resp.clone());
    }
}
// 流结束后，追加 assistant 消息并 finalize
if let Some(response) = &final_response {
    new_messages.push(response.message.clone());
    child_session.lock().await
        .finalize_turn(response, new_messages).await?;
}

事件序列：ChildCreated → Stream(StreamEvent) × N → Completed(SubTaskResult)
消费者 drop receiver → unbounded channel sender 错误 → task 自动退出

测试（预计 4 个）：

事件序列验证：收到 ChildCreated → 至少一个 Stream → Completed
错误传播：LLM 内部错误 → 正确映射到 error 事件
receiver dropped：drop receiver 后 task 正确退出，不 panic
finalize 正确性：Completed 中的 usage / summary 正确

验证：cargo test --all-targets

Step 8 — `mod.rs` + 集成验证

文件：src/engine/mod.rs 内容：追加 pub mod switch; 和 pub mod sub_agent; + pub use

pub mod checkpointer;
pub mod error;
pub mod session_manager;
pub mod snapshot;
pub mod switch;      // <-- 新增
pub mod sub_agent;   // <-- 新增

验证：

cargo test --all-targets — 374+ 测试全部通过
cargo clippy --all-targets -- -D warnings — 0 警告
cargo doc --no-deps — 0 warning

Step 9 — 示例

文件 1：examples/agent_switch_demo.rs（约 80 行）

创建 session → submit_turn（角色 A）→ switch_agent（角色 B）→ submit_turn（角色 B）→ 验证上下文保留
演示目的：证明 switch_agent 保留 slot 历史 / turn_index / session_memory

文件 2：examples/sub_agent_dispatch_demo.rs（约 150 行）

父 session → dispatch_all 3 个子 agent（研究、写作、审校）→ 收集结果 → 父汇总
树形验证：children(parent_id) 返回 3 个子 ID
演示目的：多 agent 协作完整链路

文件 3：examples/bridge_keys_demo.rs（约 140 行）

父设置 SessionMemory（key: "project_goal", "constraints"）→ dispatch + bridge_keys → 子 agent 通过 get_session_data 读取
子↔子交互：父通过 DispatchConfig.shared_namespace 设定共享命名空间，子 A 写入 shared:{parent_id}:fact_x，子 B 通过约定 key 读取
演示目的：bridge_keys 过滤机制 + 父子数据桥接 + 子↔子共享 namespace

文件 4 — 新增：examples/dispatch_stream_demo.rs（约 100 行）

父 session → dispatch_stream 单个子 agent → 消费 SubTaskStreamEvent 序列
验证收到 ChildCreated + 至少一个 Stream + Completed 事件
输出 SubTaskResult.child_id / usage / summary，验证消息重建和 finalize 正确性
演示 receiver dropped 场景：中途 drop receiver 后 task 正确退出不 panic
演示目的：dispatch_stream 的事件序列 + finalize 完整性验证

验证：4 个示例全部 cargo run --example exit 0

高层实施建议

Step 1 优先于所有步骤：Error 扩展是所有后续步骤的基础，无依赖可并行 [高]
Step 3 独立性强：switch_agent 不依赖 dispatch 的任何类型，可单独实施和测试 [高]
Step 4 是 Step 5-7 的前置：类型定义不依赖其他逻辑，建议在 Step 3 完成后立即实施 [高]
Step 5-7 按复杂度递增：dispatch → dispatch_all → dispatch_stream。dispatch_all 复用 dispatch 的核心逻辑；dispatch_stream 是最复杂的，建议最后实施 [高]
Step 8 集成验证不可跳过：clippy + doc 全量验证确保无回归 [高]
Step 9 在所有核心完成后实施：示例是验收标准的一部分，PM 确认 3 个递进示例 [中]
全量测试密码：实施过程中持续 cargo test --all-targets，不在最后统一修复 [高]

风险矩阵

#	风险	等级	可能性	对策
R1	dispatch_stream 的消息重建：从 StreamEvent 序列重建 `new_messages` 的完整性	🟡 中	低	已移除 `finalize_active_stream()` 方案。spawn task 通过追踪 `ToolExecutionCompleted` / `MessageComplete` 事件重建消息列表。完整性由 `MessageComplete` 事件保证
R2	`tokio::spawn` `'static` + SessionManager 引用	🟡 中	低	dispatch_all 和 dispatch_stream 签名已明确用 `&Arc<Self>`；调用方包装 `Arc<SessionManager>`
R3	子 session 创建成功但后续 submit_turn 失败	🟡 中	中	dispatch 内 `destroy(child_id)` 放在 `?` 前确保清理；通过 `let child_id = ...;` 先绑定，再 `let r = submit_turn(...).await`，失败时 `destroy(&child_id).await?` 清理
R4	部分失败时孤儿 checkpoint 数据	🟢 低	必然	正向利用：保留用于调试审计，存储开销可忽略
R5	并发 dispatch_all 中任务 panic	🟡 中	低	`tokio::spawn` 的 `JoinHandle` 通过 `.await` 捕获 panic；panic 传播到 `dispatch_all` 内作为 `Err` 返回
R6	bridge_keys 中不存在的 key	🟢 低	中	静默跳过（与 `List_entries` 返回全量后再过滤，不存在的 key 自然不会出现在结果中）

架构图（文本示意）

                     SessionManager
                    /      |       \
                   /       |        \
           switch_agent  dispatch  dispatch_stream
                |            |            |
                v            v            v
         AgentSession    create_child   create_child
         .agent = new    inherit_mem    inherit_mem
         SessionMeta     submit_turn    submit_turn_stream
         更新            返回结果        mpsc 转发事件
                                      finalize_on_complete

        交互层次：
        父 -> 子:  SessionMemory snapshot + bridge_keys 过滤
        子 -> 父:  SubTaskResult { child_id, response, usage, summary }
        子 <-> 子: shared:{parent_session_id} namespace

参考来源

代码路径

文件	用途
`src/agent/session.rs`	AgentSession 定义、`agent` pub 字段（L53）、`session_memory` pub 字段（L58）、`submit_turn_stream`（L384）、`finalize_turn`（L456）
`src/agent/agent.rs`	Agent trait 定义（3 个方法）
`src/agent/session_memory.rs`	SessionMemory: `set`（L50）、`set_with_meta`（L61）、`list_entries`（L81）
`src/engine/session_manager.rs`	SessionManager: `create_child`（L202）、`get`（L244）、`destroy`（L495）、`save_session_meta`（L131）、`load_session_meta`（L144）、SessionMeta（L32）、锁契约（L4-12）
`src/engine/error.rs`	EngineError 枚举（当前 6 变体，`#[non_exhaustive]`）
`src/engine/mod.rs`	模块注册
`src/engine/snapshot.rs`	SessionSnapshot（from_snapshot / to_snapshot 所需）
`src/llm/stream.rs`	StreamEvent 枚举
`Cargo.toml`	依赖声明（`futures-util` L17、`tokio-stream` L15、`futures-core` L18）

文档路径

文档	用途
`docs/roadmap-v0.3.0.md` §Phase 18	Phase 18 原始需求（交付物、交互层级、优先级）
`docs/note-opencode-agent-switching.md`	Agent 热切换调研笔记（桥接方案分析、生命周期讨论）
`docs/note-opencode-subagent-dispatch.md`	SubAgent Dispatch 调研笔记（PM/SA 建议、设计推演）
`docs/23-phase17-agent-execution-engine.md`	Phase 17 方案文档（SessionManager 设计背景）
`docs/7-agent-runtime.md`	Agent 运行时设计文档（Session 与 Agent 的关系）
`docs/17-phase10-contextslot.md`	ContextSlot 上下文管理（Phase 10）
`docs/24-phase18-agent-switch-and-dispatch.md`	本文档 — 第 1 轮审查修复记录

决策轨迹

决策	参考来源	置信度
switch_agent 在 SessionManager 而非 AgentSession	`src/agent/session.rs` AgentSession 无 store 引用	高
bridge_keys 通过 SessionMemory 副本，不碰 system_prompt	`docs/note-opencode-subagent-dispatch.md` PM/SA 建议	高
dispatch_all 返回 `Vec<Result<..>>` 部分成功	`docs/note-opencode-subagent-dispatch.md` PM/SA 一致	高
dispatch_stream 返回 `Pin<Box<dyn Stream>>`	`src/engine/session_manager.rs` `submit_turn_stream` 签名一致	高
失败时 destroy 子 session 清理全部（含 checkpoint）	`session_manager.rs:508` destroy 调 `checkpointer.delete_all`	高
dispatch_all 用 `&Arc<Self>` 签名	`tokio::spawn` `'static` 约束	高
child_memory 不做额外权限控制	`docs/note-opencode-subagent-dispatch.md` PM 明确	高
`#[non_exhaustive]` 已存在 -> 新增 EngineError 变体不是 breaking change	`src/engine/error.rs:15`	高
`futures-util` 已存在，无需新增依赖	`Cargo.toml:17`	高

审查修复轨迹（第 1 轮）

#	问题	🔴/🟡	修复内容
F1	`finalize_active_stream()` 假设不成立：`submit_turn_stream` 返回后 cycle 被 drop	🔴	移除 Step 2 的 `finalize_active_stream()`，改为 spawn task 内从 StreamEvent 重建消息列表直接调 `finalize_turn()`
F2	`SubTaskResult` 缺 child_memory 访问路径	🔴	`SubTaskResult.child_id` 可经由 `sm.get()` 读取子 session。构型 doc comment 增加说明
F3	dispatch 失败路径 `destroy` 自身 I/O 可能失败，覆盖原始错误	🔴	改用 `let _ = destroy` + `tracing::error!`，原始 `EngineError` 优先
F4	switch_agent SessionMeta 构造中 `created_at`/`parent_id` 用占位符	🔴	改用 `load_session_meta` 读取原始值，`turn_count` 从 `guard.turn_index()` 读取
F5	D9 与 `destroy()` 实现矛盾（方案说保留，代码说删除）	🟡	D9 修正为"destroy 清理全部"，否决条目同步更新
F6	`bridge_keys` 默认值安全反直觉（空=全量继承）	🟡	类型改为 `Option<Vec<String>>`，`None` = 不继承（默认），`Some(vec![])` = 全量
F7	Semaphore acquire 位置未指定	🟡	指定 `acquire_owned()` 在 spawn 内 + indexed 收集维持输入顺序
F8	dispatch_stream 缺少独立示例	🟡	Step 9 追加 `dispatch_stream_demo`
F9	mpsc channel 背压策略未指定	🟡	改用 `unbounded_channel`，与 LLM stream 内部模式一致
F10	子↔子交互层缺少实现细节	🟡	`DispatchConfig.shared_namespace` 字段 + 示例 3 演示
F11	inherit_session_memory 竞态窗口未文档化	🟡	doc comment 声明快照一致性模型
F12	switch_agent system_prompt 断裂风险未说明	🟡	doc comment 增加使用建议 + 安全提示

本文档对应的实施步骤记录在 docs/roadmap-v0.3.0.md §Phase 18，实施完成后同步更新 roadmap 状态。

37 KiB Raw Permalink Blame History Unescape Escape

Phase 18：Agent 角色热切换与子代理调度

背景与目标

问题空间

目标

依赖与优先级

当前状态分析

现有架构中的关键接入点

Agent trait 与 SessionManager 之间的关系

锁契约（需格外注意）

现有测试覆盖

调研发现

1. bridge_keys 注入位置

2. SessionMemory 继承策略

3. dispatch_all 部分成功语义

4. dispatch_stream 生命周期

5. Cargo.toml 依赖分析

可选方案

方案 A：switch_agent 作为 AgentSession 方法 vs SessionManager 方法

方案 B：bridge_keys 注入方式

方案 C：dispatch_stream 返回类型

否决方案

推荐方案

整体架构

决策清单

设计理由详述

实施建议

阶段划分

Step 1 — error.rs 扩展

Step 2 — session.rs 无变更

Step 3 — switch.rs

Step 4 — sub_agent.rs 类型

Step 5 — sub_agent.rs dispatch 核心

Step 6 — sub_agent.rs dispatch_all

Step 7 — sub_agent.rs dispatch_stream

Step 8 — mod.rs + 集成验证

Step 9 — 示例

高层实施建议

风险矩阵

架构图（文本示意）

参考来源

代码路径

文档路径

决策轨迹

审查修复轨迹（第 1 轮）

37 KiB

Raw Permalink Blame History

Step 1 — `error.rs` 扩展

Step 2 — `session.rs` 无变更

Step 3 — `switch.rs`

Step 4 — `sub_agent.rs` 类型

Step 5 — `sub_agent.rs dispatch` 核心

Step 6 — `sub_agent.rs dispatch_all`

Step 7 — `sub_agent.rs dispatch_stream`

Step 8 — `mod.rs` + 集成验证