openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support standalone function outputs in turn routing (#40991)

Browse Source 
## 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
This commit is contained in:
rka-oai
2026-08-26 22:48:37 +00:00
committed by copyberry
parent 6e008417bb
commit b9c4b9a0cf
9 changed files with 178 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex_thread.rs
+2 -2
View File
@@ -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,
codex-rs/core/src/hook_runtime.rs
+5 -1
View File
@@ -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;
codex-rs/core/src/session/input_queue.rs
+20 -5
View File
@@ -24,6 +24,7 @@ pub enum TurnInput {
content: Vec<UserInput>,
client_id: Option<String>,
},
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,
codex-rs/core/src/session/turn.rs
+6 -2
View File
@@ -660,7 +660,9 @@ fn turn_user_input(input: &[TurnInput]) -> Vec<UserInput> {
.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| {
codex-rs/core/src/session/turn_input.rs
+51 -44
View File
@@ -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<TurnInputSubmission> {
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<TurnInputSubmission> {
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<UserInput>,
input: &mut SubmittedTurnInput,
additional_context: BTreeMap<String, AdditionalContextEntry>,
expected_turn_id: Option<&str>,
required_final_output_json_schema: Option<&Value>,
client_user_message_id: Option<String>,
responsesapi_client_metadata: Option<HashMap<String, String>>,
incoming_root_turn_id: Option<Option<String>>,
) -> Result<String, NotSubmittedReason> {
@@ -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)
codex-rs/core/src/session/turn_input_tests.rs
+36
View File
@@ -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]
codex-rs/core/src/tasks/review.rs
+3 -1
View File
@@ -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(_) => {}
}
}
codex-rs/core/tests/common/responses.rs
+17 -6
View File
@@ -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()
}
codex-rs/core/tests/suite/pending_input.rs
+38
View File
@@ -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<()> {