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,