diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 603a6e4260..02d7104500 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -31,6 +31,7 @@ use codex_otel::HOOK_RUN_METRIC; use codex_protocol::items::FunctionCallOutputItem; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; +use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::AskForApproval; @@ -61,7 +62,8 @@ use tracing::instrument; use crate::context::ContextualUserFragment; use crate::context::HookAdditionalContext; use crate::environment_selection::TurnEnvironmentSnapshot; -use crate::event_mapping::parse_turn_item; +use crate::event_mapping::is_contextual_user_message_content; +use crate::event_mapping::user_input_parts; use crate::guardian::GuardianReviewContext; use crate::hook_environment_executor::with_environment_hooks; use crate::hook_environment_executor::with_tool_environment_hooks; @@ -616,21 +618,38 @@ pub(crate) async fn run_post_compact_hooks( } } +pub(crate) fn legacy_notify_input_messages(input: &[ResponseItem]) -> Vec { + input + .iter() + .filter_map(|item| { + let ResponseItem::Message { role, content, .. } = item else { + return None; + }; + if role != "user" || is_contextual_user_message_content(content) { + return None; + } + Some( + user_input_parts(content) + .filter_map(|(_, part)| match part { + ContentItem::InputText { text } => Some(text.as_str()), + ContentItem::InputImage { .. } + | ContentItem::InputAudio { .. } + | ContentItem::OutputText { .. } => None, + }) + .collect(), + ) + }) + .collect() +} + #[instrument(level = "trace", skip_all)] pub(crate) async fn run_legacy_after_agent_hook( sess: &Arc, turn_context: &Arc, - input: &[ResponseItem], + input_messages: Vec, last_assistant_message: Option, ) -> bool { let mut abort_message = None; - let input_messages = input - .iter() - .filter_map(|item| match parse_turn_item(item) { - Some(TurnItem::UserMessage(user_message)) => Some(user_message.message()), - _ => None, - }) - .collect(); let hooks = sess.hooks(); for hook_outcome in hooks .dispatch(codex_hooks::HookPayload { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 04d5d40e9e..5cd4d77e0e 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -20,6 +20,7 @@ use crate::feedback_tags; use crate::hook_runtime::drain_async_hook_results; use crate::hook_runtime::hooks_for_model_request; use crate::hook_runtime::inspect_pending_input; +use crate::hook_runtime::legacy_notify_input_messages; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::record_pending_input; use crate::hook_runtime::run_legacy_after_agent_hook; @@ -594,7 +595,7 @@ pub(crate) async fn run_turn( } .await; match sampling_request_result { - Ok((sampling_request_output, sampling_request_input)) => { + Ok((sampling_request_output, notify_messages)) => { guardian_budget_compacted = false; let SamplingRequestResult { needs_follow_up: model_needs_follow_up, @@ -790,7 +791,7 @@ pub(crate) async fn run_turn( if run_legacy_after_agent_hook( &sess, &turn_context, - &sampling_request_input, + notify_messages, last_agent_message.clone(), ) .await @@ -1744,7 +1745,7 @@ async fn run_sampling_request( client_session: &mut ModelClientSession, input: Vec, cancellation_token: CancellationToken, -) -> CodexResult<(SamplingRequestResult, Vec)> { +) -> CodexResult<(SamplingRequestResult, Vec)> { let turn_context = Arc::clone(&step_context.turn); let preempt = step_context.preempt.clone().unwrap_or_default(); let _input_watch = if let Some(preempt) = &step_context.preempt { @@ -1777,11 +1778,12 @@ async fn run_sampling_request( ); let mut retry_state = ResponsesStreamRetryState::default(); let mut initial_input = Some(input); - let mut original_input = None; + let mut notify_messages = Vec::new(); let mut executed_tool_calls_by_output = HashMap::new(); loop { // Running code-mode cells can request review while this response is in flight. // Keep the latest received ID until response.created replaces it. + let first_attempt = initial_input.is_some(); let prompt_input = if let Some(input) = initial_input.take() { input } else { @@ -1815,6 +1817,9 @@ async fn run_sampling_request( ) .await?; } + if first_attempt && turn_context.config.notify.is_some() { + notify_messages = legacy_notify_input_messages(&prompt.input); + } let mut draining_for_mailbox = false; let err = match try_run_sampling_request( tool_runtime.clone(), @@ -1824,14 +1829,14 @@ async fn run_sampling_request( client_session, &responses_metadata, Arc::clone(&turn_diff_tracker), - &prompt, + prompt, cancellation_token.child_token(), &mut draining_for_mailbox, ) .await { Ok(output) => { - return Ok((output, original_input.unwrap_or(prompt.input))); + return Ok((output, notify_messages)); } Err(err) => match err.details() { CodexErrorDetails::ContextWindowExceeded => { @@ -1849,8 +1854,6 @@ async fn run_sampling_request( }, }; - let original_input = original_input.get_or_insert(prompt.input); - let retry = handle_response_stream_error( &mut retry_state, max_retries, @@ -1874,7 +1877,7 @@ async fn run_sampling_request( needs_follow_up: true, last_agent_message: None, }, - std::mem::take(original_input), + notify_messages, )); } retry??; @@ -2677,7 +2680,7 @@ async fn try_run_sampling_request( client_session: &mut ModelClientSession, responses_metadata: &CodexResponsesMetadata, turn_diff_tracker: SharedTurnDiffTracker, - prompt: &Prompt, + prompt: Prompt, cancellation_token: CancellationToken, draining_for_mailbox: &mut bool, ) -> CodexResult { @@ -2759,9 +2762,9 @@ async fn try_run_sampling_request( let mut request_token_usage = None; sess.set_server_reasoning_included(step_context.settings.model_info.use_responses_lite) .await; - let mut stream = match client_session + let stream = client_session .stream( - prompt, + &prompt, &step_context.settings.model_info, &step_context.session_telemetry, effort, @@ -2772,8 +2775,9 @@ async fn try_run_sampling_request( ) .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) - .await? - { + .await?; + drop(prompt); + let mut stream = match stream { Ok(stream) => stream, Err(error) => { if let Some(finish_post_request) = finish_post_request { diff --git a/codex-rs/core/tests/suite/user_notification.rs b/codex-rs/core/tests/suite/user_notification.rs index 5a3dc6279d..b98da23087 100644 --- a/codex-rs/core/tests/suite/user_notification.rs +++ b/codex-rs/core/tests/suite/user_notification.rs @@ -2,6 +2,8 @@ use codex_core::TurnInputRequest; +use codex_protocol::models::ImageReference; +use codex_protocol::openai_models::InputModality; use codex_protocol::protocol::EventMsg; use codex_protocol::user_input::UserInput; use core_test_support::fs_wait; @@ -22,14 +24,14 @@ use responses::start_mock_server; use std::time::Duration; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn summarize_context_three_requests_and_instructions() -> anyhow::Result<()> { +async fn notify_receives_user_text_while_the_model_receives_attachments() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; let sse1 = sse(vec![ev_assistant_message("m1", "Done"), ev_completed("r1")]); - responses::mount_sse_once(&server, sse1).await; + let response = responses::mount_sse_once(&server, sse1).await; let notify_dir = TempDir::new()?; // write a script to the notify that touches a file next to it @@ -48,18 +50,36 @@ mv "${tmp_path}" "${payload_path}""#, let notify_script_str = notify_script.to_str().unwrap().to_string(); let TestCodex { codex, .. } = test_codex() + .with_model_info_override("gpt-5.5", |info| { + info.input_modalities.push(InputModality::Audio); + }) .with_config(move |cfg| cfg.notify = Some(vec![notify_script_str])) .build(&server) .await?; - // 1) Normal user input – should hit server once. + let image_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="; + let audio_url = "data:audio/wav;base64,AAAA"; codex - .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { - text: "hello world".into(), - text_elements: Vec::new(), - }])) + .start_or_steer_turn(TurnInputRequest::user_input(vec![ + UserInput::Text { + text: "hello world".into(), + text_elements: Vec::new(), + }, + UserInput::Image { + image: ImageReference::Inline { + image_url: image_url.into(), + }, + detail: None, + }, + UserInput::Audio { + audio_url: audio_url.into(), + }, + ])) .await?; wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + let request = response.single_request(); + assert_eq!(request.message_input_image_urls("user"), [image_url]); + assert_eq!(request.message_input_audio_urls("user"), [audio_url]); // We fork the notify script, so we need to wait for it to write to the file. fs_wait::wait_for_path_exists(¬ify_file, Duration::from_secs(5)).await?;