From b9c4b9a0cfe8544c823a5dbf6ea61fc3974500ba Mon Sep 17 00:00:00 2001 From: rka-oai Date: Wed, 26 Aug 2026 22:42:35 +0000 Subject: [PATCH] Support standalone function outputs in turn routing (#40991) ## What changed - Allow `start_or_steer_turn` to accept named `function_call_output` items without a `call_id`, starting a turn when idle or steering the active turn. - Preserve standalone outputs as pending input, assign response item IDs, and record them in conversation history without treating them as user prompts. ## Testing - Cover starting a turn with a standalone output and steering an active turn with one. - Verify passive response items do not signal pending steering activity. GitOrigin-RevId: e32b292106b3c5a5f03577b8ebf67d2d7dee4e5d --- codex-rs/core/src/codex_thread.rs | 4 +- codex-rs/core/src/hook_runtime.rs | 6 +- codex-rs/core/src/session/input_queue.rs | 25 ++++- codex-rs/core/src/session/turn.rs | 8 +- codex-rs/core/src/session/turn_input.rs | 95 ++++++++++--------- codex-rs/core/src/session/turn_input_tests.rs | 36 +++++++ codex-rs/core/src/tasks/review.rs | 4 +- codex-rs/core/tests/common/responses.rs | 23 +++-- codex-rs/core/tests/suite/pending_input.rs | 38 ++++++++ 9 files changed, 178 insertions(+), 61 deletions(-) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 41e1f9b69b..44122d6a20 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -341,8 +341,8 @@ impl CodexThread { /// Submits turn input without requiring the caller to inspect thread state. /// /// The result describes whether Core started a turn, steered an active - /// turn, or declined it without recording or enqueueing the input. Only - /// user input is accepted. + /// turn, or declined it without recording or enqueueing the input. + /// User input and named standalone function-call outputs are accepted. pub async fn start_or_steer_turn( &self, request: TurnInputRequest, diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 8cbd1517cc..b0f0f9cdc2 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -624,7 +624,7 @@ pub(crate) async fn inspect_pending_input( ) .await } - TurnInput::ResponseItem(_) => HookRuntimeOutcome { + TurnInput::ResponseItem(_) | TurnInput::FunctionCallOutput(_) => HookRuntimeOutcome { should_stop: false, additional_contexts: Vec::new(), }, @@ -656,6 +656,10 @@ pub(crate) async fn record_pending_input( sess.record_annotated_conversation_items(turn_context, vec![item]) .await; } + TurnInput::FunctionCallOutput(item) => { + sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) + .await; + } TurnInput::InterAgentCommunication(communication) => { sess.record_inter_agent_communication(turn_context, communication) .await; diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 4fa79b9e2a..e7bd6642bd 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -24,6 +24,7 @@ pub enum TurnInput { content: Vec, client_id: Option, }, + FunctionCallOutput(ResponseItem), // Preserve the existing serialized format while carrying injection API metadata // through the in-memory queue. ResponseItem(#[serde(with = "turn_input_response_item")] ResponseItemEnvelope), @@ -104,7 +105,7 @@ impl InputQueue { ) { let activity_rx = self.activity_tx.subscribe(); let has_pending_steer = if let Some(turn_state) = turn_state { - turn_state.lock().await.pending_input.has_user_input() + turn_state.lock().await.pending_input.has_pending_input() } else { false }; @@ -369,10 +370,13 @@ impl InputQueue { } impl TurnInputQueue { - fn has_user_input(&self) -> bool { - self.items - .iter() - .any(|input| matches!(input, TurnInput::UserInput { .. })) + fn has_pending_input(&self) -> bool { + self.items.iter().any(|input| { + matches!( + input, + TurnInput::UserInput { .. } | TurnInput::FunctionCallOutput(_) + ) + }) } } @@ -492,6 +496,17 @@ mod tests { async fn input_queue_reports_already_pending_steer() { let input_queue = InputQueue::new(); let turn_state = Mutex::new(TurnState::default()); + let passive_output = serde_json::from_value(serde_json::json!({ + "ResponseItem": {"type": "function_call_output", "name": "notify", "output": "passive"} + })) + .unwrap(); + input_queue + .extend_pending_input_for_turn_state(&turn_state, vec![passive_output]) + .await; + assert_eq!( + input_queue.subscribe_activity(Some(&turn_state)).await.1, + None + ); input_queue .extend_pending_input_and_accept_mailbox_delivery_for_turn_state( &turn_state, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 492bf27721..85932c0d65 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -660,7 +660,9 @@ fn turn_user_input(input: &[TurnInput]) -> Vec { .iter() .filter_map(|item| match item { TurnInput::UserInput { content, .. } => Some(content.as_slice()), - TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => None, + TurnInput::ResponseItem(_) + | TurnInput::FunctionCallOutput(_) + | TurnInput::InterAgentCommunication(_) => None, }) .flatten() .cloned() @@ -990,7 +992,9 @@ async fn track_turn_resolved_config_analytics( .iter() .filter_map(|item| match item { TurnInput::UserInput { content, .. } => Some(content.as_slice()), - TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => None, + TurnInput::ResponseItem(_) + | TurnInput::FunctionCallOutput(_) + | TurnInput::InterAgentCommunication(_) => None, }) .flatten() .filter(|item| { diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 8c6eeb0ace..5b25b67460 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -21,6 +21,7 @@ use crate::tasks::RegularTask; use codex_protocol::config_types::ModeKind; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; +use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AdditionalContextEntry; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; @@ -234,21 +235,25 @@ async fn start_or_steer( submission_id: String, ) -> CodexResult { let TurnInputRequest { - input, + mut input, thread_settings, start, additional_context, responsesapi_client_metadata, .. } = request; - let SubmittedTurnInput::UserInput { - content: mut items, - client_id, - } = input - else { - return Err(CodexErr::InvalidRequest( - "only user input can steer a turn".to_string(), - )); + let has_explicit_input = match &input { + SubmittedTurnInput::UserInput { content, .. } => !content.is_empty(), + SubmittedTurnInput::ResponseItem(ResponseItem::FunctionCallOutput { + call_id: None, + .. + }) => true, + _ => { + return Err(CodexErr::InvalidRequest( + "only user input or standalone function-call outputs can start or steer a turn" + .to_string(), + )); + } }; let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none(); let incoming_root_turn_id = start @@ -258,11 +263,10 @@ async fn start_or_steer( let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?; match session .steer_input( - &mut items, + &mut input, additional_context.clone(), /*expected_turn_id*/ None, settings.required_active_final_output_json_schema(), - client_id.clone(), responsesapi_client_metadata.clone(), incoming_root_turn_id, ) @@ -280,7 +284,7 @@ async fn start_or_steer( unreachable!("explicit user input can enter Plan mode"); }; if can_start_root_turn - && !items.is_empty() + && has_explicit_input && turn_context .turn_metadata_state .can_start_root_turn(&turn_context.session_source) @@ -297,13 +301,12 @@ async fn start_or_steer( session .maybe_emit_model_warnings_for_turn(turn_context.as_ref()) .await; - turn_context.session_telemetry.user_prompt(&items); + if let SubmittedTurnInput::UserInput { content, .. } = &input { + turn_context.session_telemetry.user_prompt(content); + } let mut task_input = merge_additional_context_input(session, additional_context).await; - if !items.is_empty() { - task_input.push(TurnInput::UserInput { - content: items, - client_id, - }); + if has_explicit_input { + task_input.push(pending_turn_input(input)); } session .spawn_task(turn_context, task_input, RegularTask::new()) @@ -447,22 +450,18 @@ async fn steer( submission_id: String, ) -> CodexResult { let TurnInputRequest { - input, + mut input, thread_settings, start, additional_context, responsesapi_client_metadata, .. } = request; - let SubmittedTurnInput::UserInput { - content: mut items, - client_id, - } = input - else { + if !matches!(&input, SubmittedTurnInput::UserInput { .. }) { return Err(CodexErr::InvalidRequest( "only user input can steer a turn".to_string(), )); - }; + } let incoming_root_turn_id = start .parent_turn_id .as_ref() @@ -470,11 +469,10 @@ async fn steer( let settings = PreparedTurnInputSettings::prepare(session, thread_settings, start).await?; match session .steer_input( - &mut items, + &mut input, additional_context, Some(expected_turn_id.as_str()), settings.required_active_final_output_json_schema(), - client_id, responsesapi_client_metadata, incoming_root_turn_id, ) @@ -538,24 +536,19 @@ impl Session { } } - /// Inject additional user input into the currently active turn. + /// Inject additional user input or a standalone tool output into the active turn. /// /// Returns the active turn id when accepted. #[expect( clippy::await_holding_invalid_type, reason = "active turn checks and turn state updates must remain atomic" )] - #[expect( - clippy::too_many_arguments, - reason = "steering carries the accepted input plus its turn-scoped metadata" - )] async fn steer_input( &self, - input: &mut Vec, + input: &mut SubmittedTurnInput, additional_context: BTreeMap, expected_turn_id: Option<&str>, required_final_output_json_schema: Option<&Value>, - client_user_message_id: Option, responsesapi_client_metadata: Option>, incoming_root_turn_id: Option>, ) -> Result { @@ -592,7 +585,7 @@ impl Session { } } - if input.is_empty() { + if matches!(input, SubmittedTurnInput::UserInput { content, .. } if content.is_empty()) { return Err(NotSubmittedReason::EmptyInput); } // Compare JSON values directly instead of serialized schema text. @@ -603,11 +596,6 @@ impl Session { { return Err(NotSubmittedReason::ActiveTurnOutputSchemaMismatch); } - active_task - .turn_context - .session_telemetry - .user_prompt(input); - let mut pending_input = merge_additional_context_input(self, additional_context).await; if let Some(responsesapi_client_metadata) = responsesapi_client_metadata { @@ -617,10 +605,20 @@ impl Session { .set_responsesapi_client_metadata(responsesapi_client_metadata); } - pending_input.push(TurnInput::UserInput { - content: std::mem::take(input), - client_id: client_user_message_id, - }); + let input = match input { + SubmittedTurnInput::UserInput { content, client_id } => { + active_task + .turn_context + .session_telemetry + .user_prompt(content); + TurnInput::UserInput { + content: std::mem::take(content), + client_id: client_id.clone(), + } + } + input => pending_turn_input(input.clone()), + }; + pending_input.push(input); if let Some(incoming_root_turn_id) = incoming_root_turn_id && active_task.turn_context.turn_metadata_state.root_turn_id() != incoming_root_turn_id { @@ -659,6 +657,15 @@ fn pending_turn_input(input: SubmittedTurnInput) -> TurnInput { SubmittedTurnInput::UserInput { content, client_id } => { TurnInput::UserInput { content, client_id } } + SubmittedTurnInput::ResponseItem(mut item) + if matches!( + &item, + ResponseItem::FunctionCallOutput { call_id: None, .. } + ) => + { + Session::assign_missing_response_item_id(&mut item); + TurnInput::FunctionCallOutput(item) + } SubmittedTurnInput::ResponseItem(item) => TurnInput::ResponseItem(item.into()), SubmittedTurnInput::InterAgentCommunication(communication) => { TurnInput::InterAgentCommunication(communication) diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index 75b305ac24..3ef6797c06 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -833,6 +833,42 @@ async fn steer_only_enforces_expected_turn_id() { }, submission ); + + let output: ResponseItem = serde_json::from_value(serde_json::json!({ + "type": "function_call_output", + "name": "send_message_to_thread", + "output": "delegated work", + })) + .expect("valid standalone output"); + + let submission = handle( + &session, + TurnInputRequest::new(SubmittedTurnInput::ResponseItem(output)), + TurnInputMode::StartOrSteer, + "test-submission".to_string(), + ) + .await + .expect("standalone output should steer the active turn"); + + assert_eq!( + submission, + TurnInputSubmission::Steered { + turn_id: turn_context.sub_id.clone() + } + ); + let turn_state = session + .input_queue + .turn_state_for_sub_id(&session.active_turn, &turn_context.sub_id) + .await + .expect("active turn state"); + assert_eq!( + session + .input_queue + .subscribe_activity(Some(turn_state.as_ref())) + .await + .1, + Some(crate::session::input_queue::InputQueueActivity::Steer) + ); } #[tokio::test] diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 2369495996..42ec842e70 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -67,7 +67,9 @@ impl SessionTask for ReviewTask { for item in input { match item { TurnInput::UserInput { mut content, .. } => user_input.append(&mut content), - TurnInput::ResponseItem(_) | TurnInput::InterAgentCommunication(_) => {} + TurnInput::ResponseItem(_) + | TurnInput::FunctionCallOutput(_) + | TurnInput::InterAgentCommunication(_) => {} } } diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 33d4284ecc..43e1413dba 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -1662,10 +1662,11 @@ pub async fn mount_compact_response_sequence( /// Validate invariants on the request body sent to `/v1/responses`. /// -/// - No `function_call_output`/`custom_tool_call_output` with missing/empty `call_id`. +/// - A `function_call_output` with missing/empty `call_id` must have a nonempty `name`. +/// - No `custom_tool_call_output` with missing/empty `call_id`. /// - `tool_search_output` must have a `call_id` unless it is a server-executed legacy item. -/// - Every `function_call_output` must match a prior `function_call` or -/// `local_shell_call` with the same `call_id` in the same `input`. +/// - Every `function_call_output` with a `call_id` must match a prior `function_call` +/// or `local_shell_call` with the same `call_id` in the same `input`. /// - Every `custom_tool_call_output` must match a prior `custom_tool_call`. /// - Every `tool_search_output` must match a prior `tool_search_call`. /// - Additionally, enforce symmetry: every `function_call`/`custom_tool_call`/ @@ -1711,9 +1712,19 @@ fn validate_request_body_invariants(request: &wiremock::Request) { items .iter() .filter(|item| item.get("type").and_then(Value::as_str) == Some(kind)) - .map(|item| { - let id = get_call_id(item).expect(missing_msg); - id.to_string() + .filter_map(|item| { + if let Some(id) = get_call_id(item) { + return Some(id.to_string()); + } + if kind == "function_call_output" + && item + .get("name") + .and_then(Value::as_str) + .is_some_and(|name| !name.is_empty()) + { + return None; + } + panic!("{missing_msg}"); }) .collect() } diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index ef4b6b98c9..aa9b2f7918 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -118,6 +118,44 @@ async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyho Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn standalone_tool_output_starts_instruction_turn() -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let response = responses::mount_sse_once( + &server, + responses::sse(vec![ev_response_created("turn"), ev_completed("turn")]), + ) + .await; + let test = test_codex().build_with_auto_env(&server).await?; + + let expected_output = json!({ + "type": "function_call_output", + "name": "send_message_to_thread", + "namespace": "codex_app", + "output": "delegated work", + }); + let output = serde_json::from_value(expected_output.clone())?; + + let submission = test + .codex + .start_or_steer_turn(TurnInputRequest::new(TurnInput::ResponseItem(output))) + .await?; + let TurnInputSubmission::Started { turn_id } = submission else { + panic!("standalone output should start a turn"); + }; + wait_for_turn_complete(test.codex.as_ref()).await; + + let request = response.single_request(); + responses::assert_root_turn(&request.body_json(), Some(&turn_id))?; + let output = &request.inputs_of_type("function_call_output")[0]; + assert_eq!(output["name"], expected_output["name"]); + assert_eq!(output["namespace"], expected_output["namespace"]); + assert_eq!(output["output"], expected_output["output"]); + assert!(output.get("call_id").is_none()); + + Ok(()) +} + async fn assert_idle_user_input_reaches_the_first_model_request( mode: ModeKind, ) -> anyhow::Result<()> {