DynamicTool处理器
Dynamic Tool 的“动态”不是运行时凭空执行一段代码，而是客户端在 thread 创建时提供工具规格，Core 把规格转换成可路由的 runtime；模型调用后，Core 暂停当前 handler，把请求交给客户端，再把客户端返回的 content items 包装成普通模型工具输出。工具定义、工具执行和工具结果因此分属不同 owner。
本文面向已经读过ToolPayload调用模型、ToolRegistry数据结构和ToolSearch工具的读者。本文回答一个具体问题：外部定义怎样成为当前 thread 的 registry 条目，schema 和命名空间在哪里被拒绝，模型调用如何经过 oneshot 到达客户端，响应如何回到原 turn，以及取消或非法多媒体结果会留下什么状态。不展开具体 App/MCP 业务实现，也不把 Dynamic Tool 与 Extension Tool 的宿主协议混为一谈。
1. 定义入口
1.1 规范形状
协议层支持两种顶层定义：无 namespace 的 Function，以及带 namespace 的 Namespace。函数定义包含 Responses API 要求的 name、description、input_schema 和 defer_loading；namespace 本身有名称和描述，内部目前只允许 Function 子项。
源码位置：codex-rs/protocol/src/dynamic_tools.rs :: DynamicToolSpec
#[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>,
}defer_loading 只表达暴露方式，不表达客户端是否已经实现了工具。真正的执行者始终是调用 Dynamic Tool 的客户端；Core 只保留规格和等待响应所需的状态。
1.2 Thread持有
Dynamic Tool 在 thread start 时进入 StartThreadOptions，随后保存到 session configuration 和 turn context。创建新 session 时，如果调用方没有提供新列表，Core 会从 rollout metadata 恢复已有列表；因此它的生效范围是 thread/session，而不是某一个 function call。
源码位置：codex-rs/core/src/session/mod.rs :: Session::spawn_internal
// Dynamic tools are defined at thread start and persisted in rollout session metadata.
let dynamic_tools = if dynamic_tools.is_empty() {
conversation_history.get_dynamic_tools().unwrap_or_default()
} else {
dynamic_tools
};这个选择发生在 Session 初始化期间。后续每个 TurnContext 从 session configuration 复制 dynamic_tools，再由 tool plan 为当前 step 建 registry；因此在 turn 中途修改客户端列表不会悄悄改变已经构造的 router。
1.3 App Server校验
App Server 是公开 thread 入口的第一道边界。它先限制 Responses API 可接受的 identifier 字符集和长度，再检查保留 namespace、重复名称、空 namespace、defer_loading 的 namespace 要求以及 input_schema 是否能被工具转换器解析。
源码位置：codex-rs/app-server/src/request_processors/thread_processor.rs :: validate_dynamic_tools
const DYNAMIC_TOOL_NAME_MAX_LEN: usize = 128;
const DYNAMIC_TOOL_NAMESPACE_MAX_LEN: usize = 64;
const DYNAMIC_TOOL_NAMESPACE_DESCRIPTION_MAX_LEN: usize = 1024;
const DYNAMIC_TOOL_IDENTIFIER_PATTERN: &str = "^[a-zA-Z0-9_-]+$";
const RESERVED_RESPONSES_NAMESPACES: &[&str] = &[
"api_tool", "browser", "computer", "container", "file_search",
"functions", "image_gen", "multi_tool_use", "python",
"python_user_visible", "submodel_delegator", "terminal",
"tool_search", "web",
];
fn validate_dynamic_tool_identifier(
value: &str,
label: &str,
max_len: usize,
) -> Result<(), String> {
if !value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
{
return Err(format!(
"{label} must match {DYNAMIC_TOOL_IDENTIFIER_PATTERN} to match Responses API"
));
}
if value.chars().count() > max_len {
return Err(format!(
"{label} must be at most {max_len} characters to match Responses API"
));
}
Ok(())
}校验拒绝发生在 thread 建立前，所以错误是 JSON-RPC invalid request，不会留下一个半初始化的 DynamicTool runtime。测试覆盖非法 schema、同 namespace 重名、空 namespace、保留 namespace、超长 identifier 和 legacy/canonical 混用。
2. 兼容归一
2.1 两种格式
rollout/session metadata 可能来自旧版本的扁平定义：旧格式使用 namespace、name、inputSchema 和 exposeToContext，新格式使用带 type 的 canonical enum。协议归一化先判断整个数组使用哪一种格式；只要同时出现 legacy 字段和 canonical type，就整体报错，不逐项猜测。
源码位置：codex-rs/protocol/src/dynamic_tools.rs :: normalize_dynamic_tool_specs
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();
}这不是单纯的反序列化便利：一致性检查防止同一个列表中 exposeToContext 和 deferLoading 表达相反含义时发生隐式优先级。
2.2 Legacy映射
legacy 的 exposeToContext=true 被映射为 defer_loading=false，false 映射为 true；如果两者都缺失，默认不延迟。随后扁平条目按 namespace 分组，保持输入顺序。
源码位置：codex-rs/protocol/src/dynamic_tools.rs :: normalize_dynamic_tool_specs
let function = DynamicToolFunctionSpec {
name: tool.name,
description: tool.description,
input_schema: tool.input_schema,
defer_loading: tool.defer_loading.unwrap_or_else(|| {
tool.expose_to_context
.map(|visible| !visible)
.unwrap_or(false)
}),
};
Ok((tool.namespace, function))2.3 分组规则
没有 namespace 的函数保持顶层 DynamicToolSpec::Function；有 namespace 的函数被包装为 namespace 子项。同一 namespace 的条目合并到首次出现的 namespace 容器，namespace description 在这个兼容路径中暂为空，后续 handler 会提供默认描述。
源码位置：codex-rs/protocol/src/dynamic_tools.rs :: group_dynamic_tools_by_namespace
for (namespace, function) in tools {
let Some(namespace) = namespace else {
grouped_tools.push(DynamicToolSpec::Function(function));
continue;
};
let function = DynamicToolNamespaceTool::Function(function);
if let Some(index) = namespace_indices.get(&namespace).copied() {
let DynamicToolSpec::Namespace(namespace) = &mut grouped_tools[index] else {
unreachable!("namespace index must point to a namespace");
};
namespace.tools.push(function);
continue;
}
namespace_indices.insert(namespace.clone(), grouped_tools.len());
grouped_tools.push(DynamicToolSpec::Namespace(DynamicToolNamespaceSpec {
name: namespace,
description: String::new(),
tools: vec![function],
}));
}这一步只改变协议形状，不做 Responses identifier 校验；公开 App Server 会先完成更严格的校验，旧 rollout 恢复则依赖归一化后的后续转换。
3. Runtime装配
3.1 名称与规格
DynamicToolHandler::from_parts 同时建立 registry key 和 model-visible spec。namespace 存在时，key 是 ToolName(namespace, tool.name)，spec 是只有一个函数的 namespace；没有 namespace 时，key 和 spec 都是顶层函数。defer_loading 被直接转成 ToolExposure::Deferred 或 Direct。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::from_parts
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()?;
// Exposure controls deferral; tool search restores this marker for deferred results.
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
},
})这里特意把 output_tool.defer_loading 清空：延迟状态由 registry exposure 负责，搜索结果重新序列化时才设置 wire-level defer_loading=true。一个字段不能同时承担内部暴露决策和 Responses 输出标记。
3.2 批量注册
tool plan 遍历当前 turn 的 DynamicToolSpec。无 namespace 和 namespace 子项都转换成独立 handler，再用 register_external 放入 registry；重复 key 会由 registry 记录 collision，而不是静默覆盖已有 runtime。
源码位置：codex-rs/core/src/tools/spec_plan.rs :: append_dynamic_tool_runtimes
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 的结果决定后续能否 dispatch；但 tool spec 是否直接可见还要经过 ToolExposure 和 build_model_visible_specs。因此看到 dynamic tool 在 thread 配置里，并不能直接推断它会出现在首个模型请求。
3.3 搜索暴露
DynamicToolHandler 实现 search_info，所以 defer_loading=true 的 namespace 工具会进入 ToolSearch；直接暴露的工具不会因为实现了 search_info 就自动变成 Deferred。两者分别由 exposure() 和 search_info() 管理。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::search_info
fn exposure(&self) -> ToolExposure {
self.exposure
}
fn search_info(&self) -> Option<ToolSearchInfo> {
ToolSearchInfo::from_tool_spec(
self.spec(),
Some(ToolSearchSourceInfo {
name: "Dynamic tools".to_string(),
description: Some("Tools provided by the current Codex thread.".to_string()),
}),
)
}Dynamic handler 没有 immutable_spec，因为它随 thread/Turn 配置重建。ToolSearch cache 因此把它归为 dynamic source，每次从 runtime 取得 ToolSearchInfo 并按值比较。相同 spec/description 的新 handler 可以复用索引；description、schema、namespace 或 exposure 变化会重建。它与 MCP 的 Weak runtime identity 是两种不同 cache 规则。
源码位置：codex-rs/core/src/tools/handlers/tool_search.rs :: ToolSearchHandlerCache::get_or_build
if tool.runtime.immutable_spec().is_some() {
Some(ToolSearchSource::Immutable(
Arc::downgrade(&tool.runtime),
))
} else {
tool.runtime
.search_info()
.map(Box::new)
.map(ToolSearchSource::Dynamic)
}4. 调用等待
4.1 参数转发
handler 只接受 ToolPayload::Function，使用通用 parse_arguments 转成 serde_json::Value，不在 DynamicToolHandler 内重复应用 input schema。schema 已在 thread 入口转换时验证；运行时这里关注的是把模型原始参数交给客户端。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::handle_call
let arguments = match payload {
ToolPayload::Function { arguments } => arguments,
_ => {
return Err(FunctionCallError::RespondToModel(
"dynamic tool handler received unsupported payload".to_string(),
));
}
};
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(),
)
})?;call_id 和 tool_name 被传入等待函数，保证客户端请求能同时表达“调用哪一个工具”和“哪个 pending sender 应该被唤醒”。
4.2 Pending所有者
request_dynamic_tool 在发送事件前，把 oneshot::Sender<DynamicToolResponse> 放进当前 active turn 的 TurnState::pending_dynamic_tools，key 是 call id。注册和 active turn 检查在同一个锁域内完成，避免事件已经发出但 sender 还没进入 map。
源码位置：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}");
}同一个 call id 的第二次注册会覆盖旧 sender 并产生 warning；因此 call id 必须由模型/Responses 保证唯一。TurnState::clear_pending_waiters 会清理该 map，使所有未完成的 receiver 结束。
4.3 请求事件
sender 注册后，Core 发出 started item。这个 item 同时是内部事件和历史可消费数据：它包含 namespace、tool、解析后的 arguments、InProgress 状态和空结果字段。客户端收到的是事件投影，不是直接拿到 oneshot。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: request_dynamic_tool
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();事件发送完成只是“客户端可见”的屏障，不代表外部工具已经完成。真正让 handler 继续的是 rx_response.await。
5. 结果回灌
5.1 Completed item
收到 response 后，Core 根据 success 选择 Completed 或 Failed，复制 content items，记录耗时；取消时 response 为 None，则构造 failed item、空 content、success=false 和固定取消错误。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: request_dynamic_tool
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;这个 item 是客户端和历史的生命周期结果；它和稍后返回模型的 FunctionToolOutput 共享 content items，但状态字段和耗时只存在于 item 层。
5.2 模型输出
handler 将 DynamicToolResponse.content_items 转成 protocol model 的 FunctionCallOutputContentItem，使用 success 作为 function output 成功标志。外部客户端返回失败并不等于 Core handler error；模型仍会收到一个结构化失败结果。
源码位置：codex-rs/core/src/tools/handlers/dynamic.rs :: DynamicToolHandler::handle_call
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),
)))普通文本、Data URL 图片和 inline Data URL 音频都可以成为 content item；App Server 会在客户端响应边界拒绝远程图片 URL 和非 Data URL 音频，防止外部响应把未允许的网络资源直接带进模型上下文。
5.3 Legacy事件
协议层把 started item 投影为 DynamicToolCallRequest，把 completed/failed item 投影为 response event；InProgress 不会产生 response event。这个投影保证旧客户端仍能看到请求/响应生命周期，而新 App Server 使用 item notification 和 server request。
源码位置：codex-rs/protocol/src/legacy_events.rs :: DynamicToolCallItem
pub(crate) fn as_legacy_request_event(
&self,
turn_id: String,
started_at_ms: i64,
) -> EventMsg {
EventMsg::DynamicToolCallRequest(DynamicToolCallRequest {
call_id: self.id.clone(),
turn_id,
started_at_ms,
namespace: self.namespace.clone(),
tool: self.tool.clone(),
arguments: self.arguments.clone(),
})
}
pub(crate) fn as_legacy_response_event(
&self,
turn_id: String,
completed_at_ms: i64,
) -> Option<EventMsg> {
if matches!(self.status, DynamicToolCallStatus::InProgress) {
return None;
}
Some(EventMsg::DynamicToolCallResponse(/* ... */))
}5.4 Hook时序
Dynamic Tool 使用 ToolPayload::Function，而 DynamicToolHandler 没有覆盖 Hook 方法，因此 registry 的默认 PreToolUse/PostToolUse 投影会生效。PreToolUse 在 pending sender 和 InProgress item 之前运行，可阻断或重写完整 arguments JSON； 当客户端返回 success=true 时，PostToolUse 在 response 已回到 Core、Completed item 已发送后运行。
客户端返回 success=false 时，FunctionToolOutput::success_for_logging() 也为 false，registry 不构造 PostToolUse payload。因此失败结果会直接回灌模型，不能依赖 PostToolUse 对失败响应做第二次过滤。
源码位置：
codex-rs/core/src/tools/registry.rs :: ToolRegistry::dispatchcodex-rs/core/src/tools/context.rs :: FunctionToolOutput::success_for_logging
let success = match &result {
Ok(result) => result.result.success_for_logging(),
Err(_) => false,
};
let post_tool_use_payload = if success {
result
.as_ref()
.ok()
.and_then(|result| result.post_tool_use_payload.clone())
} else {
None
};即使成功结果触发了 PostToolUse，block 也只能拒绝 FunctionCallOutput 回灌，不能撤销客户端调用、completed history item 或外部副作用。需要阻止客户端工具实际执行，应在 PreToolUse 阶段阻断。
6. 客户端回路
6.1 App Server请求
App Server 在收到 core 的 ItemStarted(DynamicToolCall) 后，先广播 item notification，再创建一个 ServerRequest::DynamicToolCall。请求的 call id、turn id、namespace、tool 和 arguments 都来自 Core item；App Server 不重新从 thread 配置查找工具定义。
旧 EventMsg::DynamicToolCallRequest/Response 仍可被 raw-event 和 rollout compatibility consumer 看到，但 bespoke v2 handler 明确忽略 它们；真正的 server request 只从 canonical ItemStarted 产生，避免 legacy event 再发一次客户端调用。
源码位置：codex-rs/app-server/src/bespoke_event_handling.rs :: EventMsg::ItemStarted
let dynamic_tool_call_params = match &event.item {
CoreTurnItem::DynamicToolCall(item) => Some(DynamicToolCallParams {
thread_id: conversation_id.to_string(),
turn_id: event.turn_id.clone(),
call_id: item.id.clone(),
namespace: item.namespace.clone(),
tool: item.tool.clone(),
arguments: item.arguments.clone(),
}),
_ => None,
};
if let Some(params) = dynamic_tool_call_params {
let call_id = params.call_id.clone();
let (_pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::DynamicToolCall(params))
.await;
tokio::spawn(async move {
crate::dynamic_tools::on_call_response(call_id, rx, conversation).await;
});
}发送 server request 后，Core handler 仍在等待自己的 oneshot；App Server 的 rx 是另一层 request waiter。两层 waiter 通过 call id 串起来，但不共享 sender。
6.2 响应解码
App Server 收到客户端 JSON 后，先反序列化 DynamicToolCallResponse，再检查图片必须是 inline Data URL、音频也必须是 Data URL。非法内容被替换成 success=false 的文本 fallback，然后统一提交 Op::DynamicToolResponse。
源码位置：codex-rs/app-server/src/dynamic_tools.rs :: decode_response
match serde_json::from_value::<DynamicToolCallResponse>(value) {
Ok(response)
if response.content_items.iter().any(|item| {
matches!(
item,
DynamicToolCallOutputContentItem::InputImage { image_url }
if is_remote_image_url(image_url)
)
}) => fallback_response(REMOTE_IMAGE_URL_ERROR),
Ok(response)
if response.content_items.iter().any(|item| {
matches!(
item,
DynamicToolCallOutputContentItem::InputAudio { audio_url }
if !audio_url
.get(.."data:".len())
.is_some_and(|prefix| prefix.eq_ignore_ascii_case("data:"))
)
}) => fallback_response(INVALID_AUDIO_URL_ERROR),
Ok(response) => (response, None),
Err(_) => fallback_response("dynamic tool response was invalid"),
}这里的 .."data:".len() 是 Rust 的范围到前缀长度写法，用来安全取得字符串开头的 data: 片段。为避免把校验规则隐藏在注释里，真正的调用流程如下。
源码位置：codex-rs/app-server/src/dynamic_tools.rs :: on_call_response
let response = receiver.await;
let (response, _error) = match response {
Ok(Ok(value)) => decode_response(value),
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(_)) | Err(_) => fallback_response("dynamic tool request failed"),
};
let core_response = CoreDynamicToolResponse {
content_items: response
.content_items
.into_iter()
.map(CoreDynamicToolCallOutputContentItem::from)
.collect(),
success: response.success,
};
if let Err(err) = conversation
.submit(Op::DynamicToolResponse {
id: call_id.clone(),
response: core_response,
})
.await
{
error!("failed to submit DynamicToolResponse: {err}");
}客户端 JSON 错误不会让 Core 的 handler 永远等待；fallback response 会正常唤醒原 sender，并让模型看到一个可解释的失败内容。
6.3 Core回收
Op::DynamicToolResponse 最终进入 Session::notify_dynamic_tool_response，从当前 active turn 的 pending map 移除 sender 后发送 response。没有匹配 call id 时只记录 warning；迟到响应不会写入下一个 turn。
源码位置：codex-rs/core/src/session/mod.rs :: Session::notify_dynamic_tool_response
pub async fn notify_dynamic_tool_response(
&self,
call_id: &str,
response: DynamicToolResponse,
) {
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
ts.remove_pending_dynamic_tool(call_id)
}
None => None,
}
};
match entry {
Some(tx_response) => {
tx_response.send(response).ok();
}
None => {
warn!("No pending dynamic tool call found for call_id: {call_id}");
}
}
}7. 实时历史投影
Dynamic Tool 的 completed item 还可能进入 App Server 的实时会话历史，但这里存在比普通 thread history 更严格的筛选。只有当前存在 active realtime session、item 已完成、状态为 Completed 且 success == Some(true) 时，才把整个 Dynamic Tool item 提升为 BemItemPromoted。失败调用、仅 started 的调用和 realtime session 关闭后的迟到完成都不会提升。
源码位置：codex-rs/app-server/src/realtime_history.rs :: RealtimeHistoryState::observe_item
TurnItem::DynamicToolCall(call)
if self.active_session_id.is_some()
&& completed
&& call.status == DynamicToolCallStatus::Completed
&& call.success == Some(true) =>
{
self.add_promotion(
items,
turn_id,
&call.id,
BemItemPresentation::WholeItem,
);
}add_promotion 不只是追加一个引用。它先按 presentation key 去重，再封口当前 user/assistant transcript segment，最后发出带 turn_id、item_id 和 WholeItem presentation 的 realtime item。工具前后的语音文本因此属于两个 segment，UI 可以把工具条目插在两段转录之间，而不是把工具结果混入一段连续文本。
源码位置：codex-rs/app-server/src/realtime_history.rs :: RealtimeHistoryState::add_promotion
if !self
.promoted_bem_presentation_keys
.insert(presentation_key)
{
return;
}
self.seal_segments(items, Continuation::Continue);
items.push(RealtimeItem {
id: Uuid::now_v7().to_string(),
realtime_session_id,
content: RealtimeItemContent::BemItemPromoted {
turn_id: turn_id.to_string(),
item_id: item_id.to_string(),
presentation,
},
});8. 取消边界
8.1 Turn清理
取消或 turn transition 会调用 TurnState::clear_pending_waiters，其中包括 pending_dynamic_tools.clear()。sender 被 drop 后，rx_response.await.ok() 得到 None，Core 仍发送一个 failed completed item，handler 再返回 RespondToModel 的取消消息。
源码位置：codex-rs/core/src/state/turn.rs :: TurnState::clear_pending_waiters
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();
}8.2 App Server取消
App Server 的 thread-scoped outgoing sender 还维护自己的 server request waiter。取消 thread 时，DynamicToolCall waiter 会收到内部错误并从 pending request 列表移除；这与 Core turn-state 清理是两层回收，缺一都会造成一个层面的悬挂。
源码位置：codex-rs/app-server/src/outgoing_message.rs :: cancel_requests_for_thread 测试
outgoing
.cancel_requests_for_thread(thread_id, Some(error.clone()))
.await;
let dynamic_tool_result = timeout(Duration::from_secs(1), dynamic_tool_waiter)
.await
.expect("dynamic tool waiter should resolve")
.expect("dynamic tool waiter should receive a callback");
assert_eq!(dynamic_tool_result, Err(error.clone()));
assert!(
outgoing
.pending_requests_for_thread(thread_id)
.await
.is_empty()
);9. 验证路径
App Server round-trip 测试以 namespace、tool、arguments 启动调用，先断言 started item 为 InProgress，再从 server request 取出相同字段，提交 text/image/audio content，最后检查 completed item 和下一次模型请求的 function output。它验证的是两层 request waiter、生命周期事件和模型回灌的关系。
源码位置：codex-rs/app-server/tests/suite/v2/dynamic_tools.rs :: dynamic_tool_call_round_trip_sends_text_content_items_to_model
let started = wait_for_dynamic_tool_started(&mut mcp, call_id).await?;
let ThreadItem::DynamicToolCall {
namespace,
tool,
arguments,
status,
..
} = started.item
else {
panic!("expected dynamic tool call item");
};
assert_eq!(status, DynamicToolCallStatus::InProgress);
assert_eq!(namespace.as_deref(), Some(tool_namespace));
assert_eq!(tool, tool_name);
assert_eq!(arguments, tool_args);
let request = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
)
.await??;
let ServerRequest::DynamicToolCall { params, .. } = request else {
panic!("expected dynamic tool call request");
};
assert_eq!(params.call_id, call_id);
assert_eq!(params.arguments, tool_args);非法输入测试覆盖三类边界：定义混用 canonical/legacy、隐藏顶层 Deferred 工具没有 namespace、以及 namespace/identifier/schema 违反 Responses API 约束。远程图片和远程音频测试则证明客户端返回的非法多媒体会被转换成模型可见文本失败。
源码位置：codex-rs/app-server/tests/suite/v2/dynamic_tools.rs :: dynamic_tool_remote_image_response_becomes_model_visible_error
let response = DynamicToolCallResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputImage {
image_url: "https://example.com/tool.png".to_string(),
}],
success: true,
};
mcp.send_response(request_id, serde_json::to_value(response)?).await?;
let completed = wait_for_dynamic_tool_completed(&mut mcp, call_id).await?;
assert_eq!(completed.status, DynamicToolCallStatus::Failed);
assert_eq!(completed.success, Some(false));这些测试证明的是当前协议边界和 mock App Server 的回路，不证明第三方客户端会主动遵守 content item 约束；Core/App Server 的 decode fallback 才是最终保护。
10. 源码练习
先复述完整路径：thread/start 的 DynamicToolSpec → validate_dynamic_tools/legacy normalize → Session/TurnContext snapshot → append_dynamic_tool_runtimes → DynamicToolHandler → pending sender → DynamicToolCall request → Op::DynamicToolResponse → completed item 与 FunctionCallOutput。
再做两个只读定位：
- 找到
thread_start_rejects_hidden_dynamic_tools_without_namespace，解释为什么defer_loading=true的顶层函数会被拒绝，而同一函数放进 namespace 后可以进入 ToolSearch； - 找到
dynamic_tool_remote_image_response_becomes_model_visible_error，指出错误发生在 App Server 的decode_response还是 Core handler，并说明为什么 completed item 仍然是Failed而不是取消； - 找到
promotes_successful_dynamic_tools_and_splits_active_transcripts，解释 failed item、重复 completed item 和 realtime session 关闭后的迟到 item 为什么都不会生成新的 promotion。
在 Codex 源码仓库根目录运行：
rg -n "DynamicToolHandler|validate_dynamic_tools|request_dynamic_tool|on_call_response" codex-rs/core codex-rs/app-server codex-rs/protocol
RUST_MIN_STACK=8388608 cargo test -p codex-app-server --test all dynamic_tool_call_round_trip_sends_text_content_items_to_model -- --test-threads=1
RUST_MIN_STACK=8388608 cargo test -p codex-app-server --test all dynamic_tool_remote_image_response_becomes_model_visible_error -- --test-threads=1
cargo test -p codex-app-server realtime_history_tests::promotes_successful_dynamic_tools_and_splits_active_transcripts -- --exact第一个测试检查正常请求/响应/模型回灌，第二个检查非法图片响应的降级边界，第三个检查 realtime promotion、转录切分和去重；这些测试都不把外部客户端的业务逻辑当成 Core 已知事实。