ThreadStart处理流程

thread/start 是一个容易被误读的 API。它不是把参数塞进一个 Thread 结构体后立即返回，而是启动一个由 TaskTracker 持有的后台任务：任务先合并配置并检查项目权限，再把 StartThreadOptions 交给 Core；Core 返回 NewThread 后，App Server 还要读取配置快照、建立事件监听、更新 Thread watch，最后分别写入 RPC response 和 thread/started notification。

本文面向熟悉 Rust async/await、Arc、Result 和 serde 的读者。建议先阅读AppServer架构总览、V2请求分派总表和服务端请求与客户端响应。本文只分析新建 Thread；resume、fork 和首次 turn/start 的业务处理留给后续文章。

读完后应能回答四个问题：请求在哪一层被接纳，哪些字段在构造期生效，为什么返回成功后仍可能继续产生启动通知，以及配置、Core、监听器或出站队列分别失败时 Thread 会处于什么状态。

1. 对象与出口

这里有四个同名但不同层级的对象：协议层的 ThreadStartParams 是一次请求的输入；Core 的 CodexThread 是可运行的会话；协议层的 Thread 是给客户端的快照；ThreadStartedNotification 是订阅时间线的通知。ThreadStartResponse 不是 Core 对象的引用，而是从 ThreadConfigSnapshot 和运行时状态重新投影出的结果。

图中 Config 是 Core 使用的有效配置，Thread 是对外快照，二者不应混为同一个可变对象。成功 response 只表示这些投影已经排入出站发送路径；客户端是否先收到 notification 还取决于同一个 outgoing sender 的消费顺序。

这个时序图强调处理器返回和 RPC 响应不是同一时刻：spawn 后 MessageProcessor 可以继续处理其他请求，启动任务才拥有后续所有权。

这组类型对应源码中的边界：协议输入先转成配置覆盖，再和历史、环境、扩展字段合成 Core 选项，最后从 Core 快照产生协议 response。

2. 请求入口

源码文件：codex-rs/app-server/src/message_processor.rs

相关函数：MessageProcessor::process_request 中的 ClientRequest::ThreadStart 分支。

ClientRequest::ThreadStart { params, .. } => {
    self.thread_processor
        .thread_start(
            request_id.clone(),
            params,
            app_server_client_name.clone(),
            client_version.clone(),
            client_mcp_extensions.clone(),
            request_context,
        )
        .await
}

这段分派没有直接创建 Thread。它把连接级 request id、握手阶段记录的 client info、MCP 扩展能力和 tracing 上下文一并交给 ThreadRequestProcessor。因此后续错误仍能用原 request id 返回，Core 新建的 span 也能挂在同一次请求的父 span 下。

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：ThreadRequestProcessor::thread_start、thread_start_inner。

pub(crate) async fn thread_start(
    &self,
    request_id: ConnectionRequestId,
    params: ThreadStartParams,
    app_server_client_name: Option<String>,
    app_server_client_version: Option<String>,
    client_mcp_extensions: ClientMcpExtensions,
    request_context: RequestContext,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
    self.thread_start_inner(
        request_id,
        params,
        app_server_client_name,
        app_server_client_version,
        client_mcp_extensions,
        request_context,
    )
    .await
    .map(|()| None)
}

thread_start 的返回值是 None，这是一个重要的异步边界：它只负责把启动任务交给内部调度器，真正的 ThreadStartResponse 由后台任务直接调用 send_response_with_thread_originator 写出。若只在 process_request 的返回值上设置断点，会错过真正的 response 生产点。

3. 参数与拒绝

源码文件：codex-rs/app-server-protocol/src/protocol/v2/thread.rs

相关类型：ThreadStartParams。

pub struct ThreadStartParams {
    pub model: Option<String>,
    pub model_provider: Option<String>,
    pub allow_provider_model_fallback: bool,
    pub service_tier: Option<Option<String>>,
    pub cwd: Option<String>,
    pub runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>,
    pub approval_policy: Option<AskForApproval>,
    pub approvals_reviewer: Option<ApprovalsReviewer>,
    pub sandbox: Option<SandboxMode>,
    pub permissions: Option<String>,
    pub config: Option<HashMap<String, JsonValue>>,
    pub service_name: Option<String>,
    pub base_instructions: Option<String>,
    pub developer_instructions: Option<String>,
    pub personality: Option<Personality>,
    pub ephemeral: Option<bool>,
    pub history_mode: Option<ThreadHistoryMode>,
    pub session_start_source: Option<ThreadStartSource>,
    pub thread_source: Option<ThreadSource>,
    pub project_id: Option<String>,
    pub environments: Option<Vec<TurnEnvironmentParams>>,
    pub dynamic_tools: Option<Vec<DynamicToolSpec>>,
    pub selected_capability_roots: Option<Vec<SelectedCapabilityRoot>>,
    pub experimental_raw_events: bool,
}

字段并不都由同一个消费者读取：model、cwd 和权限字段进入 ConfigOverrides；history_mode、thread_source 和 session_start_source 进入 StartThreadOptions；dynamic_tools 和 capability roots 进入扩展初始化；project_id 影响持久化 metadata；experimental_raw_events 只改变 listener 的事件投影。把它们都称为“Thread 属性”会掩盖生效时机。

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：thread_start_inner 的解构与入口校验。

let ThreadStartParams {
    model,
    model_provider,
    allow_provider_model_fallback,
    service_tier,
    cwd,
    runtime_workspace_roots,
    approval_policy,
    approvals_reviewer,
    sandbox,
    permissions,
    config,
    service_name,
    base_instructions,
    developer_instructions,
    dynamic_tools,
    selected_capability_roots,
    experimental_raw_events,
    personality,
    ephemeral,
    history_mode,
    session_start_source,
    thread_source,
    project_id,
    environments,
} = params;

if matches!(
    history_mode,
    Some(codex_app_server_protocol::ThreadHistoryMode::Paginated)
) && !self.thread_store.supports_paginated_history_lists()
{
    return Err(invalid_request(
        "paginated threads require thread/turns/list and thread/items/list support",
    ));
}
if sandbox.is_some() && permissions.is_some() {
    return Err(invalid_request(
        "`permissions` cannot be combined with `sandbox`",
    ));
}

这里的两个拒绝发生在启动 task 创建前：不支持分页历史不会创建 Core Thread；sandbox 与命名权限 profile 冲突也不会进入配置加载。它们属于请求参数错误，而不是“Thread 创建失败后再清理”的路径。

状态图把 trust 重载放在 CoreStart 之前，说明它属于构造期决策，不是 Thread 建立后的异步通知。

同一阶段还检查 project_id 是否为空、ThreadStore 是否支持 project，以及项目是否真实存在。项目不存在返回 invalid_params；ThreadStore 声明不支持该操作则返回 unsupported_thread_store_operation。这一区分让客户端知道应修改参数还是换用具备 SQLite/project 能力的存储。

4. 后台任务

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：thread_start_inner 中的 TaskTracker 提交。

let thread_start_task = async move {
    if let Err(error) = Self::thread_start_task(
        listener_task_context,
        thread_store,
        config_manager,
        request_id,
        app_server_client_name,
        app_server_client_version,
        client_mcp_extensions,
        config,
        typesafe_overrides,
        dynamic_tools,
        selected_capability_roots.unwrap_or_default(),
        history_mode.map(Into::into),
        session_start_source,
        thread_source.map(Into::into),
        project_id,
        environments,
        service_name,
        allow_provider_model_fallback,
        experimental_raw_events,
        request_trace,
        initial_config_warnings,
    ).await {
        outgoing.send_error(error_request_id, error).await;
    }
};
self.background_tasks
    .spawn(thread_start_task.instrument(request_context.span()));
Ok(())

后台任务捕获的是启动所需的所有权，而不是借用 params。TaskTracker 让 App Server 关闭时可以等待这些任务；错误由任务自己用保存下来的 ConnectionRequestId 发出。因而“处理器函数已返回”只表示任务已被接纳，不能当成 Thread 已建立。

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：thread_start_task 的配置加载与项目 trust 重载。

let requested_cwd = typesafe_overrides.cwd.clone();
let mut config = config_manager
    .load_with_overrides(config_overrides.clone(), typesafe_overrides.clone())
    .await
    .map_err(|err| config_load_error(&err))?;

let effective_permission_profile = config.permissions.effective_permission_profile();
let effective_permissions_trust_project = match &effective_permission_profile {
    PermissionProfile::Disabled | PermissionProfile::External { .. } => true,
    PermissionProfile::Managed { .. } => effective_permission_profile
        .file_system_sandbox_policy()
        .can_write_path_with_cwd(config.cwd.as_path(), config.cwd.as_path()),
};

if requested_cwd.is_some()
    && config.active_project.trust_level.is_none()
    && effective_permissions_trust_project
{
    let trust_target = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &config.cwd)
        .await
        .unwrap_or_else(|| config.cwd.clone());
    // 持久化成功后重新加载；持久化失败时使用只对当前 Thread 生效的 CLI override。
    config = config_manager
        .load_with_cli_overrides(
            cli_overrides_for_reload,
            config_overrides,
            typesafe_overrides,
            None,
        )
        .await
        .map_err(|err| config_load_error(&err))?;
}

配置不是简单的“请求覆盖默认值”。当请求指定了 cwd，代码会根据有效权限 profile判断项目是否可以被信任；受 managed profile 限制时，只有当前 cwd 具备写权限才满足条件。持久化 trust 成功后再次加载配置，失败时则把 trusted project 作为本次 Thread 的内存 override，并记录 warning。该重载发生在 Core 创建前，所以它决定本 Thread 的 sandbox、模型和指令来源。

5. 构造启动选项

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：build_thread_config_overrides、thread_start_task。

fn build_thread_config_overrides(
    &self,
    model: Option<String>,
    model_provider: Option<String>,
    service_tier: Option<Option<String>>,
    cwd: Option<String>,
    runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>,
    approval_policy: Option<AskForApproval>,
    approvals_reviewer: Option<ApprovalsReviewer>,
    sandbox: Option<SandboxMode>,
    permissions: Option<String>,
    base_instructions: Option<String>,
    developer_instructions: Option<String>,
    personality: Option<Personality>,
) -> ConfigOverrides {
    ConfigOverrides {
        model,
        model_provider,
        service_tier,
        cwd: cwd.map(PathBuf::from),
        workspace_roots: runtime_workspace_roots,
        default_permissions: permissions,
        approval_policy: approval_policy.map(AskForApproval::to_core),
        approvals_reviewer: approvals_reviewer.map(ApprovalsReviewer::to_core),
        sandbox_mode: sandbox.map(SandboxMode::to_core),
        base_instructions,
        developer_instructions,
        personality,
        ..Default::default()
    }
}

这个函数只做类型转换，不读取文件，也不创建 Thread。service_tier 的双层 Option 保留“字段省略”和“显式清空”的区别；cwd 转成 PathBuf 后由配置加载器解析；协议层的 approval/sandbox 枚举转换为 Core 枚举。真正的默认值和层级合并在 ConfigManager::load_with_overrides 内完成。

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：thread_start_task 中的环境、动态工具和 project metadata 装配。

let environments = environment_selections.unwrap_or_else(|| {
    listener_task_context
        .thread_manager
        .default_environment_selections(&config.cwd, &config.workspace_roots)
});
let dynamic_tools = dynamic_tools.unwrap_or_default();
if !dynamic_tools.is_empty() {
    validate_dynamic_tools(&dynamic_tools).map_err(invalid_request)?;
}
let mut thread_extension_init = ExtensionDataInit::new();
if !selected_capability_roots.is_empty() {
    thread_extension_init.insert(selected_capability_roots);
}
let mut start_options = StartThreadOptions::new(config);
let reserved_thread_id = if start_options.config.ephemeral {
    None
} else {
    stage_pending_project_metadata(
        listener_task_context.thread_manager.as_ref(),
        thread_store.as_ref(),
        project_id.as_deref(),
        "thread/start",
    ).await?
};
start_options.reserved_thread_id = reserved_thread_id;

省略 environments 会按 cwd 和 workspace roots 选择默认环境，传入空数组则明确关闭环境访问；动态工具必须通过名称、namespace 和参数 schema 校验；非 ephemeral Thread 在 Core 生成最终 id 前先暂存 project metadata。reserved_thread_id 是持久化协调信息，不是已经对客户端公开的 Thread id。

6. Core与历史

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：ThreadManager::start_thread 调用处。

let new_thread = listener_task_context
    .thread_manager
    .start_thread(StartThreadOptions {
        allow_provider_model_fallback,
        initial_history: match session_start_source
            .unwrap_or(codex_app_server_protocol::ThreadStartSource::Startup)
        {
            ThreadStartSource::Startup => InitialHistory::New,
            ThreadStartSource::Clear => InitialHistory::Cleared,
        },
        history_mode,
        thread_source,
        dynamic_tools,
        metrics_service_name: service_name,
        parent_trace: request_trace,
        environments: Some(environments),
        thread_extension_init,
        client_mcp_extensions,
        ..start_options
    })
    .instrument(tracing::info_span!(
        "app_server.thread_start.create_thread",
        thread_start.dynamic_tool_count = dynamic_tool_count,
    ))
    .await;

Startup 和 Clear 都是创建新 Thread，但 InitialHistory 不同：前者从空的新历史开始，后者表达“清除既有语义后开始”的来源。history_mode 决定之后使用 legacy rollout 还是分页 history API；它不是当前请求的分页参数。parent_trace、MCP 扩展和环境选择则把连接层信息带入 Core Session。

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：thread_start_task 的 NewThread 错误映射。

let NewThread {
    thread_id,
    thread,
    session_configured,
    ..
} = match new_thread {
    Ok(new_thread) => new_thread,
    Err(err) => {
        remove_pending_project_metadata(thread_store.as_ref(), reserved_thread_id).await;
        return Err(match err.details() {
            CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()),
            CodexErrorDetails::UnsupportedOperation(message) => method_not_found(message.clone()),
            _ => internal_error(format!("error creating thread: {err}")),
        });
    }
};

Core 失败时，暂存的 project metadata 会被删除。错误映射保留了 InvalidRequest 和 UnsupportedOperation 的协议语义，其余 Core 错误才折叠为 internal error。此时没有 listener、watch 或 response，客户端只会收到原 request id 对应的错误。

这里的清理只针对 Core 失败；一旦进入 NewThread，project metadata、listener 和 watch 就拥有不同的退出责任。

7. Core到快照

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：set_app_server_client_info、build_thread_from_snapshot。

Self::set_app_server_client_info(
    thread.as_ref(),
    app_server_client_name,
    app_server_client_version,
).await?;

let instruction_sources = thread.legacy_instruction_sources().await;
let config_snapshot = thread
    .config_snapshot()
    .instrument(tracing::info_span!(
        "app_server.thread_start.config_snapshot",
    ))
    .await;
let mut thread = build_thread_from_snapshot(
    thread_id,
    session_configured.session_id.to_string(),
    thread.multi_agent_version(),
    &config_snapshot,
    session_configured.rollout_path.clone(),
);
thread.project_id = project_id.clone();

Core 返回的 CodexThread 仍是内部运行对象。App Server 从它读取 client info、指令源和配置快照，再构造协议 Thread。因此 response 中的 cwd、模型 provider、权限 profile 和 instruction sources 都来自最终快照，不是未经合并的 ThreadStartParams。

8. Listener与状态

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关函数：ensure_conversation_listener、ThreadWatchManager::upsert_thread_silently、resolve_thread_status。

log_listener_attach_result(
    super::thread_lifecycle::ensure_conversation_listener(
        listener_task_context.clone(),
        thread_id,
        request_id.connection_id,
        experimental_raw_events,
    ).await,
    thread_id,
    request_id.connection_id,
    "thread",
);

listener_task_context
    .thread_watch_manager
    .upsert_thread_silently(&thread.id)
    .await;

thread.status = resolve_thread_status(
    listener_task_context
        .thread_watch_manager
        .loaded_status_for_thread(&thread.id)
        .await,
    false,
);

新 Thread 会自动绑定当前连接的 listener。experimental_raw_events 只影响 listener 是否额外发布原始 Responses item；它不改变 Thread 的创建。watch manager 先静默插入，再读取 loaded status，避免在还没有首轮 Turn 时错误地标成 active。listener attach 的错误由 log_listener_attach_result 记录，代码仍继续构造 response；这意味着“Thread 已创建但事件订阅失败”是一个可见的部分成功边界。

9. Response与通知

源码文件：codex-rs/app-server/src/request_processors/thread_processor.rs

相关类型：ThreadStartResponse；相关函数：thread_started_notification。

let response = ThreadStartResponse {
    thread: thread.clone(),
    model: config_snapshot.model,
    model_provider: config_snapshot.model_provider_id,
    service_tier: config_snapshot.service_tier,
    cwd: config_snapshot.cwd().clone(),
    runtime_workspace_roots: config_snapshot.workspace_roots,
    instruction_sources,
    approval_policy: config_snapshot.approval_policy.into(),
    approvals_reviewer: config_snapshot.approvals_reviewer.into(),
    sandbox: config_snapshot.sandbox_policy().into(),
    active_permission_profile: thread_response_active_permission_profile(
        config_snapshot.active_permission_profile,
    ),
    reasoning_effort: config_snapshot.reasoning_effort,
    multi_agent_mode: MultiAgentMode::ExplicitRequestOnly,
};
let notif = thread_started_notification(thread);

listener_task_context
    .outgoing
    .send_response_with_thread_originator(request_id, response, thread_originator)
    .await;
listener_task_context
    .outgoing
    .send_server_notification(ServerNotification::ThreadStarted(notif))
    .await;

response 额外携带最终配置字段，供当前 RPC 的等待者直接使用；notification 只携带 Thread 快照，供订阅者建立时间线。两者共享 Thread id，却不是同一个协议消息，也不能假定客户端一定先收到其中某一种。

源码文件：codex-rs/app-server/src/request_processors/thread_summary.rs

相关函数：thread_started_notification。

pub(super) fn thread_started_notification(mut thread: Thread) -> ThreadStartedNotification {
    thread.turns.clear();
    ThreadStartedNotification { thread }
}

通知会清空 turns。新 Thread 的启动事件只证明容器和配置已经建立，不把可能由 summary builder 产生的历史 turns 重复发送给订阅者；需要历史的客户端应显式调用 thread/read 或分页 history API。

10. 失败、关闭与定位

启动流程至少有四种失败位置：参数/能力校验失败，配置加载失败，Core 创建失败，listener 或出站发送失败。前两种不会创建 Core；Core 失败会删除 pending metadata；listener 失败记录诊断但可能保留已创建 Thread；出站失败发生在 Core 和 watch 已存在之后，不能回滚 Thread。

关闭时 drain_background_tasks 会关闭 TaskTracker 并等待最多十秒；因此一个已经返回“已接纳”的 thread/start 任务可能在服务器关闭期间仍处于 Configuring 或 Creating。调试时要同时查看 request error、app_server.thread_start.* spans、ThreadStore 中的 metadata 和 outgoing 队列。

关闭等待是后台任务的生命周期边界，不能据此推断已经创建的 Core Thread 会被回滚。

推荐的只读定位命令：

rg -n "ClientRequest::ThreadStart|thread_start_inner|thread_start_task|StartThreadOptions|thread_started_notification" \
  codex-rs/app-server/src/message_processor.rs \
  codex-rs/app-server/src/request_processors/thread_processor.rs \
  codex-rs/app-server/src/request_processors/thread_summary.rs

11. 测试证据

源码文件：codex-rs/app-server/src/in_process.rs

测试：in_process_start_uses_requested_session_source_for_thread_start。

let response = client
    .request(ClientRequest::ThreadStart {
        request_id: RequestId::Integer(2),
        params: ThreadStartParams {
            ephemeral: Some(true),
            ..ThreadStartParams::default()
        },
    })
    .await
    .expect("request transport should work")
    .expect("thread/start should succeed");
let parsed: ThreadStartResponse =
    serde_json::from_value(response).expect("thread/start response should parse");
assert_eq!(parsed.thread.source, expected_source);

测试在真实 in-process server 上发送 typed v2 request，断言 response 能按 schema 反序列化，并且 server 初始化时选择的 SessionSource::Cli 或 Exec 被投影到 Thread source。它证明传输、分派、Core 创建和 response schema 的闭环；没有证明 socket transport，也没有覆盖非 ephemeral project metadata。

源码文件：codex-rs/app-server/src/message_processor_tracing_tests.rs

测试：thread_start_jsonrpc_span_exports_server_span_and_parents_children。

let _: ThreadStartResponse = harness
    .start_thread(/*request_id*/ 20_003, Some(remote_trace))
    .await;
let spans = wait_for_new_exported_spans(harness.tracing, baseline_len, |spans| {
    spans.iter().any(|span| {
        span.span_kind == SpanKind::Server
            && span_attr(span, "rpc.method") == Some("thread/start")
            && span.span_context.trace_id() == remote_trace_id
    }) && spans.iter().any(|span| {
        span.name.as_ref() == "app_server.thread_start.notify_started"
            && span.span_context.trace_id() == remote_trace_id
    })
}).await;

这个测试用远端 W3C traceparent 启动 Thread，检查 server span 延续远程 trace，并且 notify_started 是其内部 descendant。它证明后台任务仍保留请求的 tracing 关系；不证明 notification 已被真实网络客户端读取，也不证明 span 导出成功时 response 一定成功送达。

还可运行：

cargo test -p codex-app-server in_process_start_uses_requested_session_source_for_thread_start
cargo test -p codex-app-server thread_start_jsonrpc_span_exports_server_span_and_parents_children
cargo test -p codex-app-server --test all thread_start

若客户端“收到 response 但没有 ThreadStarted”，先查 ensure_conversation_listener 和 outgoing consumer；若“返回错误但磁盘有残留 metadata”，查 remove_pending_project_metadata 是否执行以及错误发生在 reserved id 之前还是之后；若“状态一直 NotLoaded”，先确认这是新 Thread 的初始状态，再检查 listener 是否收到后续 Core event。

12. 学习检查

先不打开源码，复述完整链路：ClientRequest::ThreadStart 如何到达 thread_start_task，哪些字段进入 ConfigOverrides，哪些字段进入 StartThreadOptions，Core 成功后又经过哪些 snapshot、listener 和 watch 步骤，最后哪两个函数写出 response 与 notification？

再做一个故障定位练习：给 thread_manager.start_thread 注入一个 InvalidRequest，预测客户端收到的错误 code、pending project metadata 是否保留、是否会产生 thread/started；然后把故障移动到 listener attach，比较这两个结果为什么不同。

下一篇ThreadList与分页会沿相反方向读取 ThreadStore：它不创建运行对象，而是把持久化摘要、分页 cursor、过滤器和已加载 Thread 状态重新合成为列表响应。