diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index ca80687ffc..d4f8211d59 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -239,7 +239,7 @@ pub(crate) async fn handle_mcp_tool_call( let status = if result.is_ok() { "ok" } else { "error" }; let outcome = McpCallMetricOutcome::from_status(status); emit_mcp_call_metrics( - turn_context.as_ref(), + &step_context.session_telemetry, &outcome, &server, &tool_name, @@ -348,7 +348,7 @@ pub(crate) async fn handle_mcp_tool_call( let status = if result.is_ok() { "ok" } else { "error" }; let outcome = McpCallMetricOutcome::from_status(status); emit_mcp_call_metrics( - turn_context.as_ref(), + &step_context.session_telemetry, &outcome, &server, &tool_name, @@ -489,7 +489,7 @@ async fn handle_approved_mcp_tool_call( .map(|(connector_id, action_name)| HostedFileUploadContext { connector_id: connector_id.clone(), action_name: action_name.clone(), - model: turn_context.model_info().slug.clone(), + model: step_context.settings.model_info.slug.clone(), }); let rewritten_arguments = rewrite_mcp_tool_arguments_for_openai_files( sess, @@ -545,7 +545,7 @@ async fn handle_approved_mcp_tool_call( ) .await; let result = sanitize_mcp_tool_result_for_model( - &turn_context.model_info().input_modalities, + &step_context.settings.model_info.input_modalities, Ok(result), )?; Ok(maybe_request_codex_apps_auth_elicitation( @@ -590,11 +590,11 @@ async fn handle_approved_mcp_tool_call( truncate_mcp_tool_result_for_event(&result), ) .await; - maybe_track_codex_app_used(sess, turn_context, &server, &metadata).await; + maybe_track_codex_app_used(sess, step_context, &server, &metadata).await; let outcome = mcp_call_metric_outcome(&result); emit_mcp_call_metrics( - turn_context, + &step_context.session_telemetry, &outcome, &server, &tool_name, @@ -1055,7 +1055,7 @@ async fn notify_mcp_tool_call_completed( async fn maybe_track_codex_app_used( sess: &Session, - turn_context: &TurnContext, + step_context: &StepContext, server: &str, metadata: &McpToolApprovalMetadata, ) { @@ -1075,8 +1075,9 @@ async fn maybe_track_codex_app_used( InvocationType::Implicit }; + let turn_context = &step_context.turn; let tracking = build_track_events_context( - turn_context.model_info().slug.clone(), + step_context.settings.model_info.slug.clone(), sess.thread_id.to_string(), turn_context.sub_id.clone(), turn_context.originator.clone(), diff --git a/codex-rs/core/src/mcp_tool_call/telemetry.rs b/codex-rs/core/src/mcp_tool_call/telemetry.rs index 5f3600264b..1614e88a72 100644 --- a/codex-rs/core/src/mcp_tool_call/telemetry.rs +++ b/codex-rs/core/src/mcp_tool_call/telemetry.rs @@ -1,7 +1,7 @@ use std::time::Duration; -use crate::session::turn_context::TurnContext; use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY; +use codex_otel::SessionTelemetry; use codex_otel::sanitize_metric_tag_value; use codex_protocol::mcp::CallToolResult; use serde_json::Value as JsonValue; @@ -38,7 +38,7 @@ impl McpCallMetricOutcome { } pub(super) fn emit_mcp_call_metrics( - turn_context: &TurnContext, + session_telemetry: &SessionTelemetry, outcome: &McpCallMetricOutcome, server_name: &str, tool_name: &str, @@ -57,15 +57,9 @@ pub(super) fn emit_mcp_call_metrics( .iter() .map(|(key, value)| (*key, value.as_str())) .collect(); - turn_context - .session_telemetry - .counter(MCP_CALL_COUNT_METRIC, /*inc*/ 1, &tag_refs); + session_telemetry.counter(MCP_CALL_COUNT_METRIC, /*inc*/ 1, &tag_refs); if let Some(duration) = duration { - turn_context.session_telemetry.record_duration( - MCP_CALL_DURATION_METRIC, - duration, - &tag_refs, - ); + session_telemetry.record_duration(MCP_CALL_DURATION_METRIC, duration, &tag_refs); } let (Some(error_type), Some(error_code)) = (outcome.error_type, outcome.error_code.as_deref()) @@ -79,11 +73,7 @@ pub(super) fn emit_mcp_call_metrics( .iter() .map(|(key, value)| (*key, value.as_str())) .collect(); - turn_context.session_telemetry.counter( - MCP_CALL_ERROR_COUNT_METRIC, - /*inc*/ 1, - &error_tag_refs, - ); + session_telemetry.counter(MCP_CALL_ERROR_COUNT_METRIC, /*inc*/ 1, &error_tag_refs); } fn mcp_call_metric_tags( diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index b2189a6e49..0c83f723fe 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3670,13 +3670,10 @@ impl Session { }); extension_data.insert(selected_plugins.clone()); turn_context.extension_data.insert(selected_plugins); - // Tool availability still follows the admitted turn; the async message - // description comes from the captured step model. let tool_router = turn::built_tools( self.as_ref(), turn_context.as_ref(), - // TODO(CDXENT-441): use the step scoped model - turn_context.model_info(), + &settings.model_info, settings.model_info.model_messages.as_ref(), &environments, &mcp, @@ -3685,8 +3682,7 @@ impl Session { ) .or_cancel(cancellation_token) .await??; - // Publish inventory after planning rather than during finalization, so constructing - // additional candidate plans cannot overwrite turn-wide metadata. + // Publish inventory only after the step's tool plan has been finalized. if turn_context .config .tool_registry diff --git a/codex-rs/core/src/tools/code_mode/execute_handler.rs b/codex-rs/core/src/tools/code_mode/execute_handler.rs index e0cc305048..ac0383a9d2 100644 --- a/codex-rs/core/src/tools/code_mode/execute_handler.rs +++ b/codex-rs/core/src/tools/code_mode/execute_handler.rs @@ -33,7 +33,7 @@ impl CodeModeExecuteHandler { async fn execute( &self, session: std::sync::Arc, - turn: std::sync::Arc, + step_context: std::sync::Arc, call_id: String, originating_item_id: Option, code: String, @@ -41,7 +41,10 @@ impl CodeModeExecuteHandler { ) -> Result { let args = codex_code_mode::parse_exec_source(&code).map_err(FunctionCallError::RespondToModel)?; - let exec = ExecContext { session, turn }; + let exec = ExecContext { + session, + turn: Arc::clone(&step_context.turn), + }; let mut enabled_tools = Vec::with_capacity(self.nested_tool_specs.len()); for (spec, cached_runtime) in &self.nested_tool_specs { if let Some(cached_definitions) = cached_runtime @@ -138,9 +141,14 @@ impl CodeModeExecuteHandler { let wall_time = response .code_mode_host_duration() .unwrap_or_else(|| started_at.elapsed()); - handle_runtime_response(&exec, response, args.max_output_tokens, wall_time) - .await - .map_err(FunctionCallError::RespondToModel) + handle_runtime_response( + &step_context.settings.model_info, + response, + args.max_output_tokens, + wall_time, + ) + .await + .map_err(FunctionCallError::RespondToModel) } } @@ -170,6 +178,7 @@ impl CodeModeExecuteHandler { let ToolInvocation { session, turn, + step_context, call_id, tool_name, payload, @@ -188,7 +197,7 @@ impl CodeModeExecuteHandler { ToolPayload::Custom { input } if is_exec_tool_name(&tool_name) => self .execute( session, - turn, + step_context, call_id, originating_item_id, input, diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 7bb9459097..7db0e8e82b 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -248,24 +248,25 @@ impl CodeModeService { } pub(super) async fn handle_runtime_response( - exec: &ExecContext, + model_info: &codex_protocol::openai_models::ModelInfo, response: RuntimeResponse, max_output_tokens: Option, wall_time: Duration, ) -> Result { let script_status = format_script_status(&response); + let supports_original = can_request_original_image_detail(model_info); match response { RuntimeResponse::Yielded { content_items, .. } => { let mut content_items = into_function_call_output_content_items(content_items); - sanitize_runtime_image_detail(exec.turn.as_ref(), &mut content_items); + sanitize_image_detail_items(supports_original, &mut content_items); content_items = truncate_code_mode_result(content_items, max_output_tokens); prepend_script_status(&mut content_items, &script_status, wall_time); Ok(FunctionToolOutput::from_content(content_items, Some(true))) } RuntimeResponse::Terminated { content_items, .. } => { let mut content_items = into_function_call_output_content_items(content_items); - sanitize_runtime_image_detail(exec.turn.as_ref(), &mut content_items); + sanitize_image_detail_items(supports_original, &mut content_items); content_items = truncate_code_mode_result(content_items, max_output_tokens); prepend_script_status(&mut content_items, &script_status, wall_time); Ok(FunctionToolOutput::from_content(content_items, Some(true))) @@ -276,7 +277,7 @@ pub(super) async fn handle_runtime_response( .. } => { let mut content_items = into_function_call_output_content_items(content_items); - sanitize_runtime_image_detail(exec.turn.as_ref(), &mut content_items); + sanitize_image_detail_items(supports_original, &mut content_items); let success = error_text.is_none(); if let Some(error_text) = error_text { content_items.push(FunctionCallOutputContentItem::InputText { @@ -293,10 +294,6 @@ pub(super) async fn handle_runtime_response( } } -fn sanitize_runtime_image_detail(turn: &TurnContext, items: &mut [FunctionCallOutputContentItem]) { - sanitize_image_detail_items(can_request_original_image_detail(turn.model_info()), items); -} - fn format_script_status(response: &RuntimeResponse) -> String { match response { RuntimeResponse::Yielded { cell_id, .. } => { diff --git a/codex-rs/core/src/tools/code_mode/wait_handler.rs b/codex-rs/core/src/tools/code_mode/wait_handler.rs index c6880799c9..1e9e18f6d8 100644 --- a/codex-rs/core/src/tools/code_mode/wait_handler.rs +++ b/codex-rs/core/src/tools/code_mode/wait_handler.rs @@ -70,6 +70,7 @@ impl CodeModeWaitHandler { let ToolInvocation { session, turn, + step_context, call_id, tool_name, payload, @@ -159,10 +160,15 @@ impl CodeModeWaitHandler { let wall_time = wait_response .code_mode_host_duration() .unwrap_or_else(|| started_at.elapsed()); - handle_runtime_response(&exec, wait_response.into(), args.max_tokens, wall_time) - .await - .map_err(FunctionCallError::RespondToModel) - .map(boxed_tool_output) + handle_runtime_response( + &step_context.settings.model_info, + wait_response.into(), + args.max_tokens, + wall_time, + ) + .await + .map_err(FunctionCallError::RespondToModel) + .map(boxed_tool_output) } _ => Err(FunctionCallError::RespondToModel(format!( "{WAIT_TOOL_NAME} expects JSON arguments" diff --git a/codex-rs/core/src/tools/events.rs b/codex-rs/core/src/tools/events.rs index 5fa91cc82c..8da8b05288 100644 --- a/codex-rs/core/src/tools/events.rs +++ b/codex-rs/core/src/tools/events.rs @@ -18,6 +18,7 @@ use codex_protocol::items::CommandExecutionItem; use codex_protocol::items::CommandExecutionStatus; use codex_protocol::items::FileChangeItem; use codex_protocol::items::TurnItem; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::parse_command::ParsedCommand; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandSource; @@ -44,6 +45,8 @@ pub(super) fn truncate_rejection_message(message: &str) -> String { pub(crate) struct ToolEventCtx<'a> { pub session: &'a Session, pub turn: &'a TurnContext, + /// Model captured by the step that issued this call, including delayed completion events. + pub model_info: &'a ModelInfo, pub call_id: &'a str, pub turn_diff_tracker: Option<&'a SharedTurnDiffTracker>, } @@ -52,12 +55,14 @@ impl<'a> ToolEventCtx<'a> { pub fn new( session: &'a Session, turn: &'a TurnContext, + model_info: &'a ModelInfo, call_id: &'a str, turn_diff_tracker: Option<&'a SharedTurnDiffTracker>, ) -> Self { Self { session, turn, + model_info, call_id, turn_diff_tracker, } @@ -117,12 +122,17 @@ async fn emit_exec_command_begin(ctx: ToolEventCtx<'_>, exec_input: &ExecCommand ("output_format", operation.output_format), ("execution_backend", "unified_exec"), ]; - ctx.turn.session_telemetry.counter( + let session_telemetry = ctx + .turn + .session_telemetry + .clone() + .with_model(&ctx.model_info.slug, &ctx.model_info.slug); + session_telemetry.counter( ARTIFACT_OPERATION_STARTED_METRIC, /*inc*/ 1, &metric_tags, ); - ctx.turn.session_telemetry.histogram( + session_telemetry.histogram( ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC, i64::from(operation.expected_output_count), &metric_tags, @@ -132,7 +142,7 @@ async fn emit_exec_command_begin(ctx: ToolEventCtx<'_>, exec_input: &ExecCommand .analytics_events_client .track_artifact_operation( build_track_events_context( - ctx.turn.model_info().slug.clone(), + ctx.model_info.slug.clone(), ctx.session.thread_id.to_string(), ctx.turn.sub_id.clone(), ctx.turn.originator.clone(), @@ -373,7 +383,7 @@ impl ToolEmitter { output: &ExecToolCallOutput, ctx: ToolEventCtx<'_>, ) -> String { - super::format_exec_output_for_model(output, ctx.turn.model_info().truncation_policy.into()) + super::format_exec_output_for_model(output, ctx.model_info.truncation_policy.into()) } pub async fn finish( @@ -521,7 +531,7 @@ async fn emit_exec_stage( duration: output.duration, formatted_output: format_exec_output_str( &output, - ctx.turn.model_info().truncation_policy.into(), + ctx.model_info.truncation_policy.into(), ), status: if output.exit_code == 0 { ExecCommandStatus::Completed @@ -697,7 +707,13 @@ mod tests { environment_id: None, } .finish( - ToolEventCtx::new(session.as_ref(), turn.as_ref(), "call-id", Some(&tracker)), + ToolEventCtx::new( + session.as_ref(), + turn.as_ref(), + turn.model_info(), + "call-id", + Some(&tracker), + ), out, Some(&delta), ) @@ -781,7 +797,13 @@ mod tests { .expect("apply patch"); emit_patch_end( - ToolEventCtx::new(session.as_ref(), turn.as_ref(), "call-id", Some(&tracker)), + ToolEventCtx::new( + session.as_ref(), + turn.as_ref(), + turn.model_info(), + "call-id", + Some(&tracker), + ), HashMap::new(), String::new(), String::new(), @@ -830,7 +852,13 @@ mod tests { tracker.lock().await.track_delta("", &delta); emit_patch_end( - ToolEventCtx::new(session.as_ref(), turn.as_ref(), "call-id", Some(&tracker)), + ToolEventCtx::new( + session.as_ref(), + turn.as_ref(), + turn.model_info(), + "call-id", + Some(&tracker), + ), HashMap::new(), String::new(), String::new(), diff --git a/codex-rs/core/src/tools/handlers/apply_patch.rs b/codex-rs/core/src/tools/handlers/apply_patch.rs index 0f5288995b..5ab6666283 100644 --- a/codex-rs/core/src/tools/handlers/apply_patch.rs +++ b/codex-rs/core/src/tools/handlers/apply_patch.rs @@ -589,6 +589,7 @@ async fn execute_verified_patch( let event_ctx = ToolEventCtx::new( tool_ctx.session.as_ref(), tool_ctx.step_context.turn.as_ref(), + &tool_ctx.step_context.settings.model_info, &tool_ctx.call_id, tracker, ); @@ -616,6 +617,7 @@ async fn execute_verified_patch( let event_ctx = ToolEventCtx::new( tool_ctx.session.as_ref(), tool_ctx.step_context.turn.as_ref(), + &tool_ctx.step_context.settings.model_info, &tool_ctx.call_id, tracker, ); diff --git a/codex-rs/core/src/tools/handlers/extension_tools.rs b/codex-rs/core/src/tools/handlers/extension_tools.rs index 7998de0c1d..92a30d1207 100644 --- a/codex-rs/core/src/tools/handlers/extension_tools.rs +++ b/codex-rs/core/src/tools/handlers/extension_tools.rs @@ -166,13 +166,14 @@ impl TurnItemEmitter for CoreTurnItemEmitter { async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_> { let conversation_history = ConversationHistory::new(invocation.session.clone_history().await.into_raw_items()); + let settings = &invocation.step_context.settings; let codex_turn_metadata = invocation .turn .turn_metadata_state .current_meta_value_for_mcp_request(McpTurnMetadataContext { - model: invocation.turn.model_info().slug.as_str(), - reasoning_effort: invocation.turn.effective_reasoning_effort(), - node_repl_disabled: invocation.turn.model_info().node_repl_disabled, + model: settings.model_info.slug.as_str(), + reasoning_effort: settings.effective_reasoning_effort(), + node_repl_disabled: settings.model_info.node_repl_disabled, }) .and_then(|metadata| to_ascii_json_string(&metadata).ok()); let mut environments = Vec::new(); @@ -204,9 +205,9 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_> turn_id: invocation.turn.sub_id.clone(), call_id: invocation.call_id.clone(), tool_name: invocation.tool_name.clone(), - model: invocation.turn.model_info().slug.clone(), + model: settings.model_info.slug.clone(), codex_turn_metadata, - truncation_policy: invocation.turn.model_info().truncation_policy.into(), + truncation_policy: settings.model_info.truncation_policy.into(), source: extension_tool_call_source(invocation.source.clone()), conversation_history, turn_item_emitter: Arc::new(CoreTurnItemEmitter { @@ -417,7 +418,7 @@ mod tests { } #[tokio::test] - async fn passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call() { + async fn passes_step_node_repl_flag_and_scoped_turn_item_emitter_to_extension_call() { let captured_call = Arc::new(Mutex::new(None)); let captured_sandbox_cwds = Arc::new(Mutex::new(Vec::new())); let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor { @@ -464,7 +465,15 @@ mod tests { strip_response_item_id(raw_history_item.item), expected_history_item ); - let step_context = StepContext::for_test(Arc::clone(&turn)); + let mut step_context = StepContext::for_test(Arc::clone(&turn)); + let settings = Arc::make_mut( + &mut Arc::get_mut(&mut step_context) + .expect("unshared step") + .settings, + ); + let model_info = Arc::make_mut(&mut settings.model_info); + model_info.node_repl_disabled = !turn.model_info().node_repl_disabled; + let node_repl_disabled = model_info.node_repl_disabled; let invocation = ToolInvocation { session, step_context, @@ -497,6 +506,14 @@ mod tests { ); assert_eq!(captured_call.model, model); assert_eq!(captured_call.truncation_policy, truncation_policy); + let metadata: serde_json::Value = serde_json::from_str( + captured_call + .codex_turn_metadata + .as_deref() + .expect("turn metadata"), + ) + .expect("metadata JSON"); + assert_eq!(metadata["node_repl_disabled"], json!(node_repl_disabled)); assert_eq!( captured_call.source, ExtensionToolCallSource::CodeMode { diff --git a/codex-rs/core/src/tools/handlers/mcp.rs b/codex-rs/core/src/tools/handlers/mcp.rs index 7ddbb2f921..5a7642085b 100644 --- a/codex-rs/core/src/tools/handlers/mcp.rs +++ b/codex-rs/core/src/tools/handlers/mcp.rs @@ -207,7 +207,6 @@ impl McpHandler { payload, .. } = invocation; - let turn = Arc::clone(&step_context.turn); let payload = match payload { ToolPayload::Function { arguments } => arguments, @@ -223,7 +222,7 @@ impl McpHandler { .as_ref() .and_then(codex_mcp::PreparedMcpCall::output_token_limit) .map(TruncationPolicy::Tokens) - .unwrap_or(turn.model_info().truncation_policy.into()); + .unwrap_or(step_context.settings.model_info.truncation_policy.into()); let started = Instant::now(); let result = handle_mcp_tool_call( Arc::clone(&session), @@ -243,7 +242,9 @@ impl McpHandler { result: result.result, tool_input: result.tool_input, wall_time: started.elapsed(), - original_image_detail_supported: can_request_original_image_detail(turn.model_info()), + original_image_detail_supported: can_request_original_image_detail( + &step_context.settings.model_info, + ), truncation_policy, })) } diff --git a/codex-rs/core/src/tools/handlers/mcp_resource.rs b/codex-rs/core/src/tools/handlers/mcp_resource.rs index 5e18ee448f..5709d05375 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource.rs @@ -26,6 +26,7 @@ use serde_json::Value; use crate::function_tool::FunctionCallError; use crate::session::session::Session; +use crate::session::step_context::StepContext; use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolOutput; @@ -279,7 +280,7 @@ async fn emit_tool_call_end( async fn run_resource_operation( session: &Arc, - turn: &TurnContext, + step_context: &StepContext, call_id: &str, invocation: McpInvocation, operation: impl Future>, @@ -287,10 +288,14 @@ async fn run_resource_operation( where T: Serialize, { + let turn = &step_context.turn; emit_tool_call_begin(session, turn, call_id, invocation.clone()).await; let start = Instant::now(); let result = operation.await.and_then(|payload| { - serialize_function_output(payload, turn.model_info().truncation_policy.into()) + serialize_function_output( + payload, + step_context.settings.model_info.truncation_policy.into(), + ) }); match result { diff --git a/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resource_templates.rs b/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resource_templates.rs index d6baedcb91..46a9297cc0 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resource_templates.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resource_templates.rs @@ -72,7 +72,7 @@ impl ListMcpResourceTemplatesHandler { arguments: arguments.clone(), }; - run_resource_operation(&session, turn.as_ref(), &call_id, invocation, async { + run_resource_operation(&session, &step_context, &call_id, invocation, async { if let Some((server_name, params)) = args.target(turn.as_ref())? { let result = mcp .list_resource_templates(&server_name, params) diff --git a/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resources.rs b/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resources.rs index 24e8228c62..ea39895d30 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resources.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource/list_mcp_resources.rs @@ -72,7 +72,7 @@ impl ListMcpResourcesHandler { arguments: arguments.clone(), }; - run_resource_operation(&session, turn.as_ref(), &call_id, invocation, async { + run_resource_operation(&session, &step_context, &call_id, invocation, async { if let Some((server_name, params)) = args.target(turn.as_ref())? { let result = mcp .list_resources(&server_name, params) diff --git a/codex-rs/core/src/tools/handlers/mcp_resource/read_mcp_resource.rs b/codex-rs/core/src/tools/handlers/mcp_resource/read_mcp_resource.rs index ab534066e7..8a5ac7f324 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource/read_mcp_resource.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource/read_mcp_resource.rs @@ -77,7 +77,7 @@ impl ReadMcpResourceHandler { arguments: arguments.clone(), }; - run_resource_operation(&session, turn.as_ref(), &call_id, invocation, async { + run_resource_operation(&session, &step_context, &call_id, invocation, async { ensure_model_can_access_mcp_server(turn.as_ref(), &server)?; let result = mcp .read_resource(&server, ReadResourceRequestParams::new(uri.clone())) diff --git a/codex-rs/core/src/tools/handlers/request_plugin_install.rs b/codex-rs/core/src/tools/handlers/request_plugin_install.rs index ede1da318f..15bf9d90bb 100644 --- a/codex-rs/core/src/tools/handlers/request_plugin_install.rs +++ b/codex-rs/core/src/tools/handlers/request_plugin_install.rs @@ -220,7 +220,7 @@ impl RequestPluginInstallHandler { .analytics_events_client .track_plugin_install_requested( build_track_events_context( - turn.model_info().slug.clone(), + step_context.settings.model_info.slug.clone(), session.thread_id.to_string(), turn.sub_id.clone(), turn.originator.clone(), @@ -283,14 +283,16 @@ impl RequestPluginInstallHandler { Some(_) => "unknown", None => "unavailable", }; - turn.session_telemetry.record_plugin_install_suggestion( - tool_type, - tool.id(), - tool.name(), - response_action, - user_confirmed, - completed, - ); + step_context + .session_telemetry + .record_plugin_install_suggestion( + tool_type, + tool.id(), + tool.name(), + response_action, + user_confirmed, + completed, + ); } let content = serde_json::to_string(&RequestPluginInstallResult { diff --git a/codex-rs/core/src/tools/handlers/unified_exec/exec_command.rs b/codex-rs/core/src/tools/handlers/unified_exec/exec_command.rs index a8d0fb9282..37c0f3c823 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec/exec_command.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec/exec_command.rs @@ -407,7 +407,7 @@ impl ExecCommandHandler { chunk_id: String::new(), wall_time: std::time::Duration::ZERO, raw_output: output.into_text().into_bytes(), - truncation_policy: turn.model_info().truncation_policy.into(), + truncation_policy: step_context.settings.model_info.truncation_policy.into(), max_output_tokens, process_id: None, exit_code: None, @@ -417,7 +417,7 @@ impl ExecCommandHandler { })); } - emit_unified_exec_tty_metric(&turn.session_telemetry, tty); + emit_unified_exec_tty_metric(&step_context.session_telemetry, tty); let request = ExecCommandRequest { command, shell_type, @@ -461,7 +461,7 @@ impl ExecCommandHandler { chunk_id: generate_chunk_id(), wall_time: output.duration, raw_output: output_text.into_bytes(), - truncation_policy: turn.model_info().truncation_policy.into(), + truncation_policy: step_context.settings.model_info.truncation_policy.into(), max_output_tokens, // Sandbox denial is terminal, so there is no live // process for write_stdin to resume. diff --git a/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs b/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs index 657da46674..5f9d9cf8ca 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec/write_stdin.rs @@ -91,7 +91,12 @@ impl WriteStdinHandler { input: &args.chars, yield_time_ms: args.yield_time_ms, max_output_tokens: args.max_output_tokens, - truncation_policy: turn.model_info().truncation_policy.into(), + truncation_policy: context + .step_context + .settings + .model_info + .truncation_policy + .into(), interaction_event: Some(WriteStdinInteractionEvent { session: &session, turn: &turn, diff --git a/codex-rs/core/src/tools/handlers/view_image.rs b/codex-rs/core/src/tools/handlers/view_image.rs index 7f80995857..e447208fd4 100644 --- a/codex-rs/core/src/tools/handlers/view_image.rs +++ b/codex-rs/core/src/tools/handlers/view_image.rs @@ -95,8 +95,9 @@ impl ViewImageHandler { invocation: ToolInvocation, ) -> Result, FunctionCallError> { if !invocation - .turn - .model_info() + .step_context + .settings + .model_info .input_modalities .contains(&InputModality::Image) { @@ -185,7 +186,8 @@ impl ViewImageHandler { FunctionCallError::RespondToModel(VIEW_IMAGE_INVALID_MESSAGE.to_string()) })?; - let can_request_original_detail = can_request_original_image_detail(turn.model_info()); + let can_request_original_detail = + can_request_original_image_detail(&step_context.settings.model_info); let use_original_detail = self.options.unified_image_budget || can_request_original_detail && matches!(detail, Some(ViewImageDetail::Original)); let image_detail = if use_original_detail { diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index e73711f8e6..fb2ccbb36e 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -499,7 +499,7 @@ impl ToolRegistry { ) -> Result { let tool_name = invocation.tool_name.clone(); let call_id_owned = invocation.call_id.clone(); - let otel = invocation.turn.session_telemetry.clone(); + let otel = invocation.step_context.session_telemetry.clone(); // TODO(anp): Reconcile these tags with TurnEnvironment::sandbox_context // instead of reporting the thread-wide backend for environment-scoped tools. let sandbox_tags = invocation.turn.turn_metadata_state.sandbox_tags; diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index 8d5bde3085..c3526fee10 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -160,7 +160,6 @@ impl ToolRouter { } /// Whether the model can both start and interact with a terminal process. - // Consumed by the follow-up live tool-plan selection. #[cfg_attr(not(test), allow(dead_code))] pub(crate) fn has_terminal_controls(&self) -> bool { self.exposes_tool(&ToolName::plain("exec_command")) @@ -168,7 +167,6 @@ impl ToolRouter { } /// Whether the configured collaboration backend's child-management tools remain exposed. - // Consumed by the follow-up live tool-plan selection. #[cfg_attr(not(test), allow(dead_code))] pub(crate) fn can_manage_children(&self) -> bool { self.can_manage_children diff --git a/codex-rs/core/src/unified_exec/async_watcher.rs b/codex-rs/core/src/unified_exec/async_watcher.rs index 366f631052..ba91d72444 100644 --- a/codex-rs/core/src/unified_exec/async_watcher.rs +++ b/codex-rs/core/src/unified_exec/async_watcher.rs @@ -24,6 +24,7 @@ use crate::unified_exec::head_tail_buffer::HeadTailBuffer; use codex_core_plugins::PluginCommandAttribution; use codex_protocol::exec_output::ExecToolCallOutput; use codex_protocol::exec_output::StreamOutput; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandOutputDeltaEvent; use codex_protocol::protocol::ExecCommandSource; @@ -164,9 +165,7 @@ pub(crate) fn start_streaming_output( #[allow(clippy::too_many_arguments)] pub(crate) fn spawn_exit_watcher( process: Arc, - session_ref: Arc, - turn_ref: Arc, - call_id: String, + context: &UnifiedExecContext, command: Vec, cwd: PathUri, process_id: i32, @@ -176,6 +175,10 @@ pub(crate) fn spawn_exit_watcher( network_denial_monitor: Option>, plugin_metrics_sidecar: Option, ) { + let session_ref = Arc::clone(&context.session); + let turn_ref = Arc::clone(&context.step_context.turn); + let model_info = Arc::clone(&context.step_context.settings.model_info); + let call_id = context.call_id.clone(); let exit_token = process.cancellation_token(); let output_drained = process.output_drained_notify(); let interaction_lock = process.interaction_lock(); @@ -200,6 +203,7 @@ pub(crate) fn spawn_exit_watcher( emit_failed_exec_end_for_unified_exec( session_ref, turn_ref, + model_info, call_id, command, cwd, @@ -225,6 +229,7 @@ pub(crate) fn spawn_exit_watcher( emit_exec_end_for_unified_exec( session_ref, turn_ref, + model_info, call_id, command, cwd, @@ -337,6 +342,7 @@ impl Emitter { pub(crate) async fn emit_exec_end_for_unified_exec( session_ref: Arc, turn_ref: Arc, + model_info: Arc, call_id: String, command: Vec, cwd: PathUri, @@ -360,6 +366,7 @@ pub(crate) async fn emit_exec_end_for_unified_exec( let event_ctx = ToolEventCtx::new( session_ref.as_ref(), turn_ref.as_ref(), + &model_info, &call_id, /*turn_diff_tracker*/ None, ); @@ -385,6 +392,7 @@ pub(crate) async fn emit_exec_end_for_unified_exec( pub(crate) async fn emit_failed_exec_end_for_unified_exec( session_ref: Arc, turn_ref: Arc, + model_info: Arc, call_id: String, command: Vec, cwd: PathUri, @@ -416,6 +424,7 @@ pub(crate) async fn emit_failed_exec_end_for_unified_exec( let event_ctx = ToolEventCtx::new( session_ref.as_ref(), turn_ref.as_ref(), + &model_info, &call_id, /*turn_diff_tracker*/ None, ); diff --git a/codex-rs/core/src/unified_exec/async_watcher_tests.rs b/codex-rs/core/src/unified_exec/async_watcher_tests.rs index bffc5cf9f8..c2019a70ab 100644 --- a/codex-rs/core/src/unified_exec/async_watcher_tests.rs +++ b/codex-rs/core/src/unified_exec/async_watcher_tests.rs @@ -208,7 +208,7 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() -> stdout_tx, exit_tx, transcript, - context, + mut context, rx_event, } = streaming_output_harness().await?; @@ -226,11 +226,15 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() -> #[allow(deprecated)] let cwd = context.step_context.turn.cwd.clone().into(); + let step = Arc::get_mut(&mut context.step_context).expect("unshared test step"); + let model_info = Arc::make_mut(&mut Arc::make_mut(&mut step.settings).model_info); + model_info.truncation_policy = codex_protocol::openai_models::TruncationPolicyConfig { + mode: codex_protocol::openai_models::TruncationMode::Bytes, + limit: 4, + }; spawn_exit_watcher( Arc::clone(&process), - Arc::clone(&context.session), - Arc::clone(&context.step_context.turn), - context.call_id, + &context, vec!["proof".to_string()], cwd, /*process_id*/ 123, @@ -266,6 +270,13 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() -> Some("LATE_DENIAL") ) ); + assert_eq!( + item.formatted_output, + Some(codex_utils_output_truncation::formatted_truncate_text( + "LATE_DENIAL", + codex_utils_output_truncation::TruncationPolicy::Bytes(4), + )) + ); assert!( elapsed >= Duration::from_millis(10) && elapsed < TRAILING_OUTPUT_GRACE, "completion should wait for denial without falling back to the output grace: {elapsed:?}" diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index f19147f660..d68669e41b 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -406,6 +406,7 @@ async fn emit_failed_initial_exec_end_if_unstored( emit_failed_exec_end_for_unified_exec( Arc::clone(&context.session), Arc::clone(&context.step_context.turn), + Arc::clone(&context.step_context.settings.model_info), context.call_id.clone(), request.command.clone(), cwd, @@ -532,6 +533,7 @@ impl UnifiedExecProcessManager { let event_ctx = ToolEventCtx::new( context.session.as_ref(), context.step_context.turn.as_ref(), + &context.step_context.settings.model_info, &context.call_id, /*turn_diff_tracker*/ None, ); @@ -766,6 +768,7 @@ impl UnifiedExecProcessManager { emit_exec_end_for_unified_exec( Arc::clone(&context.session), Arc::clone(&context.step_context.turn), + Arc::clone(&context.step_context.settings.model_info), context.call_id.clone(), request.command.clone(), cwd.clone(), @@ -796,8 +799,8 @@ impl UnifiedExecProcessManager { raw_output: collected, truncation_policy: context .step_context - .turn - .model_info() + .settings + .model_info .truncation_policy .into(), max_output_tokens: request.max_output_tokens, @@ -1175,9 +1178,7 @@ impl UnifiedExecProcessManager { spawn_exit_watcher( Arc::clone(&process), - Arc::clone(&context.session), - Arc::clone(&context.step_context.turn), - context.call_id.clone(), + context, command.to_vec(), cwd, process_id, diff --git a/codex-rs/core/tests/suite/step_settings.rs b/codex-rs/core/tests/suite/step_settings.rs index bbfb3353bc..eb97afc5ff 100644 --- a/codex-rs/core/tests/suite/step_settings.rs +++ b/codex-rs/core/tests/suite/step_settings.rs @@ -2,13 +2,16 @@ use anyhow::Result; use codex_config::McpServerConfig; use codex_core::CodexThread; use codex_core::TurnInputRequest; +use codex_core::config::Config; use codex_core::config::Constrained; use codex_core::config::TokenBudgetConfig; use codex_extension_api::ContentItemKind; use codex_extension_api::ContextContributor; +use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::PromptFragment; +use codex_extension_api::ToolContributor; use codex_extension_api::TurnContextContributionInput; use codex_features::Feature; use codex_login::CodexAuth; @@ -18,8 +21,11 @@ use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::ServiceTier; +use codex_protocol::models::PermissionProfile; +use codex_protocol::openai_models::ApplyPatchToolType; use codex_protocol::openai_models::ApprovalMessages; use codex_protocol::openai_models::CollaborationModeMessages; +use codex_protocol::openai_models::ConfigShellToolType; use codex_protocol::openai_models::ConfirmationPolicies; use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelInstructionsVariables; @@ -32,6 +38,8 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::openai_models::ReasoningEffortPreset; use codex_protocol::openai_models::ToolMessage; use codex_protocol::openai_models::ToolMessages; +use codex_protocol::openai_models::ToolMode; +use codex_protocol::openai_models::TruncationPolicyConfig; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CONTEXT_WINDOW_GUIDANCE_CLOSE_TAG; use codex_protocol::protocol::CONTEXT_WINDOW_GUIDANCE_OPEN_TAG; @@ -45,10 +53,19 @@ use codex_protocol::request_user_input::RequestUserInputAnswer; use codex_protocol::request_user_input::RequestUserInputEvent; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; +use codex_tools::JsonToolOutput; +use codex_tools::ToolCall; +use codex_tools::ToolExecutor; +use codex_tools::ToolExecutorFuture; +use codex_tools::ToolName; +use codex_tools::ToolOutput; +use codex_tools::ToolSpec; use core_test_support::apps_test_server::AppsTestServer; use core_test_support::apps_test_server::recorded_apps_tool_calls; use core_test_support::responses::ResponsesRequest; +use core_test_support::responses::ev_apply_patch_custom_tool_call; use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_custom_tool_call; use core_test_support::responses::ev_function_call; use core_test_support::responses::ev_function_call_with_namespace; use core_test_support::responses::ev_response_created; @@ -60,6 +77,8 @@ use core_test_support::responses::sse_completed; use core_test_support::responses::sse_response; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; +use core_test_support::streaming_sse::StreamingSseChunk; +use core_test_support::streaming_sse::start_streaming_sse_server; use core_test_support::test_codex::TestCodexBuilder; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; @@ -72,6 +91,9 @@ use std::collections::HashMap; use std::sync::Arc; use test_case::test_case; +use super::rmcp_client::remote_aware_environment_id; +use super::rmcp_client::remote_aware_stdio_server_bin; + const MODEL_A: &str = "step-settings-a"; const MODEL_B: &str = "step-settings-b"; const MODEL_C: &str = "step-settings-c"; @@ -119,32 +141,55 @@ fn step_settings_test() -> TestCodexBuilder { }) } +fn direct_tool_settings_test() -> TestCodexBuilder { + step_settings_test().with_config(|config| { + for model in &mut config.model_catalog.as_mut().expect("test models").models { + model.tool_mode = Some(ToolMode::Direct); + model.use_responses_lite = false; + model.apply_patch_tool_type = + (model.slug != MODEL_A).then_some(ApplyPatchToolType::Freeform); + } + }) +} + +fn advertises_apply_patch(request: &Value) -> bool { + request["tools"] + .as_array() + .expect("provider tools") + .iter() + .any(|tool| tool["type"] == "custom" && tool["name"] == "apply_patch") +} + fn paused_response(response_id: &str, call_id: &str) -> String { sse(vec![ ev_response_created(response_id), - ev_function_call( - call_id, - "request_user_input", - &json!({ - "questions": [{ - "id": "continue", - "header": "Continue", - "question": "Continue after the settings update?", - "options": [{ - "label": "Yes (Recommended)", - "description": "Continue the current turn." - }, { - "label": "No", - "description": "Stop the current turn." - }] - }] - }) - .to_string(), - ), + pause_call(call_id), ev_completed(response_id), ]) } +fn pause_call(call_id: &str) -> Value { + ev_function_call( + call_id, + "request_user_input", + &json!({ + "questions": [{ + "id": "continue", + "header": "Continue", + "question": "Continue after the settings update?", + "options": [{ + "label": "Yes (Recommended)", + "description": "Continue the current turn." + }, { + "label": "No", + "description": "Stop the current turn." + }] + }] + }) + .to_string(), + ) +} + async fn start_paused_turn(thread: &CodexThread) -> Result { thread .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { @@ -192,6 +237,18 @@ async fn submit_turn_settings( Ok(tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), outcome).await??) } +async fn apply_turn_settings( + thread: &CodexThread, + turn_id: &str, + update: TurnSettingsUpdate, +) -> Result<()> { + assert_eq!( + submit_turn_settings(thread, turn_id, update).await?, + TurnSettingsUpdateOutcome::Applied + ); + Ok(()) +} + fn request_settings(request: &ResponsesRequest) -> Value { let body = request.body_json(); json!({ @@ -614,18 +671,15 @@ async fn active_model_switch_updates_core_context_from_captured_settings( test.codex.submit(Op::ReloadUserConfig).await?; } - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - }, - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &request.turn_id).await?; let request = wait_for_event_match(&test.codex, |event| match event { EventMsg::RequestUserInput(request) => Some(request.clone()), @@ -943,18 +997,15 @@ async fn mcp_confirmation_policy_follows_step_model_changes() -> Result<()> { let request = start_paused_turn(&test.codex).await?; assert_eq!(request.call_id, "policy-a-pending"); // The pending call must retain model A's policies after this settings update. - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - }, - ) - .await?, - TurnSettingsUpdateOutcome::Applied, - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; test.codex .submit(Op::UserInputAnswer { id: request.turn_id.clone(), @@ -1033,6 +1084,255 @@ async fn mcp_confirmation_policy_follows_step_model_changes() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_model_replans_tools_and_retains_the_issuing_request_router() -> Result<()> { + skip_if_no_network!(Ok(())); + + let call_id = "delayed-b-patch"; + let file_name = "retained-model.txt"; + let patch = + format!("*** Begin Patch\n*** Add File: {file_name}\n+written by B\n*** End Patch\n"); + let (release_patch, patch_gate) = tokio::sync::oneshot::channel(); + let (streaming_server, _) = start_streaming_sse_server(vec![ + vec![StreamingSseChunk { + gate: None, + body: paused_response("resp-a-initial", "pause-a"), + }], + vec![ + StreamingSseChunk { + gate: None, + body: sse(vec![ev_response_created("resp-b"), pause_call("pause-b")]), + }, + StreamingSseChunk { + gate: Some(patch_gate), + body: sse(vec![ + ev_apply_patch_custom_tool_call(call_id, &patch), + ev_completed("resp-b"), + ]), + }, + ], + vec![StreamingSseChunk { + gate: None, + body: sse_completed("resp-a"), + }], + ]) + .await; + let server = start_mock_server().await; + let model_api_url = format!("{}/v1", streaming_server.uri()); + let test = direct_tool_settings_test() + .with_config(move |config| { + for feature in [Feature::ShellTool, Feature::UnifiedExec] { + config.features.enable(feature).expect("enable shell tools"); + } + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.shell_type = if model.slug == MODEL_B { + ConfigShellToolType::Disabled + } else { + ConfigShellToolType::UnifiedExec + }; + } + // Use the gated model response with the automatically selected executor. + config.model_provider.base_url = Some(model_api_url); + config + .permissions + .set_permission_profile(PermissionProfile::Disabled) + .expect("test permissions"); + }) + .build_with_auto_env(&server) + .await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + let paused = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + assert_eq!(paused.call_id, "pause-b"); + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_A.to_string()), + ..Default::default() + }, + ) + .await?; + // B's response issues its patch after A is active, so dispatch must retain B's router. + release_patch.send(()).expect("release B's patch call"); + answer_paused_turn(&test.codex, &paused.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = streaming_server + .requests() + .await + .iter() + .map(|body| serde_json::from_slice::(body)) + .collect::, _>>()?; + assert_eq!( + requests + .iter() + .map(|request| request["model"].clone()) + .collect::>(), + vec![json!(MODEL_A), json!(MODEL_B), json!(MODEL_A)], + ); + assert_eq!( + requests + .iter() + .map(advertises_apply_patch) + .collect::>(), + vec![false, true, false], + ); + assert_eq!( + requests + .iter() + .map(|request| { + request["tools"] + .as_array() + .expect("provider tools") + .iter() + .filter_map(|tool| tool["name"].as_str()) + .filter(|name| matches!(*name, "exec_command" | "write_stdin")) + .collect::>() + }) + .collect::>(), + vec![ + vec!["exec_command", "write_stdin"], + vec![], + vec!["exec_command", "write_stdin"] + ], + ); + assert!( + requests[2]["input"] + .as_array() + .expect("provider input") + .iter() + .any(|item| { + item["type"] == "custom_tool_call_output" && item["call_id"] == call_id + }) + ); + assert_eq!( + test.fs() + .read_file_text( + &test.workspace_path_uri(file_name)?, + Default::default(), + /*sandbox*/ None, + ) + .await?, + "written by B\n", + ); + streaming_server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_model_enables_and_executes_code_mode() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_sse_sequence( + &server, + vec![ + paused_response("resp-a", "pause-a"), + sse(vec![ + ev_response_created("resp-b"), + ev_custom_tool_call("code-b", "exec", "const result = await tools.view_image({path: 'step.png', detail: 'original'}); image(result)"), + ev_completed("resp-b"), + ]), + sse_completed("resp-b-result"), + ], + ) + .await; + let test = direct_tool_settings_test() + .with_config(|config| { + config + .permissions + .set_permission_profile(PermissionProfile::Disabled) + .expect("test permissions"); + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.input_modalities = vec![codex_protocol::openai_models::InputModality::Text]; + model.supports_image_detail_original = false; + if model.slug == MODEL_B { + model + .input_modalities + .push(codex_protocol::openai_models::InputModality::Image); + model.supports_image_detail_original = true; + model.tool_mode = Some(ToolMode::CodeModeOnly); + } + } + }) + .build_with_auto_env(&server) + .await?; + use base64::Engine; + let png = base64::engine::general_purpose::STANDARD.decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==")?; + test.fs() + .write_file( + &test.workspace_path_uri("step.png")?, + png, + Default::default(), + /*sandbox*/ None, + ) + .await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = responses.requests(); + assert_eq!( + requests + .iter() + .map(|request| request.body_json()["model"].clone()) + .collect::>(), + vec![json!(MODEL_A), json!(MODEL_B), json!(MODEL_B)], + ); + assert_eq!( + requests + .iter() + .map(|request| { + request.body_json()["tools"] + .as_array() + .expect("provider tools") + .iter() + .any(|tool| tool["type"] == "custom" && tool["name"] == "exec") + }) + .collect::>(), + vec![false, true, true], + ); + let output = requests[2].custom_tool_call_output("code-b"); + assert_eq!( + output["output"] + .as_array() + .expect("Code Mode output items") + .last() + .expect("Code Mode image output")["detail"], + json!("original"), + ); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn response_metadata_uses_the_captured_model_after_a_turn_update() -> Result<()> { skip_if_no_network!(Ok(())); @@ -1055,18 +1355,15 @@ async fn response_metadata_uses_the_captured_model_after_a_turn_update() -> Resu .await; let test = step_settings_test().build_with_auto_env(&server).await?; let request = start_paused_turn(&test.codex).await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - }, - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &request.turn_id).await?; let mut reroutes = Vec::new(); @@ -1138,19 +1435,16 @@ async fn sparse_updates_preserve_divergent_active_and_future_models() -> Result< }, ) .await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_C.to_string()), - effort: Some(Some(ReasoningEffort::High)), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_C.to_string()), + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &request.turn_id).await?; let second_request = wait_for_event_match(&test.codex, |event| match event { EventMsg::RequestUserInput(request) => Some(request.clone()), @@ -1168,18 +1462,15 @@ async fn sparse_updates_preserve_divergent_active_and_future_models() -> Result< }, }) .await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + }, + ) + .await?; let durable_settings = wait_for_event_match(&test.codex, |event| match event { EventMsg::ThreadSettingsApplied(event) => Some(event.thread_settings.clone()), _ => None, @@ -1360,18 +1651,15 @@ async fn model_activation_uses_destination_metadata_defaults( .await?; let request = start_paused_turn(&test.codex).await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &request.turn_id).await?; let paused = wait_for_event_match(&test.codex, |event| match event { EventMsg::RequestUserInput(request) => Some(request.clone()), @@ -1380,18 +1668,15 @@ async fn model_activation_uses_destination_metadata_defaults( .await; // B cannot use the requested tier. Switching back must recover that // selection and preserve an unset summary, not reuse B's effective values. - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_A.to_string()), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_A.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &paused.turn_id).await?; wait_for_event(&test.codex, |event| match event { EventMsg::Error(error) => panic!("model activation failed: {}", error.message), @@ -1467,18 +1752,15 @@ async fn request_user_input_async_description_follows_mid_turn_model_changes() - .build_with_auto_env(&server) .await?; let paused = start_paused_turn(&test.codex).await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &paused.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &paused.turn_id).await?; wait_for_event(&test.codex, |event| match event { EventMsg::Error(error) => panic!("model activation failed: {}", error.message), @@ -1552,18 +1834,15 @@ async fn persistent_instructions_follow_mid_turn_model_changes() -> Result<()> { .build_with_auto_env(&server) .await?; let paused = start_paused_turn(&test.codex).await?; - assert_eq!( - submit_turn_settings( - &test.codex, - &paused.turn_id, - TurnSettingsUpdate { - model: Some(MODEL_B.to_string()), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &paused.turn_id).await?; wait_for_event(&test.codex, |event| match event { EventMsg::Error(error) => panic!("model activation failed: {}", error.message), @@ -1731,18 +2010,15 @@ async fn request_preference_activation_keeps_admitted_model_metadata() -> Result .await; assert_eq!(refresh.requests().len(), 1); - assert_eq!( - submit_turn_settings( - &test.codex, - &request.turn_id, - TurnSettingsUpdate { - effort: Some(Some(ReasoningEffort::High)), - ..Default::default() - } - ) - .await?, - TurnSettingsUpdateOutcome::Applied - ); + apply_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ) + .await?; answer_paused_turn(&test.codex, &request.turn_id).await?; wait_for_event(&test.codex, |event| { matches!(event, EventMsg::TurnComplete(_)) @@ -1778,5 +2054,461 @@ async fn request_preference_activation_keeps_admitted_model_metadata() -> Result Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_step_controls_exec_completion_and_write_stdin_output() -> Result<()> { + core_test_support::skip_if_target_windows!(Ok(()), "uses POSIX read and printf"); + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let output = "abcdefghij".repeat(100); + let command = format!("stty -echo; printf '{output}'; read line; printf '{output}'; exit 7"); + let responses = mount_sse_sequence( + &server, + vec![ + paused_response("resp-a", "pause-a"), + sse(vec![ + ev_response_created("resp-b"), + ev_function_call( + "exec-b", + "exec_command", + &json!({ + "cmd": command, "tty": true, "yield_time_ms": 10, + }) + .to_string(), + ), + ev_completed("resp-b"), + ]), + paused_response("resp-b-pause", "pause-b"), + sse(vec![ + ev_response_created("resp-c"), + ev_function_call( + "stdin-c", + "write_stdin", + &json!({ + "session_id": 1000, "chars": "done\n", "yield_time_ms": 1000, + }) + .to_string(), + ), + ev_completed("resp-c"), + ]), + sse_completed("resp-result"), + ], + ) + .await; + let test = direct_tool_settings_test() + .with_config(|config| { + config + .features + .enable(Feature::ShellTool) + .expect("shell tools"); + config + .features + .enable(Feature::UnifiedExec) + .expect("unified exec"); + config + .permissions + .set_permission_profile(PermissionProfile::Disabled) + .expect("test permissions"); + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.shell_type = ConfigShellToolType::UnifiedExec; + model.truncation_policy = codex_protocol::openai_models::TruncationPolicyConfig { + mode: codex_protocol::openai_models::TruncationMode::Bytes, + limit: match model.slug.as_str() { + MODEL_B => 400, + MODEL_C => 800, + _ => 8_000, + }, + }; + } + }) + .build_with_auto_env(&server) + .await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + let paused = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + assert_eq!(paused.call_id, "pause-b"); + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_C.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + let mut end = None; + let mut turn_complete = false; + wait_for_event(&test.codex, |event| { + match event { + EventMsg::ExecCommandEnd(event) if event.call_id == "exec-b" => { + end = Some(event.clone()) + } + EventMsg::TurnComplete(_) => turn_complete = true, + _ => {} + } + end.is_some() && turn_complete + }) + .await; + let end = end.expect("exec completion"); + + use codex_utils_output_truncation::TruncationPolicy; + use codex_utils_output_truncation::formatted_truncate_text; + let requests = responses.requests(); + let exec = requests[2] + .function_call_output_text("exec-b") + .expect("exec output"); + // The response budget also reserves space for command metadata and truncation notices. + assert_eq!(exec.matches("chars truncated").count(), 1, "{exec}"); + assert!( + (15..30).contains(&exec.matches("abcdefghij").count()), + "{exec}" + ); + let stdin = requests[4] + .function_call_output_text("stdin-c") + .expect("stdin output"); + assert_eq!(stdin.matches("chars truncated").count(), 1, "{stdin}"); + assert!( + (60..85).contains(&stdin.matches("abcdefghij").count()), + "{stdin}" + ); + assert_eq!(end.exit_code, 7); + // The formatting check needs an untruncated chunk, independent of how the executor + // aggregates output across the initial command and later stdin interactions. + assert!(end.aggregated_output.contains(&output)); + assert_eq!( + end.formatted_output, + formatted_truncate_text(&end.aggregated_output, TruncationPolicy::Bytes(400)) + ); + Ok(()) +} + +#[test_case(true; "model fallback and image detail")] +#[test_case(false; "text-only step omits images")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_step_controls_mcp_output_limit(supports_images: bool) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + AppsTestServer::mount(&server).await?; + let title = "abcdefghij".repeat(100); + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::body_partial_json( + json!({"method": "tools/call"}), + )) + .respond_with(|request: &wiremock::Request| { + let body: Value = serde_json::from_slice(&request.body).expect("MCP request"); + wiremock::ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": body["id"], "result": { + "content": [ + {"type": "text", "text": body["params"]["arguments"]["title"]}, + {"type": "image", "mimeType": "image/png", + "data": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==", + "_meta": {"codex/imageDetail": "original"}}, + ], "isError": false, + }, + })) + }) + .with_priority(1) + .mount(&server) + .await; + let responses = mount_sse_sequence( + &server, + vec![ + paused_response("resp-a", "pause-a"), + sse(vec![ + ev_response_created("resp-b"), + ev_function_call_with_namespace( + "mcp-b", + "mcp__calendar", + "calendar_create_event", + &json!({"title": title}).to_string(), + ), + ev_completed("resp-b"), + ]), + sse_completed("resp-result"), + ], + ) + .await; + let mcp_url = format!("{}/api/codex/ps/mcp", server.uri()); + let test = direct_tool_settings_test() + .with_config(move |config| { + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.supports_image_detail_original = model.slug != MODEL_B; + model.input_modalities = vec![codex_protocol::openai_models::InputModality::Text]; + if model.slug != MODEL_B || supports_images { + model + .input_modalities + .push(codex_protocol::openai_models::InputModality::Image); + } + model.truncation_policy = codex_protocol::openai_models::TruncationPolicyConfig { + mode: codex_protocol::openai_models::TruncationMode::Bytes, + limit: if model.slug == MODEL_B { 80 } else { 8_000 }, + }; + } + config + .mcp_servers + .set(HashMap::from([( + "calendar".to_string(), + serde_json::from_value(json!({ + "url": mcp_url, + "tools": { + "calendar_create_event": { + "approval_mode": "approve" + } + }, + })) + .expect("test MCP config"), + )])) + .expect("configure MCP"); + }) + .build_with_auto_env(&server) + .await?; + wait_for_mcp_server(&test.codex, "calendar").await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + let recorded_output = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RawResponseItem(item) => { + let item = serde_json::to_value(&item.item).expect("raw response JSON"); + (item["call_id"] == "mcp-b" && item["type"] == "function_call_output").then_some(item) + } + _ => None, + }) + .await; + let images = recorded_output["output"] + .as_array() + .expect("MCP content") + .iter() + .filter(|item| item["type"] == "input_image") + .map(|item| item["detail"].clone()) + .collect::>(); + assert_eq!( + images, + if supports_images { + vec![json!("high")] + } else { + vec![] + } + ); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let requests = responses.requests(); + let output_item = requests[2].function_call_output("mcp-b"); + let output = output_item["output"] + .as_array() + .expect("MCP content") + .iter() + .filter_map(|item| item["text"].as_str()) + .collect::>() + .join("\n"); + assert!(output.contains("truncated"), "{output}"); + assert!(output.matches("abcdefghij").count() < 10, "{output}"); + Ok(()) +} + +struct SettingsEcho; + +impl ToolContributor for SettingsEcho { + fn tools( + &self, + _session_store: &ExtensionData, + _thread_store: &ExtensionData, + ) -> Vec ToolExecutor>>> { + vec![Arc::new(Self)] + } +} + +impl<'call> ToolExecutor> for SettingsEcho { + fn tool_name(&self) -> ToolName { + ToolName::plain("extension_settings") + } + + fn spec(&self) -> ToolSpec { + ToolSpec::Function(codex_tools::ResponsesApiTool { + name: "extension_settings".to_string(), + description: "Report the settings received by this extension.".to_string(), + strict: false, + parameters: codex_tools::JsonSchema::default(), + output_schema: None, + defer_loading: None, + }) + } + + fn handle<'a>(&'a self, call: ToolCall<'call>) -> ToolExecutorFuture<'a> + where + 'call: 'a, + { + Box::pin(async move { + let metadata: Value = serde_json::from_str( + call.codex_turn_metadata + .as_deref() + .expect("extension metadata"), + ) + .expect("metadata JSON"); + Ok(Box::new(JsonToolOutput::new(json!({ + "model": call.model, + "metadata_model": metadata["model"], + "effort": metadata["reasoning_effort"], + "output_bytes": call.truncation_policy.byte_budget(), + }))) as Box) + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_step_settings_reach_extension_executor() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let responses = mount_sse_sequence( + &server, + vec![ + paused_response("resp-a", "pause-a"), + sse(vec![ + ev_response_created("resp-b"), + ev_function_call("extension-b", "extension_settings", "{}"), + ev_completed("resp-b"), + ]), + sse_completed("resp-result"), + ], + ) + .await; + let mut extensions = ExtensionRegistryBuilder::::new(); + extensions.tool_contributor(Arc::new(SettingsEcho)); + let test = direct_tool_settings_test() + .with_extensions(Arc::new(extensions.build())) + .with_config(|config| { + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.truncation_policy = + TruncationPolicyConfig::bytes(if model.slug == MODEL_B { 512 } else { 8_000 }); + } + }) + .build_with_auto_env(&server) + .await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let output = responses.requests()[2] + .function_call_output_text("extension-b") + .expect("extension output"); + assert_eq!( + serde_json::from_str::(&output)?, + json!({ + "model": MODEL_B, + "metadata_model": MODEL_B, + "effort": "high", + "output_bytes": 512, + }) + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn captured_step_controls_mcp_resource_output() -> Result<()> { + skip_if_no_network!(Ok(())); + core_test_support::skip_if_wine_exec!(Ok(()), "requires a Windows test_stdio_server binary"); + let server = start_mock_server().await; + let responses = mount_sse_sequence( + &server, + vec![ + paused_response("resp-a", "pause-a"), + sse(vec![ + ev_response_created("resp-b"), + ev_function_call( + "resource-b", + "read_mcp_resource", + &json!({ + "server": "resources", "uri": "memo://codex/example-note", + }) + .to_string(), + ), + ev_completed("resp-b"), + ]), + sse_completed("resp-result"), + ], + ) + .await; + let command = remote_aware_stdio_server_bin()?; + let test = direct_tool_settings_test() + .with_config(move |config| { + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.truncation_policy = + TruncationPolicyConfig::bytes(if model.slug == MODEL_B { 80 } else { 8_000 }); + } + config + .mcp_servers + .set(HashMap::from([( + "resources".to_string(), + serde_json::from_value(json!({ + "command": command, + "environment_id": remote_aware_environment_id(), + "cwd": config.cwd, + })) + .expect("test MCP config"), + )])) + .expect("configure MCP"); + }) + .build_with_auto_env(&server) + .await?; + wait_for_mcp_server(&test.codex, "resources").await?; + let paused = start_paused_turn(&test.codex).await?; + apply_turn_settings( + &test.codex, + &paused.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + answer_paused_turn(&test.codex, &paused.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let output = responses.requests()[2] + .function_call_output_text("resource-b") + .expect("resource output"); + assert!(output.starts_with("{\"server\":\"resources\""), "{output}"); + assert_eq!(output.matches("truncated").count(), 1, "{output}"); + // B allows 96 serialized bytes plus the truncation notice; A preserves the full resource. + assert!(output.len() < 160, "{output}"); + Ok(()) +} + #[path = "step_settings_compaction.rs"] mod compaction;