From b64de2f3ad718ebee731aa8a5e1d4ace0594efc5 Mon Sep 17 00:00:00 2001 From: Abhinav Date: Wed, 9 Sep 2026 17:25:47 +0000 Subject: [PATCH] Use the originating model when recording conversation history (#44243) ## Why History recording used the model captured at turn start, so switching models within a turn could apply stale image preparation and tool-output truncation rules. ## What changed Pass explicit model metadata through conversation recording. Record model responses and tool results using their originating step's settings, and capture the current model for pending input recorded during startup interruption or task completion. ## Testing Add regression coverage for tool results across a model switch, explicit model selection for plain and annotated history, and image input retained during interruption and completion. Extend model-switching tests to verify that prepared media and item identity survive switching back to a multimodal model in live, resumed, and forked threads. GitOrigin-RevId: 5bc750c30c334acb9b802cb80fdea01daffa84a1 --- codex-rs/core/src/agent/control_tests.rs | 10 +- codex-rs/core/src/compact.rs | 8 +- .../core/src/guardian/review_session_tests.rs | 4 +- codex-rs/core/src/guardian/tests.rs | 9 +- codex-rs/core/src/hook_runtime.rs | 17 +- codex-rs/core/src/prompt_debug.rs | 8 +- codex-rs/core/src/session/inject.rs | 20 +- codex-rs/core/src/session/inject_tests.rs | 6 +- codex-rs/core/src/session/mod.rs | 64 +++--- codex-rs/core/src/session/reasoning_effort.rs | 1 + codex-rs/core/src/session/rollout_budget.rs | 8 +- codex-rs/core/src/session/tests.rs | 163 ++++++++++++--- codex-rs/core/src/session/time_reminder.rs | 8 +- codex-rs/core/src/session/token_budget.rs | 16 +- codex-rs/core/src/session/turn.rs | 75 +++++-- codex-rs/core/src/session/turn_context.rs | 5 + codex-rs/core/src/session/turn_tests.rs | 3 +- codex-rs/core/src/stream_events_utils.rs | 38 ++-- .../core/src/stream_events_utils_tests.rs | 5 +- codex-rs/core/src/tasks/mod.rs | 2 + codex-rs/core/src/tasks/regular.rs | 9 +- codex-rs/core/src/tasks/review.rs | 2 + codex-rs/core/src/tasks/user_shell.rs | 6 +- .../src/tools/handlers/extension_tools.rs | 6 +- codex-rs/core/src/tools/router_tests.rs | 6 +- codex-rs/core/tests/suite/model_switching.rs | 86 +++++++- codex-rs/core/tests/suite/rmcp_client.rs | 72 ++++++- codex-rs/core/tests/suite/step_settings.rs | 187 ++++++++++++++++++ 28 files changed, 721 insertions(+), 123 deletions(-) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index dc5d668994..01c3737f4a 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -1173,6 +1173,7 @@ async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix() { .session .record_conversation_items( turn_context.as_ref(), + turn_context.model_info(), &[spawn_agent_call(&parent_spawn_call_id)], ) .await; @@ -1668,7 +1669,7 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { parent_thread .session .record_conversation_items( - turn_context.as_ref(), + turn_context.as_ref(), turn_context.model_info(), &[ ResponseItem::Message { id: None, @@ -2561,6 +2562,7 @@ async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { .session .record_conversation_items( turn_context.as_ref(), + turn_context.model_info(), &[ assistant_message("unflushed final answer", Some(MessagePhase::FinalAnswer)), spawn_agent_call(&parent_spawn_call_id), @@ -2633,6 +2635,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { .session .record_conversation_items( queued_turn_context.as_ref(), + queued_turn_context.model_info(), &[queued_communication.to_response_input_item().into()], ) .await; @@ -2649,6 +2652,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { .session .record_conversation_items( triggered_turn_context.as_ref(), + triggered_turn_context.model_info(), &[triggered_communication.to_response_input_item().into()], ) .await; @@ -2662,6 +2666,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { .session .record_conversation_items( spawn_turn_context.as_ref(), + spawn_turn_context.model_info(), &[spawn_agent_call(&parent_spawn_call_id)], ) .await; @@ -2774,6 +2779,7 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li .session .record_conversation_items( startup_turn_context.as_ref(), + startup_turn_context.model_info(), &[ResponseItem::Message { id: None, role: "developer".to_string(), @@ -2795,6 +2801,7 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li .session .record_conversation_items( spawn_turn_context.as_ref(), + spawn_turn_context.model_info(), &[spawn_agent_call(&parent_spawn_call_id)], ) .await; @@ -2902,6 +2909,7 @@ async fn spawn_agent_fork_last_n_turns_strips_parent_usage_hints() { .session .record_conversation_items( turn_context.as_ref(), + turn_context.model_info(), &[ ResponseItem::Message { id: None, diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 7155486919..4c6a26622a 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -791,8 +791,12 @@ async fn drain_to_completed( }; match event { Ok(ResponseEvent::OutputItemDone(item)) => { - sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&item), + ) + .await; } Ok(ResponseEvent::ServerReasoningIncluded(included)) => { sess.set_server_reasoning_included(included).await; diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 67ce82fed2..e028d32028 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -49,7 +49,9 @@ async fn run_review_preserves_evidence_during_parent_compaction() { "type": "function_call_output", "call_id": "prior-inspection", "output": EVIDENCE })) .unwrap(); - parent.record_conversation_items(&turn, &[evidence]).await; + parent + .record_conversation_items(&turn, turn.model_info(), &[evidence]) + .await; params.parent_history = parent.clone_history().await; // An idle, prewarmed reviewer is a normal manager state. diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 7494bc814b..c4d85a5e88 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -248,6 +248,7 @@ async fn seed_guardian_parent_history(session: &Arc, turn: &Arc anyh session .record_conversation_items( turn.as_ref(), + turn.model_info(), &[ ResponseItem::Message { id: None, @@ -783,6 +785,7 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() - session .record_conversation_items( turn.as_ref(), + turn.model_info(), &[ ResponseItem::Message { id: None, @@ -2129,6 +2132,7 @@ async fn guardian_review_request_layout_matches_model_visible_request_snapshot() session .record_conversation_items( turn.as_ref(), + turn.model_info(), &[ResponseItem::Message { id: None, role: "user".to_string(), @@ -2434,6 +2438,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: session .record_conversation_items( turn.as_ref(), + turn.model_info(), &[ ResponseItem::Message { id: None, @@ -3392,7 +3397,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() -> ); session .record_conversation_items( - turn.as_ref(), + turn.as_ref(), turn.model_info(), &[ ResponseItem::Message { id: None, @@ -3468,7 +3473,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() -> ); session .record_conversation_items( - turn.as_ref(), + turn.as_ref(), turn.model_info(), &[ ResponseItem::Message { id: None, diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index e17f1508e7..f6ec599c5d 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -33,6 +33,7 @@ use codex_protocol::items::FunctionCallOutputItem; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; @@ -700,6 +701,7 @@ pub(crate) async fn inspect_pending_input( pub(crate) async fn record_pending_input( sess: &Arc, turn_context: &Arc, + model_info: &ModelInfo, pending_input: TurnInput, additional_contexts: Vec, persist_context: PersistContext, @@ -712,6 +714,7 @@ pub(crate) async fn record_pending_input( } => { sess.record_user_prompt_and_emit_turn_item( turn_context.as_ref(), + model_info, content.as_slice(), client_id, acceptance_order, @@ -720,11 +723,11 @@ pub(crate) async fn record_pending_input( .await; } TurnInput::ResponseItem(item) => { - sess.record_annotated_conversation_items(turn_context, vec![item]) + sess.record_annotated_conversation_items(turn_context, model_info, vec![item]) .await; } TurnInput::FunctionCallOutput(item) => { - sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) + sess.record_conversation_items(turn_context, model_info, std::slice::from_ref(&item)) .await; if let ResponseItem::FunctionCallOutput { id: Some(id), @@ -746,7 +749,7 @@ pub(crate) async fn record_pending_input( sess.ensure_rollout_materialized(persist_context).await; } TurnInput::InterAgentCommunication(communication) => { - sess.record_inter_agent_communication(turn_context, communication) + sess.record_inter_agent_communication(turn_context, model_info, communication) .await; } } @@ -837,8 +840,12 @@ pub(crate) async fn record_additional_contexts( return; } - sess.record_conversation_items(turn_context, developer_messages.as_slice()) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + developer_messages.as_slice(), + ) + .await; } fn additional_context_messages(additional_contexts: Vec) -> Vec { diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index abb1cdd1ec..fd2b3a230e 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -95,8 +95,12 @@ pub(crate) async fn build_prompt_input_from_session( if !input.is_empty() { let response_item = sess.response_item_from_user_input(input); - sess.record_conversation_items(turn_context.as_ref(), std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + turn_context.as_ref(), + &step_context.settings.model_info, + std::slice::from_ref(&response_item), + ) + .await; } let prompt_input = sess diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index 343713bf19..b2d0c05673 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -5,6 +5,7 @@ use codex_features::Feature; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ModelInfo; impl Session { /// Returns the input if there is no active turn to inject into. @@ -92,7 +93,7 @@ impl Session { return; } drop(active); - self.record_annotated_conversation_items(turn_context, items) + self.record_annotated_conversation_items(turn_context, turn_context.model_info(), items) .await; } @@ -110,6 +111,7 @@ impl Session { pub(crate) async fn record_annotated_conversation_items( &self, turn_context: &TurnContext, + model_info: &ModelInfo, items: Vec, ) { if items.iter().all(|item| item.metadata.is_none()) { @@ -117,7 +119,8 @@ impl Session { .into_iter() .map(ResponseItemEnvelope::into_item) .collect::>(); - self.record_conversation_items(turn_context, &items).await; + self.record_conversation_items(turn_context, model_info, &items) + .await; return; } @@ -126,6 +129,7 @@ impl Session { for envelope in items { let (prepared_items, prepared_images) = self.prepare_conversation_items_for_history( turn_context, + model_info, std::slice::from_ref(&envelope.item), ); image_preparations.extend(prepared_images); @@ -138,8 +142,13 @@ impl Session { } })); } - self.record_prepared_conversation_items(turn_context, annotated_items, image_preparations) - .await; + self.record_prepared_conversation_items( + turn_context, + model_info, + annotated_items, + image_preparations, + ) + .await; } /// Injects items into active work, or records them without starting a turn. @@ -159,7 +168,8 @@ impl Session { default_turn_context.as_ref() } }; - self.record_conversation_items(turn_context, &items).await; + self.record_conversation_items(turn_context, turn_context.model_info(), &items) + .await; } } diff --git a/codex-rs/core/src/session/inject_tests.rs b/codex-rs/core/src/session/inject_tests.rs index 80796dc882..92446c1d0e 100644 --- a/codex-rs/core/src/session/inject_tests.rs +++ b/codex-rs/core/src/session/inject_tests.rs @@ -26,7 +26,11 @@ async fn harness_authored_configuration_updates_preserve_metadata_and_resume() { }), }; session - .record_annotated_conversation_items(&turn_context, vec![expected.clone()]) + .record_annotated_conversation_items( + &turn_context, + turn_context.model_info(), + vec![expected.clone()], + ) .await; let recorded = session.clone_history().await.into_annotated_items(); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 0c83f723fe..15d24f4129 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3293,22 +3293,20 @@ impl Session { item.set_create_time_if_missing(Self::response_item_create_time()); } - /// Records conversation items: append to history, persist to rollout, and - /// notify clients observing raw response items. + /// Prepares media using the originating model and preserves existing item identity. pub(crate) fn prepare_conversation_items_for_history<'a>( &self, turn_context: &TurnContext, + model_info: &ModelInfo, items: &'a [ResponseItem], ) -> (Cow<'a, [ResponseItem]>, Vec) { let mut items = items.to_vec(); - let image_preparation_mode = if unified_image_budget_enabled( - &turn_context.config.features, - turn_context.model_info(), - ) { - ImagePreparationMode::UnifiedBudget - } else { - ImagePreparationMode::DetailBased - }; + let image_preparation_mode = + if unified_image_budget_enabled(&turn_context.config.features, model_info) { + ImagePreparationMode::UnifiedBudget + } else { + ImagePreparationMode::DetailBased + }; let image_resize_notice_mode = if turn_context .config .features @@ -3402,26 +3400,35 @@ impl Session { item } + /// Appends to history, persists the prepared items, then notifies raw-item observers. + /// Execution callers supply their captured model; standalone callers need no tool runtime. #[tracing::instrument(level = "trace", skip_all, fields(item_count = items.len()))] pub(crate) async fn record_conversation_items( &self, turn_context: &TurnContext, + model_info: &ModelInfo, items: &[ResponseItem], ) { let (items, image_preparations) = - self.prepare_conversation_items_for_history(turn_context, items); + self.prepare_conversation_items_for_history(turn_context, model_info, items); let items = items .into_owned() .into_iter() .map(ResponseItemEnvelope::new) .collect(); - self.record_prepared_conversation_items(turn_context, items, image_preparations) - .await; + self.record_prepared_conversation_items( + turn_context, + model_info, + items, + image_preparations, + ) + .await; } async fn record_prepared_conversation_items( &self, turn_context: &TurnContext, + model_info: &ModelInfo, items: Vec, image_preparations: Vec, ) { @@ -3436,7 +3443,7 @@ impl Session { .note_recorded_items(&response_items); state .history - .record_annotated_items(&items, turn_context.model_info().truncation_policy.into()); + .record_annotated_items(&items, model_info.truncation_policy.into()); } for image in image_preparations { self.services @@ -3478,7 +3485,8 @@ impl Session { world_state.render_diff(&previous_snapshot), ); if !items.is_empty() { - self.record_conversation_items(turn_context, &items).await; + self.record_conversation_items(turn_context, &step_context.settings.model_info, &items) + .await; } // ContextManager remembers this for later turns; run_turn owns the live value. @@ -3713,11 +3721,13 @@ impl Session { pub(crate) async fn record_inter_agent_communication( &self, turn_context: &TurnContext, + model_info: &ModelInfo, communication: InterAgentCommunication, ) { let response_item = communication.to_model_input_item(); let (items, _) = self.prepare_conversation_items_for_history( turn_context, + model_info, std::slice::from_ref(&response_item), ); let items = items.as_ref(); @@ -3725,10 +3735,7 @@ impl Session { { let mut state = self.state.lock().await; state.current_time_reminder.note_recorded_items(items); - state.record_items( - items.iter(), - turn_context.model_info().truncation_policy.into(), - ); + state.record_items(items.iter(), model_info.truncation_policy.into()); } self.persist_rollout_items(&[ RolloutItem::InterAgentCommunicationMetadata { @@ -4390,8 +4397,12 @@ impl Session { return Ok(world_state); } if !context_items.is_empty() { - self.record_conversation_items(turn_context, &context_items) - .await; + self.record_conversation_items( + turn_context, + &step_context.settings.model_info, + &context_items, + ) + .await; } // Persist state only after any model-visible context generated from it. if let Some(world_state_item) = world_state_item { @@ -4590,11 +4601,16 @@ impl Session { pub(crate) async fn record_response_item_and_emit_turn_item( &self, turn_context: &TurnContext, + model_info: &ModelInfo, response_item: ResponseItem, ) { // Add to conversation history and persist response item to rollout. - self.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) - .await; + self.record_conversation_items( + turn_context, + model_info, + std::slice::from_ref(&response_item), + ) + .await; // Derive a turn item and emit lifecycle events if applicable. if let Some(item) = parse_turn_item(&response_item) { @@ -4606,6 +4622,7 @@ impl Session { pub(crate) async fn record_user_prompt_and_emit_turn_item( &self, turn_context: &TurnContext, + model_info: &ModelInfo, input: &[UserInput], client_id: Option, acceptance_order: Option, @@ -4617,6 +4634,7 @@ impl Session { let response_item = self.response_item_from_user_input(input.to_vec()); self.record_annotated_conversation_items( turn_context, + model_info, vec![ResponseItemEnvelope { item: response_item, metadata: acceptance_order.map(|order| CodexHarnessMetadata { diff --git a/codex-rs/core/src/session/reasoning_effort.rs b/codex-rs/core/src/session/reasoning_effort.rs index 9e61936063..ec945f94a3 100644 --- a/codex-rs/core/src/session/reasoning_effort.rs +++ b/codex-rs/core/src/session/reasoning_effort.rs @@ -69,6 +69,7 @@ impl Session { self.record_annotated_conversation_items( step_context.turn.as_ref(), + &step_context.settings.model_info, vec![ResponseItemEnvelope { item: ResponseItem::ConfigurationUpdate { reasoning: ConfigurationReasoning { effort }, diff --git a/codex-rs/core/src/session/rollout_budget.rs b/codex-rs/core/src/session/rollout_budget.rs index 1867d7e3ec..c5c640927c 100644 --- a/codex-rs/core/src/session/rollout_budget.rs +++ b/codex-rs/core/src/session/rollout_budget.rs @@ -17,8 +17,12 @@ pub(super) async fn maybe_record_reminder( let response_item = ContextualUserFragment::into(crate::context::RolloutBudgetContext { remaining_tokens: reminder.remaining_tokens, }); - sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&response_item), + ) + .await; budget.mark_reminder_delivered(sess.thread_id(), window_id, reminder); } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index b7d67a461a..b36f4329a9 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -355,6 +355,7 @@ async fn default_turn_context_assigns_missing_response_item_ids() { let (items, _) = session.prepare_conversation_items_for_history( &turn_context, + turn_context.model_info(), std::slice::from_ref(&response_item), ); @@ -2275,7 +2276,11 @@ async fn record_conversation_items_stamps_missing_turn_id_and_preserves_existing existing_item.set_turn_id_if_missing("older-turn"); session - .record_conversation_items(&turn_context, &[fresh_item.clone(), existing_item.clone()]) + .record_conversation_items( + &turn_context, + turn_context.model_info(), + &[fresh_item.clone(), existing_item.clone()], + ) .await; let history = session.clone_history().await; @@ -2308,7 +2313,11 @@ async fn record_response_item_and_emit_turn_item_emits_hook_prompt_lifecycle() { let response_item_id = response_item.id().expect("hook prompt id").to_string(); session - .record_response_item_and_emit_turn_item(&turn_context, response_item) + .record_response_item_and_emit_turn_item( + &turn_context, + turn_context.model_info(), + response_item, + ) .await; let raw_response = rx.recv().await.expect("raw response item event"); @@ -2391,7 +2400,7 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() { expected_item.set_turn_id_if_missing(&turn_context.sub_id); session - .record_inter_agent_communication(&turn_context, communication) + .record_inter_agent_communication(&turn_context, turn_context.model_info(), communication) .await; let recorded_history = session.clone_history().await; @@ -2468,7 +2477,7 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum ); session - .record_inter_agent_communication(&turn_context, communication) + .record_inter_agent_communication(&turn_context, turn_context.model_info(), communication) .await; let live_history = session.clone_history().await; @@ -2523,6 +2532,58 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum ); } +#[test_case::test_case(false; "plain items")] +#[test_case::test_case(true; "annotated items")] +#[tokio::test] +async fn annotated_history_uses_explicit_model_without_a_step(retain_metadata: bool) { + let (session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| { + config + .features + .enable(Feature::RetainClientDeveloperMessages) + .unwrap(); + }, + ) + .await; + let mut model_info = turn_context.model_info().as_ref().clone(); + model_info.truncation_policy = + codex_protocol::openai_models::TruncationPolicyConfig::tokens(/*limit*/ 4); + let text = "diagnostic line\n".repeat(50); + let mut item = ResponseItem::FunctionCallOutput { + id: Some(ResponseItemId::with_suffix("fco", "existing")), + call_id: Some("call-1".to_string()), + name: None, + namespace: None, + output: FunctionCallOutputPayload { + body: FunctionCallOutputBody::Text(text.clone()), + success: Some(true), + }, + internal_chat_message_metadata_passthrough: None, + }; + item.set_turn_id_if_missing("existing-turn"); + item.set_create_time_if_missing(123.into()); + let mut expected = ResponseItemEnvelope { + item, + metadata: retain_metadata.then_some(CodexHarnessMetadata { + client_authored: true, + ..Default::default() + }), + }; + session + .record_annotated_conversation_items(&turn_context, &model_info, vec![expected.clone()]) + .await; + let ResponseItem::FunctionCallOutput { output, .. } = &mut expected.item else { + unreachable!("fixture is a tool output"); + }; + output.body = FunctionCallOutputBody::Text(codex_utils_output_truncation::truncate_text( + &text, + codex_utils_output_truncation::TruncationPolicy::Tokens(4) * 1.2, + )); + assert_eq!(session.clone_history().await.annotated_items(), &[expected]); +} + #[tokio::test] async fn prepares_image_failures_before_history_insertion() { let (session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( @@ -2556,7 +2617,11 @@ async fn prepares_image_failures_before_history_insertion() { }; session - .record_conversation_items(turn_context.as_ref(), std::slice::from_ref(&item)) + .record_conversation_items( + turn_context.as_ref(), + turn_context.model_info(), + std::slice::from_ref(&item), + ) .await; let history = session.state.lock().await.clone_history(); @@ -2948,7 +3013,11 @@ async fn recompute_token_usage_uses_session_base_instructions() { let item = user_message("hello"); session - .record_conversation_items(&turn_context, std::slice::from_ref(&item)) + .record_conversation_items( + &turn_context, + turn_context.model_info(), + std::slice::from_ref(&item), + ) .await; let history = session.clone_history().await; @@ -3956,7 +4025,7 @@ async fn thread_rollback_fails_without_persisted_thread_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = build_initial_context(&sess, &tc).await; - sess.record_conversation_items(tc.as_ref(), &initial_context) + sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context) .await; let history_before_rollback = sess.clone_history().await; @@ -4400,7 +4469,7 @@ async fn thread_rollback_fails_when_turn_in_progress() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = build_initial_context(&sess, &tc).await; - sess.record_conversation_items(tc.as_ref(), &initial_context) + sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context) .await; let history_before_rollback = sess.clone_history().await; @@ -4425,7 +4494,7 @@ async fn thread_rollback_fails_when_num_turns_is_zero() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = build_initial_context(&sess, &tc).await; - sess.record_conversation_items(tc.as_ref(), &initial_context) + sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context) .await; let history_before_rollback = sess.clone_history().await; @@ -10658,7 +10727,11 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co internal_chat_message_metadata_passthrough: None, }; session - .record_conversation_items(&turn_context, std::slice::from_ref(&compacted_summary)) + .record_conversation_items( + &turn_context, + turn_context.model_info(), + std::slice::from_ref(&compacted_summary), + ) .await; session .record_context_updates_and_set_reference_context_item(&step_context) @@ -11414,7 +11487,11 @@ async fn legacy_compaction_retains_only_the_selected_step(first_attempt: FirstAt let server = responses::start_mock_server().await; let (session, turn, events) = make_remote_compaction_session(&server.uri()).await; session - .record_conversation_items(&turn, &[user_message("before compaction")]) + .record_conversation_items( + &turn, + turn.model_info(), + &[user_message("before compaction")], + ) .await; session .spawn_task( @@ -11535,7 +11612,11 @@ async fn interrupting_compaction_fallback_retains_last_known_step_context() { })) .await; session - .record_conversation_items(&turn, &[user_message("before compaction")]) + .record_conversation_items( + &turn, + turn.model_info(), + &[user_message("before compaction")], + ) .await; session .spawn_task(turn, Vec::new(), crate::tasks::RegularTask::new()) @@ -11870,7 +11951,22 @@ async fn submit_steer_only( #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() { - let (sess, tc, rx) = make_session_and_context_with_rx().await; + let (sess, mut tc, rx) = make_session_and_context_with_auth_and_config_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| { + config.features.enable(Feature::UnifiedImageBudget).unwrap(); + }, + ) + .await; + update_turn_settings_for_test( + Arc::get_mut(&mut tc).expect("unshared context"), + |settings| { + let model = Arc::make_mut(&mut settings.model_info); + model.use_responses_lite = false; + model.supports_image_detail_original = false; + }, + ); let input = vec![TurnInput::UserInput { acceptance_order: None, content: vec![UserInput::Text { @@ -11891,17 +11987,32 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() while rx.try_recv().is_ok() {} + let mut encoded = std::io::Cursor::new(Vec::new()); + image::DynamicImage::new_rgba8(/*w*/ 8, /*h*/ 8) + .write_to(&mut encoded, image::ImageFormat::Png) + .expect("encode image"); + let image_url = codex_utils_image::data_url_from_bytes("image/png", &encoded.into_inner()); let text_element = codex_protocol::user_input::TextElement::new( codex_protocol::user_input::ByteRange { start: 5, end: 12 }, Some("pending marker".to_string()), ); - let pending_user_input = vec![UserInput::Text { - text: "late pending input".to_string(), - text_elements: vec![text_element.clone()], - }]; + let pending_user_input = vec![ + UserInput::Text { + text: "late pending input".to_string(), + text_elements: vec![text_element.clone()], + }, + UserInput::Image { + image_url: image_url.clone(), + detail: Some(ImageDetail::High), + }, + ]; let submission = submit_steer_only(&sess, pending_user_input.clone(), &tc.sub_id).await; assert!(matches!(submission, TurnInputSubmission::Steered { .. })); + let mut current = tc.initial_settings.as_ref().clone(); + Arc::make_mut(&mut current.model_info).supports_image_detail_original = true; + tc.current_settings.store(Arc::new(current)); + sess.on_task_finished(Arc::clone(&tc), /*task_result*/ Ok(None)) .await; @@ -11909,16 +12020,22 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() let expected = ResponseItem::Message { id: None, role: "user".to_string(), - content: vec![ContentItem::InputText { - text: "late pending input".to_string(), - }], + content: vec![ + ContentItem::InputText { + text: "late pending input".to_string(), + }, + ContentItem::InputImage { + image_url: image_url.clone(), + detail: Some(ImageDetail::Original), + }, + ], phase: None, internal_chat_message_metadata_passthrough: None, }; assert!( strip_response_item_ids(&strip_metadata_from_items(&raw_history_items(&history))) .contains(&expected), - "expected pending input to be persisted into history on turn completion" + "expected pending input to use the current model's image rules on turn completion" ); let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) @@ -11965,7 +12082,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() local_images, .. }) if message == "late pending input" - && images == Some(Vec::new()) + && images == Some(vec![image_url]) && text_elements == vec![text_element] && local_images.is_empty() )); @@ -12427,7 +12544,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { }; let mut ctx = HandleOutputCtx { sess: Arc::clone(&sess), - turn_context: Arc::clone(&tc), + step_context: StepContext::for_test(Arc::clone(&tc)), turn_store: Arc::new(codex_extension_api::ExtensionData::new(tc.sub_id.clone())), tool_runtime: test_tool_runtime(Arc::clone(&sess), Arc::clone(&tc)), cancellation_token: CancellationToken::new(), diff --git a/codex-rs/core/src/session/time_reminder.rs b/codex-rs/core/src/session/time_reminder.rs index 963c8c4017..6f42343a94 100644 --- a/codex-rs/core/src/session/time_reminder.rs +++ b/codex-rs/core/src/session/time_reminder.rs @@ -132,8 +132,12 @@ pub(super) async fn maybe_record_current_time_reminder( let response_item = ContextualUserFragment::into(crate::context::CurrentTimeReminder::new(current_time)); - sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&response_item), + ) + .await; Ok(()) } diff --git a/codex-rs/core/src/session/token_budget.rs b/codex-rs/core/src/session/token_budget.rs index d20a17544e..b56e21c43a 100644 --- a/codex-rs/core/src/session/token_budget.rs +++ b/codex-rs/core/src/session/token_budget.rs @@ -189,8 +189,12 @@ pub(super) async fn maybe_record( &config.reminder_message_template, base_window_tokens_remaining, )); - sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&response_item), + ) + .await; } } @@ -211,8 +215,12 @@ pub(super) async fn maybe_record( let response_item = ContextualUserFragment::into(crate::context::AutoCompactFallbackPrompt::new(prompt)); - sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&response_item), + ) + .await; } #[cfg(test)] diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index eab4fbab2e..9127f3f031 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -93,6 +93,7 @@ use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelMessages; use codex_protocol::protocol::AgentMessageContentDeltaEvent; use codex_protocol::protocol::AgentReasoningSectionBreakEvent; @@ -186,8 +187,14 @@ pub(crate) async fn run_turn( .await { if matches!(err.details(), CodexErrorDetails::TurnAborted) { - run_hooks_and_record_inputs(&sess, &turn_context, &input, PersistContext::Standard) - .await; + run_hooks_and_record_inputs( + &sess, + &turn_context, + &turn_context.capture_current_model_info(), + &input, + PersistContext::Standard, + ) + .await; return Err(err); } if matches!(err.details(), CodexErrorDetails::ToolCollision(_)) { @@ -217,8 +224,14 @@ pub(crate) async fn run_turn( { Ok(requirements) => requirements, Err(err) => { - run_hooks_and_record_inputs(&sess, &turn_context, &input, PersistContext::Standard) - .await; + run_hooks_and_record_inputs( + &sess, + &turn_context, + &turn_context.capture_current_model_info(), + &input, + PersistContext::Standard, + ) + .await; return Err(err.into()); } }; @@ -239,8 +252,14 @@ pub(crate) async fn run_turn( { Ok(step_context) => step_context, Err(err) if matches!(err.details(), CodexErrorDetails::TurnAborted) => { - run_hooks_and_record_inputs(&sess, &turn_context, &input, PersistContext::Standard) - .await; + run_hooks_and_record_inputs( + &sess, + &turn_context, + &turn_context.capture_current_model_info(), + &input, + PersistContext::Standard, + ) + .await; return Err(err); } Err(err) => return Err(err), @@ -288,7 +307,15 @@ pub(crate) async fn run_turn( return Ok(None); } let mut can_drain_pending_input = input.is_empty(); - if run_hooks_and_record_inputs(&sess, &turn_context, &input, PersistContext::TurnStart).await { + if run_hooks_and_record_inputs( + &sess, + &turn_context, + &first_step_context.settings.model_info, + &input, + PersistContext::TurnStart, + ) + .await + { return Ok(None); } @@ -310,8 +337,12 @@ pub(crate) async fn run_turn( })) .await; for response_item in injection_items { - sess.record_conversation_items(&turn_context, std::slice::from_ref(&response_item)) - .await; + sess.record_conversation_items( + &turn_context, + &first_step_context.settings.model_info, + std::slice::from_ref(&response_item), + ) + .await; } track_turn_resolved_config_analytics(&sess, &turn_context, &input).await; @@ -348,6 +379,7 @@ pub(crate) async fn run_turn( if run_hooks_and_record_inputs( &sess, &turn_context, + &turn_context.capture_current_model_info(), &pending_input, PersistContext::Standard, ) @@ -574,6 +606,7 @@ pub(crate) async fn run_turn( { sess.record_response_item_and_emit_turn_item( &turn_context, + &step_context.settings.model_info, hook_prompt_message, ) .await; @@ -684,6 +717,7 @@ async fn turn_diff_display_roots(step_context: &StepContext) -> Vec<(String, Pat pub(crate) async fn run_hooks_and_record_inputs( sess: &Arc, turn_context: &Arc, + model_info: &ModelInfo, input: &[TurnInput], persist_context: PersistContext, ) -> bool { @@ -701,6 +735,7 @@ pub(crate) async fn run_hooks_and_record_inputs( record_pending_input( sess, turn_context, + model_info, input_item.clone(), hook_outcome.additional_contexts, persist_context, @@ -2175,13 +2210,14 @@ async fn emit_turn_item_in_plan_mode( /// Handle a completed assistant response item in plan mode, returning true if handled. async fn handle_assistant_item_done_in_plan_mode( sess: &Session, - turn_context: &TurnContext, + step_context: &StepContext, turn_store: &codex_extension_api::ExtensionData, item: &ResponseItem, state: &mut PlanModeStreamState, previously_active_item: Option<&TurnItem>, last_agent_message: &mut Option, ) -> bool { + let turn_context = &step_context.turn; if let ResponseItem::Message { role, .. } = item && role == "assistant" { @@ -2212,7 +2248,7 @@ async fn handle_assistant_item_done_in_plan_mode( record_completed_response_item_with_finalized_facts( sess, - turn_context, + step_context, item, finalized_facts.as_ref(), ) @@ -2229,8 +2265,9 @@ async fn handle_assistant_item_done_in_plan_mode( async fn drain_in_flight( in_flight: &mut FuturesOrdered>, sess: Arc, - turn_context: Arc, + step_context: &StepContext, ) -> CodexResult<()> { + let turn_context = &step_context.turn; while let Some(res) = in_flight.next().await { match res { Ok(envelope) => { @@ -2240,8 +2277,12 @@ async fn drain_in_flight( &envelope.item, ) .await; - sess.record_annotated_conversation_items(&turn_context, vec![envelope]) - .await; + sess.record_annotated_conversation_items( + turn_context, + &step_context.settings.model_info, + vec![envelope], + ) + .await; } Err(err) => { error_or_panic(format!("in-flight tool future failed during drain: {err}")); @@ -2449,7 +2490,7 @@ async fn try_run_sampling_request( if let Some(state) = plan_mode_state.as_mut() && handle_assistant_item_done_in_plan_mode( &sess, - &turn_context, + &step_context, turn_store.as_ref(), &item, state, @@ -2463,7 +2504,7 @@ async fn try_run_sampling_request( let mut ctx = HandleOutputCtx { sess: sess.clone(), - turn_context: turn_context.clone(), + step_context: Arc::clone(&step_context), turn_store: Arc::clone(&turn_store), tool_runtime: tool_runtime.clone(), cancellation_token: cancellation_token.child_token(), @@ -2858,7 +2899,7 @@ async fn try_run_sampling_request( } else { Some(turn_context.turn_timing_state.begin_tool_blocking()) }; - drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?; + drain_in_flight(&mut in_flight, sess.clone(), &step_context).await?; drop(tool_blocking_timing_guard); if should_emit_token_count { diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index ea3bfd157d..ac60990d1e 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -339,6 +339,11 @@ enum TurnMultiAgentRuntime { } impl TurnContext { + /// Captures current model metadata without preparing a step. + pub(crate) fn capture_current_model_info(&self) -> Arc { + Arc::clone(&self.current_settings.load().model_info) + } + /// Legacy: returns the frozen initial-turn model metadata. /// Step-scoped consumers should use their captured `StepContext::settings`. pub(crate) fn model_info(&self) -> &Arc { diff --git a/codex-rs/core/src/session/turn_tests.rs b/codex-rs/core/src/session/turn_tests.rs index f1829fe561..e43fb6a2ae 100644 --- a/codex-rs/core/src/session/turn_tests.rs +++ b/codex-rs/core/src/session/turn_tests.rs @@ -69,9 +69,10 @@ async fn plan_mode_uses_contributed_turn_item_for_last_agent_message() { let mut last_agent_message = None; let item = assistant_output_text("original assistant text"); + let step_context = StepContext::for_test(Arc::new(turn_context)); let handled = handle_assistant_item_done_in_plan_mode( &session, - &turn_context, + &step_context, &turn_store, &item, &mut state, diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 9de40b03e5..285e85fa14 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -12,6 +12,7 @@ use tokio_util::sync::CancellationToken; use crate::function_tool::FunctionCallError; use crate::parse_turn_item; use crate::session::session::Session; +use crate::session::step_context::StepContext; use crate::session::turn_context::TurnContext; use crate::tools::parallel::ToolCallRuntime; use crate::tools::router::ToolRouter; @@ -77,12 +78,12 @@ pub(crate) fn raw_assistant_output_text_from_item(item: &ResponseItem) -> Option /// Persist a completed model response item and record any cited memory usage. pub(crate) async fn record_completed_response_item( sess: &Session, - turn_context: &TurnContext, + step_context: &StepContext, item: &ResponseItem, ) { record_completed_response_item_with_finalized_facts( sess, - turn_context, + step_context, item, /*finalized_facts*/ None, ) @@ -91,12 +92,17 @@ pub(crate) async fn record_completed_response_item( pub(crate) async fn record_completed_response_item_with_finalized_facts( sess: &Session, - turn_context: &TurnContext, + step_context: &StepContext, item: &ResponseItem, finalized_facts: Option<&FinalizedTurnItemFacts>, ) { - sess.record_conversation_items(turn_context, std::slice::from_ref(item)) - .await; + let turn_context = &step_context.turn; + sess.record_conversation_items( + turn_context, + &step_context.settings.model_info, + std::slice::from_ref(item), + ) + .await; let defers_mailbox_delivery = finalized_facts.map_or_else( || { completed_item_defers_mailbox_delivery_to_next_turn( @@ -206,7 +212,7 @@ pub(crate) struct OutputItemResult { pub(crate) struct HandleOutputCtx { pub sess: Arc, - pub turn_context: Arc, + pub step_context: Arc, pub turn_store: Arc, pub tool_runtime: ToolCallRuntime, pub cancellation_token: CancellationToken, @@ -296,7 +302,7 @@ pub(crate) async fn handle_output_item_done( previously_active_item: Option, ) -> Result { let mut output = OutputItemResult::default(); - let plan_mode = ctx.turn_context.mode() == ModeKind::Plan; + let plan_mode = ctx.step_context.turn.mode() == ModeKind::Plan; match ToolRouter::build_tool_call(item.clone()) { // The model emitted a tool call; log it, persist the item immediately, and queue the tool execution. @@ -305,7 +311,7 @@ pub(crate) async fn handle_output_item_done( .input_queue .accept_mailbox_delivery_for_current_turn( &ctx.sess.active_turn, - &ctx.turn_context.sub_id, + &ctx.step_context.turn.sub_id, ) .await; @@ -317,7 +323,7 @@ pub(crate) async fn handle_output_item_done( payload_preview ); - record_completed_response_item(ctx.sess.as_ref(), ctx.turn_context.as_ref(), &item) + record_completed_response_item(ctx.sess.as_ref(), ctx.step_context.as_ref(), &item) .await; let cancellation_token = ctx.cancellation_token.child_token(); @@ -345,17 +351,20 @@ pub(crate) async fn handle_output_item_done( if let Some(finalized_turn_item) = finalized_turn_item { if previously_active_item.is_none() { ctx.sess - .emit_turn_item_started(&ctx.turn_context, &finalized_turn_item.turn_item) + .emit_turn_item_started( + &ctx.step_context.turn, + &finalized_turn_item.turn_item, + ) .await; } ctx.sess - .emit_turn_item_completed(&ctx.turn_context, finalized_turn_item.turn_item) + .emit_turn_item_completed(&ctx.step_context.turn, finalized_turn_item.turn_item) .await; } record_completed_response_item_with_finalized_facts( ctx.sess.as_ref(), - ctx.turn_context.as_ref(), + ctx.step_context.as_ref(), &item, finalized_facts.as_ref(), ) @@ -372,12 +381,13 @@ pub(crate) async fn handle_output_item_done( ..Default::default() }, }; - record_completed_response_item(ctx.sess.as_ref(), ctx.turn_context.as_ref(), &item) + record_completed_response_item(ctx.sess.as_ref(), ctx.step_context.as_ref(), &item) .await; if let Some(response_item) = response_input_to_response_item(&response) { ctx.sess .record_conversation_items( - &ctx.turn_context, + &ctx.step_context.turn, + &ctx.step_context.settings.model_info, std::slice::from_ref(&response_item), ) .await; diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index 3ccaea092e..0a051970c3 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/codex-rs/core/src/stream_events_utils_tests.rs @@ -298,11 +298,12 @@ async fn handle_output_item_done_returns_contributed_last_agent_message() { )); let step_context = step_context.with_tool_router_for_test(router); let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let tool_runtime = ToolCallRuntime::new(Arc::clone(&session), step_context, tracker); + let tool_runtime = + ToolCallRuntime::new(Arc::clone(&session), Arc::clone(&step_context), tracker); let item = assistant_output_text("original assistant text"); let mut ctx = HandleOutputCtx { sess: session, - turn_context: Arc::clone(&turn_context), + step_context, turn_store: Arc::new(ExtensionData::new(turn_context.sub_id.clone())), tool_runtime, cancellation_token: CancellationToken::new(), diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index d976a0ea29..2a6de503aa 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -641,6 +641,7 @@ impl Session { run_hooks_and_record_inputs( self, &turn_context, + &turn_context.capture_current_model_info(), &pending_input, PersistContext::Standard, ) @@ -951,6 +952,7 @@ impl Session { { self.record_conversation_items( task.turn_context.as_ref(), + task.turn_context.model_info(), std::slice::from_ref(&marker), ) .await; diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index fa57342185..ba4b44a8b5 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -64,7 +64,14 @@ impl SessionTask for RegularTask { .await; let prewarmed_client_session = match prewarmed_client_session { SessionStartupPrewarmResolution::Cancelled => { - run_hooks_and_record_inputs(&sess, &ctx, &input, PersistContext::Standard).await; + run_hooks_and_record_inputs( + &sess, + &ctx, + &ctx.capture_current_model_info(), + &input, + PersistContext::Standard, + ) + .await; return Ok(None); } SessionStartupPrewarmResolution::Unavailable { .. } => None, diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 42ec842e70..2435922219 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -238,6 +238,7 @@ pub(crate) async fn exit_review_mode( session .record_conversation_items( &ctx, + ctx.model_info(), &[ResponseItem::Message { id: Some(ResponseItemId::new("msg")), role: "user".to_string(), @@ -257,6 +258,7 @@ pub(crate) async fn exit_review_mode( session .record_response_item_and_emit_turn_item( ctx.as_ref(), + ctx.model_info(), ResponseItem::Message { id: Some(ResponseItemId::new("msg")), role: "assistant".to_string(), diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index 052fb4a6fc..97f43df05a 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -469,7 +469,11 @@ async fn persist_user_shell_output( if mode == UserShellCommandMode::StandaloneTurn { session - .record_conversation_items(turn_context, std::slice::from_ref(&output_item)) + .record_conversation_items( + turn_context, + turn_context.model_info(), + std::slice::from_ref(&output_item), + ) .await; // Standalone shell turns can run before any regular user turn, so // explicitly materialize rollout persistence after recording output. diff --git a/codex-rs/core/src/tools/handlers/extension_tools.rs b/codex-rs/core/src/tools/handlers/extension_tools.rs index 92a30d1207..2e209a8807 100644 --- a/codex-rs/core/src/tools/handlers/extension_tools.rs +++ b/codex-rs/core/src/tools/handlers/extension_tools.rs @@ -446,7 +446,11 @@ mod tests { internal_chat_message_metadata_passthrough: None, }; session - .record_conversation_items(&turn, std::slice::from_ref(&history_item)) + .record_conversation_items( + &turn, + turn.model_info(), + std::slice::from_ref(&history_item), + ) .await; let expected_history_item = strip_response_item_id( session diff --git a/codex-rs/core/src/tools/router_tests.rs b/codex-rs/core/src/tools/router_tests.rs index 1d6b1d322f..05a1e2e65a 100644 --- a/codex-rs/core/src/tools/router_tests.rs +++ b/codex-rs/core/src/tools/router_tests.rs @@ -543,7 +543,11 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow internal_chat_message_metadata_passthrough: None, }; session - .record_conversation_items(&turn, std::slice::from_ref(&history_item)) + .record_conversation_items( + &turn, + turn.model_info(), + std::slice::from_ref(&history_item), + ) .await; let expected_history_item = session .clone_history() diff --git a/codex-rs/core/tests/suite/model_switching.rs b/codex-rs/core/tests/suite/model_switching.rs index cd9b33f9b2..ac15cadf60 100644 --- a/codex-rs/core/tests/suite/model_switching.rs +++ b/codex-rs/core/tests/suite/model_switching.rs @@ -21,6 +21,7 @@ use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE; use codex_protocol::config_types::ServiceTier; use codex_protocol::config_types::Settings; +use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::models::BaseInstructionsProvenance; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ConfigShellToolType; @@ -942,14 +943,26 @@ async fn null_service_tier_override_is_omitted_from_http_turn_with_catalog_defau Ok(()) } +#[derive(Clone, Copy)] +enum MediaHistorySource { + Live, + Resume, + Fork, +} + +#[test_case(MediaHistorySource::Live; "live history")] +#[test_case(MediaHistorySource::Resume; "resumed history")] +#[test_case(MediaHistorySource::Fork; "forked history")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn model_change_from_multimodal_to_text_strips_prior_media_content() -> Result<()> { +async fn model_change_projects_media_without_changing_live_or_replayed_history( + source: MediaHistorySource, +) -> Result<()> { skip_if_no_network!(Ok(())); let server = MockServer::start().await; let multimodal_model_slug = "test-multimodal-model"; let text_model_slug = "test-text-only-model"; - let multimodal_model = test_model_info( + let mut multimodal_model = test_model_info( multimodal_model_slug, "Test Multimodal Model", "supports image and audio input", @@ -959,6 +972,7 @@ async fn model_change_from_multimodal_to_text_strips_prior_media_content() -> Re InputModality::Audio, ], ); + multimodal_model.supports_image_detail_original = true; let text_model = test_model_info( text_model_slug, "Test Text Model", @@ -982,9 +996,13 @@ async fn model_change_from_multimodal_to_text_strips_prior_media_content() -> Re let mut builder = test_codex() .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) .with_config(move |config| { - config.model = Some(multimodal_model_slug.to_string()); + config.model = Some(text_model_slug.to_string()); + config + .features + .enable(Feature::UnifiedImageBudget) + .expect("enable unified image budget"); }); - let test = builder.build(&server).await?; + let test = builder.build_with_auto_env(&server).await?; let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager .list_models( @@ -992,8 +1010,10 @@ async fn model_change_from_multimodal_to_text_strips_prior_media_content() -> Re codex_core::test_support::default_http_client_factory(), ) .await; - let image_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==" - .to_string(); + let mut png = std::io::Cursor::new(Vec::new()); + image::DynamicImage::new_rgba8(/*w*/ 2048, /*h*/ 2048) + .write_to(&mut png, image::ImageFormat::Png)?; + let image_url = codex_utils_image::data_url_from_bytes("image/png", &png.into_inner()); test.codex .start_or_steer_turn(read_only_user_turn( @@ -1069,6 +1089,60 @@ async fn model_change_from_multimodal_to_text_strips_prior_media_content() -> Re .any(|text| text == "audio content omitted because you do not support audio input"), "second request should include the audio-omitted placeholder text" ); + + let original = first_request + .inputs_of_type("message") + .into_iter() + .find(|item| { + item["content"] + .as_array() + .is_some_and(|content| content.iter().any(|part| part["type"] == "input_image")) + }) + .expect("original media message"); + if !matches!(source, MediaHistorySource::Live) { + test.codex.shutdown_and_wait().await?; + } + let rollout_path = test.codex.rollout_path().expect("rollout path"); + let thread = match source { + MediaHistorySource::Live => Arc::clone(&test.codex), + MediaHistorySource::Resume => { + test.thread_manager + .resume_thread_from_rollout( + test.config.clone(), + rollout_path, + test.thread_manager.auth_manager(), + /*parent_trace*/ None, + ClientMcpExtensions::default(), + ) + .await? + .thread + } + MediaHistorySource::Fork => { + test.thread_manager + .fork_thread( + ForkSnapshot::Interrupted, + codex_core::StartThreadOptions::new(test.config.clone()), + rollout_path, + ) + .await? + .thread + } + }; + let response = mount_sse_once(&server, sse_completed("resp-restored")).await; + submit_model_turn( + &thread, + multimodal_model_slug, + ThreadSettingsOverrides::default(), + ) + .await?; + let restored = response + .single_request() + .inputs_of_type("message") + .into_iter() + .find(|item| item["id"] == original["id"]); + // Preserve the prepared media, item ID, turn ID, and creation timestamp together. + assert_eq!(restored.as_ref(), Some(&original)); + thread.shutdown_and_wait().await?; Ok(()) } diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index ce4c1c66b7..a52cea710d 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -51,6 +51,7 @@ use codex_protocol::mcp_policy::McpServerIdentity; use codex_protocol::mcp_policy::McpServerRequirement; use codex_protocol::mcp_policy::PluginMcpRequirements; use codex_protocol::models::ContentItem; +use codex_protocol::models::ImageDetail; use codex_protocol::models::PermissionProfile; use codex_protocol::models::PermissionProfileSnapshot; use codex_protocol::models::ResponseItem; @@ -79,6 +80,9 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::protocol::TurnSettingsUpdate; +use codex_protocol::protocol::TurnSettingsUpdateOutcome; +use codex_protocol::turn_input::TurnInput; use codex_protocol::user_input::UserInput; use codex_utils_cargo_bin::cargo_bin; use codex_utils_path_uri::PathUri; @@ -1427,7 +1431,28 @@ async fn interrupt_during_mcp_startup_preserves_user_input_in_history( let pending_mcp_url = format!("http://{}/mcp", pending_mcp_listener.local_addr()?); let fixture = test_codex() + .with_model_info_override("gpt-5.4", |model| { + model.use_responses_lite = false; + model.supports_image_detail_original = false; + }) .with_config(move |config| { + config + .features + .enable(Feature::StepModelSwitching) + .expect("enable step model switching"); + config + .features + .enable(Feature::UnifiedImageBudget) + .expect("enable unified image budget"); + let models = &mut config.model_catalog.as_mut().expect("model catalog").models; + let mut current_model = models + .iter() + .find(|model| model.slug == "gpt-5.4") + .expect("initial model") + .clone(); + current_model.slug = "startup-image-model".to_string(); + current_model.supports_image_detail_original = true; + models.push(current_model); config.model_provider.supports_websockets = matches!(startup_phase, InterruptedMcpStartupPhase::StartupPrewarm); if matches!( @@ -1457,14 +1482,36 @@ async fn interrupt_during_mcp_startup_preserves_user_input_in_history( .await .context("MCP startup should connect before the turn is interrupted")??; let prompt = "keep this interrupted prompt in conversation history"; - fixture - .codex - .start_or_steer_turn(read_only_user_turn(&fixture, prompt)) - .await?; - wait_for_event(&fixture.codex, |event| { - matches!(event, EventMsg::TurnStarted(_)) + let mut input = read_only_user_turn(&fixture, prompt); + let TurnInput::UserInput { content, .. } = &mut input.input else { + unreachable!("read_only_user_turn creates user input"); + }; + content.push(UserInput::Image { + image_url: OPENAI_PNG.to_string(), + detail: Some(ImageDetail::High), + }); + fixture.codex.start_or_steer_turn(input).await?; + let turn_id = core_test_support::wait_for_event_match(&fixture.codex, |event| match event { + EventMsg::TurnStarted(started) => Some(started.turn_id.clone()), + _ => None, }) .await; + let (reply, outcome) = tokio::sync::oneshot::channel(); + fixture + .codex + .submit(Op::TurnSettings { + turn_id, + update: TurnSettingsUpdate { + model: Some("startup-image-model".to_string()), + ..Default::default() + }, + reply, + }) + .await?; + assert_eq!( + tokio::time::timeout(Duration::from_secs(/*secs*/ 10), outcome).await??, + TurnSettingsUpdateOutcome::Applied + ); fixture.codex.submit(Op::Interrupt).await?; wait_for_event(&fixture.codex, |event| { @@ -1494,6 +1541,19 @@ async fn interrupt_during_mcp_startup_preserves_user_input_in_history( ) }) .expect("an interrupted turn should retain its submitted user prompt"); + let RolloutItem::ResponseItem(envelope) = &history.items[user_prompt_index] else { + unreachable!("found the user prompt"); + }; + let ResponseItem::Message { content, .. } = &envelope.item else { + unreachable!("found the user message"); + }; + assert!( + content.contains(&ContentItem::InputImage { + image_url: OPENAI_PNG.to_string(), + detail: Some(ImageDetail::Original), + }), + "interrupted input must use the current model's unified image budget" + ); let interruption_marker_index = history .items .iter() diff --git a/codex-rs/core/tests/suite/step_settings.rs b/codex-rs/core/tests/suite/step_settings.rs index eb97afc5ff..998b4173ba 100644 --- a/codex-rs/core/tests/suite/step_settings.rs +++ b/codex-rs/core/tests/suite/step_settings.rs @@ -1,6 +1,9 @@ use anyhow::Result; +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use codex_config::McpServerConfig; use codex_core::CodexThread; +use codex_core::StartThreadOptions; use codex_core::TurnInputRequest; use codex_core::config::Config; use codex_core::config::Constrained; @@ -14,6 +17,7 @@ use codex_extension_api::PromptFragment; use codex_extension_api::ToolContributor; use codex_extension_api::TurnContextContributionInput; use codex_features::Feature; +use codex_history::RolloutItem; use codex_login::CodexAuth; use codex_models_manager::bundled_models_response; use codex_models_manager::manager::RefreshStrategy; @@ -21,7 +25,12 @@ 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::dynamic_tools::DynamicToolCallOutputContentItem; +use codex_protocol::dynamic_tools::DynamicToolFunctionSpec; +use codex_protocol::dynamic_tools::DynamicToolResponse; +use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::models::PermissionProfile; +use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ApplyPatchToolType; use codex_protocol::openai_models::ApprovalMessages; use codex_protocol::openai_models::CollaborationModeMessages; @@ -60,6 +69,9 @@ use codex_tools::ToolExecutorFuture; use codex_tools::ToolName; use codex_tools::ToolOutput; use codex_tools::ToolSpec; +use codex_utils_image::data_url_from_bytes; +use codex_utils_output_truncation::TruncationPolicy; +use codex_utils_output_truncation::truncate_text; 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; @@ -84,6 +96,7 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_event_match; use core_test_support::wait_for_mcp_server; +use image::GenericImageView; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; @@ -271,6 +284,180 @@ fn request_turn_id(request: &ResponsesRequest) -> String { .to_string() } +// Dynamic tools return the original payload, so handler truncation cannot hide a recorder bug. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tool_result_history_keeps_originating_model_across_switch() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mut test = step_settings_test() + .with_config(|config| { + for model in &mut config.model_catalog.as_mut().expect("models").models { + model.truncation_policy = + TruncationPolicyConfig::tokens(if model.slug == MODEL_B { 400 } else { 100 }); + model.supports_image_detail_original = model.slug == MODEL_B; + model.use_responses_lite = false; + } + config + .features + .enable(Feature::UnifiedImageBudget) + .expect("enable unified image budget"); + }) + .build_with_auto_env(&server) + .await?; + let started = test + .thread_manager + .start_thread(StartThreadOptions { + dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec { + name: "diagnostics".to_string(), + description: "Returns diagnostic text and a screenshot.".to_string(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + })], + ..StartThreadOptions::new(test.config.clone()) + }) + .await?; + test.codex = started.thread; + let responses = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-a"), + ev_function_call("call-a", "diagnostics", "{}"), + ev_completed("resp-a"), + ]), + sse(vec![ + ev_response_created("resp-b"), + ev_function_call("call-b", "diagnostics", "{}"), + ev_completed("resp-b"), + ]), + sse_completed("resp-done"), + ], + ) + .await; + let mut png = std::io::Cursor::new(Vec::new()); + image::DynamicImage::new_rgba8(/*w*/ 2048, /*h*/ 2048) + .write_to(&mut png, image::ImageFormat::Png)?; + let text = "diagnostic line\n".repeat(500); + let response = DynamicToolResponse { + content_items: vec![ + DynamicToolCallOutputContentItem::InputText { text: text.clone() }, + DynamicToolCallOutputContentItem::InputImage { + image_url: data_url_from_bytes("image/png", &png.into_inner()), + }, + ], + success: true, + }; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "collect diagnostics".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let mut raw_outputs = Vec::new(); + for call_id in ["call-a", "call-b"] { + let call = wait_for_event_match(&test.codex, |event| match event { + EventMsg::DynamicToolCallRequest(request) => Some(request.clone()), + _ => None, + }) + .await; + assert_eq!(call.call_id, call_id); + if call_id == "call-a" { + // Apply B while A's result is pending in the same turn. + assert_eq!( + submit_turn_settings( + &test.codex, + &call.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + } + test.codex + .submit(Op::DynamicToolResponse { + id: call.call_id, + response: response.clone(), + }) + .await?; + raw_outputs.push( + wait_for_event_match(&test.codex, |event| match event { + EventMsg::RawResponseItem(event) + if matches!(&event.item, ResponseItem::FunctionCallOutput { .. }) => + { + Some(serde_json::to_value(&event.item).expect("raw output")) + } + _ => None, + }) + .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)] + ); + let turn_id = request_turn_id(&requests[0]); + for request in &requests[1..] { + assert_eq!(request_turn_id(request), turn_id); + } + for (index, (call_id, limit, dimensions)) in + [("call-a", 100, (1600, 1600)), ("call-b", 400, (2048, 2048))] + .into_iter() + .enumerate() + { + let raw = &raw_outputs[index]; + assert_eq!(raw["output"][0]["text"], text); + assert!(raw["id"].is_string()); + assert!(raw["internal_chat_message_metadata_passthrough"]["create_time"].is_number()); + let url = raw["output"][1]["image_url"].as_str().expect("tool image"); + let (_, data) = url.split_once(',').expect("image data URL"); + assert_eq!( + image::load_from_memory(&BASE64_STANDARD.decode(data)?)?.dimensions(), + dimensions + ); + let mut expected = raw.clone(); + expected["output"][0]["text"] = + json!(truncate_text(&text, TruncationPolicy::Tokens(limit) * 1.2)); + assert_eq!(requests[2].function_call_output(call_id), expected); + } + assert_eq!( + requests[1].function_call_output("call-a"), + requests[2].function_call_output("call-a") + ); + + // Persistence and raw notifications retain the prepared, untruncated payload in append order. + test.codex.shutdown_and_wait().await?; + let history = codex_rollout::RolloutRecorder::get_rollout_history( + &test.codex.rollout_path().expect("rollout path"), + ) + .await?; + let saved_outputs = history + .get_rollout_items() + .iter() + .filter_map(|item| match item { + RolloutItem::ResponseItem(envelope) + if matches!(&envelope.item, ResponseItem::FunctionCallOutput { .. }) => + { + Some(serde_json::to_value(&envelope.item).expect("saved output")) + } + _ => None, + }) + .collect::>(); + assert_eq!(saved_outputs, raw_outputs); + Ok(()) +} + #[derive(Clone, Copy)] enum SettingsTarget { Thread,