Use captured step settings for tool planning and execution (#44242)

## Why

After a model change within a turn, tools could still use the turn's original model capabilities, output limits, and telemetry attribution.

## What changed

- Build each step's tool plan from its captured model settings.
- Use those settings for image support and detail, command and MCP output truncation, extension metadata, and tool telemetry.
- Preserve the issuing step's model for delayed command completion events while allowing later `write_stdin` calls to use their own step's output limits.

## Testing

Add regression coverage for model changes within a turn, including tool replanning, Code Mode execution, image handling, MCP tool and resource output limits, extension settings, and delayed command completion.

GitOrigin-RevId: ef991598e89a846889baf770a0c9e12e56886d5d
This commit is contained in:
Abhinav authored and copyberry committed 2026-09-09 17:45:37 +00:00
1 parent ed4ca07ba6
commit 205f3671e1
24 files changed
+1053 -241

No files matched your search

+9 -8
View File
@@ -239,7 +239,7 @@ pub(crate) async fn handle_mcp_tool_call(
let status = if result.is_ok() { "ok" } else { "error" }; let status = if result.is_ok() { "ok" } else { "error" };
let outcome = McpCallMetricOutcome::from_status(status); let outcome = McpCallMetricOutcome::from_status(status);
emit_mcp_call_metrics( emit_mcp_call_metrics(
turn_context.as_ref(), &step_context.session_telemetry,
&outcome, &outcome,
&server, &server,
&tool_name, &tool_name,
@@ -348,7 +348,7 @@ pub(crate) async fn handle_mcp_tool_call(
let status = if result.is_ok() { "ok" } else { "error" }; let status = if result.is_ok() { "ok" } else { "error" };
let outcome = McpCallMetricOutcome::from_status(status); let outcome = McpCallMetricOutcome::from_status(status);
emit_mcp_call_metrics( emit_mcp_call_metrics(
turn_context.as_ref(), &step_context.session_telemetry,
&outcome, &outcome,
&server, &server,
&tool_name, &tool_name,
@@ -489,7 +489,7 @@ async fn handle_approved_mcp_tool_call(
.map(|(connector_id, action_name)| HostedFileUploadContext { .map(|(connector_id, action_name)| HostedFileUploadContext {
connector_id: connector_id.clone(), connector_id: connector_id.clone(),
action_name: action_name.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( let rewritten_arguments = rewrite_mcp_tool_arguments_for_openai_files(
sess, sess,
@@ -545,7 +545,7 @@ async fn handle_approved_mcp_tool_call(
) )
.await; .await;
let result = sanitize_mcp_tool_result_for_model( let result = sanitize_mcp_tool_result_for_model(
&turn_context.model_info().input_modalities, &step_context.settings.model_info.input_modalities,
Ok(result), Ok(result),
)?; )?;
Ok(maybe_request_codex_apps_auth_elicitation( 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), truncate_mcp_tool_result_for_event(&result),
) )
.await; .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); let outcome = mcp_call_metric_outcome(&result);
emit_mcp_call_metrics( emit_mcp_call_metrics(
turn_context, &step_context.session_telemetry,
&outcome, &outcome,
&server, &server,
&tool_name, &tool_name,
@@ -1055,7 +1055,7 @@ async fn notify_mcp_tool_call_completed(
async fn maybe_track_codex_app_used( async fn maybe_track_codex_app_used(
sess: &Session, sess: &Session,
turn_context: &TurnContext, step_context: &StepContext,
server: &str, server: &str,
metadata: &McpToolApprovalMetadata, metadata: &McpToolApprovalMetadata,
) { ) {
@@ -1075,8 +1075,9 @@ async fn maybe_track_codex_app_used(
InvocationType::Implicit InvocationType::Implicit
}; };
let turn_context = &step_context.turn;
let tracking = build_track_events_context( let tracking = build_track_events_context(
turn_context.model_info().slug.clone(), step_context.settings.model_info.slug.clone(),
sess.thread_id.to_string(), sess.thread_id.to_string(),
turn_context.sub_id.clone(), turn_context.sub_id.clone(),
turn_context.originator.clone(), turn_context.originator.clone(),
+5 -15
View File
@@ -1,7 +1,7 @@
use std::time::Duration; use std::time::Duration;
use crate::session::turn_context::TurnContext;
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY; use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_otel::SessionTelemetry;
use codex_otel::sanitize_metric_tag_value; use codex_otel::sanitize_metric_tag_value;
use codex_protocol::mcp::CallToolResult; use codex_protocol::mcp::CallToolResult;
use serde_json::Value as JsonValue; use serde_json::Value as JsonValue;
@@ -38,7 +38,7 @@ impl McpCallMetricOutcome {
} }
pub(super) fn emit_mcp_call_metrics( pub(super) fn emit_mcp_call_metrics(
turn_context: &TurnContext, session_telemetry: &SessionTelemetry,
outcome: &McpCallMetricOutcome, outcome: &McpCallMetricOutcome,
server_name: &str, server_name: &str,
tool_name: &str, tool_name: &str,
@@ -57,15 +57,9 @@ pub(super) fn emit_mcp_call_metrics(
.iter() .iter()
.map(|(key, value)| (*key, value.as_str())) .map(|(key, value)| (*key, value.as_str()))
.collect(); .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 { if let Some(duration) = duration {
turn_context.session_telemetry.record_duration( session_telemetry.record_duration(MCP_CALL_DURATION_METRIC, duration, &tag_refs);
MCP_CALL_DURATION_METRIC,
duration,
&tag_refs,
);
} }
let (Some(error_type), Some(error_code)) = (outcome.error_type, outcome.error_code.as_deref()) 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() .iter()
.map(|(key, value)| (*key, value.as_str())) .map(|(key, value)| (*key, value.as_str()))
.collect(); .collect();
turn_context.session_telemetry.counter( session_telemetry.counter(MCP_CALL_ERROR_COUNT_METRIC, /*inc*/ 1, &error_tag_refs);
MCP_CALL_ERROR_COUNT_METRIC,
/*inc*/ 1,
&error_tag_refs,
);
} }
fn mcp_call_metric_tags( fn mcp_call_metric_tags(
+2 -6
View File
@@ -3670,13 +3670,10 @@ impl Session {
}); });
extension_data.insert(selected_plugins.clone()); extension_data.insert(selected_plugins.clone());
turn_context.extension_data.insert(selected_plugins); 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( let tool_router = turn::built_tools(
self.as_ref(), self.as_ref(),
turn_context.as_ref(), turn_context.as_ref(),
// TODO(CDXENT-441): use the step scoped model &settings.model_info,
turn_context.model_info(),
settings.model_info.model_messages.as_ref(), settings.model_info.model_messages.as_ref(),
&environments, &environments,
&mcp, &mcp,
@@ -3685,8 +3682,7 @@ impl Session {
) )
.or_cancel(cancellation_token) .or_cancel(cancellation_token)
.await??; .await??;
// Publish inventory after planning rather than during finalization, so constructing // Publish inventory only after the step's tool plan has been finalized.
// additional candidate plans cannot overwrite turn-wide metadata.
if turn_context if turn_context
.config .config
.tool_registry .tool_registry
@@ -33,7 +33,7 @@ impl CodeModeExecuteHandler {
async fn execute( async fn execute(
&self, &self,
session: std::sync::Arc<crate::session::session::Session>, session: std::sync::Arc<crate::session::session::Session>,
turn: std::sync::Arc<crate::session::turn_context::TurnContext>, step_context: std::sync::Arc<crate::session::step_context::StepContext>,
call_id: String, call_id: String,
originating_item_id: Option<codex_protocol::ResponseItemId>, originating_item_id: Option<codex_protocol::ResponseItemId>,
code: String, code: String,
@@ -41,7 +41,10 @@ impl CodeModeExecuteHandler {
) -> Result<FunctionToolOutput, FunctionCallError> { ) -> Result<FunctionToolOutput, FunctionCallError> {
let args = let args =
codex_code_mode::parse_exec_source(&code).map_err(FunctionCallError::RespondToModel)?; 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()); let mut enabled_tools = Vec::with_capacity(self.nested_tool_specs.len());
for (spec, cached_runtime) in &self.nested_tool_specs { for (spec, cached_runtime) in &self.nested_tool_specs {
if let Some(cached_definitions) = cached_runtime if let Some(cached_definitions) = cached_runtime
@@ -138,9 +141,14 @@ impl CodeModeExecuteHandler {
let wall_time = response let wall_time = response
.code_mode_host_duration() .code_mode_host_duration()
.unwrap_or_else(|| started_at.elapsed()); .unwrap_or_else(|| started_at.elapsed());
handle_runtime_response(&exec, response, args.max_output_tokens, wall_time) handle_runtime_response(
.await &step_context.settings.model_info,
.map_err(FunctionCallError::RespondToModel) response,
args.max_output_tokens,
wall_time,
)
.await
.map_err(FunctionCallError::RespondToModel)
} }
} }
@@ -170,6 +178,7 @@ impl CodeModeExecuteHandler {
let ToolInvocation { let ToolInvocation {
session, session,
turn, turn,
step_context,
call_id, call_id,
tool_name, tool_name,
payload, payload,
@@ -188,7 +197,7 @@ impl CodeModeExecuteHandler {
ToolPayload::Custom { input } if is_exec_tool_name(&tool_name) => self ToolPayload::Custom { input } if is_exec_tool_name(&tool_name) => self
.execute( .execute(
session, session,
turn, step_context,
call_id, call_id,
originating_item_id, originating_item_id,
input, input,
+5 -8
View File
@@ -248,24 +248,25 @@ impl CodeModeService {
} }
pub(super) async fn handle_runtime_response( pub(super) async fn handle_runtime_response(
exec: &ExecContext, model_info: &codex_protocol::openai_models::ModelInfo,
response: RuntimeResponse, response: RuntimeResponse,
max_output_tokens: Option<usize>, max_output_tokens: Option<usize>,
wall_time: Duration, wall_time: Duration,
) -> Result<FunctionToolOutput, String> { ) -> Result<FunctionToolOutput, String> {
let script_status = format_script_status(&response); let script_status = format_script_status(&response);
let supports_original = can_request_original_image_detail(model_info);
match response { match response {
RuntimeResponse::Yielded { content_items, .. } => { RuntimeResponse::Yielded { content_items, .. } => {
let mut content_items = into_function_call_output_content_items(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); content_items = truncate_code_mode_result(content_items, max_output_tokens);
prepend_script_status(&mut content_items, &script_status, wall_time); prepend_script_status(&mut content_items, &script_status, wall_time);
Ok(FunctionToolOutput::from_content(content_items, Some(true))) Ok(FunctionToolOutput::from_content(content_items, Some(true)))
} }
RuntimeResponse::Terminated { content_items, .. } => { RuntimeResponse::Terminated { content_items, .. } => {
let mut content_items = into_function_call_output_content_items(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); content_items = truncate_code_mode_result(content_items, max_output_tokens);
prepend_script_status(&mut content_items, &script_status, wall_time); prepend_script_status(&mut content_items, &script_status, wall_time);
Ok(FunctionToolOutput::from_content(content_items, Some(true))) 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); 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(); let success = error_text.is_none();
if let Some(error_text) = error_text { if let Some(error_text) = error_text {
content_items.push(FunctionCallOutputContentItem::InputText { 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 { fn format_script_status(response: &RuntimeResponse) -> String {
match response { match response {
RuntimeResponse::Yielded { cell_id, .. } => { RuntimeResponse::Yielded { cell_id, .. } => {
@@ -70,6 +70,7 @@ impl CodeModeWaitHandler {
let ToolInvocation { let ToolInvocation {
session, session,
turn, turn,
step_context,
call_id, call_id,
tool_name, tool_name,
payload, payload,
@@ -159,10 +160,15 @@ impl CodeModeWaitHandler {
let wall_time = wait_response let wall_time = wait_response
.code_mode_host_duration() .code_mode_host_duration()
.unwrap_or_else(|| started_at.elapsed()); .unwrap_or_else(|| started_at.elapsed());
handle_runtime_response(&exec, wait_response.into(), args.max_tokens, wall_time) handle_runtime_response(
.await &step_context.settings.model_info,
.map_err(FunctionCallError::RespondToModel) wait_response.into(),
.map(boxed_tool_output) args.max_tokens,
wall_time,
)
.await
.map_err(FunctionCallError::RespondToModel)
.map(boxed_tool_output)
} }
_ => Err(FunctionCallError::RespondToModel(format!( _ => Err(FunctionCallError::RespondToModel(format!(
"{WAIT_TOOL_NAME} expects JSON arguments" "{WAIT_TOOL_NAME} expects JSON arguments"
+36 -8
View File
@@ -18,6 +18,7 @@ use codex_protocol::items::CommandExecutionItem;
use codex_protocol::items::CommandExecutionStatus; use codex_protocol::items::CommandExecutionStatus;
use codex_protocol::items::FileChangeItem; use codex_protocol::items::FileChangeItem;
use codex_protocol::items::TurnItem; use codex_protocol::items::TurnItem;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::parse_command::ParsedCommand; use codex_protocol::parse_command::ParsedCommand;
use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecCommandSource; use codex_protocol::protocol::ExecCommandSource;
@@ -44,6 +45,8 @@ pub(super) fn truncate_rejection_message(message: &str) -> String {
pub(crate) struct ToolEventCtx<'a> { pub(crate) struct ToolEventCtx<'a> {
pub session: &'a Session, pub session: &'a Session,
pub turn: &'a TurnContext, 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 call_id: &'a str,
pub turn_diff_tracker: Option<&'a SharedTurnDiffTracker>, pub turn_diff_tracker: Option<&'a SharedTurnDiffTracker>,
} }
@@ -52,12 +55,14 @@ impl<'a> ToolEventCtx<'a> {
pub fn new( pub fn new(
session: &'a Session, session: &'a Session,
turn: &'a TurnContext, turn: &'a TurnContext,
model_info: &'a ModelInfo,
call_id: &'a str, call_id: &'a str,
turn_diff_tracker: Option<&'a SharedTurnDiffTracker>, turn_diff_tracker: Option<&'a SharedTurnDiffTracker>,
) -> Self { ) -> Self {
Self { Self {
session, session,
turn, turn,
model_info,
call_id, call_id,
turn_diff_tracker, turn_diff_tracker,
} }
@@ -117,12 +122,17 @@ async fn emit_exec_command_begin(ctx: ToolEventCtx<'_>, exec_input: &ExecCommand
("output_format", operation.output_format), ("output_format", operation.output_format),
("execution_backend", "unified_exec"), ("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, ARTIFACT_OPERATION_STARTED_METRIC,
/*inc*/ 1, /*inc*/ 1,
&metric_tags, &metric_tags,
); );
ctx.turn.session_telemetry.histogram( session_telemetry.histogram(
ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC, ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC,
i64::from(operation.expected_output_count), i64::from(operation.expected_output_count),
&metric_tags, &metric_tags,
@@ -132,7 +142,7 @@ async fn emit_exec_command_begin(ctx: ToolEventCtx<'_>, exec_input: &ExecCommand
.analytics_events_client .analytics_events_client
.track_artifact_operation( .track_artifact_operation(
build_track_events_context( build_track_events_context(
ctx.turn.model_info().slug.clone(), ctx.model_info.slug.clone(),
ctx.session.thread_id.to_string(), ctx.session.thread_id.to_string(),
ctx.turn.sub_id.clone(), ctx.turn.sub_id.clone(),
ctx.turn.originator.clone(), ctx.turn.originator.clone(),
@@ -373,7 +383,7 @@ impl ToolEmitter {
output: &ExecToolCallOutput, output: &ExecToolCallOutput,
ctx: ToolEventCtx<'_>, ctx: ToolEventCtx<'_>,
) -> String { ) -> 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( pub async fn finish(
@@ -521,7 +531,7 @@ async fn emit_exec_stage(
duration: output.duration, duration: output.duration,
formatted_output: format_exec_output_str( formatted_output: format_exec_output_str(
&output, &output,
ctx.turn.model_info().truncation_policy.into(), ctx.model_info.truncation_policy.into(),
), ),
status: if output.exit_code == 0 { status: if output.exit_code == 0 {
ExecCommandStatus::Completed ExecCommandStatus::Completed
@@ -697,7 +707,13 @@ mod tests {
environment_id: None, environment_id: None,
} }
.finish( .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, out,
Some(&delta), Some(&delta),
) )
@@ -781,7 +797,13 @@ mod tests {
.expect("apply patch"); .expect("apply patch");
emit_patch_end( 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(), HashMap::new(),
String::new(), String::new(),
String::new(), String::new(),
@@ -830,7 +852,13 @@ mod tests {
tracker.lock().await.track_delta("", &delta); tracker.lock().await.track_delta("", &delta);
emit_patch_end( 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(), HashMap::new(),
String::new(), String::new(),
String::new(), String::new(),
@@ -589,6 +589,7 @@ async fn execute_verified_patch(
let event_ctx = ToolEventCtx::new( let event_ctx = ToolEventCtx::new(
tool_ctx.session.as_ref(), tool_ctx.session.as_ref(),
tool_ctx.step_context.turn.as_ref(), tool_ctx.step_context.turn.as_ref(),
&tool_ctx.step_context.settings.model_info,
&tool_ctx.call_id, &tool_ctx.call_id,
tracker, tracker,
); );
@@ -616,6 +617,7 @@ async fn execute_verified_patch(
let event_ctx = ToolEventCtx::new( let event_ctx = ToolEventCtx::new(
tool_ctx.session.as_ref(), tool_ctx.session.as_ref(),
tool_ctx.step_context.turn.as_ref(), tool_ctx.step_context.turn.as_ref(),
&tool_ctx.step_context.settings.model_info,
&tool_ctx.call_id, &tool_ctx.call_id,
tracker, tracker,
); );
@@ -166,13 +166,14 @@ impl TurnItemEmitter for CoreTurnItemEmitter {
async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_> { async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_> {
let conversation_history = let conversation_history =
ConversationHistory::new(invocation.session.clone_history().await.into_raw_items()); ConversationHistory::new(invocation.session.clone_history().await.into_raw_items());
let settings = &invocation.step_context.settings;
let codex_turn_metadata = invocation let codex_turn_metadata = invocation
.turn .turn
.turn_metadata_state .turn_metadata_state
.current_meta_value_for_mcp_request(McpTurnMetadataContext { .current_meta_value_for_mcp_request(McpTurnMetadataContext {
model: invocation.turn.model_info().slug.as_str(), model: settings.model_info.slug.as_str(),
reasoning_effort: invocation.turn.effective_reasoning_effort(), reasoning_effort: settings.effective_reasoning_effort(),
node_repl_disabled: invocation.turn.model_info().node_repl_disabled, node_repl_disabled: settings.model_info.node_repl_disabled,
}) })
.and_then(|metadata| to_ascii_json_string(&metadata).ok()); .and_then(|metadata| to_ascii_json_string(&metadata).ok());
let mut environments = Vec::new(); let mut environments = Vec::new();
@@ -204,9 +205,9 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_>
turn_id: invocation.turn.sub_id.clone(), turn_id: invocation.turn.sub_id.clone(),
call_id: invocation.call_id.clone(), call_id: invocation.call_id.clone(),
tool_name: invocation.tool_name.clone(), tool_name: invocation.tool_name.clone(),
model: invocation.turn.model_info().slug.clone(), model: settings.model_info.slug.clone(),
codex_turn_metadata, 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()), source: extension_tool_call_source(invocation.source.clone()),
conversation_history, conversation_history,
turn_item_emitter: Arc::new(CoreTurnItemEmitter { turn_item_emitter: Arc::new(CoreTurnItemEmitter {
@@ -417,7 +418,7 @@ mod tests {
} }
#[tokio::test] #[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_call = Arc::new(Mutex::new(None));
let captured_sandbox_cwds = Arc::new(Mutex::new(Vec::new())); let captured_sandbox_cwds = Arc::new(Mutex::new(Vec::new()));
let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor { let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor {
@@ -464,7 +465,15 @@ mod tests {
strip_response_item_id(raw_history_item.item), strip_response_item_id(raw_history_item.item),
expected_history_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 { let invocation = ToolInvocation {
session, session,
step_context, step_context,
@@ -497,6 +506,14 @@ mod tests {
); );
assert_eq!(captured_call.model, model); assert_eq!(captured_call.model, model);
assert_eq!(captured_call.truncation_policy, truncation_policy); 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!( assert_eq!(
captured_call.source, captured_call.source,
ExtensionToolCallSource::CodeMode { ExtensionToolCallSource::CodeMode {
+4 -3
View File
@@ -207,7 +207,6 @@ impl McpHandler {
payload, payload,
.. ..
} = invocation; } = invocation;
let turn = Arc::clone(&step_context.turn);
let payload = match payload { let payload = match payload {
ToolPayload::Function { arguments } => arguments, ToolPayload::Function { arguments } => arguments,
@@ -223,7 +222,7 @@ impl McpHandler {
.as_ref() .as_ref()
.and_then(codex_mcp::PreparedMcpCall::output_token_limit) .and_then(codex_mcp::PreparedMcpCall::output_token_limit)
.map(TruncationPolicy::Tokens) .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 started = Instant::now();
let result = handle_mcp_tool_call( let result = handle_mcp_tool_call(
Arc::clone(&session), Arc::clone(&session),
@@ -243,7 +242,9 @@ impl McpHandler {
result: result.result, result: result.result,
tool_input: result.tool_input, tool_input: result.tool_input,
wall_time: started.elapsed(), 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, truncation_policy,
})) }))
} }
@@ -26,6 +26,7 @@ use serde_json::Value;
use crate::function_tool::FunctionCallError; use crate::function_tool::FunctionCallError;
use crate::session::session::Session; use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnContext;
use crate::tools::context::FunctionToolOutput; use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolOutput; use crate::tools::context::ToolOutput;
@@ -279,7 +280,7 @@ async fn emit_tool_call_end(
async fn run_resource_operation<T>( async fn run_resource_operation<T>(
session: &Arc<Session>, session: &Arc<Session>,
turn: &TurnContext, step_context: &StepContext,
call_id: &str, call_id: &str,
invocation: McpInvocation, invocation: McpInvocation,
operation: impl Future<Output = Result<T, FunctionCallError>>, operation: impl Future<Output = Result<T, FunctionCallError>>,
@@ -287,10 +288,14 @@ async fn run_resource_operation<T>(
where where
T: Serialize, T: Serialize,
{ {
let turn = &step_context.turn;
emit_tool_call_begin(session, turn, call_id, invocation.clone()).await; emit_tool_call_begin(session, turn, call_id, invocation.clone()).await;
let start = Instant::now(); let start = Instant::now();
let result = operation.await.and_then(|payload| { 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 { match result {
@@ -72,7 +72,7 @@ impl ListMcpResourceTemplatesHandler {
arguments: arguments.clone(), 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())? { if let Some((server_name, params)) = args.target(turn.as_ref())? {
let result = mcp let result = mcp
.list_resource_templates(&server_name, params) .list_resource_templates(&server_name, params)
@@ -72,7 +72,7 @@ impl ListMcpResourcesHandler {
arguments: arguments.clone(), 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())? { if let Some((server_name, params)) = args.target(turn.as_ref())? {
let result = mcp let result = mcp
.list_resources(&server_name, params) .list_resources(&server_name, params)
@@ -77,7 +77,7 @@ impl ReadMcpResourceHandler {
arguments: arguments.clone(), 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)?; ensure_model_can_access_mcp_server(turn.as_ref(), &server)?;
let result = mcp let result = mcp
.read_resource(&server, ReadResourceRequestParams::new(uri.clone())) .read_resource(&server, ReadResourceRequestParams::new(uri.clone()))
@@ -220,7 +220,7 @@ impl RequestPluginInstallHandler {
.analytics_events_client .analytics_events_client
.track_plugin_install_requested( .track_plugin_install_requested(
build_track_events_context( build_track_events_context(
turn.model_info().slug.clone(), step_context.settings.model_info.slug.clone(),
session.thread_id.to_string(), session.thread_id.to_string(),
turn.sub_id.clone(), turn.sub_id.clone(),
turn.originator.clone(), turn.originator.clone(),
@@ -283,14 +283,16 @@ impl RequestPluginInstallHandler {
Some(_) => "unknown", Some(_) => "unknown",
None => "unavailable", None => "unavailable",
}; };
turn.session_telemetry.record_plugin_install_suggestion( step_context
tool_type, .session_telemetry
tool.id(), .record_plugin_install_suggestion(
tool.name(), tool_type,
response_action, tool.id(),
user_confirmed, tool.name(),
completed, response_action,
); user_confirmed,
completed,
);
} }
let content = serde_json::to_string(&RequestPluginInstallResult { let content = serde_json::to_string(&RequestPluginInstallResult {
@@ -407,7 +407,7 @@ impl ExecCommandHandler {
chunk_id: String::new(), chunk_id: String::new(),
wall_time: std::time::Duration::ZERO, wall_time: std::time::Duration::ZERO,
raw_output: output.into_text().into_bytes(), 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, max_output_tokens,
process_id: None, process_id: None,
exit_code: 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 { let request = ExecCommandRequest {
command, command,
shell_type, shell_type,
@@ -461,7 +461,7 @@ impl ExecCommandHandler {
chunk_id: generate_chunk_id(), chunk_id: generate_chunk_id(),
wall_time: output.duration, wall_time: output.duration,
raw_output: output_text.into_bytes(), 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, max_output_tokens,
// Sandbox denial is terminal, so there is no live // Sandbox denial is terminal, so there is no live
// process for write_stdin to resume. // process for write_stdin to resume.
@@ -91,7 +91,12 @@ impl WriteStdinHandler {
input: &args.chars, input: &args.chars,
yield_time_ms: args.yield_time_ms, yield_time_ms: args.yield_time_ms,
max_output_tokens: args.max_output_tokens, 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 { interaction_event: Some(WriteStdinInteractionEvent {
session: &session, session: &session,
turn: &turn, turn: &turn,
@@ -95,8 +95,9 @@ impl ViewImageHandler {
invocation: ToolInvocation, invocation: ToolInvocation,
) -> Result<Box<dyn crate::tools::context::ToolOutput>, FunctionCallError> { ) -> Result<Box<dyn crate::tools::context::ToolOutput>, FunctionCallError> {
if !invocation if !invocation
.turn .step_context
.model_info() .settings
.model_info
.input_modalities .input_modalities
.contains(&InputModality::Image) .contains(&InputModality::Image)
{ {
@@ -185,7 +186,8 @@ impl ViewImageHandler {
FunctionCallError::RespondToModel(VIEW_IMAGE_INVALID_MESSAGE.to_string()) 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 let use_original_detail = self.options.unified_image_budget
|| can_request_original_detail && matches!(detail, Some(ViewImageDetail::Original)); || can_request_original_detail && matches!(detail, Some(ViewImageDetail::Original));
let image_detail = if use_original_detail { let image_detail = if use_original_detail {
+1 -1
View File
@@ -499,7 +499,7 @@ impl ToolRegistry {
) -> Result<AnyToolResult, FunctionCallError> { ) -> Result<AnyToolResult, FunctionCallError> {
let tool_name = invocation.tool_name.clone(); let tool_name = invocation.tool_name.clone();
let call_id_owned = invocation.call_id.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 // TODO(anp): Reconcile these tags with TurnEnvironment::sandbox_context
// instead of reporting the thread-wide backend for environment-scoped tools. // instead of reporting the thread-wide backend for environment-scoped tools.
let sandbox_tags = invocation.turn.turn_metadata_state.sandbox_tags; let sandbox_tags = invocation.turn.turn_metadata_state.sandbox_tags;
-2
View File
@@ -160,7 +160,6 @@ impl ToolRouter {
} }
/// Whether the model can both start and interact with a terminal process. /// 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))] #[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn has_terminal_controls(&self) -> bool { pub(crate) fn has_terminal_controls(&self) -> bool {
self.exposes_tool(&ToolName::plain("exec_command")) self.exposes_tool(&ToolName::plain("exec_command"))
@@ -168,7 +167,6 @@ impl ToolRouter {
} }
/// Whether the configured collaboration backend's child-management tools remain exposed. /// 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))] #[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn can_manage_children(&self) -> bool { pub(crate) fn can_manage_children(&self) -> bool {
self.can_manage_children self.can_manage_children
@@ -24,6 +24,7 @@ use crate::unified_exec::head_tail_buffer::HeadTailBuffer;
use codex_core_plugins::PluginCommandAttribution; use codex_core_plugins::PluginCommandAttribution;
use codex_protocol::exec_output::ExecToolCallOutput; use codex_protocol::exec_output::ExecToolCallOutput;
use codex_protocol::exec_output::StreamOutput; use codex_protocol::exec_output::StreamOutput;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecCommandOutputDeltaEvent; use codex_protocol::protocol::ExecCommandOutputDeltaEvent;
use codex_protocol::protocol::ExecCommandSource; use codex_protocol::protocol::ExecCommandSource;
@@ -164,9 +165,7 @@ pub(crate) fn start_streaming_output(
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
pub(crate) fn spawn_exit_watcher( pub(crate) fn spawn_exit_watcher(
process: Arc<UnifiedExecProcess>, process: Arc<UnifiedExecProcess>,
session_ref: Arc<Session>, context: &UnifiedExecContext,
turn_ref: Arc<TurnContext>,
call_id: String,
command: Vec<String>, command: Vec<String>,
cwd: PathUri, cwd: PathUri,
process_id: i32, process_id: i32,
@@ -176,6 +175,10 @@ pub(crate) fn spawn_exit_watcher(
network_denial_monitor: Option<tokio::task::JoinHandle<()>>, network_denial_monitor: Option<tokio::task::JoinHandle<()>>,
plugin_metrics_sidecar: Option<SharedPluginMetricsSidecar>, plugin_metrics_sidecar: Option<SharedPluginMetricsSidecar>,
) { ) {
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 exit_token = process.cancellation_token();
let output_drained = process.output_drained_notify(); let output_drained = process.output_drained_notify();
let interaction_lock = process.interaction_lock(); let interaction_lock = process.interaction_lock();
@@ -200,6 +203,7 @@ pub(crate) fn spawn_exit_watcher(
emit_failed_exec_end_for_unified_exec( emit_failed_exec_end_for_unified_exec(
session_ref, session_ref,
turn_ref, turn_ref,
model_info,
call_id, call_id,
command, command,
cwd, cwd,
@@ -225,6 +229,7 @@ pub(crate) fn spawn_exit_watcher(
emit_exec_end_for_unified_exec( emit_exec_end_for_unified_exec(
session_ref, session_ref,
turn_ref, turn_ref,
model_info,
call_id, call_id,
command, command,
cwd, cwd,
@@ -337,6 +342,7 @@ impl Emitter {
pub(crate) async fn emit_exec_end_for_unified_exec( pub(crate) async fn emit_exec_end_for_unified_exec(
session_ref: Arc<Session>, session_ref: Arc<Session>,
turn_ref: Arc<TurnContext>, turn_ref: Arc<TurnContext>,
model_info: Arc<ModelInfo>,
call_id: String, call_id: String,
command: Vec<String>, command: Vec<String>,
cwd: PathUri, cwd: PathUri,
@@ -360,6 +366,7 @@ pub(crate) async fn emit_exec_end_for_unified_exec(
let event_ctx = ToolEventCtx::new( let event_ctx = ToolEventCtx::new(
session_ref.as_ref(), session_ref.as_ref(),
turn_ref.as_ref(), turn_ref.as_ref(),
&model_info,
&call_id, &call_id,
/*turn_diff_tracker*/ None, /*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( pub(crate) async fn emit_failed_exec_end_for_unified_exec(
session_ref: Arc<Session>, session_ref: Arc<Session>,
turn_ref: Arc<TurnContext>, turn_ref: Arc<TurnContext>,
model_info: Arc<ModelInfo>,
call_id: String, call_id: String,
command: Vec<String>, command: Vec<String>,
cwd: PathUri, cwd: PathUri,
@@ -416,6 +424,7 @@ pub(crate) async fn emit_failed_exec_end_for_unified_exec(
let event_ctx = ToolEventCtx::new( let event_ctx = ToolEventCtx::new(
session_ref.as_ref(), session_ref.as_ref(),
turn_ref.as_ref(), turn_ref.as_ref(),
&model_info,
&call_id, &call_id,
/*turn_diff_tracker*/ None, /*turn_diff_tracker*/ None,
); );
@@ -208,7 +208,7 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() ->
stdout_tx, stdout_tx,
exit_tx, exit_tx,
transcript, transcript,
context, mut context,
rx_event, rx_event,
} = streaming_output_harness().await?; } = streaming_output_harness().await?;
@@ -226,11 +226,15 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() ->
#[allow(deprecated)] #[allow(deprecated)]
let cwd = context.step_context.turn.cwd.clone().into(); 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( spawn_exit_watcher(
Arc::clone(&process), Arc::clone(&process),
Arc::clone(&context.session), &context,
Arc::clone(&context.step_context.turn),
context.call_id,
vec!["proof".to_string()], vec!["proof".to_string()],
cwd, cwd,
/*process_id*/ 123, /*process_id*/ 123,
@@ -266,6 +270,13 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() ->
Some("LATE_DENIAL") 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!( assert!(
elapsed >= Duration::from_millis(10) && elapsed < TRAILING_OUTPUT_GRACE, elapsed >= Duration::from_millis(10) && elapsed < TRAILING_OUTPUT_GRACE,
"completion should wait for denial without falling back to the output grace: {elapsed:?}" "completion should wait for denial without falling back to the output grace: {elapsed:?}"
@@ -406,6 +406,7 @@ async fn emit_failed_initial_exec_end_if_unstored(
emit_failed_exec_end_for_unified_exec( emit_failed_exec_end_for_unified_exec(
Arc::clone(&context.session), Arc::clone(&context.session),
Arc::clone(&context.step_context.turn), Arc::clone(&context.step_context.turn),
Arc::clone(&context.step_context.settings.model_info),
context.call_id.clone(), context.call_id.clone(),
request.command.clone(), request.command.clone(),
cwd, cwd,
@@ -532,6 +533,7 @@ impl UnifiedExecProcessManager {
let event_ctx = ToolEventCtx::new( let event_ctx = ToolEventCtx::new(
context.session.as_ref(), context.session.as_ref(),
context.step_context.turn.as_ref(), context.step_context.turn.as_ref(),
&context.step_context.settings.model_info,
&context.call_id, &context.call_id,
/*turn_diff_tracker*/ None, /*turn_diff_tracker*/ None,
); );
@@ -766,6 +768,7 @@ impl UnifiedExecProcessManager {
emit_exec_end_for_unified_exec( emit_exec_end_for_unified_exec(
Arc::clone(&context.session), Arc::clone(&context.session),
Arc::clone(&context.step_context.turn), Arc::clone(&context.step_context.turn),
Arc::clone(&context.step_context.settings.model_info),
context.call_id.clone(), context.call_id.clone(),
request.command.clone(), request.command.clone(),
cwd.clone(), cwd.clone(),
@@ -796,8 +799,8 @@ impl UnifiedExecProcessManager {
raw_output: collected, raw_output: collected,
truncation_policy: context truncation_policy: context
.step_context .step_context
.turn .settings
.model_info() .model_info
.truncation_policy .truncation_policy
.into(), .into(),
max_output_tokens: request.max_output_tokens, max_output_tokens: request.max_output_tokens,
@@ -1175,9 +1178,7 @@ impl UnifiedExecProcessManager {
spawn_exit_watcher( spawn_exit_watcher(
Arc::clone(&process), Arc::clone(&process),
Arc::clone(&context.session), context,
Arc::clone(&context.step_context.turn),
context.call_id.clone(),
command.to_vec(), command.to_vec(),
cwd, cwd,
process_id, process_id,
File diff suppressed because it is too large. Load diff