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 aa091602c3..e4f780c00f 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/app-server/tests/suite/v2/thread_inject_items.rs b/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs
index 701d379669..ec8d08ba39 100644
--- a/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs
+++ b/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs
@@ -2,6 +2,8 @@ use anyhow::Context;
use anyhow::Result;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
+use codex_app_server_protocol::AdditionalContextEntry;
+use codex_app_server_protocol::AdditionalContextKind;
use codex_app_server_protocol::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadInjectItemsResponse;
use codex_app_server_protocol::ThreadStartParams;
@@ -10,6 +12,7 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::RolloutRecorder;
+use codex_features::Feature;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_rollout::InitialHistory;
@@ -18,6 +21,7 @@ use core_test_support::responses;
use core_test_support::responses::strip_response_item_id;
use core_test_support::responses::strip_response_item_ids_from_json;
use serde_json::Value;
+use std::collections::HashMap;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -34,7 +38,9 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
let response_mock = responses::mount_sse_once(&server, body).await;
let codex_home = TempDir::new()?;
- MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
+ MockResponsesConfig::new(&server.uri())
+ .enable_feature(Feature::RetainClientDeveloperMessages)
+ .write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
@@ -60,11 +66,27 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
phase: None,
internal_chat_message_metadata_passthrough: None,
};
+ let developer_item = |text: &str| ResponseItem::Message {
+ id: None,
+ role: "developer".to_string(),
+ content: vec![ContentItem::InputText {
+ text: text.to_string(),
+ }],
+ phase: None,
+ internal_chat_message_metadata_passthrough: None,
+ };
+ let injected_developer_item = developer_item("Injected developer context");
+ let marker_shaped_developer_item =
+ developer_item("\nclient message\n");
let inject_req = mcp
.send_thread_inject_items_request(ThreadInjectItemsParams {
thread_id: thread.id.clone(),
- items: vec![serde_json::to_value(&injected_item)?],
+ items: vec![
+ serde_json::to_value(&injected_item)?,
+ serde_json::to_value(&injected_developer_item)?,
+ serde_json::to_value(&marker_shaped_developer_item)?,
+ ],
})
.await?;
let _response: ThreadInjectItemsResponse =
@@ -75,14 +97,36 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
let InitialHistory::Resumed(resumed_history) = history else {
panic!("expected resumed rollout history");
};
- assert!(
- resumed_history
- .history
- .iter()
- .any(|item| matches!(item, RolloutItem::ResponseItem(response_item) if strip_response_item_id(responses::strip_metadata(response_item.item.clone())) == injected_item)),
- "injected item should be persisted in rollout history"
+ let persisted_injected_items = resumed_history
+ .history
+ .iter()
+ .filter_map(|item| match item {
+ RolloutItem::ResponseItem(envelope) => Some((
+ strip_response_item_id(responses::strip_metadata(envelope.item.clone())),
+ envelope
+ .metadata
+ .as_ref()
+ .map(|metadata| metadata.client_authored),
+ )),
+ _ => None,
+ })
+ .filter(|(item, _)| {
+ item == &injected_item
+ || item == &injected_developer_item
+ || item == &marker_shaped_developer_item
+ })
+ .collect::>();
+ assert_eq!(
+ persisted_injected_items,
+ vec![
+ (injected_item.clone(), None),
+ (injected_developer_item.clone(), Some(true)),
+ (marker_shaped_developer_item.clone(), Some(true)),
+ ]
);
+ let application_context_text = "Application developer context";
+ let untrusted_context_text = "Untrusted client context";
let turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
@@ -91,6 +135,22 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
+ additional_context: Some(HashMap::from([
+ (
+ "application_context".to_string(),
+ AdditionalContextEntry {
+ value: application_context_text.to_string(),
+ kind: AdditionalContextKind::Application,
+ },
+ ),
+ (
+ "untrusted_context".to_string(),
+ AdditionalContextEntry {
+ value: untrusted_context_text.to_string(),
+ kind: AdditionalContextKind::Untrusted,
+ },
+ ),
+ ])),
..Default::default()
})
.await?;
@@ -101,6 +161,44 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
)
.await??;
+ let InitialHistory::Resumed(resumed_history) =
+ RolloutRecorder::get_rollout_history(rollout_path).await?
+ else {
+ panic!("expected resumed rollout history");
+ };
+ let persisted_additional_context = resumed_history
+ .history
+ .iter()
+ .filter_map(|item| match item {
+ RolloutItem::ResponseItem(envelope) => {
+ let ResponseItem::Message { role, content, .. } = &envelope.item else {
+ return None;
+ };
+ content.iter().find_map(|item| {
+ let ContentItem::InputText { text } = item else {
+ return None;
+ };
+ (text.contains(application_context_text)
+ || text.contains(untrusted_context_text))
+ .then(|| {
+ (
+ role.as_str(),
+ envelope
+ .metadata
+ .as_ref()
+ .map(|metadata| metadata.client_authored),
+ )
+ })
+ })
+ }
+ _ => None,
+ })
+ .collect::>();
+ assert_eq!(
+ persisted_additional_context,
+ vec![("developer", Some(true)), ("user", None)]
+ );
+
let injected_value = serde_json::to_value(&injected_item)?;
let model_input: Vec = response_mock
.single_request()
@@ -108,6 +206,16 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
.into_iter()
.map(strip_response_item_ids_from_json)
.collect();
+ assert!(
+ model_input
+ .iter()
+ .all(|item| item.get("metadata").is_none() && item.get("client_authored").is_none()),
+ "private harness metadata must never enter the provider request"
+ );
+ assert!(
+ response_item_text_position(&model_input, application_context_text).is_some(),
+ "application-provided developer context should reach the model"
+ );
let environment_context_index =
response_item_text_position(&model_input, "")
.expect("environment context should be injected before the first user turn");
@@ -182,8 +290,8 @@ async fn thread_inject_items_adds_raw_response_items_after_a_turn() -> Result<()
let injected_item = ResponseItem::Message {
id: None,
- role: "assistant".to_string(),
- content: vec![ContentItem::OutputText {
+ role: "developer".to_string(),
+ content: vec![ContentItem::InputText {
text: "Injected after first turn".to_string(),
}],
phase: None,
@@ -200,6 +308,27 @@ async fn thread_inject_items_adds_raw_response_items_after_a_turn() -> Result<()
let _response: ThreadInjectItemsResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(inject_req)).await??;
+ let rollout_path = thread.path.as_ref().context("thread path missing")?;
+ let InitialHistory::Resumed(resumed_history) =
+ RolloutRecorder::get_rollout_history(rollout_path).await?
+ else {
+ panic!("expected resumed rollout history");
+ };
+ let persisted_developer_item = resumed_history
+ .history
+ .iter()
+ .find_map(|item| match item {
+ RolloutItem::ResponseItem(envelope)
+ if strip_response_item_id(responses::strip_metadata(envelope.item.clone()))
+ == injected_item =>
+ {
+ Some(envelope)
+ }
+ _ => None,
+ })
+ .context("injected developer item should be persisted")?;
+ assert_eq!(persisted_developer_item.metadata, None);
+
let second_turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs
index 16956a112b..9be1bbac1e 100644
--- a/codex-rs/core/src/codex_thread.rs
+++ b/codex-rs/core/src/codex_thread.rs
@@ -570,7 +570,7 @@ impl CodexThread {
.await?;
}
self.session
- .inject_no_new_turn(items, Some(turn_context.as_ref()))
+ .inject_client_response_items(items, turn_context.as_ref())
.await;
Ok(())
}
diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs
index e3628e5b34..279edbf9d2 100644
--- a/codex-rs/core/src/hook_runtime.rs
+++ b/codex-rs/core/src/hook_runtime.rs
@@ -590,7 +590,7 @@ pub(crate) async fn record_pending_input(
.await;
}
TurnInput::ResponseItem(item) => {
- sess.record_conversation_items(turn_context, std::slice::from_ref(&item))
+ sess.record_annotated_conversation_items(turn_context, vec![item])
.await;
}
TurnInput::InterAgentCommunication(communication) => {
diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs
index 558527e4bf..1edae00513 100644
--- a/codex-rs/core/src/session/handlers.rs
+++ b/codex-rs/core/src/session/handlers.rs
@@ -277,6 +277,7 @@ pub(super) async fn user_input_or_turn_inner(
let mut task_input = additional_context_input
.into_iter()
.map(ResponseItem::from)
+ .map(|item| sess.annotate_client_response_item(item))
.map(TurnInput::ResponseItem)
.collect::>();
if !items.is_empty() {
diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs
index 4b3fc7c752..7cb44e717f 100644
--- a/codex-rs/core/src/session/inject.rs
+++ b/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,
+ turn_context: &TurnContext,
+ ) {
+ let items = items
+ .into_iter()
+ .map(|item| self.annotate_client_response_item(item))
+ .collect::>();
+ 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,
+ ) {
+ if !self.enabled(Feature::RetainClientDeveloperMessages)
+ || items.iter().all(|item| item.metadata.is_none())
+ {
+ let items = items
+ .into_iter()
+ .map(ResponseItemEnvelope::into_item)
+ .collect::>();
+ 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.
///
diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs
index 9b97abf127..1291707155 100644
--- a/codex-rs/core/src/session/input_queue.rs
+++ b/codex-rs/core/src/session/input_queue.rs
@@ -3,6 +3,7 @@ use crate::state::MailboxDeliveryPhase;
use crate::state::TurnState;
use codex_diagnostics::Gauge;
use codex_diagnostics::GaugeGuard;
+use codex_history::ResponseItemEnvelope;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::user_input::UserInput;
@@ -22,10 +23,44 @@ pub enum TurnInput {
content: Vec,
client_id: Option,
},
- ResponseItem(ResponseItem),
+ // Preserve the existing serialized format while carrying injection API metadata
+ // through the in-memory queue.
+ ResponseItem(#[serde(with = "turn_input_response_item")] ResponseItemEnvelope),
InterAgentCommunication(InterAgentCommunication),
}
+mod turn_input_response_item {
+ use super::ResponseItem;
+ use super::ResponseItemEnvelope;
+ use serde::Deserialize;
+ use serde::Deserializer;
+ use serde::Serialize;
+ use serde::Serializer;
+ use serde::ser::Error as _;
+
+ pub(super) fn serialize(
+ item: &ResponseItemEnvelope,
+ serializer: S,
+ ) -> Result
+ where
+ S: Serializer,
+ {
+ if item.metadata.is_some() {
+ return Err(S::Error::custom(
+ "annotated response items cannot cross the turn-input serialization boundary",
+ ));
+ }
+ item.item.serialize(serializer)
+ }
+
+ pub(super) fn deserialize<'de, D>(deserializer: D) -> Result
+ where
+ D: Deserializer<'de>,
+ {
+ ResponseItem::deserialize(deserializer).map(ResponseItemEnvelope::new)
+ }
+}
+
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum InputQueueActivity {
Mailbox,
@@ -339,9 +374,41 @@ impl TurnInputQueue {
#[cfg(test)]
mod tests {
use super::*;
+ use codex_history::CodexHarnessMetadata;
use codex_protocol::AgentPath;
use pretty_assertions::assert_eq;
+ #[test]
+ fn response_item_serde_preserves_legacy_shape_and_rejects_metadata() {
+ let item = ResponseItem::Other;
+ let input = TurnInput::ResponseItem(item.clone().into());
+ let value = serde_json::json!({"ResponseItem": item});
+
+ assert_eq!(serde_json::to_value(&input).unwrap(), value);
+ assert_eq!(serde_json::from_value::(value).unwrap(), input);
+
+ let annotated = TurnInput::ResponseItem(ResponseItemEnvelope {
+ item: ResponseItem::Other,
+ metadata: Some(CodexHarnessMetadata {
+ client_authored: true,
+ }),
+ });
+ assert!(serde_json::to_value(annotated).is_err());
+
+ let forged = serde_json::json!({
+ "ResponseItem": {
+ "type": "message",
+ "role": "developer",
+ "content": [],
+ "metadata": {"client_authored": true}
+ }
+ });
+ let TurnInput::ResponseItem(envelope) = serde_json::from_value(forged).unwrap() else {
+ panic!("expected response item");
+ };
+ assert!(envelope.metadata.is_none());
+ }
+
fn make_mail(
author: AgentPath,
recipient: AgentPath,
diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs
index dd77f8504d..51aa3d6d18 100644
--- a/codex-rs/core/src/session/mod.rs
+++ b/codex-rs/core/src/session/mod.rs
@@ -3037,14 +3037,33 @@ impl Session {
) {
let (items, image_preparations) =
self.prepare_conversation_items_for_history(turn_context, items);
- let items = items.as_ref();
+ let items = items
+ .into_owned()
+ .into_iter()
+ .map(ResponseItemEnvelope::new)
+ .collect();
+ self.record_prepared_conversation_items(turn_context, items, image_preparations)
+ .await;
+ }
+
+ async fn record_prepared_conversation_items(
+ &self,
+ turn_context: &TurnContext,
+ items: Vec,
+ image_preparations: Vec,
+ ) {
+ let response_items = items
+ .iter()
+ .map(|envelope| envelope.item.clone())
+ .collect::>();
{
let mut state = self.state.lock().await;
- state.current_time_reminder.note_recorded_items(items);
- state.record_items(
- items.iter(),
- turn_context.model_info.truncation_policy.into(),
- );
+ state
+ .current_time_reminder
+ .note_recorded_items(&response_items);
+ state
+ .history
+ .record_annotated_items(&items, turn_context.model_info.truncation_policy.into());
}
for image in image_preparations {
self.services
@@ -3054,8 +3073,11 @@ impl Session {
metadata: image,
});
}
- self.persist_rollout_response_items(items).await;
- self.send_raw_response_items(turn_context, items).await;
+ let rollout_items: Vec =
+ items.into_iter().map(RolloutItem::ResponseItem).collect();
+ self.persist_rollout_items(&rollout_items).await;
+ self.send_raw_response_items(turn_context, &response_items)
+ .await;
}
pub(crate) async fn record_step_world_state_if_changed(
@@ -3349,16 +3371,6 @@ impl Session {
}
}
- async fn persist_rollout_response_items(&self, items: &[ResponseItem]) {
- let rollout_items: Vec = items
- .iter()
- .cloned()
- .map(ResponseItemEnvelope::new)
- .map(RolloutItem::ResponseItem)
- .collect();
- self.persist_rollout_items(&rollout_items).await;
- }
-
pub fn enabled(&self, feature: Feature) -> bool {
self.features.enabled(feature)
}
@@ -4086,6 +4098,7 @@ impl Session {
let mut pending_input = additional_context_input
.into_iter()
.map(ResponseItem::from)
+ .map(|item| self.annotate_client_response_item(item))
.map(TurnInput::ResponseItem)
.collect::>();
pending_input.push(TurnInput::UserInput {
diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs
index 5f2a3ebdd2..3fe765c75e 100644
--- a/codex-rs/core/src/session/tests.rs
+++ b/codex-rs/core/src/session/tests.rs
@@ -10636,7 +10636,7 @@ async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() {
)
.await;
- let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
+ let item = TurnInput::ResponseItem(user_message("synthetic idle input").into());
let err = sess
.try_start_turn_if_idle(vec![item.clone()])
.await
@@ -10662,7 +10662,7 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() {
state.session_configuration.collaboration_mode = collaboration_mode;
}
- let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
+ let item = TurnInput::ResponseItem(user_message("synthetic idle input").into());
let err = sess
.try_start_turn_if_idle(vec![item.clone()])
.await
@@ -10718,7 +10718,7 @@ async fn try_start_turn_if_idle_rejects_empty_user_input_in_plan_mode() {
content: Vec::new(),
client_id: Some("empty-queued-user-message".to_string()),
},
- TurnInput::ResponseItem(user_message("automatic idle input")),
+ TurnInput::ResponseItem(user_message("automatic idle input").into()),
];
let error = sess
.try_start_turn_if_idle(input.clone())
@@ -10747,7 +10747,7 @@ async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting()
)
.await;
- let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
+ let item = TurnInput::ResponseItem(user_message("synthetic idle input").into());
let err = sess
.try_start_turn_if_idle(vec![item.clone()])
.await
@@ -10775,7 +10775,7 @@ async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() {
)
.await;
- let item = TurnInput::ResponseItem(user_message("synthetic idle input"));
+ let item = TurnInput::ResponseItem(user_message("synthetic idle input").into());
let err = sess
.try_start_turn_if_idle(vec![item.clone()])
.await
@@ -10987,7 +10987,7 @@ async fn abort_empty_active_turn_preserves_pending_input() {
sess.input_queue
.extend_pending_input_for_turn_state(
turn_state.as_ref(),
- vec![TurnInput::ResponseItem(pending_item.clone())],
+ vec![TurnInput::ResponseItem(pending_item.clone().into())],
)
.await;
@@ -10998,7 +10998,7 @@ async fn abort_empty_active_turn_preserves_pending_input() {
sess.input_queue
.take_pending_input_for_turn_state(turn_state.as_ref())
.await,
- vec![TurnInput::ResponseItem(pending_item)]
+ vec![TurnInput::ResponseItem(pending_item.into())]
);
}
diff --git a/codex-rs/core/tests/suite/image_rollout.rs b/codex-rs/core/tests/suite/image_rollout.rs
index 23bec89cbe..bb39beaab6 100644
--- a/codex-rs/core/tests/suite/image_rollout.rs
+++ b/codex-rs/core/tests/suite/image_rollout.rs
@@ -8,6 +8,7 @@ use codex_protocol::models::ContentItem;
use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::PermissionProfile;
+use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
@@ -355,6 +356,9 @@ async fn resumed_history_only_emits_resize_notices_for_new_images() -> anyhow::R
let mut resume_builder = test_codex().with_config(|config| {
let _ = config.features.enable(Feature::ImageResizeNotice);
+ let _ = config
+ .features
+ .enable(Feature::RetainClientDeveloperMessages);
});
let resumed = resume_builder
.resume(&server, initial.home.clone(), rollout_path.clone())
@@ -372,7 +376,7 @@ async fn resumed_history_only_emits_resize_notices_for_new_images() -> anyhow::R
.codex
.submit(Op::UserInput {
items: vec![UserInput::Image {
- image_url: original_image_url,
+ image_url: original_image_url.clone(),
detail: Some(ImageDetail::High),
}],
final_output_json_schema: None,
@@ -446,7 +450,7 @@ async fn resumed_history_only_emits_resize_notices_for_new_images() -> anyhow::R
resumed.codex.shutdown_and_wait().await?;
let replayed = resume_builder
- .resume(&server, resumed.home.clone(), rollout_path)
+ .resume(&server, resumed.home.clone(), rollout_path.clone())
.await?;
let replayed_mock = responses::mount_sse_once(
&server,
@@ -466,5 +470,53 @@ async fn resumed_history_only_emits_resize_notices_for_new_images() -> anyhow::R
.collect::>();
assert_eq!(replayed_notices, vec![expected_notice.to_string()]);
+ let replayed = test_codex()
+ .with_config(|config| {
+ let _ = config.features.enable(Feature::ImageResizeNotice);
+ let _ = config
+ .features
+ .enable(Feature::RetainClientDeveloperMessages);
+ })
+ .restart(&server, &replayed)
+ .await?;
+ let existing_rollout_lines = fs::read_to_string(&rollout_path)?.lines().count();
+ replayed
+ .codex
+ .inject_response_items(vec![
+ ResponseInputItem::Message {
+ role: "user".to_string(),
+ content: vec![ContentItem::InputImage {
+ image_url: original_image_url,
+ detail: Some(ImageDetail::High),
+ }],
+ phase: None,
+ }
+ .into(),
+ ResponseInputItem::Message {
+ role: "developer".to_string(),
+ content: vec![ContentItem::InputText {
+ text: "client message".to_string(),
+ }],
+ phase: None,
+ }
+ .into(),
+ ])
+ .await?;
+ let persisted_developer_metadata = fs::read_to_string(&rollout_path)?
+ .lines()
+ .skip(existing_rollout_lines)
+ .filter_map(|line| {
+ let RolloutItem::ResponseItem(envelope) = serde_json::from_str::(line)
+ .expect("new rollout line should deserialize")
+ .item
+ else {
+ return None;
+ };
+ matches!(envelope.item, ResponseItem::Message { role, .. } if role == "developer")
+ .then(|| envelope.metadata.map(|metadata| metadata.client_authored))
+ })
+ .collect::>();
+ assert_eq!(persisted_developer_metadata, vec![None, Some(true)]);
+
Ok(())
}
diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs
index 4f6fd96778..9711ed45e4 100644
--- a/codex-rs/core/tests/suite/pending_input.rs
+++ b/codex-rs/core/tests/suite/pending_input.rs
@@ -71,9 +71,9 @@ async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyho
submit_queue_only_agent_mail(test.codex.as_ref(), "pending mailbox input").await;
test.codex
- .try_start_turn_if_idle(vec![TurnInput::ResponseItem(responses::user_message_item(
- "automatic response item",
- ))])
+ .try_start_turn_if_idle(vec![TurnInput::ResponseItem(
+ responses::user_message_item("automatic response item").into(),
+ )])
.await
.map_err(|error| {
anyhow::anyhow!("idle response items were rejected: {:?}", error.reason())
diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs
index 324e7a23d2..d85cfbeb77 100644
--- a/codex-rs/ext/goal/src/runtime.rs
+++ b/codex-rs/ext/goal/src/runtime.rs
@@ -404,7 +404,7 @@ impl GoalRuntimeHandle {
let item = continuation_steering_item(&protocol_goal_from_state(goal));
if let Err(err) = thread
- .try_start_turn_if_idle(vec![TurnInput::ResponseItem(item)])
+ .try_start_turn_if_idle(vec![TurnInput::ResponseItem(item.into())])
.await
{
let reason = err.reason();
diff --git a/codex-rs/ext/queue/tests/queue_service.rs b/codex-rs/ext/queue/tests/queue_service.rs
index 4c0f7a7c07..83c4c0cf82 100644
--- a/codex-rs/ext/queue/tests/queue_service.rs
+++ b/codex-rs/ext/queue/tests/queue_service.rs
@@ -616,7 +616,7 @@ async fn non_user_input_cannot_enter_the_user_message_queue() -> anyhow::Result<
let error = service
.enqueue(
ThreadId::new(),
- TurnInput::ResponseItem(ResponseItem::Other),
+ TurnInput::ResponseItem(ResponseItem::Other.into()),
)
.await
.expect_err("response item should not enter the user queue");
diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs
index 493841fd98..26788c95d0 100644
--- a/codex-rs/history/src/lib.rs
+++ b/codex-rs/history/src/lib.rs
@@ -40,10 +40,12 @@ pub struct ResponseItemEnvelope {
/// Metadata owned by the Codex harness and persisted with a response item.
///
-/// This intentionally has no fields yet. Keeping it closed prevents rollout
-/// metadata from becoming an untyped extension point.
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)]
-pub struct CodexHarnessMetadata {}
+pub struct CodexHarnessMetadata {
+ /// Whether a developer message was supplied by an app-server client.
+ #[serde(default)]
+ pub client_authored: bool,
+}
impl ResponseItemEnvelope {
/// Wraps a raw Responses API item for persisted history.
diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs
index 5db3d27f4b..b968d6ea7c 100644
--- a/codex-rs/history/src/tests.rs
+++ b/codex-rs/history/src/tests.rs
@@ -70,7 +70,9 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()>
ordinal: Some(7),
item: RolloutItem::ResponseItem(ResponseItemEnvelope {
item: response_item.clone(),
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata {
+ client_authored: true,
+ }),
}),
};
let serialized = serde_json::to_value(&line)?;
@@ -82,7 +84,7 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()>
"ordinal": 7,
"type": "response_item",
"payload": response_item,
- "metadata": {},
+ "metadata": { "client_authored": true },
})
);
assert_eq!(serialized["payload"].get("metadata"), None);
@@ -91,7 +93,12 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()>
let RolloutItem::ResponseItem(envelope) = restored.item else {
panic!("expected response item");
};
- assert_eq!(envelope.metadata, Some(CodexHarnessMetadata {}));
+ assert_eq!(
+ envelope.metadata,
+ Some(CodexHarnessMetadata {
+ client_authored: true,
+ })
+ );
Ok(())
}
@@ -118,7 +125,7 @@ fn response_item_envelope_ignores_unknown_harness_metadata_fields() -> Result<()
let RolloutItem::ResponseItem(envelope) = line.item else {
panic!("expected response item");
};
- assert_eq!(envelope.metadata, Some(CodexHarnessMetadata {}));
+ assert_eq!(envelope.metadata, Some(CodexHarnessMetadata::default()));
let compacted = serde_json::from_value::(json!({
"message": "summary",
@@ -127,7 +134,7 @@ fn response_item_envelope_ignores_unknown_harness_metadata_fields() -> Result<()
}))?;
assert_eq!(
compacted.replacement_history.expect("replacement history")[0].metadata,
- Some(CodexHarnessMetadata {})
+ Some(CodexHarnessMetadata::default())
);
Ok(())
}
@@ -175,7 +182,9 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
replacement_history: Some(vec![
ResponseItemEnvelope {
item: developer_message.clone(),
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata {
+ client_authored: true,
+ }),
},
ResponseItemEnvelope::new(compaction_item.clone()),
]),
@@ -191,7 +200,10 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
json!({
"message": "summary",
"replacement_history": [developer_message, compaction_item],
- "replacement_history_metadata": [{}, {}],
+ "replacement_history_metadata": [
+ { "client_authored": true },
+ { "client_authored": false },
+ ],
})
);
@@ -201,11 +213,13 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
Some(vec![
ResponseItemEnvelope {
item: developer_message,
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata {
+ client_authored: true,
+ }),
},
ResponseItemEnvelope {
item: compaction_item,
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata::default()),
},
])
);
@@ -260,7 +274,9 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() ->
let response_item = response_message("developer");
let envelope = ResponseItemEnvelope {
item: response_item.clone(),
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata {
+ client_authored: true,
+ }),
};
let response_line = serde_json::to_value(RolloutItem::ResponseItem(envelope.clone()))?;
let LegacyRolloutItem::ResponseItem(legacy_response) =
diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs
index 7292499451..deb3f0e04b 100644
--- a/codex-rs/rollout/src/tests.rs
+++ b/codex-rs/rollout/src/tests.rs
@@ -1331,7 +1331,7 @@ async fn read_head_for_summary_omits_harness_metadata() {
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItemEnvelope {
item: response_item.clone(),
- metadata: Some(CodexHarnessMetadata {}),
+ metadata: Some(CodexHarnessMetadata::default()),
}),
};
fs::write(