/// Immutable typed registry produced after extensions are installed. // 注释点明：build() 之后注册表不可变，只能读pub struct ExtensionRegistry<C: Sync> { // C 是宿主配置类型（core 里就是 Config），Sync 保证跨线程共享安全    event_sink: Arc<dyn ExtensionEventSink>, // 事件出口：扩展想通知 UI/客户端时走这里，而不是自己开连接    turn_start_admission: Option<Arc<dyn TurnStartAdmission>>, // 宿主级"准入门禁"：关机排空期间拒绝新轮次启动（可缺省）    thread_lifecycle_contributors: Vec<Arc<dyn ThreadLifecycleContributor<C>>>, // 线程生命周期插槽：start/ready/resume/idle/stop 五个钩子    turn_lifecycle_contributors: Vec<Arc<dyn TurnLifecycleContributor>>, // 轮次生命周期插槽：turn start/item completed/stop/abort/error    config_contributors: Vec<Arc<dyn ConfigContributor<C>>>, // 配置变更插槽：线程配置提交后回调，扩展借此刷新自己的快照    token_usage_contributors: Vec<Arc<dyn TokenUsageContributor>>, // token 用量插槽：模型每回一次包就通知（Goals 用它记账）    skill_invocation_contributors: Vec<Arc<dyn SkillInvocationContributor>>, // 技能调用插槽：显式加载/隐式命中技能时回调    context_contributors: Vec<Arc<dyn ContextContributor>>, // prompt 片段插槽：thread/turn/world-state 三个作用域各贡献一段    mcp_server_contributors: Vec<Arc<dyn McpServerContributor<C>>>, // MCP server 解析插槽：扩展可以动态声明要连哪些 MCP server    turn_input_contributors: Vec<Arc<dyn TurnInputContributor>>, // 轮次输入插槽：给本轮模型输入追加上下文片段（如技能说明）    tool_contributors: Vec<Arc<dyn ToolContributor>>, // 工具插槽：本讲主角，扩展在这里贡献原生工具    tool_lifecycle_contributors: Vec<Arc<dyn ToolLifecycleContributor>>, // 工具生命周期插槽：tool start/finish、MCP 结果后处理（只观察不改写）    turn_item_contributors: Vec<Arc<dyn TurnItemContributor>>, // 有序 item 后处理插槽：item 发出前可改写，按注册顺序串行执行    approval_review_contributors: Vec<Arc<dyn ApprovalReviewContributor>>, // 审批决策插槽：链式认领，第一个返回 Some 的说了算（guardian-v2 住这里）}

    /// Returns the attached value of type `T`, if one exists. // 文档注释：按类型取，没有就返回 None——不猜、不造    pub fn get<T>(&self) -> Option<Arc<T>> // T: Any + Send + Sync（见下方 where），返回值用 Arc 共享而非拷贝    where        T: Any + Send + Sync, // 三个约束缺一不可：Any 才能 downcast，Send+Sync 才能跨线程放进共享存储    {        let value = self.entries().get(&TypeId::of::<T>())?.clone(); // TypeId 当 key 查表；? 让"没存过"直接短路成 None        Some(downcast_data(value)) // downcast 失败会 unreachable!——同类型存取不可能错，错了就是 bug 该炸    }    /// Returns the attached value of type `T`, inserting one from `init` when absent. // get_or_init：懒初始化    pub fn get_or_init<T>(&self, init: impl FnOnce() -> T) -> Arc<T> // 返回 Arc，多次调用拿到同一个实例（单例语义）    where        T: Any + Send + Sync, // 同上：类型必须可擦除、可跨线程    {        let mut entries = self.entries(); // 先拿锁再判断——保证"检查+插入"原子，并发下不会建两份        let value = entries            .entry(TypeId::of::<T>())            .or_insert_with(|| Arc::new(init())); // entry API：存在就复用，不存在才跑 init 构造并装箱成 Arc        downcast_data(Arc::clone(value)) // clone 的是 Arc 引用计数，不是数据本身——零拷贝共享    }

/// Boxed, sendable future returned by asynchronous extension contributors. // 文档注释：所有异步插槽统一返回这个类型pub type ExtensionFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>; // RPITIT 的替代方案：trait 方法返回具体 future 类型，实现方用 async fn 满足

/// Extension contribution that adds prompt fragments during prompt assembly. // 文档注释：prompt 组装期贡献片段pub trait ContextContributor: Send + Sync { // 三个方法全有默认空实现——扩展只挑自己关心的作用域实现    /// Returns thread-scoped context using the supplied extension state. // 线程级上下文：输入稳定，适合放长期有效的说明    fn contribute_thread_context<'a>(        &'a self,        session_store: &'a ExtensionData, // 把三级存储直接递进来——扩展想读什么级别的状态自己取        thread_store: &'a ExtensionData,    ) -> ExtensionFuture<'a, Vec<PromptFragment>> { // 返回片段列表，core 按注册顺序拼进 system prompt        Box::pin(async move {            let _self = self;            let _session_store = session_store;            let _thread_store = thread_store;            Vec::new() // 默认贡献"空"：不实现就等于不参与——opt-in 语义        })    }    fn contribute_turn_context<'a>(        &'a self,        input: TurnContextContributionInput<'a>, // turn 级输入：依赖本轮宿主状态（如当前审批模式）的片段走这里    ) -> ExtensionFuture<'a, Vec<PromptFragment>> {        Box::pin(async move {            let _self = self;            let _input = input;            Vec::new() // 同上：默认空实现，保持 opt-in        })    }    fn contribute_world_state<'a>(        &'a self,        input: WorldStateContributionInput<'a>, // world-state 作用域：贡献"世界状态"章节（前几讲拆过 world_state.rs）    ) -> ExtensionFuture<'a, Vec<WorldStateSectionContribution>> {        Box::pin(async move {            let _self = self;            let _input = input;            Vec::new() // 默认空实现：不关心 world state 的扩展零成本跳过        })    }}

/// Extension contribution that exposes native tools owned by a feature. // 文档注释：扩展贡献"原生工具"（非 MCP）pub trait ToolContributor: Send + Sync { // 本讲主角插槽：web-search/生图/goal/skills 都靠它把工具塞进模型视野    /// Returns native tools bound to the supplied extension state. // 线程级工具列表：大多数扩展只需要这一档    fn tools(        &self,        session_store: &ExtensionData, // 三级存储递进来，扩展按自己的配置快照决定给不给工具        thread_store: &ExtensionData,    ) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>; // HRTB 的 for<'call>：返回的工具必须能服务任意生命周期的调用    /// Returns native tools bound to one sampling step. // 采样步级工具：更细粒度的作用域    fn tools_for_step(        &self,        session_store: &ExtensionData,        thread_store: &ExtensionData,        _step_store: &ExtensionData, // step 级存储：单步临时状态（如 code-mode 的 cell 上下文）    ) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>> {        self.tools(session_store, thread_store) // 默认退化成线程级——不需要步级差异的扩展不用实现    }}

/// Owns the complete approval decision, including whether to consult a reviewer. // 文档注释：审批决策的"完整所有权"pub trait ApprovalReviewContributor: Send + Sync { // guardian-v2 就实现这个——它有权决定"要不要问人/问哪个 agent"    /// Claims one request, including a handoff to the user. // "认领"语义：返回 Some 即终止整条链    fn decide<'a>(        &'a self,        _input: &'a ApprovalDecisionInput<'_>, // 输入含工具名、参数、风险分等——扩展据此判断该不该管    ) -> ExtensionFuture<'a, Option<ApprovalDecision>> { // None = "我不认领，交给下一个"；Some = 我说了算（含转交用户）        Box::pin(std::future::ready(None)) // 默认不认领：没实现审批逻辑的扩展自动跳过    }}

impl ThreadLifecycleContributor<Config> for WebSearchExtension { // 播种钩子：线程启动时把"当前配置快照"存进 thread_store    fn on_thread_start<'a>(        &'a self,        input: ThreadStartInput<'a, Config>, // 输入里带 config 引用 + thread_store——扩展自己决定存什么    ) -> ExtensionFuture<'a, ()> {        Box::pin(async move {            input                .thread_store                .insert(WebSearchExtensionConfig::from(input.config)); // 把 available/provider/settings 三件套固化成快照，后续 tools() 只读快照不碰活配置        })    }}impl ConfigContributor<Config> for WebSearchExtension { // 刷新钩子：线程配置提交变更后重新播种    fn on_config_changed(        &self,        _session_store: &ExtensionData,        thread_store: &'a ExtensionData, // 只关心 thread 级存储——web-search 的状态本来就是线程专属的        _previous_config: &Config, // previous 故意不用：快照整体替换，不做增量 diff        new_config: &Config,    ) {        thread_store.insert(WebSearchExtensionConfig::from(new_config)); // insert 覆盖旧快照——原子换掉，不存在半新半旧    }}impl ToolContributor for WebSearchExtension { // 工具钩子：每轮组装工具表时被 core 调用    fn tools(        &self,        session_store: &ExtensionData,        thread_store: &ExtensionData,    ) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> { // 返回空 Vec = 本轮不暴露 web.run 工具，模型根本看不见它        let Some(config) = thread_store.get::<WebSearchExtensionConfig>() else {            return Vec::new(); // 没播种过（理论上不该发生）——fail loud：不给工具而不是猜一个默认值        };        if !config.available {            return Vec::new(); // provider 不支持或用户显式 Disabled——静默缺席，不报错不打扰        }        vec![Arc::new(WebSearchTool { // 可用才构造真实工具实例            session_id: session_store.level_id().to_string(), // level_id 就是宿主对象 id（session/thread 的 uuid）——借存储拿身份            provider: create_model_provider(                config.provider.clone(),                Some(self.auth_manager.clone()), // auth_manager 是扩展构造时注入的重依赖，不塞进快照            ),            settings: config.settings.clone(), // 搜索模式/上下文大小/域名过滤等，全部来自播种时的快照            originator: thread_store                .get::<ThreadOriginator>()                .map(|originator| originator.0.clone()), // 调用方身份（谁发起的线程）——用于服务端审计与配额        })]    }}pub fn install(registry: &mut ExtensionRegistryBuilder<Config>, auth_manager: Arc<AuthManager>) { // 统一入口：宿主只调这一个函数    let extension = Arc::new(WebSearchExtension { auth_manager }); // 重依赖（auth）挂在扩展值上，三个插槽共享同一份    registry.thread_lifecycle_contributor(extension.clone()); // 同一个 Arc 注册三次——trait object 各自独立，状态却共享    registry.config_contributor(extension.clone());    registry.tool_contributor(extension);}

pub fn install_with_backend<C>( // 泛型 C：goal 不绑定具体配置类型，core/app-server 各自传自己的    registry: &mut ExtensionRegistryBuilder<C>,    state_dbs: Arc<codex_state::StateRuntime>, // 持久化后端：目标状态落 SQLite（第29讲拆过的 State Runtime）    analytics_events_client: AnalyticsEventsClient, // 分析事件出口：goal 完成/失败要上报    metrics_client: Option<MetricsClient>, // OTEL 指标（可缺省——测试环境没有 collector 也能跑）    thread_manager: Weak<ThreadManager>, // Weak 引用！避免 goal→thread_manager→extensions→goal 的引用环    goal_service: Arc<GoalService>, // 进程级服务：管理所有线程的 runtime 注册表    goal_config: impl Fn(&C) -> GoalExtensionConfig + Send + Sync + 'static, // 配置解析闭包：宿主决定"goal 开不开、预算多少") where    C: Send + Sync + 'static, // 同前：跨线程共享的硬约束{    let extension = Arc::new(GoalExtension::new_with_host_capabilities( // 构造时把 registry.event_sink() 也取走——goal 要主动推事件给 UI        state_dbs,        analytics_events_client,        registry.event_sink(),        metrics_client,        thread_manager,        Arc::clone(&goal_service),        goal_config,    ));    registry.thread_lifecycle_contributor(extension.clone()); // 插槽1：start 时建 GoalRuntimeHandle、resume 时恢复状态    registry.config_contributor(extension.clone()); // 插槽2：配置变更时 set_enabled()——开关可热切换    registry.turn_lifecycle_contributor(extension.clone()); // 插槽3：turn start/stop/abort/error 都影响目标进度    registry.token_usage_contributor(extension.clone()); // 插槽4：每次模型回包记账，逼近预算就触发 steering    registry.tool_lifecycle_contributor(extension.clone()); // 插槽5：on_tool_finish 统计"有效尝试次数"推进度    registry.tool_contributor(extension); // 插槽6：贡献 goal 相关工具（如更新目标状态）}

impl<C> ThreadLifecycleContributor<C> for QueuedItemService // queue 服务本身就是一个贡献者——服务即扩展where    C: Send + Sync + 'static, // 同前：宿主配置类型泛型化{    fn on_thread_resume<'a>(&'a self, input: ThreadResumeInput<'a>) -> ExtensionFuture<'a, ()> { // resume 钩子：记录"这个线程刚被恢复过"        Box::pin(async move {            if let Ok(thread_id) = ThreadId::from_string(input.thread_store.level_id()) { // level_id → ThreadId，解析失败就静默跳过                self.resumed_threads                    .lock()                    .unwrap_or_else(std::sync::PoisonError::into_inner) // 锁中毒恢复而非 panic——后台服务不能因一次异常永久失效                    .insert(thread_id); // 标记后，watcher 轮询时能区分"新线程"和"刚 resume 的旧线程"            }        })    }    fn on_thread_idle<'a>(&'a self, input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> { // idle 钩子：派发时机 = 线程真正空闲        Box::pin(async move {            if input.cause == ThreadIdleCause::Interrupted {                return; // 用户打断导致的"空闲"不派发——打断语义是"停下"，不是"接着跑队列"            }            let Ok(thread_id) = ThreadId::from_string(input.thread_store.level_id()) else {                tracing::warn!(                    level_id = input.thread_store.level_id(), // 日志带上原始 id——方便排查"为什么没派发"                    "queue extension received an invalid thread id"                );                return; // fail loud：记警告后放弃，不猜线程身份            };            let _guard = self.dispatch_guard(thread_id).await; // per-thread 派发锁：同一线程的队列串行消费，不同线程互不阻塞            if let Err(error) = self.dispatch_if_idle(thread_id).await { // 真正取队首 → 走 core 提交（会再过 admission 闸门）                tracing::warn!(%thread_id, %error, "failed to dispatch queued user input"); // 失败只记日志——队列项还在库里，下次 idle 重试            }        })    }}

impl ExtensionEventSink for AppServerExtensionEventSink { // 宿主实现出口 trait：扩展调 emit()，宿主决定怎么送达    fn emit(&self, event: Event) { // 同步入口——扩展不感知网络细节，也不阻塞在 IO 上        match event.msg {            EventMsg::ThreadQueueChanged(queue_event) => { // queue 扩展派发后推的事件：UI 要刷新"排队中"列表                let thread_id = queue_event.thread_id;                if let Some(listener_command_tx) = self                    .thread_state_manager                    .current_listener_command_tx(thread_id) // 优先走 listener FIFO——保证与同线程其他事件严格有序                {                    let command = ThreadListenerCommand::EmitThreadQueueChanged;                    if listener_command_tx.send(command).is_ok() {                        return; // 入队成功即返回：emit() 本身零 IO、零 await，扩展侧无感                    }                    tracing::warn!(                        "failed to enqueue extension queue update for {thread_id}: listener command channel is closed" // listener 已拆——降级到异步路径                    );                }                let outgoing = Arc::clone(&self.outgoing); // 没有 listener（如直连客户端）：spawn 任务做订阅者解析 + 发送                let thread_state_manager = self.thread_state_manager.clone();                tokio::spawn(async move {                    let subscribed_connection_ids = thread_state_manager                        .subscribed_connection_ids(thread_id)                        .await; // 异步查"谁订阅了这个线程"——可能涉及锁/IO，绝不能放在 emit() 同步路径里                    let outgoing = ThreadScopedOutgoingMessageSender::new(                        Arc::clone(outgoing),                        subscribed_connection_ids,                        thread_id,                    );                    outgoing                        .send_server_notification(ServerNotification::ThreadQueueChanged( // 翻译成 app-server 协议通知发给客户端                            ThreadQueueChangedNotification {                                thread_id: thread_id.to_string(),                            },                        ))                        .await;                });            }            msg => { // 其他事件类型（如 goal updated）走同样的 listener→spawn 双路径，此处省略                tracing::debug!(event_id = %event.id, ?msg, "dropping unsupported extension event"); // 不认识的事件记 debug 丢弃——前向兼容：新扩展发旧宿主不懂的事件不炸            }        }    }

/// A host-provided gate checked before Core starts a turn-input submission. // 文档注释：宿主提供的闸门，core 启动轮次前必查/// Implementations return a permit for work admitted before shutdown and // 准入的工作拿到"许可证"（permit）/// Core retains it through submission. `None` skips the start without consuming // core 持有它直到提交完成；None = 放弃启动且不消耗排队输入——下次还能用/// pending input. Steering an existing turn does not acquire a new permit. // steer（转向已有轮次）不占新许可——关机时允许把进行中的事做完/// Memory-only mailbox wakeups and parent-delegated subagent input bypass this // 内存信箱唤醒、父级委派的子 agent 输入绕过闸门——委托工作必须能跑完再退出/// gate so delegated work can finish before exit. Automatic starts remain gated. // 但"自动启动新轮次"（如 queue 派发）仍然受管pub trait TurnStartAdmission: std::fmt::Debug + Send + Sync { // Debug bound：宿主可以在日志里打印闸门状态排查问题    fn admit_turn_start(&self) -> Option<Box<dyn Send>>; // 返回不透明 permit——core 不需要知道许可证内部是什么，持有即有效}

#[instrument(level = "trace", skip_all)] // trace 级埋点：工具表组装是高频路径，默认不输出pub(crate) fn extension_tool_executors<'a>( // core 侧唯一认识"扩展工具"的函数——依赖方向在这里收口    session: &'a Session,    step_store: &'a ExtensionData, // step 级存储：步级差异化工具（如 code-mode）需要它) -> impl Iterator<Item = Arc<dyn for<'call> ToolExecutor<ExtensionToolCall<'call>>> + 'a { // 惰性迭代器：按需构造，不预分配大 Vec    session        .services        .extensions // SessionServices 里持有的就是 build() 后的不可变注册表（Arc）        .tool_contributors()        .iter()        .flat_map(move |contributor| { // flat_map 把每个贡献者的工具列表摊平成一个流——顺序 = 注册顺序            contributor.tools_for_step(                &session.services.session_extension_data, // 三级存储原样递入：扩展自己决定用哪一级                &session.services.thread_extension_data,                step_store,            )        })}

    /// Returns the first claimed decision in registration order. // 文档注释：注册顺序 = 优先级，第一个认领者生效    pub async fn decide_approval(        &self,        input: &crate::ApprovalDecisionInput<'_>,    ) -> Option<crate::ApprovalDecision> { // None = 没人认领 → core 走默认审批策略（问用户）        for contributor in &self.approval_review_contributors { // 串行遍历：不并发，保证决策可预测、可复现            if let Some(decision) = contributor.decide(input).await { // 每个贡献者拿到完整输入，自己判断管不管                return Some(decision); // 一旦认领立即短路——后面的贡献者根本不会被问            }        }        None // 全链无人认领：返回 None 让 core 兜底，而不是 panic    }

扩展层数据流（装配 → 运行）

▼

▼

▼

▼

▼