Thread与Session标识模型

本文承接Event与EventMsg总表。阅读重点不是记住几个 UUID newtype，而是跟着一次线程从创建到恢复的调用链，回答四个问题：哪个值代表当前线程，哪个值代表 rollout 文件，为什么子 agent 要共享 SessionId，AgentPath 又解决了什么问题。

在 rust-v0.150.0 中，这些身份分别由不同层维护：ThreadManager 分配线程并保存关系，Session 持有当前线程 ID，AgentControl 保存一棵 agent 树共享的会话 ID，rollout 元数据则把这些值写入持久化记录。它们可能都序列化成字符串，但生成时机、生命周期和恢复规则并不相同。

1. 先区分四个维度

ThreadId 是内存中 CodexThread 的稳定身份，也是普通 rollout 文件使用的 ID。RolloutId 在协议层只是 ThreadId 的类型别名，但语义已经被单独写进 HistoryPosition：普通线程通常让两者相同，thread/revert 可以创建新的 rollout 文件而保留原 ThreadId。

SessionId 是 agent control 树的共享身份。根线程通常用自己的 ThreadId 派生它；子 agent 从同一个 AgentControl 读取它，因此多个 ThreadId 可以属于同一个 SessionId。AgentPath 不参与 UUID 生成，它是 root 下的层级地址，用于消息路由和 agent 可读的寻址。

源码位置：codex-rs/protocol/src/thread_id.rs :: ThreadId、RolloutId

/// Identifier for a Codex thread.
///
/// Codex-generated thread IDs are UUIDv7, and some use cases rely on that.
pub struct ThreadId {
    pub(crate) uuid: Uuid,
}

/// Identifier encoded in a rollout filename.
///
/// Rollout IDs use the same UUID representation as thread IDs. Ordinary rollout files use the
/// thread ID as their rollout ID; `thread/revert` creates a new rollout file with a distinct
/// rollout ID while preserving the thread ID.
pub type RolloutId = ThreadId;

类型别名没有增加运行时字段；真正的区别来自调用方把它当作“文件身份”还是“线程身份”。因此读取分页历史时，HistoryPosition.thread_id 的注释要求把它按 rollout ID 理解，不能直接断言它一定等于 SessionMeta.id。

源码位置：codex-rs/protocol/src/thread_id.rs :: ThreadId::new、ThreadId::from_string、Serialize for ThreadId

impl ThreadId {
    pub fn new() -> Self {
        Self {
            uuid: Uuid::now_v7(),
        }
    }

    pub fn from_string(s: &str) -> Result<Self, uuid::Error> {
        Ok(Self {
            uuid: Uuid::parse_str(s)?,
        })
    }
}

impl Serialize for ThreadId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.collect_str(&self.uuid)
    }
}

这里有三个可观察事实：新 ID 使用 UUIDv7；输入字符串必须通过 UUID 解析；JSON 和 TypeScript schema 看到的是字符串。测试只验证默认值不是全零 UUID，不能把它解释成对分布式碰撞概率的证明。

2. SessionId来源

AgentControl 是会话树的控制平面句柄。源码注释明确要求：同一根线程树只建立一份控制对象，子 agent 拿到它的 clone。with_session_id 是把外部决定好的 SessionId 写入控制对象，而 session_id() 只是读取，不会重新生成。

源码位置：codex-rs/core/src/agent/control.rs :: AgentControl、with_session_id、session_id

/// Control-plane handle for multi-agent operations.
///
/// An `AgentControl` instance is intended to be created at most once per root thread/session
/// tree. That same `AgentControl` is then shared with every sub-agent spawned from that root,
/// which keeps the registry scoped to that root thread rather than the entire `ThreadManager`.
pub(crate) struct AgentControl {
    /// ID shared by the whole agent control session. This means every sub-agents from a common
    /// root share the same session ID.
    session_id: SessionId,
    // other control-plane state omitted
}

pub(crate) fn with_session_id(mut self, session_id: SessionId, max_threads: usize) -> Self {
    self.session_id = session_id;
    self.agent_execution_limiter.initialize(max_threads);
    self
}

pub(crate) fn session_id(&self) -> SessionId {
    self.session_id
}

Session 不把 SessionId 再复制成一个独立字段，而是通过 services.agent_control 读取。这种设计避免了子线程复制控制对象后出现两个“会话真相”。

源码位置：codex-rs/core/src/session/session.rs :: Session::thread_id、Session::session_id

/// Returns the concrete identity for this thread.
pub(crate) fn thread_id(&self) -> ThreadId {
    self.thread_id
}

/// Returns the identity shared by the root thread and all descendant threads.
pub(crate) fn session_id(&self) -> SessionId {
    self.services.agent_control.session_id()
}

3. 身份选择规则

身份的关键逻辑在 Session::new。先从 InitialHistory::Resumed 中寻找已保存的 session_id；如果是非根 agent，旧 rollout 中恰好把自身 ThreadId 当作 SessionId 的记录会被过滤掉，这是兼容早期子 agent 记录的分支。没有可恢复值时，非根 agent 使用共享 AgentControl，根线程则使用自己的 ThreadId。

源码位置：codex-rs/core/src/session/session.rs :: Session::new 中的 session identity selection

let resumed_session_id = match &initial_history {
    InitialHistory::Resumed(resumed) => {
        resumed.history.iter().find_map(|item| match item {
            RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.session_id),
            _ => None,
        })
    }
    InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => None,
};
// Legacy subagent rollouts synthesize session_id from their own thread id.
let resumed_session_id = resumed_session_id.filter(|session_id| {
    !session_configuration.session_source.is_non_root_agent()
        || *session_id != SessionId::from(thread_id)
});
let session_id = resumed_session_id.unwrap_or_else(|| {
    if session_configuration.session_source.is_non_root_agent() {
        agent_control.session_id()
    } else {
        SessionId::from(thread_id)
    }
});

这段代码解释了一个常见误判：子 agent 的 SessionId 不是“由子 ThreadId 转换而来”，除非它是一个没有历史可恢复的根线程。恢复子 agent 时，持久化的 SessionMeta 优先；新建子 agent 时，共享控制对象优先。

SessionConfigured 事件随后同时携带两个值，供实时消费者建立索引。事件中的 thread_id 是当前线程，session_id 是会话树；不要用其中一个字段替换另一个。

源码位置：codex-rs/core/src/session/session.rs :: EventMsg::SessionConfigured

msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
    session_id,
    thread_id,
    cwd: thread_config.cwd().clone(),
    forked_from_id: thread_config.forked_from_thread_id,
    parent_thread_id: thread_config.parent_thread_id,
    thread_source: thread_config.thread_source,
    thread_name: session_configuration.thread_name.clone(),
    model: thread_config.model,
    model_provider_id: thread_config.model_provider_id,
    service_tier: thread_config.service_tier,
    approval_policy: thread_config.approval_policy,
    // remaining configuration fields omitted
}),

4. 线程分配

ThreadManager 把“分配 ID”和“创建 Session”拆成两步。普通 start_thread 进入 start_thread_inner，再交给 ThreadManagerState::spawn_thread；需要提前把线程 ID 交给宿主保存时，可以调用 reserve_thread_id，然后把它放进 StartThreadOptions.reserved_thread_id。

源码位置：codex-rs/core/src/thread_manager.rs :: StartThreadOptions::reserved_thread_id、start_thread、reserve_thread_id

pub struct StartThreadOptions {
    pub config: Config,
    pub initial_history: InitialHistory,
    // other startup options omitted
    /// Thread ID reserved before startup so the caller can associate host-owned state with it.
    pub reserved_thread_id: Option<ThreadId>,
}

pub async fn start_thread(&self, options: StartThreadOptions) -> CodexResult<NewThread> {
    Box::pin(self.start_thread_inner(options, /*forked_from_thread_id*/ None)).await
}

/// Allocates a thread ID before startup so a caller can associate host-owned state with it.
pub fn reserve_thread_id(&self) -> ThreadId {
    self.state.thread_id_generator.as_ref()()
}

预留只是分配，不代表启动成功。Session::new 对新建或 fork 的历史接受这个预留值；对 InitialHistory::Resumed 则直接拒绝预留 ID，避免一个恢复请求同时携带两个线程身份。

源码位置：codex-rs/core/src/session/session.rs :: thread_id selection

let thread_id = match (&initial_history, reserved_thread_id) {
    (
        InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_),
        Some(thread_id),
    ) => thread_id,
    (InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_), None) => {
        agent_control.generate_thread_id()
    }
    (InitialHistory::Resumed(_), Some(_)) => {
        return Err(anyhow::anyhow!(
            "reserved thread ID cannot be used when resuming a thread"
        ));
    }
    (InitialHistory::Resumed(resumed), None) => resumed.conversation_id,
};

fork 和 spawn 子 agent 还要记录来源关系。forked_from_thread_id 表示历史从哪个线程复制或引用；parent_thread_id 表示控制平面上的直接父线程。两者可以同时存在，也可以只有其一：普通 fork 有来源但未必有 parent，Multi-Agent V2 子线程通常有 parent。

源码位置：codex-rs/core/src/thread_manager.rs :: spawn_subagent

pub async fn spawn_subagent(
    &self,
    forked_from_thread_id: ThreadId,
    mut options: StartThreadOptions,
) -> CodexResult<NewThread> {
    let fork_source = self.get_thread(forked_from_thread_id).await?;
    fork_source.ensure_rollout_materialized().await;
    fork_source.flush_rollout().await?;
    let stored_thread = fork_source
        .read_thread(/*include_archived*/ true, /*include_history*/ true)
        .await?;
    let history = stored_thread_to_initial_history(stored_thread, fork_source.rollout_path())?;
    options.initial_history = fork_history_from_snapshot(
        ForkSnapshot::Interrupted,
        history,
        InterruptedTurnHistoryMarker::from_config_and_version(
            &options.config,
            fork_source.multi_agent_version().unwrap_or(MultiAgentVersion::V1),
        ),
    );
    self.start_thread_inner(options, Some(forked_from_thread_id))
        .await
}

这里的 flush_rollout 很重要：fork 读取的是已经落盘的历史快照，而不是只看实时事件队列。随后 start_thread_inner 把来源 ID 交给 spawn 请求，最终由 Session 配置保存。

5. SessionMeta落盘

线程恢复依赖 rollout 中的第一条 SessionMeta。它同时记录 session_id、稳定线程 id、fork 来源、直接父线程和 agent 地址；history_base、subagent_history_start_ordinal 则描述分页历史如何继承。读取者不能只取一个 UUID 决定线程关系。

源码位置：codex-rs/protocol/src/protocol.rs :: SessionMeta

pub struct SessionMeta {
    pub session_id: SessionId,
    pub id: ThreadId,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub forked_from_id: Option<ThreadId>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent_thread_id: Option<ThreadId>,
    pub timestamp: String,
    pub cwd: PathBuf,
    pub originator: String,
    pub cli_version: String,
    #[serde(default)]
    pub source: SessionSource,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub thread_source: Option<ThreadSource>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub agent_nickname: Option<String>,
    #[serde(default, alias = "agent_type", skip_serializing_if = "Option::is_none")]
    pub agent_role: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub agent_path: Option<String>,
    // model, history and context-window fields follow
}

SessionMeta.id 是线程的稳定 ID；它不是“本次 rollout 文件名”的可靠别名，因为 revert 会引入新的 rollout ID。agent_path 以字符串保存，是因为该字段需要兼容旧记录和跨语言 schema；恢复时再通过 AgentPath::try_from 验证。

旧记录可能没有 session_id。SessionMetaLine 的反序列化会在缺失时把 id 复制到 session_id，从而让旧根线程继续可读；前一节的子 agent 过滤逻辑再负责避免把这种兼容值误当成共享会话 ID。

源码位置：codex-rs/protocol/src/protocol.rs :: SessionMetaLine 的反序列化

impl<'de> Deserialize<'de> for SessionMetaLine {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let mut value = Value::deserialize(deserializer)?;
        let fields = value
            .as_object_mut()
            .ok_or_else(|| D::Error::custom("session metadata must be an object"))?;
        if !fields.contains_key("session_id") {
            let thread_id = fields
                .get("id")
                .cloned()
                .ok_or_else(|| D::Error::missing_field("id"))?;
            fields.insert("session_id".to_string(), thread_id);
        }
        let SessionMetaLineFields { meta, git } =
            serde_json::from_value(value).map_err(D::Error::custom)?;
        Ok(Self { meta, git })
    }
}

6. AgentPath寻址

AgentPath 有固定的 root 和 morpheus 保留地址。子段只能使用小写 ASCII 字母、数字和下划线；绝对路径必须从 root 段开始，不能以 / 结尾；相对引用不能包含 ..。这些规则让 inter-agent message 可以在字符串边界上安全解析。

源码位置：codex-rs/protocol/src/agent_path.rs :: AgentPath::root、join、resolve

pub fn root() -> Self {
    Self(Self::ROOT.to_string())
}

pub fn join(&self, agent_name: &str) -> Result<Self, String> {
    validate_agent_name(agent_name)?;
    Self::from_string(format!("{self}/{agent_name}"))
}

pub fn resolve(&self, reference: &str) -> Result<Self, String> {
    if reference.is_empty() {
        return Err("agent path must not be empty".to_string());
    }
    if reference == Self::ROOT {
        return Ok(Self::root());
    }
    if reference.starts_with('/') {
        return Self::try_from(reference);
    }

    validate_relative_reference(reference)?;
    Self::from_string(format!("{self}/{reference}"))
}

因此一个 root 下的多段 AgentPath 只说明消息目标在 agent 树中的位置；它不说明对应线程的 ThreadId，也不说明该 agent 是否仍在内存中。多 agent 代码通常同时携带三者：agent_thread_id 用于查找线程，agent_path 用于寻址，session_id 用于把事件和环境变量归入同一棵树。

7. 身份跨出 Core

线程身份会进入至少三个外部边界。首先，Responses metadata 携带 session_id、thread_id、父线程和 fork 来源；其次，shell/Unified Exec 环境注入 CODEX_THREAD_ID 与 CODEX_SESSION_ID；再次，hook 和 session lifecycle payload 使用相应字段建立关联。

源码位置：codex-rs/core/src/session/session.rs :: Session::responses_metadata

pub(crate) async fn responses_metadata(
    &self,
    turn_context: &TurnContext,
    request_kind: CodexResponsesRequestKind,
) -> CodexResponsesMetadata {
    let (window_id, context_window_id) = self.current_window().await;
    CodexResponsesMetadata {
        context_window_id: Some(context_window_id),
        ..turn_context.turn_metadata_state.to_responses_metadata(
            self.installation_id.clone(),
            window_id,
            request_kind,
        )
    }
}

Session::responses_metadata 本身只补窗口身份，线程和会话字段由 Turn metadata state 组合。也就是说，看到某个 HTTP 请求带有 thread_id，还需要回到构造该 Turn metadata 的路径确认它来自哪个 Session。

源码位置：codex-rs/core/src/exec_env.rs :: inject_session_id_env

/// Exposes the shared root-session identity to model-reachable shell commands.
pub(crate) fn inject_session_id_env(env: &mut HashMap<String, String>, session_id: SessionId) {
    env.insert(CODEX_SESSION_ID_ENV_VAR.to_string(), session_id.to_string());
}

Unified Exec 同时写入 CODEX_THREAD_ID。环境变量只是诊断和关联信息，子进程可以覆盖它们，不能把它们当作权限证明或线程存活证明。

8. 边界测试

协议层测试验证字符串解析、序列化和 AgentPath 语法；Core session 测试验证新根线程和恢复子 agent 的 SessionId 选择。它们组合起来，正好覆盖“值如何生成”和“历史如何恢复”两个容易混淆的边界。

源码位置：codex-rs/protocol/src/session_id.rs :: converts_to_and_from_thread_id；codex-rs/protocol/src/agent_path.rs :: invalid_names_and_paths_are_rejected

#[test]
fn converts_to_and_from_thread_id() {
    let thread_id = ThreadId::new();
    let session_id = SessionId::from(thread_id);

    assert_eq!(ThreadId::from(session_id), thread_id);
}

#[test]
fn invalid_names_and_paths_are_rejected() {
    assert_eq!(
        AgentPath::root().join("BadName"),
        Err("agent_name must use only lowercase letters, digits, and underscores".to_string())
    );
    assert_eq!(
        AgentPath::try_from("/not-root"),
        Err("absolute agent paths must start with `/root` or be `/morpheus`".to_string())
    );
    assert_eq!(
        AgentPath::root().resolve("../sibling"),
        Err("agent_name `..` is reserved".to_string())
    );
}

源码位置：codex-rs/core/src/session/tests.rs :: resumed_root_session_uses_thread_id_as_session_id、resumed_subagent_session_restores_persisted_session_id

assert_eq!(session.thread_id(), thread_id);
assert_eq!(session.session_id(), SessionId::from(thread_id));
// resumed subagent:
assert_eq!(session.thread_id(), thread_id);
assert_eq!(session.session_id(), parent_session_id);

可以这样运行这些近场测试：

cd codex-rs
cargo test -p codex-protocol converts_to_and_from_thread_id -- --nocapture --test-threads=1
cargo test -p codex-protocol invalid_names_and_paths_are_rejected -- --nocapture --test-threads=1
cargo test -p codex-core resumed_root_session_uses_thread_id_as_session_id -- --nocapture --test-threads=1
cargo test -p codex-core resumed_subagent_session_restores_persisted_session_id -- --nocapture --test-threads=1

这些断言不能推出线程在所有进程间始终存活，也不能证明远端 App Server 会保留每个字段；它们只证明当前源码中 ID 的类型边界、恢复优先级和路径验证规则。

9. 源码定位练习

排查“两个事件看起来属于同一会话”时，按下面顺序阅读：

1. 从 SessionConfigured 取出 thread_id 和 session_id，确认当前线程与共享会话是否被混用。
2. 回到 Session::new，判断值来自 SessionMeta、AgentControl 还是当前 ThreadId。
3. 如果涉及 fork 或恢复，检查 SessionMeta.id、forked_from_id、parent_thread_id 与 HistoryPosition.thread_id，不要只看 rollout 文件名。
4. 如果涉及子 agent 消息，使用 agent_thread_id 查线程，使用 AgentPath 做路由，使用 SessionId 做树级关联。
5. 如果问题出现在模型请求或 shell 中，再分别检查 Responses metadata 和 exec_env 的注入点。

掌握这条路径后，ThreadId、RolloutId、SessionId 和 AgentPath 就不再是四个相似的字符串，而是四个有明确所有者、写入边界和恢复规则的协议概念。