动态工具与执行记录

本文承接Plan与RequestInput协议和ConversationItem类型体系。动态工具不是“把 JSON schema 塞给模型”这么简单：线程先保存 DynamicToolSpec，Core 为当前 turn 创建 handler，模型调用后由 session 发出请求并等待客户端响应，最终生成 DynamicToolCallItem，再由 history reducer 或 App Server 投影给客户端。

这条链需要区分四个身份：工具定义的 namespace/name、一次调用的 call_id、当前 turn 的 turn_id，以及执行记录的 item 状态。定义进入 registry 不代表客户端已经执行，客户端返回成功也不代表整个 turn 已完成。

1. 工具规格

源码位置：codex-rs/protocol/src/dynamic_tools.rs :: DynamicToolSpec、DynamicToolFunctionSpec、DynamicToolNamespaceSpec、DynamicToolCallRequest、DynamicToolResponse

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum DynamicToolSpec {
    Function(DynamicToolFunctionSpec),
    Namespace(DynamicToolNamespaceSpec),
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct DynamicToolFunctionSpec {
    pub name: String,
    pub description: String,
    pub input_schema: JsonValue,
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub defer_loading: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct DynamicToolNamespaceSpec {
    pub name: String,
    pub description: String,
    pub tools: Vec<DynamicToolNamespaceTool>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct DynamicToolCallRequest {
    pub call_id: String,
    pub turn_id: String,
    #[serde(default)]
    pub started_at_ms: i64,
    #[serde(default)]
    pub namespace: Option<String>,
    pub tool: String,
    pub arguments: JsonValue,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct DynamicToolResponse {
    pub content_items: Vec<DynamicToolCallOutputContentItem>,
    pub success: bool,
}

defer_loading 控制工具是否延迟暴露给模型；它不是执行结果。DynamicToolCallRequest 使用结构化 JSON arguments，响应则用 content_items + success 描述客户端结果。

2. 旧规格归一化

源码位置：codex-rs/protocol/src/dynamic_tools.rs :: normalize_dynamic_tool_specs、group_dynamic_tools_by_namespace

pub fn normalize_dynamic_tool_specs(
    values: Vec<JsonValue>,
) -> Result<Vec<DynamicToolSpec>, serde_json::Error> {
    let has_legacy_fields = |value: &JsonValue| {
        value.get("namespace").is_some()
            || value.get("exposeToContext").is_some()
            || value.get("type").is_none()
    };
    let has_legacy_format = values.iter().any(|value| {
        has_legacy_fields(value)
            || value
                .get("tools")
                .and_then(JsonValue::as_array)
                .is_some_and(|tools| tools.iter().any(&has_legacy_fields))
    });
    let has_canonical_format = values.iter().any(|value| value.get("type").is_some());
    if has_legacy_format && has_canonical_format {
        return Err(serde_json::Error::custom(
            "dynamic tools must use either canonical or legacy format consistently",
        ));
    }
    if !has_legacy_format {
        return values.into_iter().map(serde_json::from_value).collect();
    }
    // legacy entries are converted and grouped below
}

0.150.0 仍能读取旧的 flat shape，但拒绝 canonical 与 legacy 混用。兼容层把旧 exposeToContext 反向转换成 defer_loading，再通过 namespace 分组，避免同一线程出现两套工具身份规则。

3. 注册handler

源码位置：codex-rs/core/src/tools/spec_plan.rs :: append_dynamic_tool_runtimes

#[instrument(level = "trace", skip_all, fields(dynamic_tool_count = dynamic_tools.len()))]
fn append_dynamic_tool_runtimes(dynamic_tools: &[DynamicToolSpec], registry: &mut ToolRegistry) {
    for spec in dynamic_tools {
        match spec {
            DynamicToolSpec::Function(tool) => {
                let Some(handler) = DynamicToolHandler::new(tool) else {
                    tracing::error!(
                        "Failed to convert dynamic tool {:?} to OpenAI tool",
                        tool.name
                    );
                    continue;
                };
                registry.register_external(Arc::new(handler));
            }
            DynamicToolSpec::Namespace(namespace) => {
                for tool in &namespace.tools {
                    let DynamicToolNamespaceTool::Function(tool) = tool;
                    let Some(handler) = DynamicToolHandler::new_in_namespace(namespace, tool)
                    else {
                        tracing::error!(
                            "Failed to convert dynamic tool {:?}.{:?} to OpenAI tool",
                            namespace.name,
                            tool.name
                        );
                        continue;
                    };
                    registry.register_external(Arc::new(handler));
                }
            }
        }
    }
}

注册失败只记录错误并跳过该规格。register_external 说明这些 handler 来自当前线程外部提供的定义；registry 中存在 schema 才能被路由，不能从线程元数据中的定义推断执行请求一定可达。

4. Handler与暴露

源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::from_parts

fn from_parts(
    tool: &DynamicToolFunctionSpec,
    namespace: Option<&DynamicToolNamespaceSpec>,
) -> Option<Self> {
    let tool_name = ToolName::new(
        namespace.map(|namespace| namespace.name.clone()),
        tool.name.clone(),
    );
    let mut output_tool = dynamic_tool_to_responses_api_tool(tool).ok()?;
    output_tool.defer_loading = None;
    let spec = match namespace {
        Some(namespace) => ToolSpec::Namespace(ResponsesApiNamespace {
            name: namespace.name.clone(),
            description: if namespace.description.trim().is_empty() {
                default_namespace_description(&namespace.name)
            } else {
                namespace.description.clone()
            },
            tools: vec![ResponsesApiNamespaceTool::Function(output_tool)],
        }),
        None => ToolSpec::Function(output_tool),
    };
    Some(Self {
        tool_name,
        spec,
        exposure: if tool.defer_loading {
            ToolExposure::Deferred
        } else {
            ToolExposure::Direct
        },
    })
}

namespace 影响 ToolName 和 ToolSpec 的 wire 形状；defer_loading 影响 ToolExposure。handler 构造时把 Responses tool 的 defer_loading 清空，因为延迟暴露由 registry/tool-search 层管理。

5. 调用挂起与回传

源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::handle_call、request_dynamic_tool

let args: Value = parse_arguments(&arguments)?;
let response = request_dynamic_tool(
    &session,
    turn.as_ref(),
    call_id,
    self.tool_name.clone(),
    args,
)
.await
.ok_or_else(|| {
    FunctionCallError::RespondToModel(
        "dynamic tool call was cancelled before receiving a response".to_string(),
    )
})?;

let DynamicToolResponse {
    content_items,
    success,
} = response;
let body = content_items
    .into_iter()
    .map(FunctionCallOutputContentItem::from)
    .collect::<Vec<_>>();
Ok(boxed_tool_output(FunctionToolOutput::from_content(
    body,
    Some(success),
)))

handler 解析模型 arguments 后进入 session pending 流程；客户端响应中的文本、图片和音频 content item 被转换成 Responses function output。响应缺失会返回“调用被取消”的模型可见错误。

源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: request_dynamic_tool

let (tx_response, rx_response) = oneshot::channel();
let event_id = call_id.clone();
let prev_entry = {
    let mut active = session.active_turn.lock().await;
    match active.as_mut() {
        Some(at) => {
            let mut ts = at.turn_state.lock().await;
            ts.insert_pending_dynamic_tool(call_id.clone(), tx_response)
        }
        None => None,
    }
};
if prev_entry.is_some() {
    warn!("Overwriting existing pending dynamic tool call for call_id: {event_id}");
}

let started_at = Instant::now();
session
    .emit_turn_item_started(
        turn_context,
        &TurnItem::DynamicToolCall(DynamicToolCallItem {
            id: call_id.clone(),
            namespace: namespace.clone(),
            tool: tool.clone(),
            arguments: arguments.clone(),
            status: DynamicToolCallStatus::InProgress,
            content_items: None,
            success: None,
            error: None,
            duration: None,
        }),
    )
    .await;
let response = rx_response.await.ok();

pending map 以 call_id 为 key，并在等待前发送 InProgress item。重复 call id 会覆盖旧 sender 并记录 warning；没有 active turn 则 response future 不能建立有效等待关系。

6. 执行记录的终态

源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolCallItem 终态构造

let item = match &response {
    Some(response) => DynamicToolCallItem {
        id: call_id,
        namespace,
        tool,
        arguments,
        status: if response.success {
            DynamicToolCallStatus::Completed
        } else {
            DynamicToolCallStatus::Failed
        },
        content_items: Some(response.content_items.clone()),
        success: Some(response.success),
        error: None,
        duration: Some(started_at.elapsed()),
    },
    None => DynamicToolCallItem {
        id: call_id,
        namespace,
        tool,
        arguments,
        status: DynamicToolCallStatus::Failed,
        content_items: Some(Vec::new()),
        success: Some(false),
        error: Some("dynamic tool call was cancelled before receiving a response".to_string()),
        duration: Some(started_at.elapsed()),
    },
};
session
    .emit_turn_item_completed(turn_context, TurnItem::DynamicToolCall(item))
    .await;

success=false 是客户端明确返回失败；response=None 是等待被取消或 sender 关闭。两者都产生 Failed item，但只有后者有取消错误文本。执行时长从 request 注册后开始计时。

7. TurnState与回传

源码位置：codex-rs/core/src/state/turn.rs :: pending_dynamic_tools、clear_pending_waiters、insert_pending_dynamic_tool、remove_pending_dynamic_tool

pub(crate) struct TurnState {
    pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
    // other pending request maps
}

pub(crate) fn clear_pending_waiters(&mut self) {
    self.pending_approvals.clear();
    self.pending_request_permissions.clear();
    self.pending_user_input.clear();
    self.pending_elicitations.clear();
    self.mcp_tool_approval_metadata.clear();
    self.pending_dynamic_tools.clear();
}

pub(crate) fn insert_pending_dynamic_tool(
    &mut self,
    key: String,
    tx: oneshot::Sender<DynamicToolResponse>,
) -> Option<oneshot::Sender<DynamicToolResponse>> {
    self.pending_dynamic_tools.insert(key, tx)
}

pub(crate) fn remove_pending_dynamic_tool(
    &mut self,
    key: &str,
) -> Option<oneshot::Sender<DynamicToolResponse>> {
    self.pending_dynamic_tools.remove(key)
}

turn abort、关闭或完成时统一清理 pending map。回传入口只按 id 移除 sender，不能把动态工具响应投递到另一个 turn。

源码位置：codex-rs/core/src/session/handlers.rs :: dynamic_tool_response

pub async fn dynamic_tool_response(
    sess: &Arc<Session>,
    id: String,
    response: DynamicToolResponse,
) {
    sess.notify_dynamic_tool_response(&id, response).await;
}

8. History与服务端

源码位置：codex-rs/app-server-protocol/src/protocol/thread_history.rs :: handle_dynamic_tool_call_request、handle_dynamic_tool_call_response

fn handle_dynamic_tool_call_request(
    &mut self,
    payload: &codex_protocol::dynamic_tools::DynamicToolCallRequest,
) {
    let item = ThreadItem::DynamicToolCall {
        id: payload.call_id.clone(),
        namespace: payload.namespace.clone(),
        tool: payload.tool.clone(),
        arguments: payload.arguments.clone(),
        status: DynamicToolCallStatus::InProgress,
        content_items: None,
        success: None,
        duration_ms: None,
    };
    if payload.turn_id.is_empty() {
        self.upsert_item_in_current_turn(item);
    } else {
        self.upsert_item_in_turn_id(&payload.turn_id, item);
    }
}

history reducer 先建立 InProgress item，再用同一个 call id 更新终态。空 turn id 使用当前 turn，否则按显式 turn id 定位，说明历史回放和实时事件都必须处理关联缺失情况。

源码位置：codex-rs/app-server-protocol/src/protocol/thread_history.rs :: handle_dynamic_tool_call_response

fn handle_dynamic_tool_call_response(&mut self, payload: &DynamicToolCallResponseEvent) {
    let status = if payload.success {
        DynamicToolCallStatus::Completed
    } else {
        DynamicToolCallStatus::Failed
    };
    let duration_ms = i64::try_from(payload.duration.as_millis()).ok();
    let item = ThreadItem::DynamicToolCall {
        id: payload.call_id.clone(),
        namespace: payload.namespace.clone(),
        tool: payload.tool.clone(),
        arguments: payload.arguments.clone(),
        status,
        content_items: Some(convert_dynamic_tool_content_items(&payload.content_items)),
        success: Some(payload.success),
        duration_ms,
    };
    if payload.turn_id.is_empty() {
        self.upsert_item_in_current_turn(item);
    } else {
        self.upsert_item_in_turn_id(&payload.turn_id, item);
    }
}

App Server history 只做客户端 item 投影：它把动态工具 content item 转为自己的 schema，并把 Duration 转成毫秒。它不执行工具，也不验证客户端业务结果。

9. 实时通知与多媒体

源码位置：codex-rs/app-server-protocol/src/protocol/event_mapping.rs :: item_event_to_server_notification

EventMsg::DynamicToolCallResponse(response) => {
    let status = if response.success {
        DynamicToolCallStatus::Completed
    } else {
        DynamicToolCallStatus::Failed
    };
    let duration_ms = i64::try_from(response.duration.as_millis()).ok();
    let item = ThreadItem::DynamicToolCall {
        id: response.call_id,
        namespace: response.namespace,
        tool: response.tool,
        arguments: response.arguments,
        status,
        content_items: Some(
            response
                .content_items
                .into_iter()
                .map(|item| match item {
                    CoreDynamicToolCallOutputContentItem::InputText { text } => {
                        DynamicToolCallOutputContentItem::InputText { text }
                    }
                    CoreDynamicToolCallOutputContentItem::InputImage { image_url } => {
                        DynamicToolCallOutputContentItem::InputImage { image_url }
                    }
                    CoreDynamicToolCallOutputContentItem::InputAudio { audio_url } => {
                        DynamicToolCallOutputContentItem::InputAudio { audio_url }
                    }
                })
                .collect(),
        ),
        success: Some(response.success),
        duration_ms,
    };
    ServerNotification::ItemCompleted(ItemCompletedNotification {
        thread_id,
        turn_id: response.turn_id,
        item,
        completed_at_ms: response.completed_at_ms,
    })
}

这是无状态的一对一映射，适合实时 notification；history reducer 则需要维护 InProgress→Completed 的状态。两者都使用 call id，但职责不同。

10. 测试与边界

源码位置：codex-rs/core/src/tools/router_tests.rs :: specs_filter_deferred_dynamic_tools；codex-rs/tools/src/dynamic_tool_tests.rs :: 动态工具 schema 转换测试

源码位置：codex-rs/app-server-protocol/src/protocol/thread_history.rs :: reconstructs_dynamic_tool_items_from_request_and_response_events

cd codex-rs
cargo test -p codex-core specs_filter_deferred_dynamic_tools -- --nocapture --test-threads=1
cargo test -p codex-tools dynamic_tool -- --nocapture --test-threads=1
cargo test -p codex-app-server-protocol reconstructs_dynamic_tool_items_from_request_and_response_events -- --nocapture --test-threads=1

这些测试说明 deferred tool 暴露、schema 转换以及 request/response 历史重建；不能证明外部客户端一定执行了工具、响应内容满足业务约束，或所有动态工具都能跨平台注册。

11. 源码定位练习

遇到“模型看得到工具但调用无响应”，先检查 DynamicToolSpec 是否通过 normalize 进入 registry，再检查 DynamicToolHandler 的 namespace/name 和 call_id pending map，最后核对 Op::DynamicToolResponse 的 id。

遇到“历史显示失败”，区分客户端返回 success=false、等待被取消、Turn 清理 pending waiter 和 handler 参数解析失败；遇到“实时 UI 与恢复历史不一致”，分别阅读 stateless event mapper 和 stateful ThreadHistoryBuilder。

动态工具的教学主线是：规格决定暴露，registry 决定可路由，session waiter 决定回传，TurnItem 决定执行记录，history/App Server 决定客户端看到的投影。任何一层缺失，都不能仅凭 schema 或按钮状态推断工具真正完成。