diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 359302bac0..ba5b6d74f6 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 824d374559..bf00a82243 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -4833,6 +4833,7 @@ mod tests { misalignment: None, message: "Selected model is at capacity. Please try a different model.".into(), codex_error_info: Some(CodexErrorInfo::PolicyCapacityUnavailable), + rejected_audio_file_ids: Vec::new(), }), completed_at: Some(20), duration_ms: Some(10_000), @@ -5242,6 +5243,7 @@ mod tests { misalignment: None, message: "rollback failed".into(), codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed), + rejected_audio_file_ids: Vec::new(), }), ]; @@ -5289,6 +5291,7 @@ mod tests { misalignment: None, message: "request-level failure".into(), codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: Vec::new(), }), ]; @@ -5347,6 +5350,7 @@ mod tests { codex_error_info: Some(CodexErrorInfo::ResponseStreamDisconnected { http_status_code: Some(502), }), + rejected_audio_file_ids: Vec::new(), }), EventMsg::TurnComplete(TurnCompleteEvent { root_turn_id: None, @@ -5412,6 +5416,7 @@ mod tests { misalignment: None, message: "Selected model is at capacity. Please try a different model.".into(), codex_error_info: Some(CodexErrorInfo::ServerOverloaded), + rejected_audio_file_ids: Vec::new(), }), completed_at: Some(20), duration_ms: Some(10_000), diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index e96a9306c3..2f41e65197 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -79,6 +79,7 @@ fn projects_failed_turn_completion_as_snapshot() { misalignment: None, message: "request failed".to_string(), codex_error_info: None, + rejected_audio_file_ids: Vec::new(), }; let changes = project(RolloutItem::EventMsg(EventMsg::TurnComplete( diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index fd02711e5b..87792d5f5d 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -3367,7 +3367,10 @@ mod tests { }, ApiError::RateLimit(message()), ApiError::InvalidRequest { message: message() }, - ApiError::InvalidAudioInput { message: message() }, + ApiError::InvalidAudioInput { + message: message(), + per_recording: true, + }, ApiError::ModelNotFound { message: message(), status: Some(http::StatusCode::NOT_FOUND), diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index 9fcccfd99c..f73732fd6b 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -86,9 +86,13 @@ fn map_api_error_details(err: ApiError) -> CodexErr { }) } ApiError::InvalidRequest { message } => CodexErr::InvalidRequest(message), - ApiError::InvalidAudioInput { message } => { - CodexErr::new(CodexErrorDetails::InvalidAudioInput { message }) - } + ApiError::InvalidAudioInput { + message, + per_recording, + } => CodexErr::new(CodexErrorDetails::InvalidAudioInput { + message, + per_recording, + }), ApiError::ModelNotFound { message, status } => { CodexErr::new(CodexErrorDetails::ModelNotFound { message, status }) } diff --git a/codex-rs/codex-api/src/error.rs b/codex-rs/codex-api/src/error.rs index 6d32823b7d..d1ce0446f8 100644 --- a/codex-rs/codex-api/src/error.rs +++ b/codex-rs/codex-api/src/error.rs @@ -42,7 +42,11 @@ pub enum ApiError { #[error("invalid request: {message}")] InvalidRequest { message: String }, #[error("{message}")] - InvalidAudioInput { message: String }, + InvalidAudioInput { + message: String, + /// True when the provider rejected an individual clip, independent of request size. + per_recording: bool, + }, #[error("{message}")] ModelNotFound { message: String, @@ -93,6 +97,7 @@ pub(crate) fn parse_terminal_provider_error( | "missing_audio_sample_rate", None | Some(StatusCode::BAD_REQUEST), ) => Some(ApiError::InvalidAudioInput { + per_recording: code != "audio_too_large", message: error .get("message") .and_then(Value::as_str) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 93a0771986..91fcfe365f 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -622,6 +622,7 @@ async fn on_event_updates_status_from_task_complete() { misalignment: None, message: "denied".to_string(), codex_error_info: None, + rejected_audio_file_ids: Vec::new(), }), AgentStatus::Errored("denied".to_string()), ), @@ -646,6 +647,7 @@ async fn on_event_updates_status_from_error() { misalignment: None, message: "boom".to_string(), codex_error_info: None, + rejected_audio_file_ids: Vec::new(), })); let expected = AgentStatus::Errored("boom".to_string()); @@ -2701,6 +2703,7 @@ async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; let parent_resume_metadata = codex_history::CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("parent-turn".to_string()), @@ -3862,6 +3865,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc compaction_response_id: None, latest_token_usage_record: None, resume_metadata: Some(codex_history::CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, @@ -3997,6 +4001,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc assert_eq!( inherited_state, Some(codex_history::CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index bb563abf0b..7203b70f16 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -427,6 +427,7 @@ async fn run_compact_task_inner_impl( new_history, replacement_step_context.to_turn_context_item(), world_state_baseline, + /*rejected_audio_file_ids*/ Default::default(), CompactedHistoryMetadata { input_goal_ids, message: summary_text, diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 0b47a5790e..619af67d14 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -288,7 +288,8 @@ async fn run_remote_compaction_request_v2( step_context: &StepContext, prompt: &Prompt, responses_metadata: &CodexResponsesMetadata, - transport: CompactionRequestTransport<'_>, + transport: &mut CompactionRequestTransport<'_>, + submitted_audio: Option<&crate::context_manager::rejected_audio::SubmittedFileAudio>, ) -> CodexResult { let turn_context = &step_context.turn; let client_session = match transport { @@ -300,13 +301,13 @@ async fn run_remote_compaction_request_v2( .await; let trace = InferenceTraceContext::disabled(); // Replay the routing token without letting this request change foreground routing. - let turn_state = Arc::new(turn_state.map(OnceLock::from).unwrap_or_default()); + let turn_state = Arc::new(turn_state.clone().map(OnceLock::from).unwrap_or_default()); let http_executor = ResponsesHttpExecutor::new(&sess.services.model_client); let stream = tokio::select! { biased; _ = cancellation.cancelled() => return Err(CodexErr::TurnAborted), stream = http_executor.stream_responses_http( - &turn_state, auth_owner_generation, /*event_sender*/ None, prompt, &settings.model_info, &step_context.session_telemetry, effort, + &turn_state, *auth_owner_generation, /*event_sender*/ None, prompt, &settings.model_info, &step_context.session_telemetry, effort, settings.reasoning_summary, settings.service_tier.clone(), responses_metadata, &trace, ) => stream?, @@ -356,6 +357,11 @@ async fn run_remote_compaction_request_v2( }; match result { Ok(compaction_output) => return Ok(compaction_output), + Err(err) + if submitted_audio.is_some_and(|audio| audio.matches_rejection(err.details())) => + { + return Err(err); + } Err(err) => { handle_response_stream_error( &mut retry_state, @@ -493,6 +499,10 @@ fn build_v2_compacted_history( .zip(prompt_input_metadata) .map(|(item, metadata)| ResponseItemEnvelope { item, metadata }) .collect::>(); + let no_file_error_sources = + crate::context_manager::rejected_audio::checkpoint_input_has_no_file_error_sources( + &prompt_input, + ); let retained = v2_history_item_groups(prompt_input) .filter(|group| is_retained_for_remote_compaction_v2(&group.source, features)) .flat_map(HistoryItemGroup::into_items) @@ -503,7 +513,14 @@ fn build_v2_compacted_history( .iter() .map(|envelope| retained_input_image_count(&envelope.item)) .sum::(); - retained.push(ResponseItemEnvelope::new(compaction_output)); + let mut checkpoint = ResponseItemEnvelope::new(compaction_output); + if no_file_error_sources { + checkpoint + .metadata + .get_or_insert_default() + .compaction_without_file_error_sources = true; + } + retained.push(checkpoint); (retained, retained_image_count) } @@ -980,7 +997,13 @@ mod tests { metadata: Some(CodexHarnessMetadata::default()), }, ResponseItemEnvelope::new(generated_notice.clone()), - ResponseItemEnvelope::new(output.clone()), + ResponseItemEnvelope { + item: output.clone(), + metadata: Some(CodexHarnessMetadata { + compaction_without_file_error_sources: true, + ..Default::default() + }), + }, ]; if enabled { expected.insert( diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 37bbc15afb..72f00f2e30 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -3,6 +3,7 @@ use crate::context::UserGoalUpdate; use codex_protocol::ResponseItemId; use codex_protocol::openai_models::CatalogModel; +use std::collections::BTreeSet; use std::collections::HashSet; use std::sync::Arc; @@ -15,6 +16,7 @@ use crate::compact::CompactionAnalyticsDetails; use crate::compact_remote_history::trim_function_call_history_to_fit_context_window; use crate::context::world_state::WorldState; use crate::context_manager::ContextManager; +use crate::context_manager::rejected_audio; use crate::context_manager::replace_unresolvable_image_files; use crate::responses_metadata::CompactionTurnMetadata; use crate::session::CompactionPrefix; @@ -35,6 +37,7 @@ use tokio_util::sync::CancellationToken; use tracing::info; pub(super) struct RemoteCompactV2Attempt { + pub(super) rejected_audio_file_ids: BTreeSet, pub(super) input_goal_ids: HashSet, pub(super) trace_input_history: Option>, pub(super) prompt_input: Vec, @@ -83,8 +86,8 @@ pub(super) async fn run_remote_compact_v2_attempt( trace_input_history, base_instructions, mut input, - prompt_input_metadata, - transport, + mut prompt_input_metadata, + mut transport, captured_window, ) = match input { RemoteCompactV2AttemptInput::Foreground(client_session) => { @@ -169,13 +172,15 @@ pub(super) async fn run_remote_compact_v2_attempt( Some(input.window), ), }; + let mut unavailable = + rejected_audio::unavailable_file_audio(&input, sess.services.image_store.as_ref()).await; + rejected_audio::project_compaction_audio(&mut input, &mut prompt_input_metadata, &unavailable); let tools_in_history = model_info.use_responses_lite && input .iter() .any(|item| matches!(item, ResponseItem::AdditionalTools { .. })); let tool_router = &step_context.tool_router; - input.push(ResponseItem::CompactionTrigger {}); - let prompt = Prompt { + let mut prompt = Prompt { input, tools: if tools_in_history { Arc::default() @@ -199,36 +204,69 @@ pub(super) async fn run_remote_compact_v2_attempt( } }; responses_metadata.tool_namespaces_info = tool_router.tool_namespaces_info().cloned(); - let trace_attempt = compaction_trace.start_attempt(&serde_json::json!({ - "model": model_info.slug.as_str(), - "instructions": prompt.base_instructions.text.as_str(), - "input": &prompt.input, - "parallel_tool_calls": prompt.parallel_tool_calls, - })); - let compaction_output_result = run_remote_compaction_request_v2( - sess, - step_context, - &prompt, - &responses_metadata, - transport, - ) - .await; - trace_attempt.record_result( - compaction_output_result - .as_ref() - .map(|output| std::slice::from_ref(&output.compaction_output)), + // The file may disappear after the availability check, or the store may + // defer validation to Responses. Retry only an unambiguously rejected ID. + let mut submitted_audio = rejected_audio::SubmittedFileAudio::from_request( + &prompt.input, + /*accepted*/ None, + prompt + .input + .iter() + .zip(&prompt_input_metadata) + .filter_map(|(item, metadata)| { + rejected_audio::clear_checkpoint(item, metadata.as_ref()) + }), ); let RemoteCompactionV2Output { compaction_output, response_id, token_usage, - } = compaction_output_result?; - let mut prompt_input = prompt.input; - prompt_input.pop(); + } = loop { + prompt.input.push(ResponseItem::CompactionTrigger {}); + let trace_attempt = compaction_trace.start_attempt(&serde_json::json!({ + "model": model_info.slug.as_str(), + "instructions": prompt.base_instructions.text.as_str(), + "input": &prompt.input, + "parallel_tool_calls": prompt.parallel_tool_calls, + })); + let compaction_output_result = run_remote_compaction_request_v2( + sess, + step_context, + &prompt, + &responses_metadata, + &mut transport, + submitted_audio.as_ref(), + ) + .await; + trace_attempt.record_result( + compaction_output_result + .as_ref() + .map(|output| std::slice::from_ref(&output.compaction_output)), + ); + prompt.input.pop(); + match compaction_output_result { + Ok(output) => break output, + Err(error) => { + let Some(id) = submitted_audio + .take() + .and_then(|audio| audio.rejected_file_id(error.details())) + else { + return Err(error); + }; + unavailable.insert(id); + rejected_audio::project_compaction_audio( + &mut prompt.input, + &mut prompt_input_metadata, + &unavailable, + ); + } + } + }; Ok(RemoteCompactV2Attempt { + rejected_audio_file_ids: unavailable, input_goal_ids, trace_input_history, - prompt_input, + prompt_input: prompt.input, prompt_input_metadata, compaction_output, compaction_response_id: response_id, @@ -271,6 +309,7 @@ pub(crate) async fn capture_background_input( &base_instructions, ); snapshot.normalize_history(&step_context.settings.model_info.input_modalities); + snapshot.project_file_audio_for_prompt(); let (mut input, metadata): (Vec<_>, Vec<_>) = snapshot .into_annotated_items() .into_iter() diff --git a/codex-rs/core/src/compact_remote_v2_installation.rs b/codex-rs/core/src/compact_remote_v2_installation.rs index aa3726491f..0b03d89681 100644 --- a/codex-rs/core/src/compact_remote_v2_installation.rs +++ b/codex-rs/core/src/compact_remote_v2_installation.rs @@ -174,6 +174,7 @@ async fn build_and_install_compaction( (attempt, compaction_turn_context, None) }; let RemoteCompactV2Attempt { + rejected_audio_file_ids, input_goal_ids, trace_input_history, prompt_input, @@ -215,6 +216,7 @@ async fn build_and_install_compaction( compacted_history, replacement_step_context.to_turn_context_item(), world_state_baseline, + rejected_audio_file_ids, CompactedHistoryInstallation::Prefix { prefix, prefix_context, @@ -253,6 +255,7 @@ async fn build_and_install_compaction( new_history, replacement_step_context.to_turn_context_item(), world_state_baseline, + rejected_audio_file_ids, installation, ) .await diff --git a/codex-rs/core/src/compact_remote_v2_installation_tests.rs b/codex-rs/core/src/compact_remote_v2_installation_tests.rs index 5f9e290bd9..f24d9791c8 100644 --- a/codex-rs/core/src/compact_remote_v2_installation_tests.rs +++ b/codex-rs/core/src/compact_remote_v2_installation_tests.rs @@ -77,6 +77,7 @@ async fn prepared_prefix_rejection_respects_caller_intent( model_info: Arc::clone(&step.settings.model_info), cutoff_world_state: Arc::new(WorldState::default()), attempt: RemoteCompactV2Attempt { + rejected_audio_file_ids: Default::default(), input_goal_ids: Default::default(), trace_input_history: None, prompt_input: vec![original.clone()], diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 523730fd86..8735177d77 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -11,6 +11,8 @@ //! sources recovered from legacy Guardian checkpoints before their raw history is dropped. //! Prompt projections replace unavailable file images without changing retained history. +#[path = "history_rejected_audio.rs"] +pub(crate) mod rejected_audio; #[path = "history_user_authorization.rs"] mod user_authorization; @@ -62,6 +64,7 @@ use codex_protocol::openai_models::InputModality; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; +use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::WorldStateItem; use codex_utils_audio::estimate_audio_token_count; @@ -73,6 +76,7 @@ use codex_utils_output_truncation::approx_token_count; use codex_utils_output_truncation::approx_tokens_from_byte_count_i64; use codex_utils_output_truncation::truncate_function_output_payload; use codex_utils_output_truncation::with_serialization_allowance; +use std::collections::BTreeSet; use std::num::NonZeroUsize; use std::ops::Deref; use std::sync::Arc; @@ -106,6 +110,9 @@ pub(crate) struct ContextManager { review_history: Option, /// Host facts independent of the model window; snapshots share immutable state. retained_context: Arc, + /// File IDs rejected by Responses. Retained history stays append-only; only + /// request projections omit these references. + rejected_audio_file_ids: Arc>, /// Reviewer policy travels with the history snapshot, independently of capture. guardian_review_mode: GuardianContextMode, retain_inherited_user_messages: bool, @@ -251,6 +258,7 @@ impl ContextManager { items: Arc::new(Vec::new()), review_history: None, retained_context: Arc::default(), + rejected_audio_file_ids: Arc::default(), guardian_review_mode: GuardianContextMode::ThreadOwned, retain_inherited_user_messages: false, history_version: 0, @@ -307,6 +315,7 @@ impl ContextManager { return false; } match event { + RetainedContextEvent::AcceptedFileAudio { .. } => {} RetainedContextEvent::VerifiedAnswer { .. } => { self.user_message_revision = self.user_message_revision.saturating_add(/*rhs*/ 1); @@ -656,9 +665,22 @@ impl ContextManager { image_store, ) .await; + self.project_file_audio_for_prompt(); Arc::unwrap_or_clone(self.items) } + /// Applies the file-audio request projection to a normalized snapshot whose + /// remaining preparation is performed by another request path. + pub(crate) fn project_file_audio_for_prompt(&mut self) { + if self.rejected_audio_file_ids.is_empty() { + return; + } + rejected_audio::project_file_audio( + Arc::make_mut(&mut self.items), + &self.rejected_audio_file_ids, + ); + } + /// Iterates over raw response items without exposing their history envelopes. pub(crate) fn raw_items( &self, @@ -676,6 +698,37 @@ impl ContextManager { &self.items } + /// Apply the durable failed-turn record without rewriting retained user history. + pub(crate) fn record_rejected_file_audio(&mut self, event: &TurnCompleteEvent) { + let Some(error) = &event.error else { + return; + }; + self.extend_rejected_audio_file_ids(error.rejected_audio_file_ids.iter().cloned()); + } + + pub(crate) fn extend_rejected_audio_file_ids(&mut self, ids: impl IntoIterator) { + let mut ids = ids.into_iter().peekable(); + if ids.peek().is_none() { + return; + } + let rejected = Arc::make_mut(&mut self.rejected_audio_file_ids); + let previous_len = rejected.len(); + rejected.extend(ids); + if rejected.len() != previous_len { + self.history_version = self.history_version.saturating_add(1); + self.guardian_review_context_revision = next_guardian_review_context_revision(); + } + } + + pub(crate) fn rejected_audio_file_ids(&self) -> Arc> { + Arc::clone(&self.rejected_audio_file_ids) + } + + /// Restores durable rejection facts from a checkpoint and subsequent failed turns. + pub(crate) fn restore_rejected_audio_file_ids(&mut self, rejected: BTreeSet) { + self.rejected_audio_file_ids = Arc::new(rejected); + } + /// Returns annotated history items and consumes the snapshot. pub(crate) fn into_annotated_items(self) -> Vec { Arc::unwrap_or_clone(self.into_shared_annotated_items()) diff --git a/codex-rs/core/src/context_manager/history_rejected_audio.rs b/codex-rs/core/src/context_manager/history_rejected_audio.rs new file mode 100644 index 0000000000..75e7f3d236 --- /dev/null +++ b/codex-rs/core/src/context_manager/history_rejected_audio.rs @@ -0,0 +1,360 @@ +//! Non-mutating request projection for uploaded audio rejected by Responses. +//! Provider errors are attributed only against the immutable submitted request. + +use std::collections::BTreeSet; +use std::time::Duration; + +use codex_attachment_store::AttachmentStore; +use codex_history::AcceptedFileAudio; +use codex_history::CodexHarnessMetadata; +use codex_history::ResponseItemEnvelope; +use codex_protocol::error::CodexErrorDetails; +use codex_protocol::models::AudioReference; +use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputContentItem; +use codex_protocol::models::ImageReference; +use codex_protocol::models::ResponseItem; +use futures::StreamExt; +use futures::stream; + +/// Exact file identity proven by a terminal provider error and its submitted request. +#[derive(Clone)] +pub(crate) struct RejectedAudioFileIds(pub(crate) Vec); + +pub(crate) fn pending_audio_confirmation( + previous: Option<&AcceptedFileAudio>, + turn_id: &str, + model: &str, + provider: &str, + items: &[ResponseItem], +) -> Option { + let ids = items + .iter() + .filter_map(|item| match item { + ResponseItem::Message { content, .. } => Some(content.iter()), + _ => None, + }) + .flatten() + .filter_map(|part| match part { + ContentItem::InputAudio { + audio: AudioReference::File { file_id }, + } => Some(file_id), + _ => None, + }); + if ids + .clone() + .all(|id| previous.is_some_and(|known| known.file_ids.contains(id))) + { + return None; + } + Some(AcceptedFileAudio { + turn_id: turn_id.to_owned(), + model: model.to_owned(), + provider: provider.to_owned(), + file_ids: ids.cloned().collect(), + }) +} + +enum FileErrorSource<'a> { + Audio(&'a str), + InlineAudio, + FileImage, + Opaque, +} + +fn visit_sources<'a>(item: &'a ResponseItem, mut visit: impl FnMut(FileErrorSource<'a>)) { + match item { + ResponseItem::Message { content, .. } => { + for part in content { + match part { + ContentItem::InputAudio { + audio: AudioReference::File { file_id }, + } => visit(FileErrorSource::Audio(file_id)), + ContentItem::InputAudio { + audio: AudioReference::Inline { .. }, + } => visit(FileErrorSource::InlineAudio), + ContentItem::InputImage { + image: ImageReference::File { .. }, + .. + } => visit(FileErrorSource::FileImage), + _ => {} + } + } + } + ResponseItem::FunctionCallOutput { output, .. } + | ResponseItem::CustomToolCallOutput { output, .. } => { + for part in output.content_items().unwrap_or_default() { + match part { + FunctionCallOutputContentItem::InputAudio { .. } => { + visit(FileErrorSource::InlineAudio) + } + FunctionCallOutputContentItem::InputImage { + image: ImageReference::File { .. }, + .. + } => visit(FileErrorSource::FileImage), + _ => {} + } + } + } + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } => { + visit(FileErrorSource::Opaque) + } + _ => {} + } +} + +/// Older opaque checkpoints carry no provenance and continue to compete for attribution. +pub(crate) fn checkpoint_input_has_no_file_error_sources(items: &[ResponseItemEnvelope]) -> bool { + items.iter().all(|envelope| { + let mut clear = true; + visit_sources(&envelope.item, |source| { + clear &= matches!(source, FileErrorSource::Opaque) + && clear_checkpoint(&envelope.item, envelope.metadata.as_ref()).is_some(); + }); + clear + }) +} + +/// Only the host marker on this ciphertext proves that the checkpoint had no competing files. +pub(crate) fn clear_checkpoint<'a>( + item: &'a ResponseItem, + metadata: Option<&CodexHarnessMetadata>, +) -> Option<&'a str> { + metadata.filter(|metadata| metadata.compaction_without_file_error_sources)?; + opaque_content(item) +} + +fn opaque_content(item: &ResponseItem) -> Option<&str> { + match item { + ResponseItem::Compaction { + encrypted_content, .. + } => Some(encrypted_content), + ResponseItem::ContextCompaction { + encrypted_content, .. + } => encrypted_content.as_deref(), + _ => None, + } +} + +/// Audio IDs and competing sources in the immutable submitted request. The prompt itself +/// can be released while the provider streams. +pub(crate) struct SubmittedFileAudio { + file_ids: Vec, + candidate: Option, + has_inline_audio: bool, + has_file_image: bool, +} + +impl SubmittedFileAudio { + pub(crate) fn from_request<'a>( + items: &[ResponseItem], + accepted: Option<&AcceptedFileAudio>, + clear_checkpoints: impl Iterator + Clone, + ) -> Option { + let mut file_ids = Vec::new(); + let mut has_inline_audio = false; + let mut has_file_image = false; + let mut unknown_checkpoint = false; + for item in items { + visit_sources(item, |source| match source { + FileErrorSource::Audio(id) => file_ids.push(id), + FileErrorSource::InlineAudio => has_inline_audio = true, + FileErrorSource::FileImage => has_file_image = true, + FileErrorSource::Opaque => { + unknown_checkpoint |= !opaque_content(item).is_some_and(|ciphertext| { + clear_checkpoints.clone().any(|known| known == ciphertext) + }); + } + }); + } + if file_ids.is_empty() { + return None; + } + let candidate = if unknown_checkpoint { + None + } else if file_ids.len() == 1 { + Some(0) + } else { + let mut unseen = file_ids + .iter() + .enumerate() + .filter(|(_, id)| !accepted.is_some_and(|known| known.file_ids.contains(**id))); + let first = unseen.next().map(|(index, _)| index); + first.filter(|_| unseen.next().is_none()) + }; + Some(Self { + file_ids: file_ids.into_iter().map(str::to_owned).collect(), + candidate, + has_inline_audio, + has_file_image, + }) + } + + fn rejected_index(&self, error: &CodexErrorDetails) -> Option { + let index = self.candidate?; + let unique_file = self.file_ids.len() == 1; + let attributable = match error { + CodexErrorDetails::InvalidAudioInput { per_recording, .. } => { + !self.has_inline_audio && (unique_file || *per_recording) + } + _ => unique_file && !self.has_file_image && is_unknown_file_error(error), + }; + attributable.then_some(index) + } + + pub(crate) fn matches_rejection(&self, error: &CodexErrorDetails) -> bool { + self.rejected_index(error).is_some() + } + + pub(crate) fn rejected_file_id(mut self, error: &CodexErrorDetails) -> Option { + let index = self.rejected_index(error)?; + Some(self.file_ids.swap_remove(index)) + } + + pub(crate) async fn rejected_file_ids( + mut self, + error: &CodexErrorDetails, + store: &dyn AttachmentStore, + ) -> Vec { + if let Some(index) = self.rejected_index(error) { + return vec![self.file_ids.swap_remove(index)]; + } + if is_unknown_file_error(error) { + // A file that decoded successfully may since have expired. Probe only IDs from + // this submission; later queued input cannot be blamed for the provider's 404. + self.file_ids.sort_unstable(); + self.file_ids.dedup(); + return unavailable_file_audio_ids(self.file_ids, store) + .await + .into_iter() + .collect(); + } + Vec::new() + } +} + +fn is_unknown_file_error(error: &CodexErrorDetails) -> bool { + const UNKNOWN_FILE: &str = "Invalid input: Unknown `file_id` in input content."; + matches!(error, CodexErrorDetails::UnexpectedStatus(response) + if response.status == http::StatusCode::NOT_FOUND + && (response.body.trim() == UNKNOWN_FILE + || response.extract_error_message().as_deref() == Some(UNKNOWN_FILE))) +} + +/// A long-lived thread can reach compaction after an earlier upload has expired. +/// Only an authoritative negative availability check permits omitting the file. +pub(crate) async fn unavailable_file_audio( + items: &[ResponseItem], + store: &dyn AttachmentStore, +) -> BTreeSet { + let mut ids = BTreeSet::new(); + for item in items { + if let ResponseItem::Message { content, .. } = item { + for part in content { + if let ContentItem::InputAudio { + audio: AudioReference::File { file_id }, + } = part + { + ids.insert(file_id.clone()); + } + } + } + } + unavailable_file_audio_ids(ids, store).await +} + +async fn unavailable_file_audio_ids( + ids: impl IntoIterator, + store: &dyn AttachmentStore, +) -> BTreeSet { + let mut checks = stream::iter(ids) + .map(|id| async move { (!store.is_resolvable(&id).await).then_some(id) }) + .buffer_unordered(8); + let mut missing = BTreeSet::new(); + // Bound the entire preflight. Responses can still validate unchecked IDs. + let _ = tokio::time::timeout(Duration::from_millis(250), async { + while let Some(result) = checks.next().await { + missing.extend(result); + } + }) + .await; + missing +} + +/// Keep per-item metadata paired with the submitted input when an audio-only +/// message disappears. The resulting projection also becomes the checkpoint. +pub(crate) fn project_compaction_audio( + input: &mut Vec, + metadata: &mut Vec>, + unavailable: &BTreeSet, +) { + if unavailable.is_empty() { + return; + } + debug_assert_eq!(input.len(), metadata.len()); + let mut envelopes = input + .drain(..) + .zip(metadata.drain(..)) + .map(|(item, metadata)| ResponseItemEnvelope { item, metadata }) + .collect(); + project_file_audio(&mut envelopes, unavailable); + for envelope in envelopes { + input.push(envelope.item); + metadata.push(envelope.metadata); + } +} + +pub(super) fn project_file_audio( + items: &mut Vec, + rejected_file_ids: &BTreeSet, +) { + items.retain_mut(|envelope| { + let ResponseItem::Message { + role, + content, + internal_chat_message_metadata_passthrough, + .. + } = &mut envelope.item + else { + return true; + }; + let keep = content + .iter() + .map(|part| { + !matches!(part, ContentItem::InputAudio { + audio: AudioReference::File { file_id }, + } if rejected_file_ids.contains(file_id)) + }) + .collect::>(); + if keep.iter().all(|keep| *keep) { + return true; + } + retain_aligned(content, &keep); + if let Some(metadata) = internal_chat_message_metadata_passthrough { + if let Some(kinds) = &mut metadata.content_item_kinds { + retain_aligned(kinds, &keep); + } + if let Some(sources) = &mut metadata.content_item_metadata { + retain_aligned(sources, &keep); + } + } + role != "user" || !content.is_empty() + }); +} + +fn retain_aligned(items: &mut Vec, keep: &[bool]) { + if items.len() != keep.len() { + items.clear(); + return; + } + let mut index = 0; + items.retain(|_| { + let retain = keep[index]; + index += 1; + retain + }); +} + +#[cfg(test)] +#[path = "history_rejected_audio_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs b/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs new file mode 100644 index 0000000000..4ec0467901 --- /dev/null +++ b/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs @@ -0,0 +1,236 @@ +#![cfg(test)] + +use super::*; +use codex_protocol::models::ContentItemKind; +use codex_protocol::models::InternalChatMessageMetadataPassthrough; +use pretty_assertions::assert_eq; + +fn file_audio(file_id: &str) -> ContentItem { + ContentItem::InputAudio { + audio: AudioReference::File { + file_id: file_id.to_string(), + }, + } +} + +fn inline_audio() -> ContentItem { + ContentItem::InputAudio { + audio: AudioReference::Inline { + audio_url: "data:audio/wav;base64,AA==".to_string(), + }, + } +} + +fn message(content: Vec) -> ResponseItem { + ResponseItem::Message { + encrypted_content: None, + status: None, + id: None, + role: "user".to_string(), + content, + phase: None, + internal_chat_message_metadata_passthrough: None, + } +} + +fn audio_error() -> CodexErrorDetails { + CodexErrorDetails::InvalidAudioInput { + per_recording: true, + message: "invalid recording".to_string(), + } +} + +fn rejected_file_audio_id(error: &CodexErrorDetails, items: &[ResponseItem]) -> Option { + SubmittedFileAudio::from_request(items, /*accepted*/ None, std::iter::empty())? + .rejected_file_id(error) +} + +#[test] +fn previously_accepted_file_only_disambiguates_clip_specific_errors() { + let mut context = codex_history::RetainedContext::default(); + let good = message(vec![file_audio("file_a")]); + let evidence = pending_audio_confirmation( + /*previous*/ None, + "turn", + "model", + "provider", + std::slice::from_ref(&good), + ) + .expect("new audio"); + context.record(&codex_history::RetainedContextEvent::AcceptedFileAudio { + evidence: evidence.into(), + }); + assert!( + context + .accepted_file_audio("another model", "provider") + .is_none() + ); + assert!( + context + .accepted_file_audio("model", "another provider") + .is_none() + ); + let accepted = context.accepted_file_audio("model", "provider"); + let items = [good, message(vec![file_audio("file_b")])]; + let candidate = SubmittedFileAudio::from_request(&items, accepted, std::iter::empty()) + .expect("one new recording"); + assert!(candidate.matches_rejection(&audio_error())); + assert!( + !candidate.matches_rejection(&CodexErrorDetails::InvalidAudioInput { + message: "The combined audio input is too large.".into(), + per_recording: false, + }) + ); + assert_eq!( + candidate.rejected_file_id(&audio_error()).as_deref(), + Some("file_b") + ); +} + +#[test] +fn tool_output_audio_makes_an_invalid_audio_error_ambiguous() { + for kind in ["function_call_output", "custom_tool_call_output"] { + let output = serde_json::from_value(serde_json::json!({ + "type": kind, + "call_id": "call", + "output": [{"type": "input_audio", "audio_url": "data:audio/wav;base64,AA=="}], + })) + .expect("valid tool output"); + let items = [message(vec![file_audio("valid_upload")]), output]; + assert_eq!( + rejected_file_audio_id(&audio_error(), &items), + None, + "{kind}" + ); + } +} + +#[test] +fn attributes_audio_error_only_to_one_visible_file_audio() { + let error = audio_error(); + assert_eq!( + rejected_file_audio_id(&error, &[message(vec![file_audio("file_a")])]), + Some("file_a".to_string()) + ); + assert_eq!( + rejected_file_audio_id( + &error, + &[message(vec![file_audio("file_a"), file_audio("file_b")])] + ), + None + ); + assert_eq!( + rejected_file_audio_id( + &error, + &[message(vec![file_audio("file_a"), inline_audio()])] + ), + None + ); + assert_eq!( + rejected_file_audio_id( + &error, + &[ + message(vec![file_audio("file_a")]), + ResponseItem::ContextCompaction { + id: None, + encrypted_content: None, + internal_chat_message_metadata_passthrough: None, + }, + ] + ), + None + ); +} + +#[test] +fn unknown_file_error_counts_the_submitted_projection_not_raw_history() { + let error = CodexErrorDetails::UnexpectedStatus(codex_protocol::error::UnexpectedResponseError { + status: http::StatusCode::NOT_FOUND, + body: serde_json::json!({"error": {"message": "Invalid input: Unknown `file_id` in input content."}}).to_string(), + url: None, + user_message: None, + cf_ray: None, + request_id: None, + identity_authorization_error: None, + identity_error_code: None, + }); + let unavailable_image_projection = message(vec![ + ContentItem::InputText { + text: "[Image unavailable]".to_string(), + }, + file_audio("file_audio"), + ]); + assert_eq!( + rejected_file_audio_id(&error, &[unavailable_image_projection]), + Some("file_audio".to_string()) + ); + let unresolved_image = message(vec![ + ContentItem::InputImage { + image: ImageReference::File { + file_id: "file_image".to_string(), + }, + detail: None, + }, + file_audio("file_audio"), + ]); + assert_eq!(rejected_file_audio_id(&error, &[unresolved_image]), None); +} + +#[test] +fn projection_removes_only_rejected_audio_and_keeps_metadata_aligned() { + let mut first = message(vec![ + ContentItem::InputText { + text: "first".to_string(), + }, + file_audio("file_a"), + ]); + let ResponseItem::Message { + internal_chat_message_metadata_passthrough, + .. + } = &mut first + else { + unreachable!() + }; + *internal_chat_message_metadata_passthrough = Some(InternalChatMessageMetadataPassthrough { + content_item_kinds: Some(vec![ + ContentItemKind("user.text".to_string()), + ContentItemKind("user.audio".to_string()), + ]), + ..Default::default() + }); + let second = message(vec![file_audio("file_b")]); + let rejected_only = message(vec![file_audio("file_a")]); + let raw = vec![first, second.clone(), rejected_only]; + let mut projected = raw + .clone() + .into_iter() + .map(ResponseItemEnvelope::new) + .collect(); + project_file_audio(&mut projected, &BTreeSet::from(["file_a".to_string()])); + let projected = projected + .into_iter() + .map(ResponseItemEnvelope::into_item) + .collect::>(); + + let ResponseItem::Message { + content, + internal_chat_message_metadata_passthrough: Some(metadata), + .. + } = &projected[0] + else { + panic!("expected retained first message") + }; + assert_eq!( + content, + &[ContentItem::InputText { + text: "first".to_string() + }] + ); + assert_eq!( + metadata.content_item_kinds, + Some(vec![ContentItemKind("user.text".to_string())]) + ); + assert_eq!(projected[1], second); + assert_eq!(projected.len(), 2); + assert_eq!(raw.len(), 3, "the retained history was not rewritten"); +} diff --git a/codex-rs/core/src/context_manager/mod.rs b/codex-rs/core/src/context_manager/mod.rs index 2bd7eb87c0..3ffc67c061 100644 --- a/codex-rs/core/src/context_manager/mod.rs +++ b/codex-rs/core/src/context_manager/mod.rs @@ -8,4 +8,5 @@ pub(crate) use history::HistoryReplacement; pub(crate) use history::estimate_image_reference_bytes; pub(crate) use history::estimate_item_token_count; pub(crate) use history::is_user_turn_boundary; +pub(crate) use history::rejected_audio; pub(crate) use image_availability::replace_unresolvable_image_files; diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index d5e1062e57..fcaa4b1d5d 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -107,6 +107,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() { vec![checkpoint.into()], turn.to_turn_context_item(), crate::context::world_state::WorldStateSnapshot::default(), + /*rejected_audio_file_ids*/ Default::default(), crate::compact::CompactedHistoryMetadata { input_goal_ids: Default::default(), message: String::new(), @@ -808,6 +809,7 @@ async fn wait_for_guardian_review_ignores_prior_turn_errors() { misalignment: None, message: "stale guardian error".to_string(), codex_error_info: None, + rejected_audio_file_ids: Vec::new(), }), }) .await @@ -856,6 +858,7 @@ async fn wait_for_guardian_review_preserves_structured_session_error() { misalignment: None, message: "temporary failure".to_string(), codex_error_info: Some(CodexErrorInfo::ServerOverloaded), + rejected_audio_file_ids: Vec::new(), }), }) .await diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 5183c2509d..e18a9acb10 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -2112,6 +2112,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: .collect(), turn.to_turn_context_item(), crate::context::world_state::WorldStateSnapshot::default(), + /*rejected_audio_file_ids*/ Default::default(), crate::compact::CompactedHistoryMetadata { input_goal_ids: Default::default(), message: String::new(), diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index d1d4a0ed9d..09235da17d 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -699,6 +699,7 @@ pub(crate) async fn run_legacy_after_agent_hook( misalignment: None, message, codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }); sess.send_event(turn_context, event).await; true diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 6f26c1b228..a139b0e96f 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -2916,6 +2916,7 @@ async fn send_conversation_error( misalignment: None, message, codex_error_info: Some(codex_error_info), + rejected_audio_file_ids: Vec::new(), }), }) .await; diff --git a/codex-rs/core/src/session/async_compaction_tests.rs b/codex-rs/core/src/session/async_compaction_tests.rs index d73b96c657..e6ad3dc549 100644 --- a/codex-rs/core/src/session/async_compaction_tests.rs +++ b/codex-rs/core/src/session/async_compaction_tests.rs @@ -4,7 +4,10 @@ use super::*; use codex_analytics::CompactionReason; use codex_analytics::CompactionTrigger; use codex_login::CodexAuth; +use codex_protocol::models::AudioReference; +use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::InputModality; use codex_protocol::protocol::TokenUsage; use core_test_support::responses; use pretty_assertions::assert_eq; @@ -21,7 +24,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac provider.base_url = Some(format!("{}/v1", server.uri())); provider.supports_websockets = false; let server_uri = server.uri(); - let (sess, turn, _events) = + let (sess, mut turn, _events) = crate::session::tests::make_session_and_context_with_auth_and_config_and_rx( CodexAuth::create_dummy_chatgpt_auth_for_testing(), Vec::new(), @@ -29,6 +32,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac config.model = Some("gpt-5.2".to_string()); config.model_context_window = Some(100_000); config.model_auto_compact_token_limit = Some(90_000); + config.async_compaction_threshold_percent = 50; config.model_provider = provider; config.chatgpt_base_url = server_uri; config.features.enable(Feature::AsyncCompaction).unwrap(); @@ -36,14 +40,27 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac }, ) .await; - let request = responses::mount_sse_once( + crate::session::tests::update_turn_settings_for_test( + Arc::get_mut(&mut turn).expect("unshared turn context"), + |settings| { + Arc::make_mut(&mut settings.model_info) + .input_modalities + .push(InputModality::Audio); + }, + ); + let request = responses::mount_response_sequence( &server, - responses::sse(vec![ - serde_json::json!({"type":"response.output_item.done", "item": { - "type":"compaction", "encrypted_content":"prepared-summary" - }}), - responses::ev_completed("prepared"), - ]), + vec![ + wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({ + "error": {"message": "Invalid input: Unknown `file_id` in input content."} + })), + responses::sse_response(responses::sse(vec![ + serde_json::json!({"type":"response.output_item.done", "item": { + "type":"compaction", "encrypted_content":"prepared-summary" + }}), + responses::ev_completed("prepared"), + ])), + ], ) .await; let turn_cancellation = CancellationToken::new(); @@ -51,18 +68,44 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac .capture_step_context(Arc::clone(&turn), &turn_cancellation) .await .expect("capture step"); - let original: ResponseItem = serde_json::from_value(serde_json::json!({ - "type": "message", "role": "user", - "content": [{"type": "input_text", "text": "original turn"}] - })) - .expect("user message"); + let rejected_file_id = "file_rejected_background_compaction"; + let original = ResponseItem::Message { + encrypted_content: None, + status: None, + id: None, + role: "user".to_string(), + content: vec![ + ContentItem::InputText { + text: "original turn".to_string(), + }, + ContentItem::InputAudio { + audio: AudioReference::File { + file_id: rejected_file_id.to_string(), + }, + }, + ContentItem::InputAudio { + audio: AudioReference::File { + file_id: "file_expired_since_last_turn".to_string(), + }, + }, + ], + phase: None, + internal_chat_message_metadata_passthrough: None, + }; sess.replace_history(vec![original], /*reference_context_item*/ None) .await; + sess.state + .lock() + .await + .history + .restore_rejected_audio_file_ids(std::collections::BTreeSet::from([ + rejected_file_id.to_string() + ])); sess.update_token_usage_info( &turn, Some(&TokenUsage { - input_tokens: 70_000, - total_tokens: 70_000, + input_tokens: 60_000, + total_tokens: 60_000, ..Default::default() }), /*includes_reasoning*/ false, @@ -70,6 +113,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac .await .expect("record usage"); let status = super::super::context_window::context_window_token_status(&sess, &turn).await; + assert!(!status.token_limit_reached); let world_state = Arc::new(WorldState::default()); let cancelled = CancellationToken::new(); cancelled.cancel(); @@ -118,6 +162,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac .await; // Wait for client publication before exercising either ready-result consumer. wait_for_background_compaction_for_test(&sess).await; + assert_eq!(request.requests().len(), 2, "background retry completed"); let next_step = sess .capture_step_context(Arc::clone(&turn), &CancellationToken::new()) .await @@ -163,6 +208,13 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac }) .collect::>(); assert_eq!(summaries, ["prepared-summary"]); - assert_eq!(request.requests().len(), 1); + let requests = request.requests(); + assert_eq!(requests.len(), 2); + for (index, request) in requests.into_iter().enumerate() { + let body = request.body_json().to_string(); + assert!(body.contains("original turn")); + assert!(!body.contains(rejected_file_id)); + assert_eq!(body.contains("file_expired_since_last_turn"), index == 0); + } shutdown(&sess).await; } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 4b9a79e3ab..c85206ac66 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -300,6 +300,7 @@ pub async fn set_thread_memory_mode(sess: &Arc, sub_id: String, mode: T misalignment: None, message: err.to_string(), codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }), }; sess.send_event_raw(event).await; @@ -400,6 +401,7 @@ pub async fn shutdown(sess: &Arc, sub_id: String) -> bool { misalignment: None, message: "Failed to shutdown thread persistence".to_string(), codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }), }; sess.send_event_raw(event).await; @@ -449,6 +451,7 @@ pub async fn review( misalignment: None, message: err.to_string(), codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }), }; sess.send_event(&turn_context, event.msg).await; @@ -552,6 +555,7 @@ pub(super) async fn submission_loop( misalignment: None, message: err.to_string(), codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }), }) .await; @@ -681,6 +685,7 @@ pub(super) async fn submission_loop( misalignment: None, message, codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: Vec::new(), }), }) .await; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 4113bfbea4..0c2f12e450 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1909,6 +1909,7 @@ impl Session { ) -> Option { let rollout_reconstruction::RolloutReconstruction { mut history, + rejected_audio_file_ids, retained_context, guardian_history, last_started_turn_id, @@ -1977,6 +1978,9 @@ impl Session { reference_context_item, HistoryReplacement::Reset, ); + state + .history + .restore_rejected_audio_file_ids(rejected_audio_file_ids); state.history.restore_review_context( Some(&retained_context), guardian_history.as_ref(), @@ -2582,7 +2586,7 @@ impl Session { } /// Persist the event to rollout and send it to clients. - pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) { + pub(crate) async fn send_event(&self, turn_context: &TurnContext, mut msg: EventMsg) { if let Some((item_id, started_at_ms)) = msg.command_approval_start() { // The client can see the command before execution. Keep its first start // for the canonical lifecycle events that will be saved in history. @@ -2591,19 +2595,32 @@ impl Session { .record_item_started(item_id.to_owned(), started_at_ms) .await; } - let legacy_source = msg.clone(); - if let EventMsg::Error(error) = &legacy_source + if let EventMsg::TurnComplete(completed) = &msg + && completed.error.is_some() + { + self.state + .lock() + .await + .history + .record_rejected_file_audio(completed); + } + if let EventMsg::Error(error) = &mut msg && error .codex_error_info .as_ref() .is_some_and(CodexErrorInfo::affects_turn_status) { - turn_context - .terminal_error - .lock() - .await - .replace(error.clone()); + let mut terminal_error = turn_context.terminal_error.lock().await; + if let Some(previous) = terminal_error.as_ref() { + for file_id in &previous.rejected_audio_file_ids { + if !error.rejected_audio_file_ids.contains(file_id) { + error.rejected_audio_file_ids.push(file_id.clone()); + } + } + } + terminal_error.replace(error.clone()); } + let legacy_source = msg.clone(); self.services .rollout_thread_trace .record_codex_turn_event(&turn_context.sub_id, &legacy_source); @@ -4608,6 +4625,7 @@ impl Session { mut items: Vec, reference_context_item: TurnContextItem, mut world_state_baseline: WorldStateSnapshot, + rejected_audio_file_ids: std::collections::BTreeSet, installation: impl Into>, ) -> bool { let installation = installation.into(); @@ -4737,6 +4755,9 @@ impl Session { ); } let replacement_history = items.clone(); + state + .history + .extend_rejected_audio_file_ids(rejected_audio_file_ids); state.replace_annotated_history( items, Some(reference_context_item.clone()), @@ -4764,6 +4785,7 @@ impl Session { compaction_response_id: metadata.compaction_response_id, latest_token_usage_record: state.latest_token_usage_record.clone(), resume_metadata: Some(CompactionResumeMetadata { + rejected_audio_file_ids: state.history.rejected_audio_file_ids(), dynamic_tools: Some(Box::new(DynamicToolsSnapshot { thread_id: self.thread_id(), tools: state.persisted_dynamic_tools.to_vec(), @@ -5289,6 +5311,7 @@ impl Session { context_items, turn_context_item, world_state_snapshot, + /*rejected_audio_file_ids*/ Default::default(), CompactedHistoryMetadata { input_goal_ids, message: String::new(), diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 130d5e9ddc..888981f782 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -6,6 +6,7 @@ use codex_history::TurnAttribution; use codex_protocol::protocol::SessionContextWindow; use codex_protocol::protocol::ThreadHistoryMode; use codex_utils_output_truncation::TruncationPolicy; +use std::collections::BTreeSet; use uuid::Uuid; // Return value of `Session::reconstruct_history_from_rollout`, bundling the rebuilt history with @@ -13,6 +14,7 @@ use uuid::Uuid; #[derive(Debug, PartialEq)] pub(crate) struct RolloutReconstruction { pub(super) history: Vec, + pub(super) rejected_audio_file_ids: BTreeSet, pub(crate) retained_context: codex_history::RetainedContext, pub(super) guardian_history: Option, pub(super) last_started_turn_id: Option, @@ -211,6 +213,27 @@ impl ContextManager { mut history: Self, truncation_policy: TruncationPolicy, ) -> RolloutReconstruction { + // Rejection facts are durable across compaction boundaries. File IDs are immutable, + // so retaining a fact after rollback is conservative and cannot suppress a different + // future upload. + let mut rejected_audio_file_ids = BTreeSet::new(); + for item in rollout_items { + match item { + RolloutItem::Compacted(compacted) => { + if let Some(metadata) = &compacted.resume_metadata { + rejected_audio_file_ids + .extend(metadata.rejected_audio_file_ids.iter().cloned()); + } + } + RolloutItem::EventMsg(EventMsg::TurnComplete(completed)) => { + if let Some(error) = &completed.error { + rejected_audio_file_ids + .extend(error.rejected_audio_file_ids.iter().cloned()); + } + } + _ => {} + } + } // Select the compaction and suffix that can affect reconstruction. let has_legacy_compaction_without_window_number = rollout_items.iter().any(|item| { @@ -616,6 +639,7 @@ impl ContextManager { guardian_history: history.guardian_history_checkpoint(), last_started_turn_id, history: history.into_annotated_items(), + rejected_audio_file_ids, previous_turn_settings, reference_context_item, turn_attribution, diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index ebc312eafa..27c657d4c4 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -1622,6 +1622,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( compaction_response_id: None, latest_token_usage_record: None, resume_metadata: current.then(|| codex_history::CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: None, last_started_turn_id: Some(format!("wake-{window_number}")), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 1ab26f39fb..f9c25a35c0 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6121,6 +6121,7 @@ async fn settings_checkpoint_waits_for_accepted_settings_persistence() { vec![ResponseItemEnvelope::new(user_message("compacted history"))], turn_context.to_turn_context_item(), WorldStateSnapshot::default(), + /*rejected_audio_file_ids*/ Default::default(), CompactedHistoryMetadata { input_goal_ids: Default::default(), message: "summary".to_string(), @@ -6255,6 +6256,7 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore vec![ResponseItemEnvelope::new(user_message("compacted history"))], turn_context.to_turn_context_item(), WorldStateSnapshot::default(), + /*rejected_audio_file_ids*/ Default::default(), CompactedHistoryMetadata { input_goal_ids: Default::default(), message: "summary".to_string(), @@ -6393,6 +6395,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() { session.multi_agent_version = std::sync::OnceLock::from(MultiAgentVersion::V2); let expected = CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: Some(Box::new(DynamicToolsSnapshot { thread_id: session.thread_id(), tools: Vec::new(), @@ -6450,6 +6453,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() { vec![ResponseItemEnvelope::new(user_message("compacted context"))], turn_context_baseline.clone(), world_state.render_full().0, + /*rejected_audio_file_ids*/ Default::default(), CompactedHistoryMetadata { input_goal_ids, message: String::new(), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 9d1d5caa60..769917524f 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -16,6 +16,7 @@ use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_ use crate::connectors; use crate::context::ContextualUserFragment; use crate::context::UserVerificationNotice; +use crate::context_manager::rejected_audio; use crate::cyber_access_program; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::feedback_tags; @@ -105,6 +106,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::InputModality; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::AgentMessageContentDeltaEvent; use codex_protocol::protocol::AgentReasoningSectionBreakEvent; @@ -183,18 +185,21 @@ pub(crate) async fn run_turn( prewarmed_client_session: Option, cancellation_token: CancellationToken, ) -> CodexResult> { - if input.iter().any(turn_input_has_uploaded_audio_file) { + let model_info = turn_context.capture_current_model_info(); + if !model_info.input_modalities.contains(&InputModality::Audio) + && input.iter().any(turn_input_has_uploaded_audio_file) + { let result = run_hooks_and_record_inputs( &sess, &turn_context, - &turn_context.capture_current_model_info(), + &model_info, &input, PersistContext::TurnStart, ) .await; return match result { Err(err) => Err(err), - Ok(_) => Err(uploaded_audio_disabled_error()), + Ok(_) => Err(unsupported_uploaded_audio_error()), }; } let auto_compaction_enabled = sess @@ -912,6 +917,7 @@ pub(crate) async fn run_turn( message: "Invalid image in your last message. Please remove it and try again." .to_string(), codex_error_info: Some(error), + rejected_audio_file_ids: Vec::new(), }); sess.send_event(&turn_context, event).await; break; @@ -928,7 +934,23 @@ pub(crate) async fn run_turn( sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone(), e.details()) .await; sess.track_turn_codex_error(turn_context.as_ref(), &e); - let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None)); + let mut error_event = e.to_error_event(/*message_prefix*/ None); + if let Some(rejected) = turn_context + .extension_data + .remove::() + { + error_event + .rejected_audio_file_ids + .extend(Arc::unwrap_or_clone(rejected).0); + } + if matches!(e.details(), CodexErrorDetails::InvalidAudioInput { .. }) + || !error_event.rejected_audio_file_ids.is_empty() + { + error_event.message.push_str( + " Please remove or replace the affected audio attachment and try again.", + ); + } + let event = EventMsg::Error(error_event); sess.send_event(&turn_context, event).await; // let the user continue the conversation break; @@ -972,7 +994,7 @@ pub(crate) async fn run_hooks_and_record_inputs( ) -> CodexResult { let mut blocked_input = false; let mut accepted_user_input = false; - for input_item in recordable_inputs(sess, input) { + for input_item in recordable_inputs(sess, model_info, input) { let input_item = input_item?; let hook_outcome = inspect_pending_input(sess, turn_context, &input_item).await; if hook_outcome.should_stop { @@ -1008,9 +1030,11 @@ pub(crate) async fn run_hooks_and_record_inputs( /// Yields usable input in order, then reports unsupported audio after valid siblings can be recorded. pub(crate) fn recordable_inputs<'a>( sess: &Session, + model_info: &ModelInfo, input: &'a [TurnInput], ) -> impl Iterator>> { - let rejected_audio = input.iter().any(turn_input_has_uploaded_audio_file); + let rejected_audio = !model_info.input_modalities.contains(&InputModality::Audio) + && input.iter().any(turn_input_has_uploaded_audio_file); // Only finalized evidence may enter reusable Guardian history. let items: &[TurnInput] = if sess .services @@ -1024,9 +1048,10 @@ pub(crate) fn recordable_inputs<'a>( }; items .iter() - .filter_map(|item| { + .filter_map(move |item| { let mut item = Cow::Borrowed(item); - if turn_input_has_uploaded_audio_file(&item) + if rejected_audio + && turn_input_has_uploaded_audio_file(&item) && let TurnInput::UserInput { content, .. } = item.to_mut() { content.retain(|part| !is_uploaded_audio_file(part)); @@ -1036,16 +1061,16 @@ pub(crate) fn recordable_inputs<'a>( } Some(Ok(item)) }) - .chain(rejected_audio.then(|| Err(uploaded_audio_disabled_error()))) + .chain(rejected_audio.then(|| Err(unsupported_uploaded_audio_error()))) } fn turn_input_has_uploaded_audio_file(input: &TurnInput) -> bool { matches!(input, TurnInput::UserInput { content, .. } if content.iter().any(is_uploaded_audio_file)) } -fn uploaded_audio_disabled_error() -> CodexErr { +fn unsupported_uploaded_audio_error() -> CodexErr { CodexErr::UnsupportedOperation( - "Uploaded audio file inputs are not enabled in this build.".to_string(), + "The selected model does not support uploaded audio. Choose an audio-capable model or remove the attachment.".to_string(), ) } @@ -1058,7 +1083,7 @@ fn is_uploaded_audio_file(input: &UserInput) -> bool { ) } -fn turn_user_input(input: &[TurnInput]) -> Vec { +pub(crate) fn turn_user_input(input: &[TurnInput]) -> Vec { input .iter() .filter_map(|item| match item { @@ -1877,6 +1902,30 @@ async fn run_sampling_request( &sess.services.thread_extension_data, &mut prompt.tools, ); + let (submitted_audio, pending_audio) = { + let model = &step_context.settings.model_info.slug; + let provider = &turn_context.config.model_provider_id; + let history = sess.clone_history().await; + let accepted = history + .retained_context() + .accepted_file_audio(model, provider); + let pending = rejected_audio::pending_audio_confirmation( + accepted, + &turn_context.sub_id, + model, + provider, + &prompt.input, + ); + let clear_checkpoints = history.annotated_items().iter().filter_map(|envelope| { + rejected_audio::clear_checkpoint(&envelope.item, envelope.metadata.as_ref()) + }); + let submitted = rejected_audio::SubmittedFileAudio::from_request( + &prompt.input, + accepted, + clear_checkpoints, + ); + (submitted, pending) + }; let mut draining_for_mailbox = false; let err = match try_run_sampling_request( tool_runtime.clone(), @@ -1887,6 +1936,7 @@ async fn run_sampling_request( &responses_metadata, Arc::clone(&turn_diff_tracker), prompt, + pending_audio, input_item_count, cancellation_token.child_token(), &mut draining_for_mailbox, @@ -1915,6 +1965,19 @@ async fn run_sampling_request( }, }; + // A steer can preempt retry handling. Finish this terminal failure with the + // identity from the submitted request before draining any new recordings. + if let Some(audio) = submitted_audio { + let ids = audio + .rejected_file_ids(err.details(), sess.services.image_store.as_ref()) + .await; + if !ids.is_empty() { + turn_context + .extension_data + .insert(rejected_audio::RejectedAudioFileIds(ids)); + return Err(err); + } + } let retry = handle_response_stream_error( &mut retry_state, max_retries, @@ -2757,6 +2820,7 @@ async fn try_run_sampling_request( responses_metadata: &CodexResponsesMetadata, turn_diff_tracker: SharedTurnDiffTracker, prompt: Prompt, + mut pending_audio: Option, input_item_count: usize, cancellation_token: CancellationToken, draining_for_mailbox: &mut bool, @@ -3351,6 +3415,14 @@ async fn try_run_sampling_request( usage_metadata, end_turn, } => { + if let Some(accepted) = pending_audio.take() { + sess.record_retained_context( + codex_history::RetainedContextEvent::AcceptedFileAudio { + evidence: Arc::new(accepted), + }, + ) + .await; + } response_completed = true; request_token_usage = token_usage.clone(); sess.services diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index c6a6081750..637904c2bb 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -1237,6 +1237,7 @@ impl Session { misalignment: None, message: message.clone(), codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: Vec::new(), }), }) .await; diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 94f1664e12..fec9bd7a1a 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -849,6 +849,7 @@ impl Session { misalignment: None, message: format!("failed to submit turn input: {reason:?}"), codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: Vec::new(), }), }) .await; diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index c8df7c18d2..78cd5d7f01 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -734,6 +734,7 @@ async fn admission_revalidates_constraints_before_committing(kind: TurnStartKind misalignment: None, message: expected_message, codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: Vec::new(), }] ); assert_eq!(session.thread_settings_snapshot().await, desired_settings); diff --git a/codex-rs/core/src/tasks/admission.rs b/codex-rs/core/src/tasks/admission.rs index fc27d02f83..e796ed93dc 100644 --- a/codex-rs/core/src/tasks/admission.rs +++ b/codex-rs/core/src/tasks/admission.rs @@ -105,7 +105,7 @@ impl Session { return Ok(()); } - for item in recordable_inputs(self, &input) { + for item in recordable_inputs(self, &model_info, &input) { let item = item?; // The user must resubmit a prompt that still needs a submit hook. if matches!(item.as_ref(), TurnInput::UserInput { .. }) diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index eb4ae1023e..5e1774ff14 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -759,10 +759,11 @@ impl Session { std::mem::take(&mut ts.token_usage_by_model), ) }; + let model_info = turn_context.capture_current_model_info(); let record_result = run_hooks_and_record_inputs( self, &turn_context, - &turn_context.capture_current_model_info(), + &model_info, &pending_input, PersistContext::Standard, ) diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index f3915ed5c7..1ce6ebfbf4 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -370,6 +370,7 @@ async fn send_user_shell_error(session: &Session, turn_context: &TurnContext, me misalignment: None, message: message.to_string(), codex_error_info: None, + rejected_audio_file_ids: Vec::new(), }), ) .await; diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 13b446e291..f9e819ddd4 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -1143,6 +1143,20 @@ where response_mock } +/// Choose a response from the decoded request, including compressed requests. +pub async fn mount_response_handler( + server: &MockServer, + respond: impl Fn(ResponsesRequest) -> ResponseTemplate + Send + Sync + 'static, +) -> ResponseMock { + let (mock, requests) = base_mock(); + mock.respond_with(move |request: &wiremock::Request| { + respond(ResponsesRequest(request.clone())) + }) + .mount(server) + .await; + requests +} + fn base_mock() -> (MockBuilder, ResponseMock) { let response_mock = ResponseMock::new(); let mock = Mock::given(method("POST")) diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 7bfae31a60..5880b9e151 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -30,6 +30,7 @@ use codex_protocol::ThreadId; use codex_protocol::account::PlanType; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::ServiceTier; +use codex_protocol::models::AudioReference; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; use codex_protocol::models::ExecutedToolCall; @@ -1999,9 +2000,14 @@ async fn responses_websocket_uses_incremental_create_on_prefix() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); - let prompt_one = prompt_with_input(vec![message_item("hello")]); + let audio = ResponseItem::from(ResponseInputItem::from(vec![UserInput::Audio { + audio: AudioReference::File { + file_id: "file_audio".to_string(), + }, + }])); + let prompt_one = prompt_with_input(vec![audio.clone()]); let prompt_two = prompt_with_input(vec![ - message_item("hello"), + audio, assistant_message_item("1", "assistant output"), message_item("second"), ]); @@ -2018,6 +2024,10 @@ async fn responses_websocket_uses_incremental_create_on_prefix() { assert_eq!(first["model"].as_str(), Some(MODEL)); assert_eq!(first["stream"], serde_json::Value::Bool(true)); assert_eq!(first["input"].as_array().map(Vec::len), Some(1)); + assert_eq!( + first["input"][0]["content"], + serde_json::json!([{"type": "input_audio", "file_id": "file_audio"}]), + ); assert_eq!(second["type"].as_str(), Some("response.create")); assert_eq!(second["previous_response_id"].as_str(), Some("resp-1")); assert_eq!( diff --git a/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs b/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs index 495ae98065..ed4b7ce3bb 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs @@ -168,6 +168,7 @@ async fn guardian_circuit_breaker_notifies_parent(action: CircuitBreakAction) -> (action == CircuitBreakAction::Strict).then_some(ErrorEvent { message: warning.clone(), codex_error_info: Some(CodexErrorInfo::TooManyDenials), + rejected_audio_file_ids: Vec::new(), misalignment: None, }), AgentStatus::Interrupted, diff --git a/codex-rs/core/tests/suite/model_switching.rs b/codex-rs/core/tests/suite/model_switching.rs index 3e33e14d2c..2d540d7a58 100644 --- a/codex-rs/core/tests/suite/model_switching.rs +++ b/codex-rs/core/tests/suite/model_switching.rs @@ -910,8 +910,8 @@ async fn model_change_projects_media_without_changing_live_or_replayed_history( detail: None, }, UserInput::Audio { - audio: AudioReference::Inline { - audio_url: "data:audio/wav;base64,YXVkaW8=".to_string(), + audio: AudioReference::File { + file_id: "file_model_switch_audio".to_string(), }, }, UserInput::Text { @@ -945,9 +945,16 @@ async fn model_change_projects_media_without_changing_live_or_replayed_history( !first_request.message_input_image_urls("user").is_empty(), "first request should include the uploaded image" ); + let first_audio = first_request.body_json()["input"] + .as_array() + .into_iter() + .flatten() + .flat_map(|item| item["content"].as_array().into_iter().flatten()) + .find(|part| part["type"] == "input_audio") + .cloned(); assert_eq!( - first_request.message_input_audio_urls("user"), - vec!["data:audio/wav;base64,YXVkaW8=".to_string()] + first_audio, + Some(json!({"type": "input_audio", "file_id": "file_model_switch_audio"})) ); let second_request = requests.last().expect("expected second request"); @@ -961,7 +968,12 @@ async fn model_change_projects_media_without_changing_live_or_replayed_history( "second request should strip unsupported image content" ); assert!( - second_request.message_input_audio_urls("user").is_empty(), + second_request.body_json()["input"] + .as_array() + .into_iter() + .flatten() + .flat_map(|item| item["content"].as_array().into_iter().flatten()) + .all(|part| part["type"] != "input_audio"), "second request should strip unsupported audio content" ); let second_user_texts = second_request.message_input_texts("user"); diff --git a/codex-rs/core/tests/suite/scenarios_audio_file_tests.rs b/codex-rs/core/tests/suite/scenarios_audio_file_tests.rs index c10ebf8f19..00b4a4efd7 100644 --- a/codex-rs/core/tests/suite/scenarios_audio_file_tests.rs +++ b/codex-rs/core/tests/suite/scenarios_audio_file_tests.rs @@ -1,9 +1,1175 @@ -//! New uploaded-audio inputs stay inactive until the runtime integration is installed. +//! Uploaded audio references remain intact across model requests and user-message history. use super::*; use codex_protocol::models::AudioReference; use codex_protocol::openai_models::InputModality; +use core_test_support::responses; +use core_test_support::responses::mount_response_sequence; use pretty_assertions::assert_eq; +use test_case::test_case; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn uploaded_audio_survives_follow_up_and_resume() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_assistant_message("first", "I heard the recording."), + ev_completed("first"), + ]), + sse(vec![ + ev_assistant_message("second", "The recording contains speech."), + ev_completed("second"), + ]), + sse(vec![ + ev_assistant_message("third", "The same recording is still available."), + ev_completed("third"), + ]), + ], + ) + .await; + let builder = || { + test_codex().with_model_info_override("gpt-5.5", |model_info| { + model_info.input_modalities.push(InputModality::Audio); + }) + }; + let test = builder().build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Listen to this recording."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_audio".into(), + }, + }, + ])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_text_turn("What did you hear?").await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed + .submit_text_turn("Listen to the same recording again.") + .await?; + let requests = mock.requests(); + assert_eq!(requests.len(), 3); + for (index, request) in requests.iter().enumerate() { + let audio = request + .input() + .into_iter() + .flat_map(|item| item["content"].as_array().cloned().unwrap_or_default()) + .find(|part| part["type"] == "input_audio"); + assert_eq!( + audio, + Some(json!({"type": "input_audio", "file_id": "file_audio"})), + "request {index}", + ); + } + insta::assert_snapshot!( + "uploaded_audio_survives_follow_up", + context_snapshot::format_request_history_snapshot( + "An uploaded audio file ID is forwarded to Responses and retained across follow-up and resume.", + &requests, + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + Ok(()) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum AudioAtCompaction { + Available, + MissingAtStore, + MissingAtResponses, +} + +#[test_case(AudioAtCompaction::Available, &[true, true, true]; "available")] +#[test_case(AudioAtCompaction::MissingAtStore, &[true, false, false]; "expired before compaction")] +#[test_case(AudioAtCompaction::MissingAtResponses, &[true, true, false, false]; "expired during compaction")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn uploaded_audio_survives_compaction_and_cold_resume( + availability: AudioAtCompaction, + expected_audio: &[bool], +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mut replies = vec![responses::sse_response(sse(vec![ev_completed("initial")]))]; + if availability == AudioAtCompaction::MissingAtResponses { + replies.push(wiremock::ResponseTemplate::new(404).set_body_json(json!({ + "error": {"message": "Invalid input: Unknown `file_id` in input content."} + }))); + } + replies.extend([ + responses::sse_response(sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { + "type": "compaction", + "encrypted_content": "UPLOADED_AUDIO_COMPACTION_CHECKPOINT" + } + }), + ev_completed("compacted"), + ])), + responses::sse_response(sse(vec![ev_completed("resumed")])), + ]); + let responses = mount_response_sequence(&server, replies).await; + let store = Arc::new(RecordingFileAttachmentStore::default()); + let builder = || { + test_codex() + .with_image_store(store.clone()) + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio); + }) + }; + let test = builder().build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Retain this recording through compaction."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_audio_compacted".to_string(), + }, + }, + ])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + if availability == AudioAtCompaction::MissingAtStore { + store + .resolve_errors + .lock() + .expect("attachment store errors") + .insert( + "file_audio_compacted".to_string(), + AttachmentStoreErrorKind::NotFound, + ); + } + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let resumed = builder().restart_with_auto_env(&server, &test).await?; + let mut input = vec![text("Use the retained recording after cold resume.")]; + if availability != AudioAtCompaction::Available { + // Reintroduce the ID to exercise the checkpoint's rejection record. + input.push(UserInput::Audio { + audio: AudioReference::File { + file_id: "file_audio_compacted".into(), + }, + }); + } + resumed + .codex + .start_or_steer_turn(TurnInputRequest::user_input(input)) + .await?; + wait_for_event(&resumed.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = responses.requests(); + assert_eq!(requests.len(), expected_audio.len()); + for (request, expect_audio) in requests.iter().zip(expected_audio) { + let audio = request + .input() + .into_iter() + .flat_map(|item| item["content"].as_array().cloned().unwrap_or_default()) + .find(|part| part["type"] == "input_audio"); + assert_eq!( + audio, + expect_audio.then(|| json!({"type": "input_audio", "file_id": "file_audio_compacted"})), + ); + assert!( + request + .body_json() + .to_string() + .contains("Retain this recording through compaction.") + ); + } + assert_eq!( + requests + .last() + .expect("resumed request") + .inputs_of_type("compaction")[0]["encrypted_content"], + "UPLOADED_AUDIO_COMPACTION_CHECKPOINT" + ); + if availability == AudioAtCompaction::Available { + insta::assert_snapshot!( + "uploaded_audio_compaction_and_cold_resume", + context_snapshot::format_request_history_snapshot( + "The uploaded audio file ID survives remote compaction and a cold resume in canonical message content.", + &requests, + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn expired_audio_during_required_compaction_after_a_text_only_checkpoint() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let requests = responses::mount_response_handler(&server, |request| { + let input = request.body_json()["input"].to_string(); + let has_file = input.contains("file_expired"); + if !request.inputs_of_type("compaction_trigger").is_empty() { + if has_file { + return wiremock::ResponseTemplate::new(404).set_body_json(json!({ + "error": {"message": "Invalid input: Unknown `file_id` in input content."} + })); + } + let checkpoint = if input.contains("FIRST_CHECKPOINT") { + "SECOND_CHECKPOINT" + } else { + "FIRST_CHECKPOINT" + }; + return responses::sse_response(sse(vec![ + json!({"type":"response.output_item.done","item":{ + "type":"compaction","encrypted_content":checkpoint + }}), + ev_completed("compacted"), + ])); + } + let tokens = if has_file { 500_000 } else { 80 }; + responses::sse_response(sse(vec![ + responses::ev_assistant_message("message", "I can continue."), + responses::ev_completed_with_tokens("answer", tokens), + ])) + }) + .await; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| config.model_auto_compact_token_limit = Some(200_000)) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio) + }) + .build_with_auto_env(&server) + .await?; + test.submit_text_turn("Begin with text only.").await?; + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Listen to the recording."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_expired".into(), + }, + }, + ])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + for prompt in ["Continue after the file expired.", "Continue again."] { + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![text(prompt)])) + .await?; + let done = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => Some(done.clone()), + _ => None, + }) + .await; + assert!(done.error.is_none(), "{done:?}"); + assert_eq!(done.last_agent_message.as_deref(), Some("I can continue.")); + } + let requests = requests.requests(); + let compactions: Vec<_> = requests + .iter() + .filter(|r| !r.inputs_of_type("compaction_trigger").is_empty()) + .collect(); + assert_eq!( + compactions.len(), + 3, + "initial compaction, missing-file rejection, and single retry" + ); + for (r, contains_file) in compactions.iter().skip(1).zip([true, false]) { + let input = r.body_json()["input"].to_string(); + assert!(input.contains("FIRST_CHECKPOINT")); + assert_eq!(input.contains("file_expired"), contains_file); + } + for r in requests.iter().rev().take(2) { + assert!(r.inputs_of_type("compaction_trigger").is_empty()); + let input = r.body_json()["input"].to_string(); + assert!(input.contains("SECOND_CHECKPOINT")); + assert!(!input.contains("file_expired")); + } + Ok(()) +} + +#[test_case(400, Some("audio_too_large"), "The audio input is too long."; "invalid audio")] +#[test_case(404, None, "Invalid input: Unknown `file_id` in input content."; "unknown file")] +#[test_case(200, Some("audio_too_large"), "The audio input is too long."; "streamed invalid audio")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejected_uploaded_audio_does_not_poison_follow_up_or_resume( + status: u16, + code: Option<&str>, + message: &str, +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let error = json!({ "type": "invalid_request_error", "code": code, "message": message }); + let rejection = if status == 200 { + wiremock::ResponseTemplate::new(status) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![json!({ + "type": "response.failed", "response": {"error": error} + })])) + } else { + wiremock::ResponseTemplate::new(status).set_body_json(json!({"error": error})) + }; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v1/responses")) + .respond_with(rejection) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + let builder = || { + test_codex() + .with_config(|config| config.model_provider.stream_max_retries = Some(0)) + .with_model_info_override("gpt-5.5", |model_info| { + model_info.input_modalities.push(InputModality::Audio); + }) + }; + let test = builder().build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Listen to this recording."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_rejected".into(), + }, + }, + ])) + .await?; + let EventMsg::TurnComplete(completed) = wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await + else { + unreachable!() + }; + let error = completed.error.as_ref().expect("rejected turn error"); + assert!( + error + .message + .contains("remove or replace the affected audio attachment") + ); + assert_eq!( + error.rejected_audio_file_ids, + vec!["file_rejected".to_string()], + ); + let mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("follow_up")]), + sse(vec![ev_completed("resumed")]), + ], + ) + .await; + test.submit_text_turn("Continue without the recording.") + .await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed.submit_text_turn("Continue after restart.").await?; + let requests = mock.requests(); + assert_eq!(requests.len(), 2); + for request in &requests { + let input = request.input(); + let projected = input + .iter() + .find(|item| item["content"][0]["text"] == "Listen to this recording.") + .expect("the rejected turn's user message remains in history"); + assert_eq!( + ( + &projected["content"], + &projected["internal_chat_message_metadata_passthrough"]["content_item_kinds"], + ), + ( + &json!([{"type": "input_text", "text": "Listen to this recording."}]), + &json!(["user.text"]), + ), + ); + let content = input + .into_iter() + .flat_map(|item| item["content"].as_array().cloned().unwrap_or_default()) + .collect::>(); + assert!( + content.iter().all(|part| part["type"] != "input_audio"), + "failed turn error: {:?}", + completed.error + ); + } + insta::assert_snapshot!( + "rejected_uploaded_audio_follow_up_and_resume", + context_snapshot::format_request_history_snapshot( + "A rejected uploaded recording is omitted only from request projections on follow-up and after resume, preserving surrounding text and aligned annotations without inventing user content.", + &requests, + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejection_is_attributed_against_projection_after_unavailable_image() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let image_store = Arc::new(RecordingFileAttachmentStore::default()); + let responses = mount_response_sequence( + &server, + vec![ + wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![ev_completed("image_turn")])), + wiremock::ResponseTemplate::new(404).set_body_json(json!({ + "error": { + "type": "invalid_request_error", + "code": null, + "message": "Invalid input: Unknown `file_id` in input content." + } + })), + wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![ev_completed("follow_up")])), + wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![ev_completed("resumed")])), + ], + ) + .await; + let builder = || { + test_codex() + .with_image_store(image_store.clone()) + .with_config(|config| config.model_provider.stream_max_retries = Some(0)) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio); + }) + }; + let test = builder().build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Remember this historical image."), + UserInput::Image { + image: ImageReference::File { + file_id: "file_image_unavailable_later".to_string(), + }, + detail: None, + }, + ])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + image_store + .resolve_errors + .lock() + .expect("resolve errors lock") + .insert( + "file_image_unavailable_later".to_string(), + AttachmentStoreErrorKind::NotFound, + ); + + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Check this recording."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_audio_rejected_after_projection".to_string(), + }, + }, + ])) + .await?; + let rejected = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(completed) => Some(completed.clone()), + _ => None, + }) + .await; + assert!(rejected.error.is_some()); + test.submit_text_turn("Continue without unavailable files.") + .await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed.submit_text_turn("Continue after restart.").await?; + + let requests = responses.requests(); + assert_eq!(requests.len(), 4); + assert!( + requests[0] + .body_json() + .to_string() + .contains("file_image_unavailable_later") + ); + let submitted = requests[1].body_json().to_string(); + assert!(submitted.contains("[image unavailable]")); + assert!(!submitted.contains("file_image_unavailable_later")); + assert!(submitted.contains("file_audio_rejected_after_projection")); + for request in &requests[2..] { + let body = request.body_json().to_string(); + assert!(body.contains("[image unavailable]")); + assert!(!body.contains("file_image_unavailable_later")); + assert!(!body.contains("file_audio_rejected_after_projection")); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn ambiguous_multi_audio_rejection_preserves_all_audio_on_follow_up_and_resume() -> Result<()> +{ + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let responses = mount_response_sequence( + &server, + vec![ + wiremock::ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "type": "invalid_request_error", + "code": "invalid_audio", + "message": "one recording is invalid" + } + })), + wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![ev_completed("follow_up")])), + wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse(vec![ev_completed("resumed")])), + ], + ) + .await; + let builder = || { + test_codex() + .with_config(|config| config.model_provider.stream_max_retries = Some(0)) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio); + }) + }; + let test = builder().build_with_auto_env(&server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Compare both recordings."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_ambiguous_a".to_string(), + }, + }, + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_ambiguous_b".to_string(), + }, + }, + ])) + .await?; + let rejected = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(completed) => Some(completed.clone()), + _ => None, + }) + .await; + let error = rejected.error.expect("ambiguous audio rejection"); + assert!( + error + .message + .contains("remove or replace the affected audio attachment") + ); + assert!(error.rejected_audio_file_ids.is_empty()); + test.submit_text_turn("Retry with the retained recordings.") + .await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed.submit_text_turn("Retry after restart.").await?; + + let requests = responses.requests(); + assert_eq!(requests.len(), 3); + for request in requests { + let body = request.body_json().to_string(); + assert!(body.contains("file_ambiguous_a")); + assert!(body.contains("file_ambiguous_b")); + } + Ok(()) +} + +#[test_case(AttachmentStoreErrorKind::NotFound, true; "store confirms B expired")] +#[test_case(AttachmentStoreErrorKind::Backend, false; "transient storage error preserves B")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unknown_file_after_two_recordings_were_accepted( + storage_error: AttachmentStoreErrorKind, + confirmed: bool, +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let expired = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let reject = Arc::clone(&expired); + let requests = responses::mount_response_handler(&server, move |r| { + if reject.load(std::sync::atomic::Ordering::Relaxed) + && r.body_json()["input"].to_string().contains("file_b") + { + return wiremock::ResponseTemplate::new(404).set_body_json(json!({ + "error": {"message": "Invalid input: Unknown `file_id` in input content."} + })); + } + responses::sse_response(sse(vec![ + ev_assistant_message("answer", "I can continue."), + ev_completed("completed"), + ])) + }) + .await; + let store = Arc::new(RecordingFileAttachmentStore::default()); + let builder = || { + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_image_store(store.clone()) + .with_config(|config| config.model_provider.stream_max_retries = Some(0)) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio) + }) + }; + let test = builder().build_with_auto_env(&server).await?; + for id in ["file_a", "file_b"] { + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Audio { + audio: AudioReference::File { file_id: id.into() }, + }])) + .await?; + let done = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => Some(done.clone()), + _ => None, + }) + .await; + assert!(done.error.is_none(), "{done:?}"); + } + assert!(store.resolves.lock().expect("probes").is_empty()); + store + .resolve_errors + .lock() + .expect("store") + .insert("file_b".into(), storage_error); + expired.store(true, std::sync::atomic::Ordering::Relaxed); + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![text( + "Recall the recordings.", + )])) + .await?; + let error = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => done.error.clone(), + _ => None, + }) + .await; + assert_eq!( + error.rejected_audio_file_ids, + if confirmed { + vec!["file_b"] + } else { + Vec::new() + } + ); + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![text( + "Continue in this chat.", + )])) + .await?; + let follow_up = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => Some(done.clone()), + _ => None, + }) + .await; + assert_eq!(follow_up.error.is_none(), confirmed); + if confirmed { + assert_eq!( + follow_up.last_agent_message.as_deref(), + Some("I can continue.") + ); + } + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed + .submit_text_turn("Continue after reopening.") + .await?; + let all = requests.requests(); + assert_eq!(all.len(), 5); + for request in &all[3..] { + let input = request.body_json()["input"].to_string(); + assert!(input.contains("file_a")); + assert_eq!(input.contains("file_b"), !confirmed); + } + let probes = store.resolves.lock().expect("probes"); + assert!(probes.contains_key("file_a") && probes.contains_key("file_b")); + Ok(()) +} + +#[test_case(false, false; "accepted recording in a previous turn")] +#[test_case(false, true; "accepted recording before cold resume")] +#[test_case(true, false; "first recording after opaque text-only compaction")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn persistently_invalid_recording_is_removed_on_follow_up_and_resume( + compact_text: bool, + resume_before_rejection: bool, +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let request_count = std::sync::atomic::AtomicUsize::new(0); + let requests = responses::mount_response_handler(&server, move |request| { + let index = request_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let body = request.body_json(); + let input = body["input"].to_string(); + if input.contains("file_bad_b") { + return wiremock::ResponseTemplate::new(400).set_body_json(json!({ + "error": {"code": "audio_too_short", "message": "The recording is too short."} + })); + } + if compact_text && index == 1 { + return responses::sse_response(sse(vec![ + json!({"type": "response.output_item.done", "item": { + "type": "compaction", "encrypted_content": "TEXT_ONLY_CHECKPOINT" + }}), + ev_completed("compacted"), + ])); + } + responses::sse_response(sse(vec![ev_completed("accepted")])) + }) + .await; + let builder = || { + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio) + }) + }; + let recording = |id: &str| UserInput::Audio { + audio: AudioReference::File { file_id: id.into() }, + }; + let test = builder().build_with_auto_env(&server).await?; + let mut initial = vec![text("Start this conversation.")]; + if !compact_text { + initial.push(recording("file_good_a")); + } + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(initial)) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + if compact_text { + test.codex.submit(Op::Compact).await?; + let compacted = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => Some(done.clone()), + _ => None, + }) + .await; + assert!(compacted.error.is_none(), "{compacted:?}"); + } + let test = if resume_before_rejection { + builder().restart_with_auto_env(&server, &test).await? + } else { + test + }; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Check this new recording."), + recording("file_bad_b"), + ])) + .await?; + let error = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(done) => done.error.clone(), + _ => None, + }) + .await; + assert_eq!(error.rejected_audio_file_ids, ["file_bad_b"]); + test.submit_text_turn("Continue without the rejected upload.") + .await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed + .submit_text_turn("Continue after reopening.") + .await?; + let requests = requests.requests(); + let rejected = requests.len() - 3; + assert!( + requests[rejected].body_json()["input"] + .to_string() + .contains("file_bad_b") + ); + for request in &requests[rejected..] { + let input = request.body_json()["input"].to_string(); + assert!(input.contains(if compact_text { + "TEXT_ONLY_CHECKPOINT" + } else { + "file_good_a" + })); + } + for request in &requests[rejected + 1..] { + assert!( + !request.body_json()["input"] + .to_string() + .contains("file_bad_b") + ); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn paginated_resume_filters_audio_rejected_before_and_after_compaction() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let rejection = wiremock::ResponseTemplate::new(400).set_body_json(json!({ + "error": {"type": "invalid_request_error", "code": "invalid_audio", "message": "bad recording"} + })); + let requests = mount_response_sequence( + &server, + vec![ + rejection.clone(), + responses::sse_response(sse(vec![ + json!({"type": "response.output_item.done", "item": { + "type": "compaction", "encrypted_content": "PAGINATED_AUDIO_CHECKPOINT" + }}), + ev_completed("compacted"), + ])), + rejection, + responses::sse_response(sse(vec![ev_completed("resumed")])), + ], + ) + .await; + let builder = || { + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio) + }) + }; + let test = builder().build_with_auto_env(&server).await?; + let recording = |id: &str| UserInput::Audio { + audio: AudioReference::File { file_id: id.into() }, + }; + for id in ["file_A", "file_B"] { + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Check the recording."), + recording(id), + ])) + .await?; + let error = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(completed) => completed.error.clone(), + _ => None, + }) + .await; + assert_eq!(error.rejected_audio_file_ids, [id]); + if id == "file_A" { + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + } + } + test.codex.shutdown_and_wait().await?; + let path = test + .session_configured + .rollout_path + .clone() + .expect("rollout path"); + // The bounded CCA page contains the session header, checkpoint and suffix. + // Remove A's failed turn entirely, so only the checkpoint can remember its ID. + let mut page = String::new(); + let mut at_checkpoint = false; + for line in fs::read_to_string(&path)?.lines() { + let record = codex_rollout::parse_rollout_line(line)?; + if let RolloutItem::Compacted(compacted) = &record.item { + assert_eq!( + compacted + .resume_metadata + .as_ref() + .expect("resume metadata") + .rejected_audio_file_ids + .as_ref(), + &std::collections::BTreeSet::from(["file_A".to_string()]) + ); + at_checkpoint = true; + } + if at_checkpoint || matches!(record.item, RolloutItem::SessionMeta(_)) { + page.push_str(line); + page.push('\n'); + } + } + assert!(at_checkpoint); + fs::write(&path, page)?; + let resumed = builder().resume(&server, test.home.clone(), path).await?; + resumed + .codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Check these recordings again."), + recording("file_A"), + recording("file_B"), + recording("file_C"), + ])) + .await?; + wait_for_event(&resumed.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let requests = requests.requests(); + assert_eq!(requests.len(), 4); + let body = requests[3].body_json().to_string(); + assert!(body.contains("PAGINATED_AUDIO_CHECKPOINT")); + assert!(body.contains("file_C")); + assert!(!body.contains("file_A")); + assert!(!body.contains("file_B")); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn instant_interrupt_preserves_delayed_http_audio_rejection() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let builder = || { + test_codex() + .with_config(|config| { + config + .features + .enable(Feature::InstantInterrupt) + .expect("enable InstantInterrupt"); + config.model_provider.supports_websockets = false; + }) + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio) + }) + }; + let test = builder().build_with_auto_env(&server).await?; + let received = Arc::new(tokio::sync::Notify::new()); + let started = Arc::clone(&received); + let (release, gate) = std::sync::mpsc::channel(); + let gate = std::sync::Mutex::new(gate); + let rejection = wiremock::ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "type": "invalid_request_error", + "code": "invalid_audio", + "message": "bad recording" + } + })); + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v1/responses")) + .respond_with(move |_: &wiremock::Request| { + started.notify_one(); + gate.lock() + .expect("HTTP response gate") + .recv_timeout(Duration::from_secs(15)) + .expect("steer B before rejecting A"); + rejection.clone() + }) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + let recording = |id: &str| UserInput::Audio { + audio: AudioReference::File { file_id: id.into() }, + }; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![recording("file_A")])) + .await?; + let turn_id = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), + _ => None, + }) + .await; + tokio::time::timeout(Duration::from_secs(15), received.notified()).await?; + test.codex + .steer_turn( + TurnInputRequest::user_input(vec![recording("file_B")]), + turn_id, + ) + .await?; + release.send(())?; + let completed = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(completed) => Some(completed.clone()), + _ => None, + }) + .await; + assert_eq!( + completed + .error + .expect("terminal rejection of A") + .rejected_audio_file_ids, + ["file_A"] + ); + mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("live")]), + sse(vec![ev_completed("resumed")]), + ], + ) + .await; + test.submit_text_turn("Continue with the replacement.") + .await?; + let resumed = builder().restart_with_auto_env(&server, &test).await?; + resumed.submit_text_turn("Use the same recording.").await?; + let requests = responses::received_responses_requests(&server).await; + assert_eq!(requests.len(), 3); + for (request, expected_id) in requests.iter().zip(["file_A", "file_B", "file_B"]) { + let parts: Vec<_> = request + .input() + .iter() + .flat_map(|item| item["content"].as_array().into_iter().flatten()) + .filter(|part| part["type"] == "input_audio") + .cloned() + .collect(); + assert_eq!( + parts, + [json!({"type": "input_audio", "file_id": expected_id})] + ); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn provider_rejection_preserves_a_different_queued_audio_file() -> Result<()> { + skip_if_no_network!(Ok(())); + let (release, gate) = tokio::sync::oneshot::channel(); + let failure = json!({ + "type": "response.failed", + "response": { + "error": { + "type": "invalid_request_error", + "code": "invalid_audio", + "message": "invalid recording" + } + } + }); + let (streaming, _completions) = start_streaming_sse_server(vec![ + vec![ + StreamingSseChunk { + gate: None, + body: sse(vec![ + ev_response_created("active"), + ev_assistant_message("working", "I am checking the recording."), + ]), + }, + StreamingSseChunk { + gate: Some(gate), + body: sse(vec![failure]), + }, + ], + vec![StreamingSseChunk { + gate: None, + body: sse(vec![ev_completed("follow_up")]), + }], + vec![StreamingSseChunk { + gate: None, + body: sse(vec![ev_completed("resumed")]), + }], + ]) + .await; + let config_server = start_mock_server().await; + let base_url = format!("{}/v1", streaming.uri()); + let builder = || { + let base_url = base_url.clone(); + test_codex() + .with_model_info_override("gpt-5.5", |model| { + model.input_modalities.push(InputModality::Audio); + }) + .with_config(move |config| { + config.model_provider.base_url = Some(base_url); + config.model_provider.stream_max_retries = Some(0); + config.model_provider.supports_websockets = false; + config + .features + .disable(Feature::InstantInterrupt) + .expect("queue steering"); + config + .features + .disable(Feature::EnableRequestCompression) + .expect("record request bodies"); + }) + }; + let test = builder().build_with_auto_env(&config_server).await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + text("Check submitted recording A."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_a_rejected".into(), + }, + }, + ])) + .await?; + let turn_id = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), + _ => None, + }) + .await; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::ItemCompleted(event) + if matches!(&event.item, TurnItem::AgentMessage(message) if message.id == "working")) + }) + .await; + test.codex + .steer_turn( + TurnInputRequest::user_input(vec![ + text("Keep queued recording B."), + UserInput::Audio { + audio: AudioReference::File { + file_id: "file_b_queued".into(), + }, + }, + ]), + turn_id, + ) + .await?; + release.send(()).expect("release rejection"); + let completed = wait_for_event_match(&test.codex, |event| match event { + EventMsg::TurnComplete(completed) => Some(completed.clone()), + _ => None, + }) + .await; + assert_eq!( + completed + .error + .as_ref() + .map(|error| error.rejected_audio_file_ids.clone()), + Some(vec!["file_a_rejected".to_string()]) + ); + + test.submit_text_turn("Continue with queued recording B.") + .await?; + let resumed = builder() + .restart_with_auto_env(&config_server, &test) + .await?; + resumed + .submit_text_turn("Continue after restart with recording B.") + .await?; + + let requests = streaming + .requests() + .await + .iter() + .map(|body| serde_json::from_slice::(body)) + .collect::>>()?; + assert_eq!(requests.len(), 3); + let audio_ids = |request: &serde_json::Value| { + request["input"] + .as_array() + .into_iter() + .flatten() + .flat_map(|item| item["content"].as_array().into_iter().flatten()) + .filter_map(|part| part["file_id"].as_str().map(str::to_string)) + .collect::>() + }; + assert_eq!(audio_ids(&requests[0]), vec!["file_a_rejected".to_string()]); + for request in &requests[1..] { + assert_eq!(audio_ids(request), vec!["file_b_queued".to_string()]); + } + Ok(()) +} #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn unsupported_steered_audio_is_rejected_before_continuation_and_resume() -> Result<()> { @@ -39,7 +1205,9 @@ async fn unsupported_steered_audio_is_rejected_before_continuation_and_resume() let base_url = base_url.clone(); test_codex() .with_model_info_override("gpt-5.5", |model| { - model.input_modalities.push(InputModality::Audio); + model + .input_modalities + .retain(|modality| *modality != InputModality::Audio); }) .with_config(move |config| { config.model_provider.base_url = Some(base_url); @@ -134,7 +1302,9 @@ async fn unsupported_uploaded_audio_fails_before_inference_without_poisoning_fol let mock = mount_sse_sequence(&server, vec![sse(vec![ev_completed("follow_up")])]).await; let test = test_codex() .with_model_info_override("gpt-5.5", |model| { - model.input_modalities.push(InputModality::Audio); + model + .input_modalities + .retain(|modality| *modality != InputModality::Audio); }) .build_with_auto_env(&server) .await?; diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__rejected_uploaded_audio_follow_up_and_resume.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__rejected_uploaded_audio_follow_up_and_resume.snap new file mode 100644 index 0000000000..8fd07d88aa --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__rejected_uploaded_audio_follow_up_and_resume.snap @@ -0,0 +1,27 @@ +--- +source: core/tests/suite/scenarios_audio_file_tests.rs +expression: "context_snapshot::format_request_history_snapshot(\"A rejected uploaded recording is omitted only from request projections on follow-up and after resume, preserving surrounding text and aligned annotations without inventing user content.\",\n&requests, &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: A rejected uploaded recording is omitted only from request projections on follow-up and after resume, preserving surrounding text and aligned annotations without inventing user content. + +## Window 1 +-- request 1 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - multi_agent_v1: Tools for spawning and managing sub-agents. + +02:message/user: + +03:message/user: + Listen to this recording. +04:message/user: + Continue without the recording. +-- request 2 (turn) -- +05:message/user: + +06:message/user: + Continue after restart. diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_compaction_and_cold_resume.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_compaction_and_cold_resume.snap new file mode 100644 index 0000000000..3e48a81ec4 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_compaction_and_cold_resume.snap @@ -0,0 +1,44 @@ +--- +source: core/tests/suite/scenarios_audio_file_tests.rs +expression: "context_snapshot::format_request_history_snapshot(\"The uploaded audio file ID survives remote compaction and a cold resume in canonical message content.\",\n&requests, &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: The uploaded audio file ID survives remote compaction and a cold resume in canonical message content. + +## Window 1 +-- request 1 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - multi_agent_v1: Tools for spawning and managing sub-agents. + +02:message/user: + +03:message/user[2]: + [01] Retain this recording through compaction. + [02] +-- request 2 (compaction) -- +04:compaction_trigger + +## Window 2 (after request 2: input diverged at item 04) +-- request 3 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - multi_agent_v1: Tools for spawning and managing sub-agents. + +02:message/user: + +03:message/user[2]: + [01] Retain this recording through compaction. + [02] +04:compaction:encrypted=true; chars=36; hash=09ef79034851b5bf +05:message/user: + +06:message/user: + Use the retained recording after cold resume. diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_survives_follow_up.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_survives_follow_up.snap new file mode 100644 index 0000000000..1e5691820f --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__audio_file__uploaded_audio_survives_follow_up.snap @@ -0,0 +1,33 @@ +--- +source: core/tests/suite/scenarios_audio_file_tests.rs +expression: "context_snapshot::format_request_history_snapshot(\"An uploaded audio file ID is forwarded to Responses and retained across follow-up and resume.\",\n&requests, &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: An uploaded audio file ID is forwarded to Responses and retained across follow-up and resume. + +## Window 1 +-- request 1 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - multi_agent_v1: Tools for spawning and managing sub-agents. + +02:message/user: + +03:message/user[2]: + [01] Listen to this recording. + [02] +-- request 2 (turn) -- +04:message/assistant: + I heard the recording. +05:message/user: + What did you hear? +-- request 3 (turn) -- +06:message/assistant: + The recording contains speech. +07:message/user: + +08:message/user: + Listen to the same recording again. diff --git a/codex-rs/ext/guardian-reviewer/src/review.rs b/codex-rs/ext/guardian-reviewer/src/review.rs index 08f0433b2d..2f18f505a5 100644 --- a/codex-rs/ext/guardian-reviewer/src/review.rs +++ b/codex-rs/ext/guardian-reviewer/src/review.rs @@ -237,6 +237,7 @@ impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { ErrorEvent { message, codex_error_info: Some(CodexErrorInfo::TooManyDenials), + rejected_audio_file_ids: Vec::new(), misalignment: None, }, ) diff --git a/codex-rs/history/src/compaction_resume_metadata.rs b/codex-rs/history/src/compaction_resume_metadata.rs index ca22476e1a..6368889632 100644 --- a/codex-rs/history/src/compaction_resume_metadata.rs +++ b/codex-rs/history/src/compaction_resume_metadata.rs @@ -9,6 +9,8 @@ use codex_protocol::turn_input::TurnAttribution; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; +use std::collections::BTreeSet; +use std::sync::Arc; /// Resume metadata that is not represented by the companion records after a compaction. /// @@ -16,6 +18,9 @@ use serde::Serialize; /// compactions that did not. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] pub struct CompactionResumeMetadata { + /// Immutable file references rejected before this checkpoint. + #[serde(default, skip_serializing_if = "BTreeSet::is_empty")] + pub rejected_audio_file_ids: Arc>, /// Catalog owned by the producing thread; absent in older checkpoints. #[serde(default, skip_serializing_if = "Option::is_none")] pub dynamic_tools: Option>, diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 10f7089e6a..a475172c77 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -108,6 +108,10 @@ pub struct CodexHarnessMetadata { #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub harness_authored_configuration: bool, + /// The exact input to this opaque checkpoint contained no audio or file-backed images. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub compaction_without_file_error_sources: bool, + /// Producer compatibility for an opaque compaction item, never the currently selected model. #[serde(default, skip_serializing_if = "Option::is_none")] pub compaction_model_hash: Option, @@ -280,6 +284,7 @@ mod sender_user_messages; pub use sender_user_messages::SenderUserMessages; pub use reconciled_retained_context::ReconciledRetainedContext; +pub use retained_context::AcceptedFileAudio; pub use retained_context::RetainedContext; pub use retained_context::RetainedContextEntry; pub use retained_context::RetainedContextEvent; diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index 6418bd6907..f827425649 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -11,7 +11,9 @@ pub use source::RetainedSource; pub use source::RetainedSourceId; pub use source::RetainedSourceRole; +use std::collections::BTreeSet; use std::collections::VecDeque; +use std::sync::Arc; use schemars::JsonSchema; use serde::Deserialize; @@ -141,10 +143,29 @@ impl std::fmt::Debug for VerifiedAnswer { } } +/// File IDs in a completed Responses request, scoped to its model and provider. +/// This proves decoding only. The same ID may subsequently expire. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct AcceptedFileAudio { + pub turn_id: String, + pub model: String, + pub provider: String, + pub file_ids: BTreeSet, +} + +impl AcceptedFileAudio { + fn fits_record_budget(&self) -> bool { + serde_json::to_vec(self).is_ok_and(|bytes| bytes.len() <= MAX_RECORD_BYTES) + } +} + /// Sparse, model-invisible updates. Only the host may produce these records. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] #[serde(tag = "type", rename_all = "snake_case")] pub enum RetainedContextEvent { + AcceptedFileAudio { + evidence: Arc, + }, VerifiedAnswer { #[serde(flatten)] answer: VerifiedAnswer, @@ -164,6 +185,8 @@ pub enum RetainedContextEvent { /// Facts live until their instruction boundary is rolled back; compaction does not expire them. #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] pub struct RetainedContext { + #[serde(default, skip_serializing_if = "Option::is_none")] + accepted_file_audio: Option>, verified_answers: VecDeque>, /// Lost evidence cannot be treated as a complete authorization history. #[serde(rename = "incomplete")] @@ -233,11 +256,19 @@ impl RetainedContextEvent { } } Self::DeliveredAssistantMessage { message, .. } => message.bound(), + // An oversized audio record is omitted in record(); IDs must never be truncated. + Self::AcceptedFileAudio { .. } => {} } } } impl RetainedContext { + pub fn accepted_file_audio(&self, model: &str, provider: &str) -> Option<&AcceptedFileAudio> { + self.accepted_file_audio + .as_deref() + .filter(|evidence| evidence.model == model && evidence.provider == provider) + } + /// Context for the latest delivery still present in this task's retained history. pub fn sender_user_messages(&self) -> Option<&SenderUserMessages> { self.sender_deliveries.back().map(|entry| &entry.value) @@ -443,6 +474,15 @@ impl RetainedContext { let mut event = event.clone(); event.bound(); match event { + RetainedContextEvent::AcceptedFileAudio { evidence } => { + if !evidence.fits_record_budget() + || self.accepted_file_audio.as_ref() == Some(&evidence) + { + return false; + } + self.accepted_file_audio = Some(evidence); + true + } RetainedContextEvent::VerifiedAnswer { answer, acceptance_order, @@ -498,6 +538,8 @@ impl RetainedContext { user_messages_incomplete: true, ..Self::default() }); + self.accepted_file_audio + .take_if(|evidence| !evidence.fits_record_budget()); for entry in &mut self.verified_answers { let mut event = RetainedContextEvent::VerifiedAnswer { answer: entry.value.clone(), @@ -567,6 +609,8 @@ impl RetainedContext { first_removed_message_id: Option<&str>, source: RetainedInputSource, ) { + self.accepted_file_audio + .take_if(|evidence| turn_ids.contains(&evidence.turn_id.as_str())); let boundary = source.acceptance_order().map(RetainedContextOrder::Local); if let Some(boundary) = boundary.or_else(|| { first_removed_message_id.and_then(|id| { diff --git a/codex-rs/history/src/retained_context_tests.rs b/codex-rs/history/src/retained_context_tests.rs index 6cb0a35704..4d569d1102 100644 --- a/codex-rs/history/src/retained_context_tests.rs +++ b/codex-rs/history/src/retained_context_tests.rs @@ -4,6 +4,35 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use pretty_assertions::assert_eq; +#[test] +fn accepted_audio_checkpoint_discards_oversized_and_rolled_back_evidence() { + let mut retained = RetainedContext::default(); + let proof = |id: String| RetainedContextEvent::AcceptedFileAudio { + evidence: Arc::new(AcceptedFileAudio { + turn_id: "accepted".into(), + model: "model".into(), + provider: "provider".into(), + file_ids: BTreeSet::from([id]), + }), + }; + assert!(retained.record(&proof("file_a".into()))); + assert!(!retained.record(&proof("x".repeat(MAX_RECORD_BYTES)))); + let bytes = serde_json::to_vec(&retained).expect("encode checkpoint"); + let mut restored: RetainedContext = serde_json::from_slice(&bytes).expect("decode checkpoint"); + let get_ids = |state: &RetainedContext| { + state + .accepted_file_audio("model", "provider") + .map(|proof| proof.file_ids.clone()) + }; + assert_eq!(get_ids(&restored), Some(BTreeSet::from(["file_a".into()]))); + restored.rollback( + &["accepted"], + /*first_removed_message_id*/ None, + RetainedInputSource::Local(None), + ); + assert_eq!(get_ids(&restored), None); +} + fn publish_answer() -> RetainedContextEvent { RetainedContextEvent::VerifiedAnswer { answer: VerifiedAnswer { diff --git a/codex-rs/history/src/rollout_payload.rs b/codex-rs/history/src/rollout_payload.rs index 301c7d3c06..75295ee774 100644 --- a/codex-rs/history/src/rollout_payload.rs +++ b/codex-rs/history/src/rollout_payload.rs @@ -49,7 +49,7 @@ pub(super) enum RolloutItemWire<'a> { payload: InterAgentCommunicationMetadataPayload, }, Compacted { - payload: Cow<'a, CompactedItem>, + payload: Cow<'a, Box>, }, TurnContext { payload: Cow<'a, TurnContextItem>, @@ -239,9 +239,7 @@ impl From> for RolloutItem { trigger_turn: payload.trigger_turn, } } - RolloutItemWire::Compacted { payload } => { - Self::Compacted(Box::new(payload.into_owned())) - } + RolloutItemWire::Compacted { payload } => Self::Compacted(payload.into_owned()), RolloutItemWire::TurnContext { payload } => Self::TurnContext(payload.into_owned()), RolloutItemWire::TokenUsageRecord { payload } => { Self::TokenUsageRecord(payload.into_owned()) diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index c02aa200cf..f7a8357cef 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -585,6 +585,7 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu #[test] fn compacted_resume_metadata_presence_round_trips_empty_values() -> Result<()> { let resume_metadata = CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: None, last_started_turn_id: None, @@ -697,6 +698,7 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> compaction_response_id: None, latest_token_usage_record: None, resume_metadata: Some(CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("turn-1".to_string()), diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index bd4a60866d..3dff3b4a75 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -155,7 +155,11 @@ pub enum CodexErrorDetails { #[error("{0}")] InvalidRequest(String), #[error("{message}")] - InvalidAudioInput { message: String }, + InvalidAudioInput { + message: String, + /// True when the provider rejected an individual clip, independent of request size. + per_recording: bool, + }, #[error("{message}")] ModelNotFound { message: String, @@ -558,6 +562,7 @@ impl CodexErr { ErrorEvent { message, codex_error_info: Some(self.to_codex_protocol_error()), + rejected_audio_file_ids: Vec::new(), misalignment: match &self.details { CodexErrorDetails::MisalignmentPolicyViolation { misalignment, .. } => { misalignment.clone() @@ -640,7 +645,8 @@ impl UnexpectedResponseError { truncate_with_ellipsis(trimmed_body, UNEXPECTED_RESPONSE_BODY_MAX_BYTES) } - fn extract_error_message(&self) -> Option { + /// Returns the provider message from a structured HTTP error body. + pub fn extract_error_message(&self) -> Option { let json = serde_json::from_str::(&self.body).ok()?; let message = json .get("error") diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 0634952ec5..6a6bef52a5 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2331,6 +2331,11 @@ pub struct ErrorEvent { pub message: String, #[serde(default)] pub codex_error_info: Option, + /// Host-only file identities proven rejected against the submitted request. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[schemars(skip)] + #[ts(skip)] + pub rejected_audio_file_ids: Vec, /// Sensitive explanation and steering are delivered live but never enter rollout storage. #[serde(skip)] #[schemars(skip)] @@ -6172,6 +6177,7 @@ mod tests { misalignment: None, message: "rollback failed".into(), codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed), + rejected_audio_file_ids: Vec::new(), }; assert!(!event.affects_turn_status()); } @@ -6184,6 +6190,7 @@ mod tests { codex_error_info: Some(CodexErrorInfo::ActiveTurnNotSteerable { turn_kind: NonSteerableTurnKind::Review, }), + rejected_audio_file_ids: Vec::new(), }; assert!(!event.affects_turn_status()); } @@ -6194,6 +6201,7 @@ mod tests { misalignment: None, message: "generic".into(), codex_error_info: Some(CodexErrorInfo::Other), + rejected_audio_file_ids: Vec::new(), }; assert!(event.affects_turn_status()); } @@ -6203,6 +6211,7 @@ mod tests { let event = ErrorEvent { message: "This request violated the misalignment policy.".to_string(), codex_error_info: Some(CodexErrorInfo::MisalignmentPolicyViolation), + rejected_audio_file_ids: Vec::new(), misalignment: Some(MisalignmentErrorDetails { error_type: Some("unauthorized_data_transfer".to_string()), detailed_explanation: Some("Sensitive customer explanation".to_string()), @@ -6231,6 +6240,22 @@ mod tests { assert!(!debug.contains("sensitive-review-target")); } + #[test] + fn rejected_audio_identity_round_trips_for_host_recovery() { + let event = ErrorEvent { + message: "invalid recording".to_string(), + codex_error_info: Some(CodexErrorInfo::BadRequest), + rejected_audio_file_ids: vec!["file_audio".to_string()], + misalignment: None, + }; + + let serialized = serde_json::to_value(&event).expect("serialize error event"); + assert_eq!(serialized["rejected_audio_file_ids"], json!(["file_audio"])); + let restored: ErrorEvent = + serde_json::from_value(serialized).expect("deserialize error event"); + assert_eq!(restored.rejected_audio_file_ids, vec!["file_audio"]); + } + #[test] fn realtime_conversation_started_event_uses_realtime_session_id() { let event = RealtimeConversationStartedEvent { diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 514b36b0b1..d60b119df6 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -273,6 +273,7 @@ async fn fork_version_stops_before_older_segments_once_resolved() { unreachable!(); }; compacted.resume_metadata = Some(codex_rollout::CompactionResumeMetadata { + rejected_audio_file_ids: Default::default(), dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs index f23d057ae7..4b52fbe44d 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs @@ -279,6 +279,11 @@ impl RollbackPlanner { self.pending_turn_records.push(index); } } + RolloutItem::RetainedContext( + codex_rollout::RetainedContextEvent::AcceptedFileAudio { evidence }, + ) => { + self.assign_targeted_record(index, Some(&evidence.turn_id)); + } RolloutItem::TokenUsageRecord(record) => { self.assign_targeted_record(index, Some(record.turn_id.as_str())); }