雨 雨天小六

读懂 Codex（2.6）：Session 初始化时并行建立的服务和失败收尾

· 更新于 2026-08-02 · 专栏：读懂 Codex

#Codex#Agent Runtime#软件架构#Session#Initialization

创建 Session 不是给结构体字段逐个赋值。Codex 在这一阶段要同时确定模型合同、Thread 身份、持久化、 环境、指令、技能、MCP、网络代理、Hook 与 Extension。部分步骤可以并行，部分必须严格排序；更麻烦的 是，持久化可能已经建立，而后面的网络或配置锁校验仍会失败。

源码用“并行准备 + 首事件发布屏障 + LiveThread 初始化守卫”把这段过程组织成可回滚的启动事务。

spawn_internal 先冻结 SessionConfiguration

在进入 Session::new 前，spawn_internal 完成不会因单个 Turn 改变的基础解析：

  • 建立容量 512 的 Submission Channel 和无界 Event Channel；
  • 校验父 W3C trace，非法 carrier 被忽略并记录 warning；
  • 加载 ExecPolicy，子 Thread 可继承父策略；
  • 根据历史、配置和模型目录选择 model/model info；
  • 解析 multi-agent version 与 history mode；
  • 生成 collaboration mode、service tier、审批与权限状态；
  • 创建 AgentStatus::PendingInit 的 watch channel。

模型信息刷新策略有意区分根 Thread 与非根 Agent。根 Thread 可以 OnlineIfUncached，非根 Agent 使用 Offline，避免每次派生都触发独立网络刷新并改变父子模型合同。

指令与动态工具的恢复优先级

Base instructions 的顺序是：

显式 config.base_instructions
    > 历史 SessionMeta 中的 base instructions
    > 当前 ModelInfo 按 personality 生成的默认指令

Dynamic tools 则采用“显式非空优先，否则从历史恢复”。这里没有把两个来源盲目合并，因为恢复旧 Thread 时必须维持当时的工具合同，而新启动方显式提供工具时又需要完整替换。

def resolve_session_contract(config, history, model_info, supplied_tools):
    base = (
        config.base_instructions
        or history.base_instructions
        or model_info.instructions_for(config.personality)
    )
    tools = supplied_tools if supplied_tools else history.dynamic_tools_or_empty()
    return SessionConfiguration(base_instructions=base, dynamic_tools=tools)

身份必须先于服务确定

Fresh/Fork 创建新 Thread ID，Resume 使用历史 conversation ID。Session ID 对根 Thread 通常由 Thread ID 派生；非根 Agent 使用 AgentControl 的 Session ID，从而把多条子 Thread 归入同一 Agent Session。 旧历史中错误地用自身 Thread ID 合成的 subagent Session ID 会被过滤。

Multi-agent version、selected capability roots、originator 与 Extension initial data 也在此处固化。后面的 MCP projection、工具面和持久 SessionMeta 都依赖这些值。

第一组并行任务：持久化、State DB、Auth 与 MCP 投影

三个互不依赖的 Future 用 tokio::join! 并行：

FutureFresh/ForkResumeEphemeral
Thread persistencecreate，必要时写 inherited contextresume 指定 history/path直接 None
State DBLocalThreadStore 时读取 handle同左None
Auth + MCP projection读取当前认证并计算初始配置投影同左仍需要
Session 初始化的两组并行任务以及后续网络代理、服务组装和 SessionConfigured 发布顺序
图 2.6-1：只有无数据依赖的准备工作并行；需要稳定 Session 或事件顺序的步骤仍保持串行。

join! 不是后台放任执行：三个分支都完成后，初始化才继续。如果持久化分支失败，错误映射为 Session 初始化错误；已经得到的其他结果随当前启动流程丢弃。

LiveThreadInitGuard 保护半完成的持久化

create/resume 成功时，ThreadStore 可能已经有可见资源，但 Session 还没有完成 shell、AGENTS.md、 network proxy 等步骤。LiveThreadInitGuard 暂时拥有 LiveThread：

  • 成功：SessionServices 已保存 LiveThread clone，guard commit() 清空自己；
  • 失败：discard().await 删除/放弃这次未完成的 live persistence；
  • 意外提前 Drop：尝试在当前 Tokio runtime 中异步 discard，作为最后兜底。

这不是数据库全局事务，却给 Session 启动定义了明确的资源提交点。

环境建立后才刷新指令和技能

默认 shell、可选 ShellSnapshot 和 ThreadEnvironments 建立后，Runtime 才能解析选中的环境。第二组 并行任务随后执行：

  1. 根据 resolved environments 刷新 AGENTS.md；
  2. 预热 Plugins 与 Skills；
  3. 从 ThreadStore 查询 Thread title。

技能预热错误会被收集和记录，并不自动让整个 Session 启动失败；而 config lock 校验、导出或 managed network proxy 启动失败会终止初始化。区分硬失败和可降级失败能避免一个损坏技能屏蔽整个 Agent， 同时不放松企业约束和网络安全边界。

managed network proxy 必须在 Session 发布前成功

若权限配置要求受管代理，Session::new 会把 ExecPolicy 网络规则叠加到 proxy spec，连接 network approval 服务，启动 HTTP/SOCKS proxy，并生成可公开的 runtime address。只有符合当前 PermissionProfile 时， 地址才进入 SessionConfigured。

代理需要在发布前成功，是因为工具执行一旦开始就必须遵守同一个网络边界。先发布、后补代理会产生 短暂但真实的策略空窗。

Extension lifecycle 先拿到稳定的 Thread 资源

Session 对象尚未存在时，Runtime 先创建空的 MCP Runtime、Session/Thread ExtensionData 和稳定的 MCP Resource Client，再调用各 ThreadLifecycleContributor::on_thread_start。贡献者可以写 Extension Data，但不能假设 MCP 已经发出连接事件。

随后组装 SessionServices，包括：

  • MCP Runtime、Models/Auth、Skills/Plugins/Extensions；
  • Unified Exec、Elicitation、Tool approvals、Hooks；
  • ThreadEnvironments、Shell、Code Mode；
  • Network proxy、State DB、LiveThread、ThreadStore；
  • ModelClient、Telemetry、AgentControl。

SessionConfigured 必须是第一个公开事件

Session 建好后，Runtime 首先发送 ID 为空的 SessionConfiguredEvent。它包含 Session/Thread ID、父子 关系、模型和 Provider、service tier、权限、cwd、初始消息、proxy 地址与 rollout path。

随后才发送 deprecation/startup/unstable warnings，启动环境连接事件转发，并安装初始 MCP Runtime。 源码特意让 MCP Services 先以空连接集进入 SessionServices，就是为了避免 MCP startup event 抢在 SessionConfigured 前面。

async def publish_initialized_session(session, prepared):
    await session.events.send(configured_event(prepared))  # 必须第一条
    for warning in prepared.post_configured_warnings:
        await session.events.send(warning)

    prepared.environments.start_event_forwarding(session.events)
    await session.install_initial_mcp_runtime(prepared.mcp_projection)
    session.start_mcp_prewarm_worker()
    await session.record_initial_history_after_configured()

SessionConfigured 是“身份与基础服务可解释”的屏障，不是“所有 MCP 都连接完成”的屏障。MCP 有自己的 StartupUpdate/Complete 事件，环境也有 Connected/Disconnected 事件。

历史为什么在配置事件之后记录

Resume/Fork 的 record_initial_history 可能发出 RawResponseItem、TokenCount 或其他回放事件。如果先回放， 客户端甚至不知道这些 Item 属于哪个已配置 Thread。源码因此在 SessionConfigured 发出后才记录历史。

Referenced fork 还要确保 rollout materialized，直到子 Thread 对父历史的引用变得持久，才能释放 source reservation。这是跨 Thread 历史所有权的 durability barrier。

成功提交与失败回滚

LiveThreadInitGuard 在 Session 初始化成功时 commit、任一后续步骤失败时 discard 的事务状态图
图 2.6-2：持久化创建成功不等于 Thread 创建成功；guard 一直持有回滚责任直到整个 Session 可用。
async def new_session(args) -> Session:
    live_result, state_db, auth_mcp = await join(
        open_persistence(args),
        lookup_state_db(args),
        resolve_auth_and_mcp(args),
    )
    guard = LiveThreadInitGuard(live_result)
    try:
        session = await build_remaining_services(guard.view(), state_db, auth_mcp)
        await emit_configured_first(session)
        await finish_post_publish_initialization(session)
    except Exception:
        await guard.discard()
        raise
    else:
        guard.commit()
        return session

注意：只有 Session::new 成功返回后，spawn_internal 才 tokio::spawn(submission_loop)。因此启动中没有 普通 Submission 与初始化步骤竞争同一 Session 状态。

失败矩阵

失败点行为是否启动 submission loop
Parent trace 非法忽略 trace、warning，继续成功后才启动
Model/ExecPolicy 解析失败spawn 失败否
Thread persistence create/resume 失败初始化失败否
Skill/Plugin warmup 单项失败记录错误，可继续是
Config lock 校验/导出失败guard discard否
Managed proxy 启动失败guard discard否
MCP initial install 失败已发 Configured，但 guard discard，spawn 失败否
Session 创建成功guard commit，启动 loop是

MCP install 失败这一行说明了一个细节：Event Channel 内可能短暂存在 Configured，但 ThreadManager 只有在 Session::spawn 整体成功返回后才取得 IO 并发布 Thread，因此失败 Session 的事件不会成为对外已注册对象。

测试应观察顺序与资源归属

async def test_configured_is_first_event(spawned):
    first = await spawned.io.next_event()
    assert first.id == ""
    assert first.type == "session_configured"


async def test_failed_init_discards_live_thread(store):
    store.inject_failure("network_proxy_start")
    with raises(SessionInitError):
        await spawn_session(store=store)
    assert not await store.has_active_writer_for_attempt()


async def test_submission_loop_starts_after_init(spawner):
    session, io = await spawner.spawn()
    assert await io.submit(Op.ThreadSettings(...))

Session 初始化的设计重点不是把所有工作都并行，而是明确哪些值必须先冻结、哪些 IO 可以并发、哪个事件 构成发布屏障，以及失败时谁负责清理已经打开的资源。

评论

← 返回文章列表