Track client-authored developer messages in rollout history (#38243)

## What changed

- Mark developer messages received from clients with `client_authored` harness
  metadata when `retain_client_developer_messages` is enabled.
- Carry that provenance through pending input and persist it alongside rollout
  response items without adding it to provider requests or the serialized
  `TurnInput` payload.
- Preserve existing behavior for unannotated response items and when the
  feature is disabled.

## Testing

- Cover client item injection before and during turns, additional application
  context, rollout replay, image resize notices, and serialization boundaries.

GitOrigin-RevId: f385c177732becde732272e4b8440eb91f5c1039

codex-rs/core/src/session/inject.rs

@@ -7,6 +7,9 @@ use crate::state::ActiveTurn;
 use crate::state::TurnState;
 use crate::tasks::MailboxParentProvenance;
 use crate::tasks::RegularTask;
+use codex_features::Feature;
+use codex_history::CodexHarnessMetadata;
+use codex_history::ResponseItemEnvelope;
 use codex_protocol::config_types::ModeKind;
 use codex_protocol::models::ResponseItem;
 use std::sync::Arc;
@@ -27,7 +30,11 @@ impl Session {
                 self.input_queue
                     .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
                         active_turn.turn_state.as_ref(),
-                        input.into_iter().map(TurnInput::ResponseItem).collect(),
+                        input
+                            .into_iter()
+                            .map(ResponseItemEnvelope::new)
+                            .map(TurnInput::ResponseItem)
+                            .collect(),
                     )
                     .await;
                 Ok(())
@@ -36,6 +43,82 @@ impl Session {
         }
     }
 
+    /// Preserves trusted client provenance while items wait for an active turn.
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "active turn checks and turn state updates must remain atomic"
+    )]
+    pub(crate) async fn inject_client_response_items(
+        &self,
+        items: Vec<ResponseItem>,
+        turn_context: &TurnContext,
+    ) {
+        let items = items
+            .into_iter()
+            .map(|item| self.annotate_client_response_item(item))
+            .collect::<Vec<_>>();
+        let mut active = self.active_turn.lock().await;
+        if let Some(active_turn) = active.as_mut() {
+            self.input_queue
+                .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
+                    active_turn.turn_state.as_ref(),
+                    items.into_iter().map(TurnInput::ResponseItem).collect(),
+                )
+                .await;
+            return;
+        }
+        drop(active);
+        self.record_annotated_conversation_items(turn_context, items)
+            .await;
+    }
+
+    pub(crate) fn annotate_client_response_item(&self, item: ResponseItem) -> ResponseItemEnvelope {
+        let metadata = (self.enabled(Feature::RetainClientDeveloperMessages)
+            && matches!(&item, ResponseItem::Message { role, .. } if role == "developer"))
+        .then_some(CodexHarnessMetadata {
+            client_authored: true,
+        });
+
+        ResponseItemEnvelope { item, metadata }
+    }
+
+    pub(crate) async fn record_annotated_conversation_items(
+        &self,
+        turn_context: &TurnContext,
+        items: Vec<ResponseItemEnvelope>,
+    ) {
+        if !self.enabled(Feature::RetainClientDeveloperMessages)
+            || items.iter().all(|item| item.metadata.is_none())
+        {
+            let items = items
+                .into_iter()
+                .map(ResponseItemEnvelope::into_item)
+                .collect::<Vec<_>>();
+            self.record_conversation_items(turn_context, &items).await;
+            return;
+        }
+
+        let mut annotated_items = Vec::with_capacity(items.len());
+        let mut image_preparations = Vec::new();
+        for envelope in items {
+            let (prepared_items, prepared_images) = self.prepare_conversation_items_for_history(
+                turn_context,
+                std::slice::from_ref(&envelope.item),
+            );
+            image_preparations.extend(prepared_images);
+
+            let mut metadata = envelope.metadata;
+            annotated_items.extend(prepared_items.into_owned().into_iter().map(|item| {
+                ResponseItemEnvelope {
+                    item,
+                    metadata: metadata.take(),
+                }
+            }));
+        }
+        self.record_prepared_conversation_items(turn_context, annotated_items, image_preparations)
+            .await;
+    }
+
     /// Starts a regular turn with the provided input only if automatic idle work
     /// is allowed for the current session state.
     ///