Add retention annotations to submitted tool outputs (#52671)

## What changed

Add `TurnInput::AnnotatedResponseItem` with a caller-supplied `retain` annotation. For function-call outputs, `retain: true` sets `client_authored` history metadata while keeping annotations out of model requests. Accept annotated items through the existing turn submission paths, including continuation.

## Testing

Add an integration test covering retention metadata during turn start, steering, idle submission, and rollout storage. Verify that ordinary outputs remain unmarked and model requests contain neither `retain` nor `client_authored`.

GitOrigin-RevId: 7285dde31149b17b0dcf31d4ea1c7cba66d9efe6

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<GuardianSenderMessages> {
-    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
 }

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::<Vec<_>>();
+    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(())
+}

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<RetainedSource>,
 
-    /// 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,
 

codex-rs/protocol/src/turn_input.rs

@@ -36,9 +36,26 @@ pub enum TurnInput {
         client_id: Option<String>,
     },
     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