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 96be2b3865..bf9172bec2 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/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 3a95e42a12..74a7592de8 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -42,6 +42,7 @@ use codex_protocol::protocol::NonSteerableTurnKind; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::turn_input::AnnotatedResponseItem; use codex_protocol::turn_input::NotSubmittedReason; use codex_protocol::turn_input::TurnInput as SubmittedTurnInput; use codex_protocol::turn_input::TurnInputMode; @@ -242,6 +243,7 @@ pub(super) async fn handle( } SubmittedTurnInput::UserInput { .. } | SubmittedTurnInput::ResponseItem(_) + | SubmittedTurnInput::AnnotatedResponseItem(_) | SubmittedTurnInput::InterAgentCommunication(_) => TurnStartKind::Automatic, }; start_if_idle( @@ -256,7 +258,10 @@ pub(super) async fn handle( TurnInputMode::ContinueIfIdle { expected_previous_turn_id, } => { - if !matches!(&request.request.input, SubmittedTurnInput::ResponseItem(_)) { + if !matches!( + &request.request.input, + SubmittedTurnInput::ResponseItem(_) | SubmittedTurnInput::AnnotatedResponseItem(_) + ) { return Err(CodexErr::InvalidRequest( "continuation requires internal response input".to_string(), )); @@ -343,6 +348,10 @@ async fn start_or_steer( SubmittedTurnInput::ResponseItem(ResponseItem::FunctionCallOutput { call_id: None, .. + }) + | SubmittedTurnInput::AnnotatedResponseItem(AnnotatedResponseItem { + item: ResponseItem::FunctionCallOutput { call_id: None, .. }, + .. }) => true, _ => { return Err(CodexErr::InvalidRequest( @@ -907,12 +916,14 @@ async fn capture_sender_user_messages( session: &Session, input: &mut SubmittedTurnInput, ) -> Option { - let SubmittedTurnInput::ResponseItem( - item @ ResponseItem::FunctionCallOutput { call_id: None, .. }, - ) = input + let (SubmittedTurnInput::ResponseItem(item) + | SubmittedTurnInput::AnnotatedResponseItem(AnnotatedResponseItem { item, .. })) = input else { return None; }; + if !matches!(item, ResponseItem::FunctionCallOutput { call_id: None, .. }) { + return None; + } Session::assign_missing_response_item_id(item); session .services @@ -928,7 +939,11 @@ async fn pending_turn_input( turn_id: &str, origin: UserInputOrigin, ) -> TurnInput { - match input { + let retain = match &input { + SubmittedTurnInput::AnnotatedResponseItem(input) => input.annotations.retain, + _ => false, + }; + let mut input = match input { SubmittedTurnInput::UserInput { content, client_id } => TurnInput::UserInput { content, client_id, @@ -938,6 +953,7 @@ async fn pending_turn_input( }, }, SubmittedTurnInput::ResponseItem(item) + | SubmittedTurnInput::AnnotatedResponseItem(AnnotatedResponseItem { item, .. }) if matches!( &item, ResponseItem::FunctionCallOutput { call_id: None, .. } @@ -960,9 +976,20 @@ async fn pending_turn_input( }; TurnInput::FunctionCallOutput(ResponseItemEnvelope { item, metadata }) } - SubmittedTurnInput::ResponseItem(item) => TurnInput::ResponseItem(item.into()), + SubmittedTurnInput::ResponseItem(item) + | SubmittedTurnInput::AnnotatedResponseItem(AnnotatedResponseItem { item, .. }) => { + TurnInput::ResponseItem(item.into()) + } SubmittedTurnInput::InterAgentCommunication(communication) => { TurnInput::InterAgentCommunication(communication) } + }; + if retain + && let TurnInput::ResponseItem(envelope) | TurnInput::FunctionCallOutput(envelope) = + &mut input + && matches!(envelope.item, ResponseItem::FunctionCallOutput { .. }) + { + envelope.metadata.get_or_insert_default().client_authored = true; } + input } diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index af7fab2fa8..920d864531 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -29,8 +29,11 @@ use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnEnvironmentRequests; use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::turn_input::AnnotatedResponseItem; +use codex_protocol::turn_input::ResponseItemAnnotations; use codex_protocol::turn_input::TurnAttribution; use codex_protocol::user_input::UserInput; +use core_test_support::ThreadIdle; use core_test_support::responses; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; @@ -1578,3 +1581,120 @@ async fn daemon_recovery_includes_local_environment_that_finished_starting() -> .await; Ok(()) } + +/// Retention follows accepted outputs through idle submission, steering, and rollout storage. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tool_output_retention_is_submission_metadata() -> anyhow::Result<()> { + let (release, gate) = oneshot::channel(); + let (server, _completions) = start_streaming_sse_server(vec![ + vec![StreamingSseChunk { + gate: Some(gate), + body: responses::sse_completed("started"), + }], + vec![StreamingSseChunk { + gate: None, + body: responses::sse_completed("steered"), + }], + vec![StreamingSseChunk { + gate: None, + body: responses::sse_completed("idle"), + }], + ]) + .await; + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle)); + let test = test_codex() + .with_extensions(Arc::new(extensions.build())) + .build_with_streaming_server(&server) + .await?; + for (name, retain) in [("started", true), ("steered", true), ("ordinary", false)] { + let item = serde_json::from_value(serde_json::json!({ + "type": "function_call_output", "name": name, + "namespace": "example", "output": "Inspect the failing tests.", + }))?; + let input = if retain { + TurnInput::AnnotatedResponseItem(AnnotatedResponseItem { + item, + annotations: ResponseItemAnnotations { retain }, + }) + } else { + TurnInput::ResponseItem(item) + }; + let request = TurnInputRequest::new(input); + let submission = test.codex.start_or_steer_turn(request).await?; + if name == "started" { + assert!(matches!(submission, TurnInputSubmission::Started { .. })); + server.wait_for_request_count(/*count*/ 1).await; + } else { + assert!(matches!(submission, TurnInputSubmission::Steered { .. })); + } + } + release.send(()).expect("release active response"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + ThreadIdle::wait(&test.codex).await; + + let request = TurnInputRequest::new(TurnInput::AnnotatedResponseItem(AnnotatedResponseItem { + item: serde_json::from_value(serde_json::json!({ + "type": "function_call_output", "name": "idle", + "namespace": "example", "output": "Report your findings.", + }))?, + annotations: ResponseItemAnnotations { retain: true }, + })); + assert!(matches!( + test.codex.start_turn_if_idle(request).await?, + StartIfIdleSubmission::Started { .. } + )); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex.shutdown_and_wait().await?; + + let history = codex_rollout::RolloutRecorder::get_rollout_history( + &test.codex.rollout_path().expect("durable rollout"), + ) + .await?; + let retained = history + .get_rollout_items() + .iter() + .filter_map(|item| { + let codex_history::RolloutItem::ResponseItem(envelope) = item else { + return None; + }; + let codex_protocol::models::ResponseItem::FunctionCallOutput { name, .. } = + &envelope.item + else { + return None; + }; + Some(( + name.as_deref().unwrap(), + envelope + .metadata + .as_ref() + .is_some_and(|metadata| metadata.client_authored), + )) + }) + .collect::>(); + assert_eq!( + retained, + vec![ + ("started", true), + ("steered", true), + ("ordinary", false), + ("idle", true) + ] + ); + for request in server.requests().await { + let body: Value = serde_json::from_slice(&request)?; + for item in body["input"].as_array().unwrap() { + assert!(item.get("retain").is_none()); + assert!(item.get("client_authored").is_none()); + } + } + server.shutdown().await; + Ok(()) +} diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 5ce19f056a..e639c12831 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -85,7 +85,7 @@ pub struct CodexHarnessMetadata { #[serde(default, skip_serializing_if = "Option::is_none")] pub retained_source: Option, - /// Whether a developer message was supplied by an app-server client. + /// Whether a developer message was supplied by a client or a client requested tool-output retention. #[serde(default)] pub client_authored: bool, diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index 3069a5f6ff..7e9977959c 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -36,9 +36,26 @@ pub enum TurnInput { client_id: Option, }, ResponseItem(ResponseItem), + AnnotatedResponseItem(AnnotatedResponseItem), InterAgentCommunication(InterAgentCommunication), } +/// A Responses API item with caller-supplied, harness-only annotations. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct AnnotatedResponseItem { + pub item: ResponseItem, + #[serde(default)] + pub annotations: ResponseItemAnnotations, +} + +/// Public annotations accepted by Core; internal history metadata stays private. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ResponseItemAnnotations { + /// Request compaction retention for a function-call output. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub retain: bool, +} + /// One turn input and the context that follows it through submission. /// /// Callers choose start-or-steer, idle-start, or steer-only behavior through