读懂 Codex（2.4）：ThreadManager 的创建、恢复、派生与注册

当 App Server 收到 thread/start 时，它不能只执行 threads[id] = Session()。Thread 可能来自全新 配置、持久 Rollout、一个正在运行的同 ID Session，或另一个 Thread 的截断快照。创建期间还要启动 MCP、Extensions、环境和存储服务。任何一步失败，都不能把半初始化对象暴露给客户端。

ThreadManager 解决的核心问题是：把“构造 Session”与“发布一个可查询 Thread”分成两个阶段，并 让新建、恢复和派生最终汇入同一注册屏障。

ThreadManager 是共享服务与活对象注册表

ThreadManager 内部的 State 由 Arc 共享，主要持有两类东西：

类别	代表成员	生命周期
进程级共享服务	Auth、Models、Environment、Skills、Plugins、MCP、Extensions、ThreadStore	通常跨多个 Thread
活 Thread 控制面	live map、thread-created broadcast、启动中 MCP 标记	随当前进程中的 Thread 变化

Live map 保存 ThreadId → Arc<CodexThread>，它不等于持久数据库。一个 Thread 可以存在于 ThreadStore 但尚未加载进 map；也可以从 map 移除后仍被其他 Arc 持有。

这个区分解释了为什么“列出历史 Thread”和“取得当前运行对象”需要不同路径。

StartThreadOptions 是创建合同

新 Thread 不只需要 Config。创建合同还携带：

• InitialHistory：New、Resumed、Forked 或其他历史形态；
• History mode 与 fork persistence；
• SessionSource、ThreadSource 和 parent trace；
• 动态工具与初始 Extension Data；
• 工作环境选择；
• 是否支持 OpenAI form elicitation；
• 是否允许 Provider model fallback。

将这些信息打包的意义是：Session 启动后不必再从 UI 全局变量猜来源。来源还会影响内部 Thread 可见性、AgentControl、Rollout metadata 和生命周期 Hook。

Fresh Start 先补齐派生值

Fresh start 会在进入通用 spawn 前补默认环境、按 Config 选择 AgentControl，并从历史或 Manager 默认 值解析 Session/Thread source。

async def start_thread(manager: ThreadManager, options: StartThreadOptions) -> NewThread:
    environments = options.environments or default_environments(
        manager.environment_manager,
        cwd=options.config.cwd,
        workspace_roots=options.config.workspace_roots,
    )
    agent_control = manager.agent_control_for(options.config)
    session_source, thread_source = resolve_sources(
        options.initial_history,
        explicit_session_source=options.session_source,
        explicit_thread_source=options.thread_source,
        manager_default=manager.session_source,
    )
    return await manager.state.spawn_with_source(
        options=options,
        environments=environments,
        agent_control=agent_control,
        session_source=session_source,
        thread_source=thread_source,
    )

默认值计算发生在发布前，因此失败不会留下一个只有 ID、没有执行环境的可见 Thread。

Resume 的第一步是判断活对象还是冷历史

从 rollout path 恢复时，ThreadStore 先返回 StoredThread 和历史，再转成 Resumed InitialHistory。进入 spawn 后，系统检查 live map 中是否已经有同 conversation ID 的对象。

有三种结果：

1. 对象仍在运行，且请求的 rollout path 相容：直接返回现有 Arc；
2. 对象仍在运行，但 path 不同：拒绝请求，不能让同 ID 指向两份历史；
3. 对象已经停止：先移出 stale entry，再从历史重建。

async def resolve_resume_target(
    live: LiveThreadMap,
    history: ResumedHistory,
) -> ExistingThread | SpawnRequired:
    current = await live.get(history.thread_id)
    if current is None:
        return SpawnRequired(history)
    if current.is_running():
        if history.requested_rollout_path not in {None, current.rollout_path}:
            raise InvalidRequest("same thread id is running with another rollout")
        return ExistingThread(current)
    await live.remove(history.thread_id)
    return SpawnRequired(history)

Resume 是恢复同一个 Thread 身份，不是复制。V2 根 Agent 还需要从 AgentGraphStore 恢复元数据；真正 完成注册后，Runtime 才发送 thread-resume lifecycle。

Fork 先定义快照语义，再创建新身份

Fork 与 Resume 的区别不只是换 ID。Fork 必须决定 source 正处于 mid-turn 时如何得到一致历史。

当前有两个明确快照合同：

快照	行为
TruncateBeforeNthUserMessage	在第 n 个 user boundary 前截断；越界且正在 mid-turn 时丢弃未完成 suffix
Interrupted	保留当前持久前缀；若末尾仍在 Turn 内，追加与真实 interrupt 同形的 aborted marker

Interrupted 不是执行一次真实取消。它是在派生历史中表达“这个快照到这里被中止”，保证模型不会 把孤立 Tool Call 或半个 Turn 当成正常完成。

Subagent 从 live parent 派生时，还要先 materialize 并 flush Rollout，再从 Store 读取。否则内存里 刚写的 Item 可能尚未进入可复制快照。

async def fork_thread(source: CodexThread, snapshot: ForkSnapshot, config: Config) -> NewThread:
    await source.ensure_rollout_materialized()
    await source.flush_rollout()
    stored = await source.read_thread(include_history=True)
    history = convert_to_initial_history(stored)

    if snapshot.kind == "interrupted":
        fork_history = append_abort_marker_if_mid_turn(history, config)
    else:
        fork_history = truncate_at_user_boundary(history, snapshot.user_index)

    return await spawn_new_identity(config=config, history=fork_history)

Copied persistence 与 Referenced persistence 也不能混为一谈。前者复制历史，后者引用已有 history base 并记录 inherited item count；二者后续存储成本和释放时机不同。

通用 spawn 组装 Session 需要的全部服务

在真正创建 Session 前，State 还会解析：

• user instructions 来源；
• parent/forked Thread 关系；
• multi-agent version；
• originator；
• parent rollout trace；
• 继承的 environment/exec policy；
• Extension 初始数据。

然后一次性构造 SessionSpawnArgs，把共享 Managers 与这次 Thread 的值交给 Session::spawn。这里的 失败仍属于“未发布启动失败”。

SessionConfigured 是发布屏障

Session::spawn 返回 Session 和 SessionIo，并不代表客户端现在就能安全提交 Turn。ThreadManager 会先 从 Event Channel 读取第一个事件，并验证：

• Event ID 是初始化专用 ID；
• Event 类型是 SessionConfigured；
• 其中包含 Thread ID、配置快照与 Rollout path 等初始化结果。

只有验证通过，才构造 CodexThread 并尝试插入 live map。

async def finalize_spawn(
    state: ThreadManagerState,
    session: Session,
    io: SessionIo,
    source: SessionSource,
) -> NewThread:
    first = await io.next_event()
    if first.id != INITIAL_SUBMIT_ID or not isinstance(first.msg, SessionConfigured):
        raise SessionConfiguredNotFirstEvent()

    candidate = CodexThread(
        session=session,
        io=io,
        configured=first.msg,
        source=source,
    )
    inserted = await state.live_threads.insert_if_vacant(session.thread_id, candidate)
    if inserted:
        return NewThread(candidate, first.msg)

    await io.shutdown_and_wait()
    raise InvalidRequest("thread id is already running")

“先插 map，再等配置”看似能减少一步等待，但会让其他请求取得一个尚未建立 Rollout 或服务状态的 Thread。当前顺序用一次首事件等待换取清晰的发布不变式。

重复注册必须关闭败方 Session

即使启动前检查过 map，两个并发请求仍可能同时完成 Session::spawn。最终插入使用 Vacant entry 判断。只有一个 Candidate 成功；另一个必须 shutdown_and_wait，否则它虽然不可查询，却仍可能持有 后台任务、通道和外部服务。

这也是为什么注册不能只做 map.insert(id, new)：覆盖旧 Arc 会让正在服务客户端的 Thread 从 Manager 消失，却不一定真正停止。

Internal Thread 的“不可见”不等于“不受管理”

Memory consolidation 等内部 Session 可以进入完整 map，以便生命周期和全局关闭统一管理。但普通 get_thread 和 list_thread_ids 会隐藏它们，防止公开客户端向内部维护任务提交用户 Turn。

全局 bounded shutdown 读取完整 map，并发向每个 Thread 提交 Shutdown，再分成：

• completed：移出 map；
• submit_failed：保留，便于诊断；
• timed_out：保留，允许后续重试或检查。

async def shutdown_all(manager: ThreadManager, timeout: float) -> ShutdownReport:
    snapshot = await manager.live_threads.clone_entries()
    outcomes = await gather(
        *(shutdown_one(thread, timeout) for thread in snapshot.values())
    )
    report = classify(outcomes)
    await manager.live_threads.remove_many(report.completed)
    return report

remove_thread 只改变注册所有权

remove_thread 返回被移出的 Arc。其他组件可能还持有 clone，所以这不是内存销毁屏障；也不会删除 ThreadStore 中的历史。永久删除、归档、冷存储读取是另外的操作。

失败矩阵

条件	结果	是否进入 live map
StoredThread 无历史	Fatal/读取错误	否
Resume live Thread，path 相容	返回现有 Arc	已存在
Resume live Thread，path 冲突	InvalidRequest	不新增
Fork source rollout flush 失败	Fork 失败	否
Session::spawn 失败	启动失败	否
首事件不是初始化 SessionConfigured	专用错误	否
map 插入竞争失败	关闭新 Session，返回 duplicate	只有胜方
Internal Thread 普通查询	ThreadNotFound	实际仍被追踪
bounded shutdown 超时	报告 timed_out	保留

测试重点

ThreadManager 测试必须观察身份与所有权，而不是只看“能否发消息”：

async def test_resume_running_thread_reuses_arc() -> None:
    first = await manager.start_thread(options())
    resumed = await manager.resume(first.rollout_path)
    assert resumed.thread is first.thread


async def test_publish_requires_configured_first_event() -> None:
    session, io = fake_spawn(first_event=TurnStarted("bad"))
    with raises(SessionConfiguredNotFirstEvent):
        await manager.finalize(session, io)
    assert not manager.live_threads.contains(session.thread_id)


async def test_internal_thread_hidden_but_shutdown() -> None:
    internal = await manager.start_thread(internal_options())
    assert await manager.get_thread(internal.id) is None
    report = await manager.shutdown_all(timeout=5)
    assert internal.id in report.completed

当前生产测试还覆盖 Extension initial data、fork aborted marker、环境继承边界、active resume 和 bounded shutdown。它们共同保护“历史身份、活对象身份和公开可见性不是同一个概念”。

小结

ThreadManager 的工作不是创建一个 hashmap entry，而是把共享服务、历史、来源和环境组装为 Session， 等待初始化事件证明它已就绪，再通过唯一性检查发布 CodexThread。Resume 复用身份，Fork 创建身份， Internal source 决定可见性，ThreadStore 保留冷状态。

下一节将打开发布后的 CodexThread，区分它向调用方暴露的提交端、事件端、状态 watcher 和关闭 协议。ThreadManager 决定“哪个 Thread 存在”，CodexThread 决定“怎样与这个 Thread 交互”。