From 7b2fb1a6f412c7de49ed15802f4ee2f7a9c5a3ef Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 7 Jun 2026 11:10:36 -0700 Subject: [PATCH] use stable item IDs for Responses API calls --- .../app-server/tests/suite/v2/compaction.rs | 1 + .../app-server/tests/suite/v2/web_search.rs | 9 +- codex-rs/codex-api/src/endpoint/compact.rs | 7 +- codex-rs/codex-api/src/endpoint/responses.rs | 12 +- .../src/endpoint/responses_websocket.rs | 8 +- codex-rs/codex-api/src/requests/mod.rs | 3 +- codex-rs/codex-api/src/requests/responses.rs | 94 ++++++- codex-rs/codex-api/tests/clients.rs | 101 ++++++- codex-rs/context-fragments/src/fragment.rs | 20 +- codex-rs/core/src/agent/control_tests.rs | 28 +- codex-rs/core/src/client.rs | 29 +- codex-rs/core/src/client_common.rs | 7 +- codex-rs/core/src/codex_thread.rs | 8 +- codex-rs/core/src/compact.rs | 51 +++- codex-rs/core/src/compact_remote.rs | 11 +- codex-rs/core/src/compact_remote_v2.rs | 4 + codex-rs/core/src/compact_tests.rs | 40 ++- codex-rs/core/src/context_manager/history.rs | 21 +- .../core/src/context_manager/history_tests.rs | 87 ++++-- .../core/src/context_manager/normalize.rs | 37 ++- codex-rs/core/src/context_manager/updates.rs | 7 +- codex-rs/core/src/guardian/tests.rs | 2 + codex-rs/core/src/session/input_queue.rs | 28 +- .../src/session/rollout_reconstruction.rs | 2 +- codex-rs/core/src/session/tests.rs | 249 ++++++++++++------ codex-rs/core/src/stream_events_utils.rs | 36 +-- .../core/src/stream_events_utils_tests.rs | 3 + codex-rs/core/src/tasks/mod.rs | 10 +- codex-rs/core/src/tasks/review.rs | 22 +- codex-rs/core/src/tools/code_mode/delegate.rs | 8 +- codex-rs/core/src/tools/context_tests.rs | 17 +- .../src/tools/handlers/extension_tools.rs | 21 +- codex-rs/core/src/tools/router_tests.rs | 4 +- codex-rs/core/src/turn_timing_tests.rs | 1 + codex-rs/core/tests/suite/client.rs | 176 ++++++++----- .../core/tests/suite/client_websockets.rs | 94 +++++++ codex-rs/core/tests/suite/compact.rs | 3 + codex-rs/core/tests/suite/compact_remote.rs | 11 +- .../core/tests/suite/compact_remote_parity.rs | 18 +- codex-rs/core/tests/suite/hooks.rs | 1 + codex-rs/core/tests/suite/image_rollout.rs | 14 +- codex-rs/core/tests/suite/prompt_caching.rs | 109 +++++--- .../core/tests/suite/prompt_debug_tests.rs | 7 +- ...pi_auth_prompt_cache_key_request_diff.snap | 2 + ...ce_tier_prompt_cache_key_request_diff.snap | 2 + .../tests/suite/subagent_notifications.rs | 13 +- codex-rs/core/tests/suite/unified_exec.rs | 1 + codex-rs/ext/image-generation/src/tests.rs | 3 + codex-rs/ext/image-generation/src/tool.rs | 6 +- codex-rs/memories/write/src/phase1.rs | 1 + codex-rs/protocol/src/items.rs | 7 +- codex-rs/protocol/src/models.rs | 201 ++++++++++++-- codex-rs/protocol/src/protocol.rs | 87 +++++- codex-rs/rollout/src/recorder.rs | 5 +- codex-rs/rollout/src/recorder_tests.rs | 75 ++++++ 55 files changed, 1415 insertions(+), 409 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/compaction.rs b/codex-rs/app-server/tests/suite/v2/compaction.rs index 690c6d8bff..4a229ffdb0 100644 --- a/codex-rs/app-server/tests/suite/v2/compaction.rs +++ b/codex-rs/app-server/tests/suite/v2/compaction.rs @@ -137,6 +137,7 @@ async fn auto_compaction_remote_emits_started_and_completed_items() -> Result<() phase: None, }, ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; diff --git a/codex-rs/app-server/tests/suite/v2/web_search.rs b/codex-rs/app-server/tests/suite/v2/web_search.rs index 626767132b..b4818dd0c4 100644 --- a/codex-rs/app-server/tests/suite/v2/web_search.rs +++ b/codex-rs/app-server/tests/suite/v2/web_search.rs @@ -164,8 +164,14 @@ async fn standalone_web_search_round_trips_output() -> Result<()> { })) ); + let function_call_output = requests[1].function_call_output(call_id); + let function_call_output_id = function_call_output["id"] + .as_str() + .context("function call output should have an id")? + .to_string(); + assert!(function_call_output_id.starts_with("fco_")); assert_eq!( - requests[1].function_call_output(call_id), + function_call_output, json!({ "type": "function_call_output", "call_id": call_id, @@ -173,6 +179,7 @@ async fn standalone_web_search_round_trips_output() -> Result<()> { "type": "input_text", "text": "Search result", }], + "id": function_call_output_id, }) ); assert_eq!( diff --git a/codex-rs/codex-api/src/endpoint/compact.rs b/codex-rs/codex-api/src/endpoint/compact.rs index a3da854844..ebcf1a5722 100644 --- a/codex-rs/codex-api/src/endpoint/compact.rs +++ b/codex-rs/codex-api/src/endpoint/compact.rs @@ -3,6 +3,7 @@ use crate::common::CompactionInput; use crate::endpoint::session::EndpointSession; use crate::error::ApiError; use crate::provider::Provider; +use crate::requests::attach_stateless_response_item_ids; use codex_client::HttpTransport; use codex_client::RequestTelemetry; use codex_protocol::models::ResponseItem; @@ -62,9 +63,13 @@ impl CompactClient { input: &CompactionInput<'_>, extra_headers: HeaderMap, request_timeout: Duration, + include_item_ids_for_stateless_mode: bool, ) -> Result, ApiError> { - let body = to_value(input) + let mut body = to_value(input) .map_err(|e| ApiError::Stream(format!("failed to encode compaction input: {e}")))?; + if include_item_ids_for_stateless_mode { + attach_stateless_response_item_ids(&mut body, input.input); + } self.compact(body, extra_headers, request_timeout).await } } diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index cc1be2846a..7b5f3f6137 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -5,7 +5,8 @@ use crate::endpoint::session::EndpointSession; use crate::error::ApiError; use crate::provider::Provider; use crate::requests::Compression; -use crate::requests::attach_item_ids; +use crate::requests::attach_stateful_response_item_ids; +use crate::requests::attach_stateless_response_item_ids; use crate::requests::headers::build_session_headers; use crate::requests::headers::insert_header; use crate::requests::headers::subagent_header; @@ -36,6 +37,7 @@ pub struct ResponsesOptions { pub extra_headers: HeaderMap, pub compression: Compression, pub turn_state: Option>>, + pub include_item_ids_for_stateless_mode: bool, } impl ResponsesClient { @@ -79,12 +81,16 @@ impl ResponsesClient { extra_headers, compression, turn_state, + include_item_ids_for_stateless_mode, } = options; let mut body = serde_json::to_value(&request) .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?; - if request.store && self.session.provider().is_azure_responses_endpoint() { - attach_item_ids(&mut body, &request.input); + let is_azure_responses_endpoint = self.session.provider().is_azure_responses_endpoint(); + if is_azure_responses_endpoint && request.store { + attach_stateful_response_item_ids(&mut body, &request.input); + } else if !is_azure_responses_endpoint && include_item_ids_for_stateless_mode { + attach_stateless_response_item_ids(&mut body, &request.input); } let mut headers = extra_headers; diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 44cb6a544b..a9339b9678 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -5,6 +5,7 @@ use crate::common::ResponsesWsRequest; use crate::error::ApiError; use crate::provider::Provider; use crate::rate_limits::parse_rate_limit_event; +use crate::requests::attach_stateless_response_item_ids; use crate::sse::ResponsesStreamEvent; use crate::sse::process_responses_event; use crate::telemetry::WebsocketTelemetry; @@ -215,6 +216,7 @@ impl ResponsesWebsocketConnection { &self, request: ResponsesWsRequest, connection_reused: bool, + include_item_ids_for_stateless_mode: bool, ) -> Result { let (tx_event, rx_event) = mpsc::channel::>(1600); @@ -224,9 +226,13 @@ impl ResponsesWebsocketConnection { let models_etag = self.models_etag.clone(); let server_model = self.server_model.clone(); let telemetry = self.telemetry.clone(); - let request_body = serde_json::to_value(&request).map_err(|err| { + let mut request_body = serde_json::to_value(&request).map_err(|err| { ApiError::Stream(format!("failed to encode websocket request: {err}")) })?; + if include_item_ids_for_stateless_mode { + let ResponsesWsRequest::ResponseCreate(request) = &request; + attach_stateless_response_item_ids(&mut request_body, &request.input); + } let current_span = Span::current(); tokio::spawn( diff --git a/codex-rs/codex-api/src/requests/mod.rs b/codex-rs/codex-api/src/requests/mod.rs index 1c357b2a61..8bd0d915f8 100644 --- a/codex-rs/codex-api/src/requests/mod.rs +++ b/codex-rs/codex-api/src/requests/mod.rs @@ -2,4 +2,5 @@ pub(crate) mod headers; pub(crate) mod responses; pub use responses::Compression; -pub(crate) use responses::attach_item_ids; +pub(crate) use responses::attach_stateful_response_item_ids; +pub(crate) use responses::attach_stateless_response_item_ids; diff --git a/codex-rs/codex-api/src/requests/responses.rs b/codex-rs/codex-api/src/requests/responses.rs index 5f3e1ba876..1a888b3add 100644 --- a/codex-rs/codex-api/src/requests/responses.rs +++ b/codex-rs/codex-api/src/requests/responses.rs @@ -8,7 +8,16 @@ pub enum Compression { Zstd, } -pub(crate) fn attach_item_ids(payload_json: &mut Value, original_items: &[ResponseItem]) { +/// Reattaches the item IDs historically sent with stored/stateful Responses requests. +/// +/// `ResponseItem` IDs are skipped by serde so request modes can choose their attachment policy +/// after serializing the common request body. Stateful requests use the legacy narrower policy: +/// attach IDs for prompt items the stateful Responses path may treat as existing stored items, +/// but do not opt every Codex-owned output item into the newer stateless stable-ID behavior. +pub(crate) fn attach_stateful_response_item_ids( + payload_json: &mut Value, + original_items: &[ResponseItem], +) { let Some(input_value) = payload_json.get_mut("input") else { return; }; @@ -35,3 +44,86 @@ pub(crate) fn attach_item_ids(payload_json: &mut Value, original_items: &[Respon } } } + +/// Reattaches every non-empty stable item ID before sending a stateless Responses request. +/// +/// Stateless requests resend the full prompt history by value, so the server only sees stable +/// item identity when Codex explicitly puts each known ID back into the serialized `input` array. +/// This includes Codex-generated IDs for new local prompt items and IDs preserved from server +/// output, compaction, and rollout replay. +pub(crate) fn attach_stateless_response_item_ids( + payload_json: &mut Value, + original_items: &[ResponseItem], +) { + let Some(Value::Array(items)) = payload_json.get_mut("input") else { + return; + }; + + for (value, item) in items.iter_mut().zip(original_items.iter()) { + item.attach_id_to_json(value); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + + #[test] + fn attaches_stateless_response_item_ids_to_input_json() { + let items = serde_json::from_str::>( + r#"[ + {"type":"message","id":"msg_1","role":"user","content":[]}, + {"type":"agent_message","id":"amsg_1","author":"/root","recipient":"/root/worker","content":[]}, + {"type":"reasoning","id":"rs_1","summary":[]}, + {"type":"local_shell_call","id":"lsh_1","call_id":"call_shell","status":"completed","action":{"type":"exec","command":["echo","ok"]}}, + {"type":"function_call","id":"fc_1","name":"shell","arguments":"{}","call_id":"call_function"}, + {"type":"tool_search_call","id":"tsc_1","call_id":"call_search","status":"completed","execution":"client","arguments":{}}, + {"type":"function_call_output","id":"fco_1","call_id":"call_1","output":"ok"}, + {"type":"custom_tool_call","id":"ctc_1","status":"completed","call_id":"call_custom","name":"apply_patch","input":"{}"}, + {"type":"custom_tool_call_output","id":"ctco_1","call_id":"call_custom","name":"apply_patch","output":"ok"}, + {"type":"tool_search_output","id":"tso_1","call_id":"call_search","status":"completed","execution":"client","tools":[]}, + {"type":"web_search_call","id":"ws_1","status":"completed","action":{"type":"search","query":"weather"}}, + {"type":"image_generation_call","id":"ig_1","status":"completed","result":"image"}, + {"type":"compaction","id":"cmp_1","encrypted_content":"opaque"} + ]"#, + ) + .expect("deserialize response items"); + let mut payload = serde_json::json!({ + "input": serde_json::to_value(&items).expect("serialize input"), + }); + + attach_stateless_response_item_ids(&mut payload, &items); + + let ids = payload["input"] + .as_array() + .expect("input array") + .iter() + .map(|item| item["id"].as_str().expect("item id")) + .collect::>(); + assert_eq!( + ids, + vec![ + "msg_1", "amsg_1", "rs_1", "lsh_1", "fc_1", "tsc_1", "fco_1", "ctc_1", "ctco_1", + "tso_1", "ws_1", "ig_1", "cmp_1", + ] + ); + } + + #[test] + fn attach_stateless_response_item_ids_leaves_missing_ids_unchanged() { + let items = vec![ResponseItem::Message { + id: None, + role: "user".to_string(), + content: Vec::new(), + phase: None, + }]; + let mut payload = serde_json::json!({ + "input": serde_json::to_value(&items).expect("serialize input"), + }); + + attach_stateless_response_item_ids(&mut payload, &items); + + assert_eq!(payload["input"][0].get("id"), None); + } +} diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index afaf82389f..f04100d787 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -20,6 +20,7 @@ use codex_client::Response; use codex_client::StreamResponse; use codex_client::TransportError; use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -419,12 +420,19 @@ async fn azure_default_store_attaches_ids_and_headers() -> Result<()> { let request = ResponsesApiRequest { model: "gpt-test".into(), instructions: "Say hi".into(), - input: vec![ResponseItem::Message { - id: Some("msg_1".into()), - role: "user".into(), - content: vec![ContentItem::InputText { text: "hi".into() }], - phase: None, - }], + input: vec![ + ResponseItem::Message { + id: Some("msg_1".into()), + role: "user".into(), + content: vec![ContentItem::InputText { text: "hi".into() }], + phase: None, + }, + ResponseItem::FunctionCallOutput { + id: Some("fco_1".into()), + call_id: "call_1".into(), + output: FunctionCallOutputPayload::from_text("ok".into()), + }, + ], tools: Vec::new(), tool_choice: "auto".into(), parallel_tool_calls: false, @@ -450,6 +458,7 @@ async fn azure_default_store_attaches_ids_and_headers() -> Result<()> { extra_headers, compression: Compression::None, turn_state: None, + include_item_ids_for_stateless_mode: false, }, ) .await?; @@ -485,15 +494,85 @@ async fn azure_default_store_attaches_ids_and_headers() -> Result<()> { Some("present") ); - let input_id = req + let input = req .body .as_ref() .and_then(RequestBody::json) .and_then(|body| body.get("input")) - .and_then(|input| input.get(0)) - .and_then(|item| item.get("id")) - .and_then(|id| id.as_str()); - assert_eq!(input_id, Some("msg_1")); + .expect("input array"); + assert_eq!( + input + .get(0) + .and_then(|item| item.get("id")) + .and_then(|id| id.as_str()), + Some("msg_1") + ); + assert_eq!( + input + .get(1) + .and_then(|item| item.get("id")) + .and_then(|id| id.as_str()), + None + ); + + Ok(()) +} + +#[tokio::test] +async fn response_item_ids_require_explicit_request_option() -> Result<()> { + let state = RecordingState::default(); + let transport = RecordingTransport::new(state.clone()); + let client = ResponsesClient::new(transport, provider("openai"), Arc::new(NoAuth)); + let request = ResponsesApiRequest { + model: "gpt-test".into(), + instructions: "Say hi".into(), + input: vec![ResponseItem::Message { + id: Some("msg_1".into()), + role: "user".into(), + content: vec![ContentItem::InputText { text: "hi".into() }], + phase: None, + }], + tools: Vec::new(), + tool_choice: "auto".into(), + parallel_tool_calls: false, + reasoning: None, + store: false, + stream: true, + include: Vec::new(), + service_tier: None, + prompt_cache_key: None, + text: None, + client_metadata: None, + }; + + let _stream = client + .stream_request(request.clone(), ResponsesOptions::default()) + .await?; + let _stream = client + .stream_request( + request, + ResponsesOptions { + include_item_ids_for_stateless_mode: true, + ..Default::default() + }, + ) + .await?; + + let requests = state.take_stream_requests(); + assert_eq!(requests.len(), 2); + let input_id = |request: &Request| { + request + .body + .as_ref() + .and_then(RequestBody::json) + .and_then(|body| body.get("input")) + .and_then(|input| input.get(0)) + .and_then(|item| item.get("id")) + .and_then(serde_json::Value::as_str) + .map(str::to_string) + }; + assert_eq!(input_id(&requests[0]), None); + assert_eq!(input_id(&requests[1]), Some("msg_1".to_string())); Ok(()) } diff --git a/codex-rs/context-fragments/src/fragment.rs b/codex-rs/context-fragments/src/fragment.rs index bd0e06772b..bc79fec83d 100644 --- a/codex-rs/context-fragments/src/fragment.rs +++ b/codex-rs/context-fragments/src/fragment.rs @@ -76,25 +76,21 @@ pub trait ContextualUserFragment { where Self: Sized, { - ResponseItem::Message { - id: None, - role: self.role().to_string(), - content: vec![ContentItem::InputText { + ResponseItem::new_message( + self.role(), + vec![ContentItem::InputText { text: self.render(), }], - phase: None, - } + ) } fn into_boxed_response_item(self: Box) -> ResponseItem { - ResponseItem::Message { - id: None, - role: self.role().to_string(), - content: vec![ContentItem::InputText { + ResponseItem::new_message( + self.role(), + vec![ContentItem::InputText { text: self.render(), }], - phase: None, - } + ) } fn into_response_input_item(self) -> ResponseInputItem diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index c626dda4a5..8420771c07 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -871,7 +871,14 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { phase: None, }, assistant_message("parent commentary", Some(MessagePhase::Commentary)), - assistant_message("parent final answer", Some(MessagePhase::FinalAnswer)), + ResponseItem::Message { + id: Some("msg_parent_final_answer".to_string()), + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: "parent final answer".to_string(), + }], + phase: Some(MessagePhase::FinalAnswer), + }, assistant_message("parent unknown phase", /*phase*/ None), ResponseItem::Reasoning { id: "parent-reasoning".to_string(), @@ -933,18 +940,31 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { .expect("child thread should be registered"); assert_ne!(child_thread_id, parent_thread_id); let history = child_thread.codex.session.clone_history().await; + let history_ids = history + .raw_items() + .iter() + .map(|item| item.id().expect("forked messages should have ids")) + .collect::>(); + assert!(history_ids.iter().all(|id| id.starts_with("msg_"))); let expected_history = [ ResponseItem::Message { - id: None, + id: Some(history_ids[0].to_string()), role: "user".to_string(), content: vec![ContentItem::InputText { text: "parent seed context".to_string(), }], phase: None, }, - assistant_message("parent final answer", Some(MessagePhase::FinalAnswer)), ResponseItem::Message { - id: None, + id: Some(history_ids[1].to_string()), + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: "parent final answer".to_string(), + }], + phase: Some(MessagePhase::FinalAnswer), + }, + ResponseItem::Message { + id: Some(history_ids[2].to_string()), role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Child subagent guidance.".to_string(), diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 02d231c962..d7f12d7a0f 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -535,12 +535,19 @@ impl ModelClient { .api_provider .stream_idle_timeout .saturating_mul(COMPACT_REQUEST_TIMEOUT_IDLE_MULTIPLIER); + let include_item_ids_for_stateless_mode = + self.should_include_response_item_ids_for_stateless_mode(); let client = ApiCompactClient::new(transport, client_setup.api_provider, client_setup.api_auth) .with_telemetry(Some(request_telemetry)); let trace_attempt = compaction_trace.start_attempt(&payload); let result = client - .compact_input(&payload, extra_headers, compact_request_timeout) + .compact_input( + &payload, + extra_headers, + compact_request_timeout, + include_item_ids_for_stateless_mode, + ) .await .map_err(map_api_error); trace_attempt.record_result(result.as_deref()); @@ -816,6 +823,10 @@ impl ModelClient { Ok(request) } + fn should_include_response_item_ids_for_stateless_mode(&self) -> bool { + self.state.provider.info().is_openai() + } + /// Returns whether the Responses-over-WebSocket transport is active for this session. /// /// WebSocket use is controlled by provider capability and session-scoped fallback state. @@ -1001,6 +1012,7 @@ impl ModelClientSession { turn_metadata_header: Option<&str>, compression: Compression, use_responses_lite: bool, + include_item_ids_for_stateless_mode: bool, ) -> ApiResponsesOptions { let turn_metadata_header = parse_turn_metadata_header(turn_metadata_header); let session_id = self.client.state.session_id.to_string(); @@ -1024,6 +1036,7 @@ impl ModelClientSession { }, compression, turn_state: Some(Arc::clone(&self.turn_state)), + include_item_ids_for_stateless_mode, } } @@ -1286,11 +1299,15 @@ impl ModelClientSession { self.client.state.auth_env_telemetry.clone(), ); let compression = self.responses_request_compression(client_setup.auth.as_ref()); + let include_item_ids_for_stateless_mode = self + .client + .should_include_response_item_ids_for_stateless_mode(); let mut options = self .build_responses_options( turn_metadata_header, compression, model_info.use_responses_lite, + include_item_ids_for_stateless_mode, ) .await; @@ -1399,12 +1416,16 @@ impl ModelClientSession { pending_retry, ); let compression = self.responses_request_compression(client_setup.auth.as_ref()); + let include_item_ids_for_stateless_mode = self + .client + .should_include_response_item_ids_for_stateless_mode(); let options = self .build_responses_options( turn_metadata_header, compression, model_info.use_responses_lite, + include_item_ids_for_stateless_mode, ) .await; let request = self.client.build_responses_request( @@ -1492,7 +1513,11 @@ impl ModelClientSession { )) })?; let stream_result = websocket_connection - .stream_request(ws_request, self.websocket_session.connection_reused()) + .stream_request( + ws_request, + self.websocket_session.connection_reused(), + include_item_ids_for_stateless_mode, + ) .await .map_err(|err| { let response_debug_context = diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 48d0523df4..34ad8186c1 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -58,7 +58,10 @@ impl Prompt { .iter() .cloned() .map(|item| { - let ResponseItem::Message { role, content, .. } = &item else { + let ResponseItem::Message { + id, role, content, .. + } = &item + else { return item; }; if role != "assistant" { @@ -66,7 +69,7 @@ impl Prompt { } InterAgentCommunication::from_message_content(content) .filter(|communication| communication.encrypted_content.is_some()) - .map(|communication| communication.to_model_input_item()) + .map(|communication| communication.to_model_input_item(id.as_deref())) .unwrap_or(item) }) .collect() diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index ab1f6df03e..34923fa1ee 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -413,12 +413,8 @@ impl CodexThread { /// Records a user-role session-prefix message without creating a new user turn boundary. pub(crate) async fn inject_user_message_without_turn(&self, message: String) { - let item = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { text: message }], - phase: None, - }; + let item = + ResponseItem::new_message("user", vec![ContentItem::InputText { text: message }]); self.codex .session .inject_no_new_turn(vec![item], /*current_turn_context*/ None) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index e2573d6d6d..59332d66d2 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -519,14 +519,36 @@ pub(crate) fn build_compacted_history( user_messages, summary_text, COMPACT_USER_MESSAGE_MAX_TOKENS, + CompactedHistoryMessageIds::New, ) } +pub(crate) fn rebuild_legacy_compacted_history( + initial_context: Vec, + user_messages: &[String], + summary_text: &str, +) -> Vec { + build_compacted_history_with_limit( + initial_context, + user_messages, + summary_text, + COMPACT_USER_MESSAGE_MAX_TOKENS, + CompactedHistoryMessageIds::Missing, + ) +} + +#[derive(Clone, Copy)] +enum CompactedHistoryMessageIds { + New, + Missing, +} + fn build_compacted_history_with_limit( mut history: Vec, user_messages: &[String], summary_text: &str, max_tokens: usize, + message_ids: CompactedHistoryMessageIds, ) -> Vec { let mut selected_messages: Vec = Vec::new(); if max_tokens > 0 { @@ -549,14 +571,7 @@ fn build_compacted_history_with_limit( } for message in &selected_messages { - history.push(ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { - text: message.clone(), - }], - phase: None, - }); + history.push(compacted_user_message(message.clone(), message_ids)); } let summary_text = if summary_text.is_empty() { @@ -565,16 +580,24 @@ fn build_compacted_history_with_limit( summary_text.to_string() }; - history.push(ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { text: summary_text }], - phase: None, - }); + history.push(compacted_user_message(summary_text, message_ids)); history } +fn compacted_user_message(text: String, message_ids: CompactedHistoryMessageIds) -> ResponseItem { + let content = vec![ContentItem::InputText { text }]; + match message_ids { + CompactedHistoryMessageIds::New => ResponseItem::new_message("user", content), + CompactedHistoryMessageIds::Missing => ResponseItem::Message { + id: None, + role: "user".to_string(), + content, + phase: None, + }, + } +} + async fn drain_to_completed( sess: &Session, turn_context: &TurnContext, diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index 5fca191b84..c0492fb1a8 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -448,25 +448,34 @@ pub(crate) fn trim_function_call_history_to_fit_context_window( fn rewritten_output_for_context_window(item: &ResponseItem) -> Option { Some(match item { - ResponseItem::FunctionCallOutput { call_id, output } => ResponseItem::FunctionCallOutput { + ResponseItem::FunctionCallOutput { + id, + call_id, + output, + } => ResponseItem::FunctionCallOutput { + id: id.clone(), call_id: call_id.clone(), output: truncated_output_payload(output), }, ResponseItem::CustomToolCallOutput { + id, call_id, name, output, } => ResponseItem::CustomToolCallOutput { + id: id.clone(), call_id: call_id.clone(), name: name.clone(), output: truncated_output_payload(output), }, ResponseItem::ToolSearchOutput { + id, call_id, status, execution, .. } => ResponseItem::ToolSearchOutput { + id: id.clone(), call_id: call_id.clone(), status: status.clone(), execution: execution.clone(), diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 1d4eb8ad5a..80c93369fc 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -626,10 +626,12 @@ mod tests { call_id: "call_1".to_string(), }, ResponseItem::Compaction { + id: None, encrypted_content: "old".to_string(), }, ]; let output = ResponseItem::Compaction { + id: None, encrypted_content: "new".to_string(), }; @@ -657,6 +659,7 @@ mod tests { new.clone(), ]; let output = ResponseItem::Compaction { + id: None, encrypted_content: "new".to_string(), }; @@ -808,6 +811,7 @@ mod tests { #[tokio::test] async fn collect_compaction_output_accepts_additional_output_items() { let compaction = ResponseItem::Compaction { + id: None, encrypted_content: "encrypted".to_string(), }; let stream = response_stream(vec![ diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 69bd95f2df..bc3f32e181 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/codex-rs/core/src/compact_tests.rs @@ -162,6 +162,7 @@ fn build_token_limited_compacted_history_truncates_overlong_user_messages() { std::slice::from_ref(&big), "SUMMARY", max_tokens, + super::CompactedHistoryMessageIds::New, ); assert_eq!(history.len(), 2); @@ -215,6 +216,21 @@ fn build_token_limited_compacted_history_appends_summary_message() { assert_eq!(summary, summary_text); } +#[test] +fn build_compacted_history_marks_new_compacted_messages() { + let history = build_compacted_history( + Vec::new(), + &["first user message".to_string()], + "summary text", + ); + + assert!( + history + .iter() + .all(|item| { item.id().is_some_and(|id| id.starts_with("msg_")) }) + ); +} + #[test] fn should_use_remote_compact_task_for_azure_provider() { let provider = ModelProviderInfo { @@ -280,7 +296,7 @@ async fn process_compacted_history_replaces_developer_messages() { }], phase: None, }); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); } #[tokio::test] @@ -306,7 +322,17 @@ async fn process_compacted_history_reinjects_full_initial_context() { }], phase: None, }); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); + let reinjected_initial_context = &refreshed[..expected.len() - 1]; + crate::session::tests::assert_response_items_eq_ignoring_ids( + reinjected_initial_context, + &expected[..expected.len() - 1], + ); + assert!( + reinjected_initial_context + .iter() + .all(|item| { item.id().is_some_and(|id| id.starts_with("msg_")) }) + ); } #[tokio::test] @@ -379,7 +405,7 @@ keep me updated }], phase: None, }); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); } #[tokio::test] @@ -404,7 +430,7 @@ async fn process_compacted_history_drops_legacy_warnings() { .await; let mut expected = initial_context; expected.push(latest_user); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); } #[tokio::test] @@ -468,7 +494,7 @@ async fn process_compacted_history_inserts_context_before_last_real_user_message }], phase: None, }); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); } #[tokio::test] @@ -510,7 +536,7 @@ async fn process_compacted_history_reinjects_model_switch_message() { }], phase: None, }); - assert_eq!(refreshed, expected); + crate::session::tests::assert_response_items_eq_ignoring_ids(&refreshed, &expected); } #[test] @@ -592,6 +618,7 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_summary_last() #[test] fn insert_initial_context_before_last_real_user_or_summary_keeps_compaction_last() { let compacted_history = vec![ResponseItem::Compaction { + id: None, encrypted_content: "encrypted".to_string(), }]; let initial_context = vec![ResponseItem::Message { @@ -615,6 +642,7 @@ fn insert_initial_context_before_last_real_user_or_summary_keeps_compaction_last phase: None, }, ResponseItem::Compaction { + id: None, encrypted_content: "encrypted".to_string(), }, ]; diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index b8bcf9c846..bfc5890526 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -367,20 +367,22 @@ impl ContextManager { fn process_item(&self, item: &ResponseItem, policy: TruncationPolicy) -> ResponseItem { let policy_with_serialization_budget = policy * 1.2; match item { - ResponseItem::FunctionCallOutput { call_id, output } => { - ResponseItem::FunctionCallOutput { - call_id: call_id.clone(), - output: truncate_function_output_payload( - output, - policy_with_serialization_budget, - ), - } - } + ResponseItem::FunctionCallOutput { + id, + call_id, + output, + } => ResponseItem::FunctionCallOutput { + id: id.clone(), + call_id: call_id.clone(), + output: truncate_function_output_payload(output, policy_with_serialization_budget), + }, ResponseItem::CustomToolCallOutput { + id, call_id, name, output, } => ResponseItem::CustomToolCallOutput { + id: id.clone(), call_id: call_id.clone(), name: name.clone(), output: truncate_function_output_payload(output, policy_with_serialization_budget), @@ -540,6 +542,7 @@ pub(crate) fn estimate_response_item_model_visible_bytes(item: &ResponseItem) -> } | ResponseItem::Compaction { encrypted_content: content, + .. } | ResponseItem::ContextCompaction { encrypted_content: Some(content), diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index bb22749cb3..e35989d5f4 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -141,6 +141,7 @@ fn reference_context_item() -> TurnContextItem { fn custom_tool_call_output(call_id: &str, output: &str) -> ResponseItem { ResponseItem::CustomToolCallOutput { + id: None, call_id: call_id.to_string(), name: None, output: FunctionCallOutputPayload::from_text(output.to_string()), @@ -387,6 +388,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() { call_id: "call-1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputText { @@ -406,6 +408,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() { input: "view_image".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "tool-1".to_string(), name: None, output: FunctionCallOutputPayload::from_content_items(vec![ @@ -449,6 +452,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() { call_id: "call-1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputText { @@ -468,6 +472,7 @@ fn for_prompt_strips_images_when_model_does_not_support_images() { input: "view_image".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "tool-1".to_string(), name: None, output: FunctionCallOutputPayload::from_content_items(vec![ @@ -622,6 +627,7 @@ fn remove_first_item_removes_matching_output_for_function_call() { call_id: "call-1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -635,6 +641,7 @@ fn remove_first_item_removes_matching_output_for_function_call() { fn remove_first_item_removes_matching_call_for_output() { let items = vec![ ResponseItem::FunctionCallOutput { + id: None, call_id: "call-2".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -656,6 +663,7 @@ fn replace_last_turn_images_replaces_tool_output_images() { let items = vec![ user_input_text_msg("hi"), ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::ContentItems(vec![ @@ -677,6 +685,7 @@ fn replace_last_turn_images_replaces_tool_output_images() { vec![ user_input_text_msg("hi"), ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::ContentItems(vec![ @@ -724,6 +733,7 @@ fn remove_first_item_handles_local_shell_pair() { }), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-3".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -949,6 +959,7 @@ fn remove_first_item_handles_custom_tool_pair() { input: "{}".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "tool-1".to_string(), name: None, output: FunctionCallOutputPayload::from_text("ok".to_string()), @@ -975,6 +986,7 @@ fn normalization_retains_local_shell_outputs() { }), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "shell-1".to_string(), output: FunctionCallOutputPayload::from_text("Total output lines: 1\n\nok".to_string()), }, @@ -995,6 +1007,7 @@ fn record_items_truncates_function_call_output_content() { let long_line = "a very long line to trigger truncation\n"; let long_output = long_line.repeat(2_500); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-100".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::Text(long_output.clone()), @@ -1029,6 +1042,7 @@ fn record_items_truncates_custom_tool_call_output_content() { let line = "custom output that is very long\n"; let long_output = line.repeat(2_500); let item = ResponseItem::CustomToolCallOutput { + id: None, call_id: "tool-200".to_string(), name: None, output: FunctionCallOutputPayload::from_text(long_output.clone()), @@ -1060,6 +1074,7 @@ fn record_items_respects_custom_token_limit() { let policy = TruncationPolicy::Tokens(10); let long_output = "tokenized content repeated many times ".repeat(200); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-custom-limit".to_string(), output: FunctionCallOutputPayload { body: FunctionCallOutputBody::Text(long_output), @@ -1189,9 +1204,9 @@ fn normalize_adds_missing_output_for_function_call() { h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::FunctionCall { id: None, name: "do_it".to_string(), @@ -1200,10 +1215,11 @@ fn normalize_adds_missing_output_for_function_call() { call_id: "call-x".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-x".to_string(), output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, - ] + ], ); } @@ -1221,9 +1237,9 @@ fn normalize_adds_missing_output_for_custom_tool_call() { h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::CustomToolCall { id: None, status: None, @@ -1232,11 +1248,12 @@ fn normalize_adds_missing_output_for_custom_tool_call() { input: "{}".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "tool-x".to_string(), name: None, output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, - ] + ], ); } @@ -1259,9 +1276,9 @@ fn normalize_adds_missing_output_for_local_shell_call_with_id() { h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::LocalShellCall { id: None, call_id: Some("shell-1".to_string()), @@ -1275,10 +1292,11 @@ fn normalize_adds_missing_output_for_local_shell_call_with_id() { }), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "shell-1".to_string(), output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, - ] + ], ); } @@ -1286,6 +1304,7 @@ fn normalize_adds_missing_output_for_local_shell_call_with_id() { #[test] fn normalize_removes_orphan_function_call_output() { let items = vec![ResponseItem::FunctionCallOutput { + id: None, call_id: "orphan-1".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }]; @@ -1300,6 +1319,7 @@ fn normalize_removes_orphan_function_call_output() { #[test] fn normalize_removes_orphan_custom_tool_call_output() { let items = vec![ResponseItem::CustomToolCallOutput { + id: None, call_id: "orphan-2".to_string(), name: None, output: FunctionCallOutputPayload::from_text("ok".to_string()), @@ -1325,6 +1345,7 @@ fn normalize_mixed_inserts_and_removals() { }, // Orphan output that should be removed ResponseItem::FunctionCallOutput { + id: None, call_id: "c2".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -1354,9 +1375,9 @@ fn normalize_mixed_inserts_and_removals() { h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::FunctionCall { id: None, name: "f1".to_string(), @@ -1365,6 +1386,7 @@ fn normalize_mixed_inserts_and_removals() { call_id: "c1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "c1".to_string(), output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, @@ -1376,6 +1398,7 @@ fn normalize_mixed_inserts_and_removals() { input: "{}".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "t1".to_string(), name: None, output: FunctionCallOutputPayload::from_text("aborted".to_string()), @@ -1393,10 +1416,11 @@ fn normalize_mixed_inserts_and_removals() { }), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "s1".to_string(), output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, - ] + ], ); } @@ -1411,9 +1435,9 @@ fn normalize_adds_missing_output_for_function_call_inserts_output() { }]; let mut h = create_history_with_items(items); h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::FunctionCall { id: None, name: "do_it".to_string(), @@ -1422,10 +1446,16 @@ fn normalize_adds_missing_output_for_function_call_inserts_output() { call_id: "call-x".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-x".to_string(), output: FunctionCallOutputPayload::from_text("aborted".to_string()), }, - ] + ], + ); + assert!( + h.raw_items()[1] + .id() + .is_some_and(|id| id.starts_with("fco_")) ); } @@ -1442,9 +1472,9 @@ fn normalize_adds_missing_output_for_tool_search_call() { h.normalize_history(&default_input_modalities()); - assert_eq!( + crate::session::tests::assert_response_items_eq_ignoring_ids( h.raw_items(), - vec![ + &[ ResponseItem::ToolSearchCall { id: None, call_id: Some("search-call-x".to_string()), @@ -1453,12 +1483,18 @@ fn normalize_adds_missing_output_for_tool_search_call() { arguments: "{}".into(), }, ResponseItem::ToolSearchOutput { + id: None, call_id: Some("search-call-x".to_string()), status: "completed".to_string(), execution: "client".to_string(), tools: Vec::new(), }, - ] + ], + ); + assert!( + h.raw_items()[1] + .id() + .is_some_and(|id| id.starts_with("tso_")) ); } @@ -1502,6 +1538,7 @@ fn normalize_adds_missing_output_for_local_shell_call_with_id_panics_in_debug() #[should_panic] fn normalize_removes_orphan_function_call_output_panics_in_debug() { let items = vec![ResponseItem::FunctionCallOutput { + id: None, call_id: "orphan-1".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }]; @@ -1514,6 +1551,7 @@ fn normalize_removes_orphan_function_call_output_panics_in_debug() { #[should_panic] fn normalize_removes_orphan_custom_tool_call_output_panics_in_debug() { let items = vec![ResponseItem::CustomToolCallOutput { + id: None, call_id: "orphan-2".to_string(), name: None, output: FunctionCallOutputPayload::from_text("ok".to_string()), @@ -1526,6 +1564,7 @@ fn normalize_removes_orphan_custom_tool_call_output_panics_in_debug() { #[test] fn normalize_removes_orphan_client_tool_search_output() { let items = vec![ResponseItem::ToolSearchOutput { + id: None, call_id: Some("orphan-search".to_string()), status: "completed".to_string(), execution: "client".to_string(), @@ -1543,6 +1582,7 @@ fn normalize_removes_orphan_client_tool_search_output() { #[should_panic] fn normalize_removes_orphan_client_tool_search_output_panics_in_debug() { let items = vec![ResponseItem::ToolSearchOutput { + id: None, call_id: Some("orphan-search".to_string()), status: "completed".to_string(), execution: "client".to_string(), @@ -1555,6 +1595,7 @@ fn normalize_removes_orphan_client_tool_search_output_panics_in_debug() { #[test] fn normalize_keeps_server_tool_search_output_without_matching_call() { let items = vec![ResponseItem::ToolSearchOutput { + id: None, call_id: Some("server-search".to_string()), status: "completed".to_string(), execution: "server".to_string(), @@ -1567,6 +1608,7 @@ fn normalize_keeps_server_tool_search_output_without_matching_call() { assert_eq!( h.raw_items(), vec![ResponseItem::ToolSearchOutput { + id: None, call_id: Some("server-search".to_string()), status: "completed".to_string(), execution: "server".to_string(), @@ -1588,6 +1630,7 @@ fn normalize_mixed_inserts_and_removals_panics_in_debug() { call_id: "c1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "c2".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -1657,6 +1700,7 @@ fn image_data_url_payload_does_not_dominate_function_call_output_estimate() { let payload = "B".repeat(50_000); let image_url = format!("data:image/png;base64,{payload}"); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-abc".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputText { @@ -1682,6 +1726,7 @@ fn image_data_url_payload_does_not_dominate_custom_tool_call_output_estimate() { let payload = "C".repeat(50_000); let image_url = format!("data:image/png;base64,{payload}"); let item = ResponseItem::CustomToolCallOutput { + id: None, call_id: "call-js-repl".to_string(), name: None, output: FunctionCallOutputPayload::from_content_items(vec![ @@ -1715,6 +1760,7 @@ fn non_base64_image_urls_are_unchanged() { phase: None, }; let function_output_item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { @@ -1738,6 +1784,7 @@ fn non_base64_image_urls_are_unchanged() { fn encrypted_function_output_uses_plaintext_byte_estimate() { let encrypted_content = "A".repeat(1_868); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-encrypted".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::EncryptedContent { @@ -1777,6 +1824,7 @@ fn non_image_base64_data_url_is_unchanged() { let payload = "C".repeat(4_096); let image_url = format!("data:application/octet-stream;base64,{payload}"); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-octet".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { @@ -1862,6 +1910,7 @@ fn original_detail_images_scale_with_dimensions() { let payload = BASE64_STANDARD.encode(bytes.get_ref()); let image_url = format!("data:image/png;base64,{payload}"); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-original".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { @@ -1892,6 +1941,7 @@ fn original_detail_images_are_capped_at_max_patch_count() { let payload = BASE64_STANDARD.encode(bytes.get_ref()); let image_url = format!("data:image/png;base64,{payload}"); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-original-capped".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { @@ -1925,6 +1975,7 @@ fn original_detail_webp_images_scale_with_dimensions() { let payload = BASE64_STANDARD.encode(bytes.get_ref()); let image_url = format!("data:image/webp;base64,{payload}"); let item = ResponseItem::FunctionCallOutput { + id: None, call_id: "call-original-webp".to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { diff --git a/codex-rs/core/src/context_manager/normalize.rs b/codex-rs/core/src/context_manager/normalize.rs index 839bae331e..7e9a0eea5a 100644 --- a/codex-rs/core/src/context_manager/normalize.rs +++ b/codex-rs/core/src/context_manager/normalize.rs @@ -31,10 +31,10 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec) { info!("Function call output is missing for call id: {call_id}"); missing_outputs_to_insert.push(( idx, - ResponseItem::FunctionCallOutput { - call_id: call_id.clone(), - output: FunctionCallOutputPayload::from_text("aborted".to_string()), - }, + ResponseItem::new_function_call_output( + call_id.clone(), + FunctionCallOutputPayload::from_text("aborted".to_string()), + ), )); } } @@ -54,12 +54,12 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec) { info!("Tool search output is missing for call id: {call_id}"); missing_outputs_to_insert.push(( idx, - ResponseItem::ToolSearchOutput { - call_id: Some(call_id.clone()), - status: "completed".to_string(), - execution: "client".to_string(), - tools: Vec::new(), - }, + ResponseItem::new_tool_search_output( + call_id.clone(), + "completed", + "client", + Vec::new(), + ), )); } } @@ -77,11 +77,10 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec) { )); missing_outputs_to_insert.push(( idx, - ResponseItem::CustomToolCallOutput { - call_id: call_id.clone(), - name: None, - output: FunctionCallOutputPayload::from_text("aborted".to_string()), - }, + ResponseItem::new_custom_tool_call_output( + call_id.clone(), + FunctionCallOutputPayload::from_text("aborted".to_string()), + ), )); } } @@ -101,10 +100,10 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec) { )); missing_outputs_to_insert.push(( idx, - ResponseItem::FunctionCallOutput { - call_id: call_id.clone(), - output: FunctionCallOutputPayload::from_text("aborted".to_string()), - }, + ResponseItem::new_function_call_output( + call_id.clone(), + FunctionCallOutputPayload::from_text("aborted".to_string()), + ), )); } } diff --git a/codex-rs/core/src/context_manager/updates.rs b/codex-rs/core/src/context_manager/updates.rs index d7302bbbf8..8b7194e6da 100644 --- a/codex-rs/core/src/context_manager/updates.rs +++ b/codex-rs/core/src/context_manager/updates.rs @@ -198,12 +198,7 @@ fn build_text_message(role: &str, text_sections: Vec) -> Option, turn: &Arc ResponseItem::Message { + id: None, + role, + content, + phase, + }, + item => item, + }; assert_eq!( - input_queue.drain_mailbox_input_items().await, + actual + .into_iter() + .map(without_message_id) + .collect::>(), vec![ ResponseItem::from(mail_one.to_response_input_item()), ResponseItem::from(mail_two.to_response_input_item()) ] + .into_iter() + .map(without_message_id) + .collect::>() ); assert!(!input_queue.has_pending_mailbox_items().await); } diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index a4c042af0c..4012d1be61 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -263,7 +263,7 @@ impl Session { // TODO(ccunningham): if we drop support for None replacement_history compaction items, // we can get rid of this second loop entirely and just build `history` directly in the first loop. let user_messages = collect_user_messages(history.raw_items()); - let rebuilt = compact::build_compacted_history( + let rebuilt = compact::rebuild_legacy_compacted_history( Vec::new(), &user_messages, &compacted.message, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d44c7fbdcc..b707511368 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -201,6 +201,72 @@ fn assistant_message(text: &str) -> ResponseItem { } } +fn without_response_item_id(mut item: ResponseItem) -> ResponseItem { + match &mut item { + ResponseItem::Message { id, .. } + | ResponseItem::AgentMessage { id, .. } + | ResponseItem::LocalShellCall { id, .. } + | ResponseItem::FunctionCall { id, .. } + | ResponseItem::ToolSearchCall { id, .. } + | ResponseItem::FunctionCallOutput { id, .. } + | ResponseItem::CustomToolCall { id, .. } + | ResponseItem::CustomToolCallOutput { id, .. } + | ResponseItem::ToolSearchOutput { id, .. } + | ResponseItem::WebSearchCall { id, .. } + | ResponseItem::Compaction { id, .. } => *id = None, + ResponseItem::Reasoning { id, .. } | ResponseItem::ImageGenerationCall { id, .. } => { + id.clear(); + } + ResponseItem::CompactionTrigger + | ResponseItem::ContextCompaction { .. } + | ResponseItem::Other => {} + } + item +} + +fn response_items_without_ids(items: &[ResponseItem]) -> Vec { + items + .iter() + .cloned() + .map(without_response_item_id) + .collect() +} + +pub(crate) fn assert_response_items_eq_ignoring_ids( + actual: &[ResponseItem], + expected: &[ResponseItem], +) { + assert_eq!( + response_items_without_ids(actual), + response_items_without_ids(expected) + ); +} + +fn assert_turn_inputs_eq_ignoring_response_item_ids( + actual: Vec, + expected: Vec, +) { + let without_response_item_ids = |inputs: Vec| { + inputs + .into_iter() + .map(|input| match input { + TurnInput::ResponseItem(item) => { + TurnInput::ResponseItem(without_response_item_id(item)) + } + input => input, + }) + .collect::>() + }; + assert_eq!( + without_response_item_ids(actual), + without_response_item_ids(expected) + ); +} + +fn response_item_eq_ignoring_id(actual: &ResponseItem, expected: &ResponseItem) -> bool { + without_response_item_id(actual.clone()) == without_response_item_id(expected.clone()) +} + fn test_session_telemetry_without_metadata() -> SessionTelemetry { let exporter = InMemoryMetricExporter::default(); let metrics = MetricsClient::new( @@ -1592,7 +1658,7 @@ async fn reconstruct_history_matches_live_compactions() { async fn reconstruct_history_uses_replacement_history_verbatim() { let (session, turn_context) = make_session_and_context().await; let summary_item = ResponseItem::Message { - id: None, + id: Some("msg_summary".to_string()), role: "user".to_string(), content: vec![ContentItem::InputText { text: "summary".to_string(), @@ -1602,7 +1668,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { let replacement_history = vec![ summary_item.clone(), ResponseItem::Message { - id: None, + id: Some("msg_developer".to_string()), role: "developer".to_string(), content: vec![ContentItem::InputText { text: "stale developer instructions".to_string(), @@ -1636,7 +1702,43 @@ async fn record_initial_history_reconstructs_resumed_transcript() { .await; let history = session.state.lock().await.clone_history(); - assert_eq!(expected, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &expected); +} + +#[tokio::test] +async fn record_initial_history_preserves_missing_historical_response_item_ids() { + let (session, _turn_context) = make_session_and_context().await; + let expected = vec![ + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: "legacy user message".to_string(), + }], + phase: None, + }, + ResponseItem::FunctionCallOutput { + id: None, + call_id: "legacy_call".to_string(), + output: FunctionCallOutputPayload::from_text("legacy output".to_string()), + }, + ]; + let rollout_items = expected + .iter() + .cloned() + .map(RolloutItem::ResponseItem) + .collect(); + + session + .record_initial_history(InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: rollout_items, + rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")), + })) + .await; + + let history = session.state.lock().await.clone_history(); + assert_eq!(history.raw_items(), expected.as_slice()); } #[test] @@ -1745,14 +1847,14 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only() .await; let history_before_seed = session.state.lock().await.clone_history(); - assert_eq!(expected, history_before_seed.raw_items()); + assert_response_items_eq_ignoring_ids(history_before_seed.raw_items(), &expected); session .record_context_updates_and_set_reference_context_item(&turn_context) .await; expected.extend(session.build_initial_context(&turn_context).await); let history_after_seed = session.clone_history().await; - assert_eq!(expected, history_after_seed.raw_items()); + assert_response_items_eq_ignoring_ids(history_after_seed.raw_items(), &expected); session .record_context_updates_and_set_reference_context_item(&turn_context) @@ -2334,7 +2436,7 @@ async fn record_initial_history_reconstructs_forked_transcript() { .await; let history = session.state.lock().await.clone_history(); - assert_eq!(expected, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &expected); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -2654,7 +2756,7 @@ async fn thread_rollback_drops_last_turn_from_history() { expected.extend(turn_1); let history = sess.clone_history().await; - assert_eq!(expected, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &expected); assert_eq!(sess.previous_turn_settings().await, None); assert!(sess.reference_context_item().await.is_none()); @@ -2700,7 +2802,7 @@ async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns() assert_eq!(rollback_event.num_turns, 99); let history = sess.clone_history().await; - assert_eq!(initial_context, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &initial_context); } #[tokio::test] @@ -2722,7 +2824,7 @@ async fn thread_rollback_fails_without_persisted_thread_history() { error_event.codex_error_info, Some(CodexErrorInfo::ThreadRollbackFailed) ); - assert_eq!(sess.clone_history().await.raw_items(), initial_context); + assert_response_items_eq_ignoring_ids(sess.clone_history().await.raw_items(), &initial_context); } #[tokio::test] @@ -2826,9 +2928,9 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 1); - assert_eq!( + assert_response_items_eq_ignoring_ids( sess.clone_history().await.raw_items(), - vec![turn_one_user, turn_one_assistant] + &[turn_one_user, turn_one_assistant], ); assert_eq!( sess.previous_turn_settings().await, @@ -2956,7 +3058,10 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio let rollback_event = wait_for_thread_rolled_back(&rx).await; assert_eq!(rollback_event.num_turns, 1); - assert_eq!(sess.clone_history().await.raw_items(), compacted_history); + assert_response_items_eq_ignoring_ids( + sess.clone_history().await.raw_items(), + &compacted_history, + ); assert!(sess.reference_context_item().await.is_none()); } @@ -3061,12 +3166,12 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() { let second_rollback = wait_for_thread_rolled_back(&rx).await; assert_eq!(second_rollback.num_turns, 1); - assert_eq!( + assert_response_items_eq_ignoring_ids( sess.clone_history().await.raw_items(), - vec![ + &[ user_message("turn 1 user"), - assistant_message("turn 1 assistant") - ] + assistant_message("turn 1 assistant"), + ], ); let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path) @@ -3101,7 +3206,7 @@ async fn thread_rollback_fails_when_turn_in_progress() { ); let history = sess.clone_history().await; - assert_eq!(initial_context, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &initial_context); } #[tokio::test] @@ -3122,7 +3227,7 @@ async fn thread_rollback_fails_when_num_turns_is_zero() { ); let history = sess.clone_history().await; - assert_eq!(initial_context, history.raw_items()); + assert_response_items_eq_ignoring_ids(history.raw_items(), &initial_context); } #[tokio::test] @@ -7850,7 +7955,7 @@ async fn handle_output_item_done_records_image_save_history_message() { image_output_path.display(), ), ); - assert_eq!(history.raw_items(), &[image_message, item]); + assert_response_items_eq_ignoring_ids(history.raw_items(), &[image_message, item]); assert_eq!( std::fs::read(&expected_saved_path).expect("saved file"), b"foo" @@ -7999,7 +8104,7 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont .await; let history = session.clone_history().await; let initial_context = session.build_initial_context(&turn_context).await; - assert_eq!(history.raw_items().to_vec(), initial_context); + assert_response_items_eq_ignoring_ids(history.raw_items(), &initial_context); let current_context = session.reference_context_item().await; assert_eq!( @@ -8045,7 +8150,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co let history = session.clone_history().await; let mut expected_history = vec![compacted_summary]; expected_history.extend(session.build_initial_context(&turn_context).await); - assert_eq!(history.raw_items().to_vec(), expected_history); + assert_response_items_eq_ignoring_ids(history.raw_items(), &expected_history); } #[tokio::test] @@ -8616,9 +8721,15 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() }], phase: None, }; + let pending_item = history + .raw_items() + .iter() + .find(|item| response_item_eq_ignoring_id(item, &expected)) + .expect("expected pending input to be persisted into history on turn completion"); assert!( - history.raw_items().iter().any(|item| item == &expected), - "expected pending input to be persisted into history on turn completion" + pending_item + .id() + .is_some_and(|item_id| item_id.starts_with("msg_")) ); let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) @@ -8785,7 +8896,7 @@ async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() { .expect_err("active turn should reject idle-only input"); assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason()); - assert_eq!(vec![item], err.into_input()); + assert_response_items_eq_ignoring_ids(&err.into_input(), &[item]); assert_eq!( Vec::::new(), sess.input_queue.get_pending_input(&sess.active_turn).await @@ -8811,7 +8922,7 @@ async fn try_start_turn_if_idle_rejects_plan_mode_without_injecting() { .expect_err("plan mode should reject automatic idle input"); assert_eq!(TryStartTurnIfIdleRejectionReason::PlanMode, err.reason()); - assert_eq!(vec![item], err.into_input()); + assert_response_items_eq_ignoring_ids(&err.into_input(), &[item]); assert!(sess.active_turn.lock().await.is_none()); assert_eq!( Vec::::new(), @@ -8842,7 +8953,7 @@ async fn try_start_turn_if_idle_rejects_pending_trigger_turn_without_injecting() TryStartTurnIfIdleRejectionReason::PendingTriggerTurn, err.reason() ); - assert_eq!(vec![item], err.into_input()); + assert_response_items_eq_ignoring_ids(&err.into_input(), &[item]); assert!(sess.active_turn.lock().await.is_none()); assert!(sess.input_queue.has_trigger_turn_mailbox_items().await); } @@ -8867,7 +8978,7 @@ async fn try_start_turn_if_idle_rejects_active_review_turn_without_injecting() { .expect_err("active review turn should reject automatic idle input"); assert_eq!(TryStartTurnIfIdleRejectionReason::Busy, err.reason()); - assert_eq!(vec![item], err.into_input()); + assert_response_items_eq_ignoring_ids(&err.into_input(), &[item]); assert_eq!( Vec::::new(), sess.input_queue.get_pending_input(&sess.active_turn).await @@ -9110,10 +9221,10 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { sess.abort_all_tasks(TurnAbortReason::Replaced).await; - assert_eq!( + assert_turn_inputs_eq_ignoring_response_item_ids( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![TurnInput::ResponseItem(ResponseItem::from( - communication.to_response_input_item() + communication.to_response_input_item(), ))], ); } @@ -9193,7 +9304,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { .await .expect("steered input should be accepted"); - assert_eq!( + assert_turn_inputs_eq_ignoring_response_item_ids( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![ TurnInput::UserInput { @@ -9201,7 +9312,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { text: "follow up".to_string(), text_elements: Vec::new(), }], - client_id: None + client_id: None, }, TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), ], @@ -9251,7 +9362,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { .defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id) .await; - assert_eq!( + assert_turn_inputs_eq_ignoring_response_item_ids( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![ TurnInput::UserInput { @@ -9259,7 +9370,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { text: "follow up".to_string(), text_elements: Vec::new(), }], - client_id: None + client_id: None, }, TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), ], @@ -9314,10 +9425,10 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { assert!(output.needs_follow_up); assert!(output.tool_future.is_some()); - assert_eq!( + assert_turn_inputs_eq_ignoring_response_item_ids( sess.input_queue.get_pending_input(&sess.active_turn).await, vec![TurnInput::ResponseItem(ResponseItem::from( - communication.to_response_input_item() + communication.to_response_input_item(), ))], ); } @@ -9509,28 +9620,24 @@ async fn sample_rollout( reconstruction_turn.truncation_policy, ); - let user1 = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { + let user1 = ResponseItem::new_message( + "user", + vec![ContentItem::InputText { text: "first user".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&user1), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user1.clone())); - let assistant1 = ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { + let assistant1 = ResponseItem::new_message( + "assistant", + vec![ContentItem::OutputText { text: "assistant reply one".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&assistant1), reconstruction_turn.truncation_policy, @@ -9543,34 +9650,30 @@ async fn sample_rollout( .for_prompt(&reconstruction_turn.model_info.input_modalities); let user_messages1 = collect_user_messages(&snapshot1); let rebuilt1 = compact::build_compacted_history(Vec::new(), &user_messages1, summary1); - live_history.replace(rebuilt1); + live_history.replace(rebuilt1.clone()); rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary1.to_string(), - replacement_history: None, + replacement_history: Some(rebuilt1), })); - let user2 = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { + let user2 = ResponseItem::new_message( + "user", + vec![ContentItem::InputText { text: "second user".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&user2), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user2.clone())); - let assistant2 = ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { + let assistant2 = ResponseItem::new_message( + "assistant", + vec![ContentItem::OutputText { text: "assistant reply two".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&assistant2), reconstruction_turn.truncation_policy, @@ -9583,34 +9686,30 @@ async fn sample_rollout( .for_prompt(&reconstruction_turn.model_info.input_modalities); let user_messages2 = collect_user_messages(&snapshot2); let rebuilt2 = compact::build_compacted_history(Vec::new(), &user_messages2, summary2); - live_history.replace(rebuilt2); + live_history.replace(rebuilt2.clone()); rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary2.to_string(), - replacement_history: None, + replacement_history: Some(rebuilt2), })); - let user3 = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { + let user3 = ResponseItem::new_message( + "user", + vec![ContentItem::InputText { text: "third user".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&user3), reconstruction_turn.truncation_policy, ); rollout_items.push(RolloutItem::ResponseItem(user3)); - let assistant3 = ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { + let assistant3 = ResponseItem::new_message( + "assistant", + vec![ContentItem::OutputText { text: "assistant reply three".to_string(), }], - phase: None, - }; + ); live_history.record_items( std::iter::once(&assistant3), reconstruction_turn.truncation_policy, diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 81beea5be1..7a977c52f4 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -620,40 +620,8 @@ fn completed_item_defers_mailbox_delivery_to_next_turn( pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option { match input { - ResponseInputItem::FunctionCallOutput { call_id, output } => { - Some(ResponseItem::FunctionCallOutput { - call_id: call_id.clone(), - output: output.clone(), - }) - } - ResponseInputItem::CustomToolCallOutput { - call_id, - name, - output, - } => Some(ResponseItem::CustomToolCallOutput { - call_id: call_id.clone(), - name: name.clone(), - output: output.clone(), - }), - ResponseInputItem::McpToolCallOutput { call_id, output } => { - let output = output.as_function_call_output_payload(); - Some(ResponseItem::FunctionCallOutput { - call_id: call_id.clone(), - output, - }) - } - ResponseInputItem::ToolSearchOutput { - call_id, - status, - execution, - tools, - } => Some(ResponseItem::ToolSearchOutput { - call_id: Some(call_id.clone()), - status: status.clone(), - execution: execution.clone(), - tools: tools.clone(), - }), - _ => None, + ResponseInputItem::Message { .. } => None, + _ => Some(input.clone().into()), } } diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index 9c4be94576..b8c3bf9c76 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/codex-rs/core/src/stream_events_utils_tests.rs @@ -61,6 +61,7 @@ fn external_context_pollution_items_include_web_search_and_tool_search() { arguments: serde_json::json!({"query": "calendar"}), }, ResponseItem::ToolSearchOutput { + id: None, call_id: Some("search-1".to_string()), status: "completed".to_string(), execution: "client".to_string(), @@ -98,6 +99,7 @@ fn external_context_pollution_items_exclude_local_tool_calls() { call_id: "call-1".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), }, @@ -109,6 +111,7 @@ fn external_context_pollution_items_exclude_local_tool_calls() { input: "*** Begin Patch\n*** End Patch\n".to_string(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "custom-1".to_string(), name: Some("apply_patch".to_string()), output: FunctionCallOutputPayload::from_text("ok".to_string()), diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 911f30b319..e546186d75 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -100,14 +100,12 @@ pub(crate) fn interrupted_turn_history_marker( let marker = crate::context::TurnAborted::new( crate::context::TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE, ); - Some(ResponseItem::Message { - id: None, - role: "developer".to_string(), - content: vec![ContentItem::InputText { + Some(ResponseItem::new_message( + "developer", + vec![ContentItem::InputText { text: marker.render(), }], - phase: None, - }) + )) } } } diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 9f7eba9573..e2db114c41 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -216,8 +216,6 @@ pub(crate) async fn exit_review_mode( review_output: Option, ctx: Arc, ) { - const REVIEW_USER_MESSAGE_ID: &str = "review_rollout_user"; - const REVIEW_ASSISTANT_MESSAGE_ID: &str = "review_rollout_assistant"; let (user_message, assistant_message) = if let Some(out) = review_output.clone() { let mut findings_str = String::new(); let text = out.overall_explanation.trim(); @@ -242,12 +240,10 @@ pub(crate) async fn exit_review_mode( session .record_conversation_items( &ctx, - &[ResponseItem::Message { - id: Some(REVIEW_USER_MESSAGE_ID.to_string()), - role: "user".to_string(), - content: vec![ContentItem::InputText { text: user_message }], - phase: None, - }], + &[ResponseItem::new_message( + "user", + vec![ContentItem::InputText { text: user_message }], + )], ) .await; @@ -260,14 +256,12 @@ pub(crate) async fn exit_review_mode( session .record_response_item_and_emit_turn_item( ctx.as_ref(), - ResponseItem::Message { - id: Some(REVIEW_ASSISTANT_MESSAGE_ID.to_string()), - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { + ResponseItem::new_message( + "assistant", + vec![ContentItem::OutputText { text: assistant_message, }], - phase: None, - }, + ), ) .await; diff --git a/codex-rs/core/src/tools/code_mode/delegate.rs b/codex-rs/core/src/tools/code_mode/delegate.rs index 06bb71a280..df1b8de882 100644 --- a/codex-rs/core/src/tools/code_mode/delegate.rs +++ b/codex-rs/core/src/tools/code_mode/delegate.rs @@ -298,11 +298,11 @@ impl CoreTurnHost { } self.exec .session - .inject_if_running(vec![ResponseItem::CustomToolCallOutput { + .inject_if_running(vec![ResponseItem::new_named_custom_tool_call_output( call_id, - name: Some(PUBLIC_TOOL_NAME.to_string()), - output: FunctionCallOutputPayload::from_text(text), - }]) + PUBLIC_TOOL_NAME, + FunctionCallOutputPayload::from_text(text), + )]) .await .map_err(|_| { format!("failed to inject exec notify message for cell {cell_id}: no active turn") diff --git a/codex-rs/core/src/tools/context_tests.rs b/codex-rs/core/src/tools/context_tests.rs index 6ac5b6978f..5daf1883e3 100644 --- a/codex-rs/core/src/tools/context_tests.rs +++ b/codex-rs/core/src/tools/context_tests.rs @@ -35,7 +35,9 @@ fn function_payloads_remain_function_outputs() { .to_response_item("fn-1", &payload); match response { - ResponseInputItem::FunctionCallOutput { call_id, output } => { + ResponseInputItem::FunctionCallOutput { + call_id, output, .. + } => { assert_eq!(call_id, "fn-1"); assert_eq!(output.content_items(), None); assert_eq!(output.body.to_text().as_deref(), Some("ok")); @@ -111,7 +113,9 @@ fn mcp_tool_output_response_item_includes_wall_time() { ); match response { - ResponseInputItem::FunctionCallOutput { call_id, output } => { + ResponseInputItem::FunctionCallOutput { + call_id, output, .. + } => { assert_eq!(call_id, "mcp-call-1"); assert_eq!(output.success, Some(true)); let Some(text) = output.body.to_text() else { @@ -163,7 +167,9 @@ fn mcp_tool_output_response_item_truncates_large_structured_content() { ); match response { - ResponseInputItem::FunctionCallOutput { call_id, output } => { + ResponseInputItem::FunctionCallOutput { + call_id, output, .. + } => { assert_eq!(call_id, "mcp-call-large"); assert_eq!(output.success, Some(true)); let text = output @@ -348,6 +354,7 @@ fn tool_search_payloads_roundtrip_as_tool_search_outputs() { status, execution, tools, + .. } => { assert_eq!(call_id, "search-1"); assert_eq!(status, "completed"); @@ -439,7 +446,9 @@ fn exec_command_tool_output_formats_truncated_response() { .to_response_item("call-42", &payload); match response { - ResponseInputItem::FunctionCallOutput { call_id, output } => { + ResponseInputItem::FunctionCallOutput { + call_id, output, .. + } => { assert_eq!(call_id, "call-42"); assert_eq!(output.success, Some(true)); let text = output diff --git a/codex-rs/core/src/tools/handlers/extension_tools.rs b/codex-rs/core/src/tools/handlers/extension_tools.rs index ecbf772e90..5d04bb5f48 100644 --- a/codex-rs/core/src/tools/handlers/extension_tools.rs +++ b/codex-rs/core/src/tools/handlers/extension_tools.rs @@ -287,14 +287,12 @@ mod tests { let turn_id = turn.sub_id.clone(); let model = turn.model_info.slug.clone(); let truncation_policy = turn.truncation_policy; - let history_item = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { + let history_item = ResponseItem::new_message( + "user", + vec![ContentItem::InputText { text: "extension history".to_string(), }], - phase: None, - }; + ); session .record_conversation_items(&turn, std::slice::from_ref(&history_item)) .await; @@ -331,9 +329,18 @@ mod tests { ); assert_eq!(captured_call.model, model); assert_eq!(captured_call.truncation_policy, truncation_policy); + let history_item_id = captured_call.conversation_history.items()[0] + .id() + .expect("recorded history item should have an id"); + assert!(history_item_id.starts_with("msg_")); + let mut expected_history_item = history_item; + let ResponseItem::Message { id, .. } = &mut expected_history_item else { + panic!("test history item should be a message"); + }; + *id = Some(history_item_id.to_string()); assert_eq!( captured_call.conversation_history.items(), - std::slice::from_ref(&history_item) + std::slice::from_ref(&expected_history_item) ); match captured_call.payload { ToolPayload::Function { arguments } => { diff --git a/codex-rs/core/src/tools/router_tests.rs b/codex-rs/core/src/tools/router_tests.rs index cece7fc171..b8e0a82087 100644 --- a/codex-rs/core/src/tools/router_tests.rs +++ b/codex-rs/core/src/tools/router_tests.rs @@ -379,7 +379,9 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow let response = result.into_response(); match response { - ResponseInputItem::FunctionCallOutput { call_id, output } => { + ResponseInputItem::FunctionCallOutput { + call_id, output, .. + } => { assert_eq!(call_id, "call-extension"); let FunctionCallOutputBody::Text(text) = output.body else { panic!("expected text function call output") diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index 7146f537be..450924f0c9 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -145,6 +145,7 @@ fn response_item_records_turn_ttft_ignores_empty_non_output_items() { })); assert!(!response_item_records_turn_ttft( &ResponseItem::FunctionCallOutput { + id: None, call_id: "call-1".to_string(), output: FunctionCallOutputPayload::from_text("ok".to_string()), } diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 9a6f104d0c..fb242c4dae 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -290,14 +290,15 @@ async fn resume_includes_initial_messages_and_sends_prior_items() { // Prior item: user message (should be delivered) let prior_user = codex_protocol::models::ResponseItem::Message { - id: None, + id: Some("msg_resumed_user".to_string()), role: "user".to_string(), content: vec![codex_protocol::models::ContentItem::InputText { text: "resumed user message".to_string(), }], phase: None, }; - let prior_user_json = serde_json::to_value(&prior_user).unwrap(); + let mut prior_user_json = serde_json::to_value(&prior_user).unwrap(); + prior_user.attach_id_to_json(&mut prior_user_json); writeln!( f, "{}", @@ -311,14 +312,15 @@ async fn resume_includes_initial_messages_and_sends_prior_items() { // Prior item: system message (excluded from API history) let prior_system = codex_protocol::models::ResponseItem::Message { - id: None, + id: Some("msg_resumed_system".to_string()), role: "system".to_string(), content: vec![codex_protocol::models::ContentItem::OutputText { text: "resumed system instruction".to_string(), }], phase: None, }; - let prior_system_json = serde_json::to_value(&prior_system).unwrap(); + let mut prior_system_json = serde_json::to_value(&prior_system).unwrap(); + prior_system.attach_id_to_json(&mut prior_system_json); writeln!( f, "{}", @@ -332,14 +334,15 @@ async fn resume_includes_initial_messages_and_sends_prior_items() { // Prior item: assistant message let prior_item = codex_protocol::models::ResponseItem::Message { - id: None, + id: Some("msg_resumed_assistant".to_string()), role: "assistant".to_string(), content: vec![codex_protocol::models::ContentItem::OutputText { text: "resumed assistant message".to_string(), }], phase: Some(MessagePhase::Commentary), }; - let prior_item_json = serde_json::to_value(&prior_item).unwrap(); + let mut prior_item_json = serde_json::to_value(&prior_item).unwrap(); + prior_item.attach_id_to_json(&mut prior_item_json); writeln!( f, "{}", @@ -403,6 +406,15 @@ async fn resume_includes_initial_messages_and_sends_prior_items() { let request = resp_mock.single_request(); let request_body = request.body_json(); let input = request_body["input"].as_array().expect("input array"); + let message_item = |role: &str, text: &str| { + input + .iter() + .find(|item| { + item.get("role").and_then(|role| role.as_str()) == Some(role) + && message_input_texts(item).contains(&text) + }) + .expect("message item") + }; let mut messages: Vec<(String, String)> = Vec::new(); for item in input { let Some(role) = item.get("role").and_then(|role| role.as_str()) else { @@ -420,19 +432,23 @@ async fn resume_includes_initial_messages_and_sends_prior_items() { .iter() .position(|(role, text)| role == "assistant" && text == "resumed assistant message") .expect("prior assistant message"); - let prior_assistant = input - .iter() - .find(|item| { - item.get("role").and_then(|role| role.as_str()) == Some("assistant") - && item - .get("content") - .and_then(|content| content.as_array()) - .and_then(|content| content.first()) - .and_then(|entry| entry.get("text")) - .and_then(|text| text.as_str()) - == Some("resumed assistant message") - }) - .expect("resumed assistant message request item"); + let prior_user = message_item("user", "resumed user message"); + let prior_assistant = message_item("assistant", "resumed assistant message"); + let new_user = message_item("user", "hello"); + assert_eq!( + prior_user.get("id").and_then(|id| id.as_str()), + Some("msg_resumed_user") + ); + assert_eq!( + prior_assistant.get("id").and_then(|id| id.as_str()), + Some("msg_resumed_assistant") + ); + assert!( + new_user + .get("id") + .and_then(|id| id.as_str()) + .is_some_and(|id| id.starts_with("msg_")) + ); assert_eq!( prior_assistant .get("phase") @@ -507,6 +523,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() { RolloutLine { timestamp: "2024-01-01T00:00:02.000Z".to_string(), item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput { + id: None, call_id: "legacy-js-call".to_string(), name: None, output: FunctionCallOutputPayload::from_text("legacy js_repl stdout".to_string()), @@ -644,6 +661,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { RolloutLine { timestamp: "2024-01-01T00:00:01.500Z".to_string(), item: RolloutItem::ResponseItem(ResponseItem::FunctionCallOutput { + id: None, call_id: function_call_id.to_string(), output: FunctionCallOutputPayload::from_content_items(vec![ FunctionCallOutputContentItem::InputImage { @@ -666,6 +684,7 @@ async fn resume_replays_image_tool_outputs_with_detail() { RolloutLine { timestamp: "2024-01-01T00:00:02.500Z".to_string(), item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput { + id: None, call_id: custom_call_id.to_string(), name: None, output: FunctionCallOutputPayload::from_content_items(vec![ @@ -2437,6 +2456,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { call_id: "function-call-id".into(), }); prompt.input.push(ResponseItem::FunctionCallOutput { + id: None, call_id: "function-call-id".into(), output: FunctionCallOutputPayload::from_text("ok".into()), }); @@ -2460,6 +2480,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { input: "{}".into(), }); prompt.input.push(ResponseItem::CustomToolCallOutput { + id: None, call_id: "custom-tool-call-id".into(), name: None, output: FunctionCallOutputPayload::from_text("ok".into()), @@ -3105,8 +3126,7 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() { // each time: a few deltas, then a final assistant message, then completed. let server = MockServer::start().await; - // Build a small SSE stream with deltas and a final assistant message. - // We emit the same body for all 3 turns. + // Build small SSE streams with deltas and distinct final assistant items. let sse1 = sse(vec![ ev_message_item_added("msg-1", ""), ev_output_text_delta("Hey "), @@ -3115,8 +3135,24 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() { ev_assistant_message("msg-1", "Hey there!\n"), ev_completed("resp1"), ]); + let sse2 = sse(vec![ + ev_message_item_added("msg-2", ""), + ev_output_text_delta("Second "), + ev_output_text_delta("reply"), + ev_output_text_delta("!\n"), + ev_assistant_message("msg-2", "Second reply!\n"), + ev_completed("resp2"), + ]); + let sse3 = sse(vec![ + ev_message_item_added("msg-3", ""), + ev_output_text_delta("Third "), + ev_output_text_delta("reply"), + ev_output_text_delta("!\n"), + ev_assistant_message("msg-3", "Third reply!\n"), + ev_completed("resp3"), + ]); - let request_log = mount_sse_sequence(&server, vec![sse1.clone(), sse1.clone(), sse1]).await; + let request_log = mount_sse_sequence(&server, vec![sse1, sse2, sse3]).await; let mut builder = test_codex().with_auth(CodexAuth::from_api_key("Test API Key")); let codex = builder @@ -3180,47 +3216,63 @@ async fn history_dedupes_streamed_and_final_messages_across_turns() { assert_eq!(request.path(), "/v1/responses"); } - // Replace full-array compare with tail-only raw JSON compare using a single hard-coded value. - let r3_tail_expected = json!([ - { - "type": "message", - "role": "user", - "content": [{"type":"input_text","text":"U1"}] - }, - { - "type": "message", - "role": "assistant", - "content": [{"type":"output_text","text":"Hey there!\n"}] - }, - { - "type": "message", - "role": "user", - "content": [{"type":"input_text","text":"U2"}] - }, - { - "type": "message", - "role": "assistant", - "content": [{"type":"output_text","text":"Hey there!\n"}] - }, - { - "type": "message", - "role": "user", - "content": [{"type":"input_text","text":"U3"}] - } - ]); + let message_id = |request: &ResponsesRequest, role: &str, text: &str| { + request + .inputs_of_type("message") + .into_iter() + .find(|item| { + item.get("role").and_then(serde_json::Value::as_str) == Some(role) + && item + .get("content") + .and_then(serde_json::Value::as_array) + .is_some_and(|content| { + content.iter().any(|entry| { + entry.get("text").and_then(serde_json::Value::as_str) == Some(text) + }) + }) + }) + .and_then(|item| item.get("id").cloned()) + .and_then(|id| id.as_str().map(str::to_string)) + .expect("message id") + }; + let message_ids_by_role = |request: &ResponsesRequest, role: &str| { + request + .inputs_of_type("message") + .into_iter() + .filter(|item| item.get("role").and_then(serde_json::Value::as_str) == Some(role)) + .map(|item| { + item.get("id") + .and_then(serde_json::Value::as_str) + .expect("message id") + .to_string() + }) + .collect::>() + }; + let u1_id = message_id(&requests[0], "user", "U1"); + let u2_id = message_id(&requests[1], "user", "U2"); + let u3_id = message_id(&requests[2], "user", "U3"); + let developer_ids = message_ids_by_role(&requests[0], "developer"); - let r3_input_array = requests[2] - .body_json() - .get("input") - .and_then(|v| v.as_array()) - .cloned() - .expect("r3 missing input array"); - // skipping earlier context and developer messages - let tail_len = r3_tail_expected.as_array().unwrap().len(); - let actual_tail = &r3_input_array[r3_input_array.len() - tail_len..]; + assert!(u1_id.starts_with("msg_")); + assert!(u2_id.starts_with("msg_")); + assert!(u3_id.starts_with("msg_")); + assert!(!developer_ids.is_empty()); + assert!(developer_ids.iter().all(|id| id.starts_with("msg_"))); + assert!(message_ids_by_role(&requests[1], "developer").starts_with(&developer_ids)); + assert!(message_ids_by_role(&requests[2], "developer").starts_with(&developer_ids)); + assert_eq!(message_id(&requests[1], "user", "U1"), u1_id); + assert_eq!(message_id(&requests[2], "user", "U1"), u1_id); + assert_eq!(message_id(&requests[2], "user", "U2"), u2_id); assert_eq!( - serde_json::Value::Array(actual_tail.to_vec()), - r3_tail_expected, - "request 3 tail mismatch", + message_id(&requests[1], "assistant", "Hey there!\n"), + "msg-1" + ); + assert_eq!( + message_id(&requests[2], "assistant", "Hey there!\n"), + "msg-1" + ); + assert_eq!( + message_id(&requests[2], "assistant", "Second reply!\n"), + "msg-2" ); } diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 4dc674ec30..14b372fb5b 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -1644,6 +1644,81 @@ async fn responses_websocket_creates_on_non_prefix() { server.shutdown().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn responses_websocket_openai_sends_item_ids_for_incremental_and_full_create() { + skip_if_no_network!(); + + let server = start_websocket_server(vec![vec![ + vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg_assistant_1", "assistant output 1"), + ev_completed("resp-1"), + ], + vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg_assistant_2", "assistant output 2"), + ev_completed("resp-2"), + ], + vec![ev_response_created("resp-3"), ev_completed("resp-3")], + ]]) + .await; + + let harness = websocket_harness_with_provider_options( + openai_websocket_provider(&server), + /*runtime_metrics_enabled*/ false, + ) + .await; + let mut session = harness.client.new_session(); + let prompt_one = prompt_with_input_and_instructions( + vec![message_item_with_id("msg_user_1", "hello")], + "base instructions one", + ); + let prompt_two = prompt_with_input_and_instructions( + vec![ + message_item_with_id("msg_user_1", "hello"), + assistant_message_item("msg_assistant_1", "assistant output 1"), + message_item_with_id("msg_user_2", "second"), + ], + "base instructions one", + ); + let prompt_three = prompt_with_input_and_instructions( + vec![ + message_item_with_id("msg_user_1", "hello"), + assistant_message_item("msg_assistant_1", "assistant output 1"), + message_item_with_id("msg_user_2", "second"), + assistant_message_item("msg_assistant_2", "assistant output 2"), + message_item_with_id("msg_user_3", "third"), + ], + "base instructions two", + ); + + stream_until_complete(&mut session, &harness, &prompt_one).await; + stream_until_complete(&mut session, &harness, &prompt_two).await; + stream_until_complete(&mut session, &harness, &prompt_three).await; + + let second = server + .single_connection() + .get(1) + .expect("missing second request") + .body_json(); + assert_eq!(second["previous_response_id"].as_str(), Some("resp-1")); + assert_eq!(second["input"][0]["id"].as_str(), Some("msg_user_2")); + + let third = server + .single_connection() + .get(2) + .expect("missing third request") + .body_json(); + assert_eq!(third.get("previous_response_id"), None); + assert_eq!(third["input"][0]["id"].as_str(), Some("msg_user_1")); + assert_eq!(third["input"][1]["id"].as_str(), Some("msg_assistant_1")); + assert_eq!(third["input"][2]["id"].as_str(), Some("msg_user_2")); + assert_eq!(third["input"][3]["id"].as_str(), Some("msg_assistant_2")); + assert_eq!(third["input"][4]["id"].as_str(), Some("msg_user_3")); + + server.shutdown().await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn responses_websocket_creates_when_non_input_request_fields_change() { skip_if_no_network!(); @@ -1948,6 +2023,15 @@ fn message_item(text: &str) -> ResponseItem { } } +fn message_item_with_id(id: &str, text: &str) -> ResponseItem { + ResponseItem::Message { + id: Some(id.to_string()), + role: "user".into(), + content: vec![ContentItem::InputText { text: text.into() }], + phase: None, + } +} + fn assistant_message_item(id: &str, text: &str) -> ResponseItem { ResponseItem::Message { id: Some(id.to_string()), @@ -1975,6 +2059,16 @@ fn websocket_provider(server: &WebSocketTestServer) -> ModelProviderInfo { websocket_provider_with_connect_timeout(server, /*websocket_connect_timeout_ms*/ None) } +fn openai_websocket_provider(server: &WebSocketTestServer) -> ModelProviderInfo { + let mut provider = + ModelProviderInfo::create_openai_provider(Some(format!("{}/v1", server.uri()))); + provider.request_max_retries = Some(0); + provider.stream_max_retries = Some(0); + provider.stream_idle_timeout_ms = Some(5_000); + provider.supports_websockets = true; + provider +} + fn websocket_provider_with_connect_timeout( server: &WebSocketTestServer, websocket_connect_timeout_ms: Option, diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index ac497aff3b..56c404c7d7 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -1903,6 +1903,7 @@ async fn auto_compact_runs_after_resume_when_token_usage_is_over_limit() { phase: None, }, codex_protocol::models::ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; @@ -3450,6 +3451,7 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { phase: None, }, codex_protocol::models::ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; @@ -3574,6 +3576,7 @@ async fn auto_compact_runs_when_reasoning_header_clears_between_turns() { phase: None, }, codex_protocol::models::ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index 73a500f04d..4497803924 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -156,6 +156,7 @@ fn format_labeled_requests_snapshot( fn compacted_summary_only_output(summary: &str) -> Vec { vec![ResponseItem::Compaction { + id: None, encrypted_content: summary_with_prefix(summary), }] } @@ -315,6 +316,7 @@ async fn remote_compact_replaces_history_for_followups() -> Result<()> { .await; let compacted_history = vec![ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }]; let compact_mock = responses::mount_compact_json_once( @@ -2330,6 +2332,7 @@ async fn remote_compact_persists_replacement_history_in_rollout() -> Result<()> let compacted_history = vec![ ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ResponseItem::Message { @@ -2387,7 +2390,10 @@ async fn remote_compact_persists_replacement_history_in_rollout() -> Result<()> let has_compaction_item = replacement_history.iter().any(|item| { matches!( item, - ResponseItem::Compaction { encrypted_content } + ResponseItem::Compaction { + encrypted_content, + .. + } if encrypted_content == "ENCRYPTED_COMPACTION_SUMMARY" ) }); @@ -2480,6 +2486,7 @@ async fn remote_compact_and_resume_refresh_stale_developer_instructions() -> Res phase: None, }, ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; @@ -2620,6 +2627,7 @@ async fn remote_compact_refreshes_stale_developer_instructions_without_resume() phase: None, }, ResponseItem::Compaction { + id: None, encrypted_content: "ENCRYPTED_COMPACTION_SUMMARY".to_string(), }, ]; @@ -3676,6 +3684,7 @@ async fn snapshot_request_shape_remote_mid_turn_compaction_summary_only_reinject .await; let compacted_history = vec![ResponseItem::Compaction { + id: None, encrypted_content: summary_with_prefix("REMOTE_SUMMARY_ONLY"), }]; let compact_mock = responses::mount_compact_json_once( diff --git a/codex-rs/core/tests/suite/compact_remote_parity.rs b/codex-rs/core/tests/suite/compact_remote_parity.rs index 1b7c50936c..fd53e4aafd 100644 --- a/codex-rs/core/tests/suite/compact_remote_parity.rs +++ b/codex-rs/core/tests/suite/compact_remote_parity.rs @@ -762,13 +762,13 @@ fn compact_request_view(body: &Value, mode: Mode) -> Value { } let mut selected = selected_request_fields(body, SelectedFieldsMode::Compact); - selected["input"] = normalize_value(Value::Array(input)); + selected["input"] = normalize_response_items(Value::Array(input)); canonical_json(&normalize_value(selected)) } fn follow_up_request_view(body: &Value) -> Value { let mut selected = selected_request_fields(body, SelectedFieldsMode::FollowUp); - selected["input"] = normalize_value( + selected["input"] = normalize_response_items( body.get("input") .cloned() .expect("follow-up request should include input"), @@ -800,7 +800,9 @@ fn replacement_history_from_rollout(path: &Path) -> Result { } let replacement_history = replacement_history.expect("expected compacted rollout replacement history"); - Ok(canonical_json(&normalize_value(replacement_history))) + Ok(canonical_json(&normalize_response_items( + replacement_history, + ))) } fn write_manual_compact_hooks(home: &Path) { @@ -931,6 +933,16 @@ fn normalize_value(value: Value) -> Value { } } +fn normalize_response_items(mut value: Value) -> Value { + let Value::Array(items) = &mut value else { + panic!("response items should be an array"); + }; + for item in items { + remove_object_field(item, "id"); + } + normalize_value(value) +} + fn normalize_string(value: &str) -> String { if is_uuid_like(value) { return "".to_string(); diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 16412c2d27..ae28fb723d 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -1451,6 +1451,7 @@ async fn resumed_thread_runs_resume_then_compact_session_start_hooks() -> Result phase: None, }, ResponseItem::Compaction { + id: None, encrypted_content: "encrypted compact summary".to_string(), }, ]; diff --git a/codex-rs/core/tests/suite/image_rollout.rs b/codex-rs/core/tests/suite/image_rollout.rs index 199a7d1db6..1efbcabcd5 100644 --- a/codex-rs/core/tests/suite/image_rollout.rs +++ b/codex-rs/core/tests/suite/image_rollout.rs @@ -157,8 +157,13 @@ async fn copy_paste_local_image_persists_rollout_request_shape() -> anyhow::Resu .expect("expected user message with input image in rollout"); let image_url = extract_image_url(&actual).expect("expected image url in rollout"); + let actual_id = actual + .id() + .expect("expected user message id in rollout") + .to_string(); + assert!(actual_id.starts_with("msg_")); let expected = ResponseItem::Message { - id: None, + id: Some(actual_id), role: "user".to_string(), content: vec![ ContentItem::InputText { @@ -255,8 +260,13 @@ async fn drag_drop_image_persists_rollout_request_shape() -> anyhow::Result<()> .expect("expected user message with input image in rollout"); let image_url = extract_image_url(&actual).expect("expected image url in rollout"); + let actual_id = actual + .id() + .expect("expected user message id in rollout") + .to_string(); + assert!(actual_id.starts_with("msg_")); let expected = ResponseItem::Message { - id: None, + id: Some(actual_id), role: "user".to_string(), content: vec![ ContentItem::InputImage { diff --git a/codex-rs/core/tests/suite/prompt_caching.rs b/codex-rs/core/tests/suite/prompt_caching.rs index 80dee582ad..a97d50d267 100644 --- a/codex-rs/core/tests/suite/prompt_caching.rs +++ b/codex-rs/core/tests/suite/prompt_caching.rs @@ -47,6 +47,19 @@ fn text_user_input_parts(texts: Vec) -> serde_json::Value { }) } +fn generated_message_id(item: &serde_json::Value) -> &str { + let Some(id) = item["id"].as_str() else { + panic!("message id"); + }; + assert!(id.starts_with("msg_")); + id +} + +fn with_message_id(mut item: serde_json::Value, id: &str) -> serde_json::Value { + item["id"] = serde_json::json!(id); + item +} + fn assert_default_env_context(text: &str, cwd: &str) { assert_env_context_fragment(text); assert!( @@ -383,7 +396,13 @@ async fn prefixes_context_and_instructions_once_and_consistently_across_requests Some("input_text"), "expected environment context bundled after UI message in cached contextual message" ); - assert_eq!(input1[2], text_user_input("hello 1".to_string())); + assert_eq!( + input1[2], + with_message_id( + text_user_input("hello 1".to_string()), + generated_message_id(&input1[2]), + ) + ); let body2 = req2.single_request().body_json(); let input2 = body2["input"].as_array().expect("input array"); @@ -392,7 +411,13 @@ async fn prefixes_context_and_instructions_once_and_consistently_across_requests input1.as_slice(), "expected cached prefix to be reused" ); - assert_eq!(input2[input1.len()], text_user_input("hello 2".to_string())); + assert_eq!( + input2[input1.len()], + with_message_id( + text_user_input("hello 2".to_string()), + generated_message_id(&input2[input1.len()]), + ) + ); Ok(()) } @@ -492,13 +517,16 @@ async fn overrides_turn_context_but_keeps_cached_prefix_and_key_constant() -> an // The entire prefix from the first request should be identical and reused // as the prefix of the second request, ensuring cache hit potential. - let expected_user_message_2 = serde_json::json!({ - "type": "message", - "role": "user", - "content": [ { "type": "input_text", "text": "hello 2" } ] - }); let expected_permissions_msg = body1["input"][0].clone(); let body1_input = body1["input"].as_array().expect("input array"); + let expected_user_message_2 = with_message_id( + serde_json::json!({ + "type": "message", + "role": "user", + "content": [ { "type": "input_text", "text": "hello 2" } ] + }), + generated_message_id(&body2["input"][body1_input.len() + 2]), + ); // After overriding the thread settings, emit one updated permissions message. let expected_permissions_msg_2 = body2["input"][body1_input.len()].clone(); assert_ne!( @@ -780,13 +808,16 @@ async fn per_turn_overrides_keep_cached_prefix_and_key_constant() -> anyhow::Res // The entire prefix from the first request should be identical and reused // as the prefix of the second request. - let expected_user_message_2 = serde_json::json!({ - "type": "message", - "role": "user", - "content": [ { "type": "input_text", "text": "hello 2" } ] - }); let expected_permissions_msg = body1["input"][0].clone(); let body1_input = body1["input"].as_array().expect("input array"); + let expected_user_message_2 = with_message_id( + serde_json::json!({ + "type": "message", + "role": "user", + "content": [ { "type": "input_text", "text": "hello 2" } ] + }), + generated_message_id(&body2["input"][body1_input.len() + 2]), + ); let expected_settings_update_msg = body2["input"][body1_input.len()].clone(); assert_ne!( expected_settings_update_msg, expected_permissions_msg, @@ -931,14 +962,20 @@ async fn send_user_turn_with_no_changes_does_not_send_environment_context() -> a .to_string(); assert_default_env_context(&expected_env_text_1, &default_cwd_lossy); - let expected_contextual_user_msg_1 = text_user_input_parts(vec![ - expected_ui_msg["content"][0]["text"] - .as_str() - .expect("cached user instructions text") - .to_string(), - expected_env_text_1, - ]); - let expected_user_message_1 = text_user_input("hello 1".to_string()); + let expected_contextual_user_msg_1 = with_message_id( + text_user_input_parts(vec![ + expected_ui_msg["content"][0]["text"] + .as_str() + .expect("cached user instructions text") + .to_string(), + expected_env_text_1, + ]), + generated_message_id(&expected_ui_msg), + ); + let expected_user_message_1 = with_message_id( + text_user_input("hello 1".to_string()), + generated_message_id(&body1["input"][2]), + ); let expected_input_1 = serde_json::Value::Array(vec![ expected_permissions_msg.clone(), @@ -947,7 +984,10 @@ async fn send_user_turn_with_no_changes_does_not_send_environment_context() -> a ]); assert_eq!(body1["input"], expected_input_1); - let expected_user_message_2 = text_user_input("hello 2".to_string()); + let expected_user_message_2 = with_message_id( + text_user_input("hello 2".to_string()), + generated_message_id(&body2["input"][3]), + ); let expected_input_2 = serde_json::Value::Array(vec![ expected_permissions_msg, expected_contextual_user_msg_1, @@ -1073,14 +1113,20 @@ async fn send_user_turn_with_changes_sends_environment_context() -> anyhow::Resu .expect("cached environment context text") .to_string(); assert_default_env_context(&expected_env_text_1, &default_cwd.to_string_lossy()); - let expected_contextual_user_msg_1 = text_user_input_parts(vec![ - expected_ui_msg["content"][0]["text"] - .as_str() - .expect("cached user instructions text") - .to_string(), - expected_env_text_1, - ]); - let expected_user_message_1 = text_user_input("hello 1".to_string()); + let expected_contextual_user_msg_1 = with_message_id( + text_user_input_parts(vec![ + expected_ui_msg["content"][0]["text"] + .as_str() + .expect("cached user instructions text") + .to_string(), + expected_env_text_1, + ]), + generated_message_id(&expected_ui_msg), + ); + let expected_user_message_1 = with_message_id( + text_user_input("hello 1".to_string()), + generated_message_id(&body1["input"][2]), + ); let expected_input_1 = serde_json::Value::Array(vec![ expected_permissions_msg.clone(), expected_contextual_user_msg_1.clone(), @@ -1116,7 +1162,10 @@ async fn send_user_turn_with_changes_sends_environment_context() -> anyhow::Resu ), "expected disabled filesystem profile in environment context: {expected_env_update_text}" ); - let expected_user_message_2 = text_user_input("hello 2".to_string()); + let expected_user_message_2 = with_message_id( + text_user_input("hello 2".to_string()), + generated_message_id(&body2["input"][body1_input.len() + 2]), + ); let expected_input_2 = serde_json::Value::Array(vec![ expected_permissions_msg, expected_contextual_user_msg_1, diff --git a/codex-rs/core/tests/suite/prompt_debug_tests.rs b/codex-rs/core/tests/suite/prompt_debug_tests.rs index bad7e1ab5a..963051ea44 100644 --- a/codex-rs/core/tests/suite/prompt_debug_tests.rs +++ b/codex-rs/core/tests/suite/prompt_debug_tests.rs @@ -36,8 +36,13 @@ async fn build_prompt_input_includes_context_and_user_message() -> Result<()> { ) .await?; + let user_message_id = input + .last() + .and_then(ResponseItem::id) + .expect("debug prompt user message should have an id"); + assert!(user_message_id.starts_with("msg_")); let expected_user_message = ResponseItem::Message { - id: None, + id: Some(user_message_id.to_string()), role: "user".to_string(), content: vec![ContentItem::InputText { text: "hello from debug prompt".to_string(), diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap index 89796cb918..6ba5f0b7bc 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap @@ -22,6 +22,7 @@ Scenario: After five varied API-key-auth turns, remote manual compaction omits s + } + ], + "encrypted_content": "YmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYnR1cm4gZml2ZSByYXcgY29udGVudA==", ++ "id": "turn-five-reasoning", + "summary": [ + { + "text": "TURN_FIVE_REASONING", @@ -37,6 +38,7 @@ Scenario: After five varied API-key-auth turns, remote manual compaction omits s + "type": "output_text" + } + ], ++ "id": "turn-five-assistant", + "role": "assistant", + "type": "message" - "service_tier": "priority", diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap index ab6c9d1b37..fea321fe07 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap @@ -22,6 +22,7 @@ Scenario: After five varied ChatGPT-auth turns, remote manual compaction reuses + } + ], + "encrypted_content": "YmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYnR1cm4gZml2ZSByYXcgY29udGVudA==", ++ "id": "turn-five-reasoning", + "summary": [ + { + "text": "TURN_FIVE_REASONING", @@ -37,6 +38,7 @@ Scenario: After five varied ChatGPT-auth turns, remote manual compaction reuses + "type": "output_text" + } + ], ++ "id": "turn-five-assistant", + "role": "assistant", + "type": "message" - "store": false, diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 4d6cd17bb8..0e2db95b56 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -1085,9 +1085,16 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result .await? .pop() .expect("child request"); + let agent_messages = child_request.inputs_of_type("agent_message"); + let [agent_message] = agent_messages.as_slice() else { + panic!("expected one agent message input"); + }; + let agent_message_id = agent_message["id"].as_str().expect("agent message id"); + assert!(agent_message_id.starts_with("amsg_")); assert_eq!( - child_request.inputs_of_type("agent_message"), - vec![json!({ + agent_message, + &json!({ + "id": agent_message_id, "type": "agent_message", "author": "/root", "recipient": "/root/worker", @@ -1095,7 +1102,7 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result "type": "encrypted_content", "encrypted_content": encrypted_message, }], - })] + }) ); Ok(()) diff --git a/codex-rs/core/tests/suite/unified_exec.rs b/codex-rs/core/tests/suite/unified_exec.rs index caea5ecf55..40cb7406da 100644 --- a/codex-rs/core/tests/suite/unified_exec.rs +++ b/codex-rs/core/tests/suite/unified_exec.rs @@ -173,6 +173,7 @@ async fn wait_for_raw_unified_exec_output( ResponseItem::FunctionCallOutput { call_id: output_call_id, output, + .. } if output_call_id == call_id => output.text_content().map(str::to_string), _ => None, }, diff --git a/codex-rs/ext/image-generation/src/tests.rs b/codex-rs/ext/image-generation/src/tests.rs index 3f72972656..bc21329c76 100644 --- a/codex-rs/ext/image-generation/src/tests.rs +++ b/codex-rs/ext/image-generation/src/tests.rs @@ -83,6 +83,7 @@ fn recent_image_fallback_selects_newest_images_in_chronological_order() { call_id: "mcp-call".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "mcp-call".to_string(), output: image_output("mcp"), }, @@ -94,6 +95,7 @@ fn recent_image_fallback_selects_newest_images_in_chronological_order() { input: String::new(), }, ResponseItem::CustomToolCallOutput { + id: None, call_id: "code-mode-call".to_string(), name: Some("exec".to_string()), output: image_output("code-mode"), @@ -105,6 +107,7 @@ fn recent_image_fallback_selects_newest_images_in_chronological_order() { result: "generated".to_string(), }, ResponseItem::FunctionCallOutput { + id: None, call_id: "orphan-call".to_string(), output: image_output("orphan"), }, diff --git a/codex-rs/ext/image-generation/src/tool.rs b/codex-rs/ext/image-generation/src/tool.rs index 69b99bdbc4..616548e5c2 100644 --- a/codex-rs/ext/image-generation/src/tool.rs +++ b/codex-rs/ext/image-generation/src/tool.rs @@ -248,9 +248,9 @@ fn recent_images(history: &[ResponseItem], count: usize) -> Vec { ContentItem::InputText { .. } | ContentItem::OutputText { .. } => None, })); } - ResponseItem::FunctionCallOutput { call_id, output } - if function_call_ids.contains(call_id.as_str()) => - { + ResponseItem::FunctionCallOutput { + call_id, output, .. + } if function_call_ids.contains(call_id.as_str()) => { image_urls.extend(output_image_urls(output)); } ResponseItem::CustomToolCallOutput { diff --git a/codex-rs/memories/write/src/phase1.rs b/codex-rs/memories/write/src/phase1.rs index 21b20759b1..4a157718e9 100644 --- a/codex-rs/memories/write/src/phase1.rs +++ b/codex-rs/memories/write/src/phase1.rs @@ -721,6 +721,7 @@ mod tests { let serialized = job::serialize_filtered_rollout_response_items(&[RolloutItem::ResponseItem( ResponseItem::FunctionCallOutput { + id: None, call_id: "call_123".to_string(), output: codex_protocol::models::FunctionCallOutputPayload { body: codex_protocol::models::FunctionCallOutputBody::Text( diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index ee61dc3949..2b89a11df0 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -386,12 +386,7 @@ pub fn build_hook_prompt_message(fragments: &[HookPromptFragment]) -> Option, }, AgentMessage { + #[serde(default, skip_serializing)] + #[ts(skip)] + #[schemars(skip)] + id: Option, author: String, recipient: String, content: Vec, @@ -822,6 +828,10 @@ pub enum ResponseItem { // - an array of structured content items (`content_items`) // We keep this behavior centralized in `FunctionCallOutputPayload`. FunctionCallOutput { + #[serde(default, skip_serializing)] + #[ts(skip)] + #[schemars(skip)] + id: Option, call_id: String, #[ts(as = "FunctionCallOutputBody")] #[schemars(with = "FunctionCallOutputBody")] @@ -843,6 +853,10 @@ pub enum ResponseItem { // `function_call_output.output` so freeform tools can return either plain // text or structured content items. CustomToolCallOutput { + #[serde(default, skip_serializing)] + #[ts(skip)] + #[schemars(skip)] + id: Option, call_id: String, #[serde(default, skip_serializing_if = "Option::is_none")] #[ts(optional)] @@ -852,6 +866,10 @@ pub enum ResponseItem { output: FunctionCallOutputPayload, }, ToolSearchOutput { + #[serde(default, skip_serializing)] + #[ts(skip)] + #[schemars(skip)] + id: Option, call_id: Option, status: String, execution: String, @@ -896,6 +914,10 @@ pub enum ResponseItem { }, #[serde(alias = "compaction_summary")] Compaction { + #[serde(default, skip_serializing)] + #[ts(skip)] + #[schemars(skip)] + id: Option, encrypted_content: String, }, CompactionTrigger, @@ -908,13 +930,137 @@ pub enum ResponseItem { Other, } +/// Responses API item kinds for stable IDs Codex assigns to new local response items. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ResponseItemIdKind { + Message, + FunctionCallOutput, + CustomToolCallOutput, + ToolSearchOutput, +} + +impl ResponseItemIdKind { + fn prefix(self) -> &'static str { + match self { + Self::Message => "msg", + Self::FunctionCallOutput => "fco", + Self::CustomToolCallOutput => "ctco", + Self::ToolSearchOutput => "tso", + } + } + + /// Generates a new stable Responses API item ID with this kind's server-recognized prefix. + pub fn new_id(self) -> String { + format!("{}_{}", self.prefix(), Uuid::new_v4().simple()) + } +} + impl ResponseItem { + /// Constructs a new Codex-owned message with a fresh stable Responses API ID. + pub fn new_message(role: impl Into, content: Vec) -> Self { + Self::Message { + id: Some(ResponseItemIdKind::Message.new_id()), + role: role.into(), + content, + phase: None, + } + } + + /// Constructs a new Codex-owned function call output with a fresh stable Responses API ID. + pub fn new_function_call_output( + call_id: impl Into, + output: FunctionCallOutputPayload, + ) -> Self { + Self::FunctionCallOutput { + id: Some(ResponseItemIdKind::FunctionCallOutput.new_id()), + call_id: call_id.into(), + output, + } + } + + /// Constructs a new Codex-owned unnamed custom tool output with a fresh stable Responses API ID. + pub fn new_custom_tool_call_output( + call_id: impl Into, + output: FunctionCallOutputPayload, + ) -> Self { + Self::CustomToolCallOutput { + id: Some(ResponseItemIdKind::CustomToolCallOutput.new_id()), + call_id: call_id.into(), + name: None, + output, + } + } + + /// Constructs a new Codex-owned named custom tool output with a fresh stable Responses API ID. + pub fn new_named_custom_tool_call_output( + call_id: impl Into, + name: impl Into, + output: FunctionCallOutputPayload, + ) -> Self { + Self::CustomToolCallOutput { + id: Some(ResponseItemIdKind::CustomToolCallOutput.new_id()), + call_id: call_id.into(), + name: Some(name.into()), + output, + } + } + + /// Constructs a new Codex-owned tool search output with a fresh stable Responses API ID. + pub fn new_tool_search_output( + call_id: impl Into, + status: impl Into, + execution: impl Into, + tools: Vec, + ) -> Self { + Self::ToolSearchOutput { + id: Some(ResponseItemIdKind::ToolSearchOutput.new_id()), + call_id: Some(call_id.into()), + status: status.into(), + execution: execution.into(), + tools, + } + } + + pub fn id(&self) -> Option<&str> { + match self { + Self::Message { id, .. } + | Self::AgentMessage { id, .. } + | Self::LocalShellCall { id, .. } + | Self::FunctionCall { id, .. } + | Self::ToolSearchCall { id, .. } + | Self::FunctionCallOutput { id, .. } + | Self::CustomToolCall { id, .. } + | Self::CustomToolCallOutput { id, .. } + | Self::ToolSearchOutput { id, .. } + | Self::WebSearchCall { id, .. } + | Self::Compaction { id, .. } => id.as_deref(), + Self::Reasoning { id, .. } | Self::ImageGenerationCall { id, .. } => Some(id), + Self::CompactionTrigger | Self::ContextCompaction { .. } | Self::Other => None, + } + } + + pub fn attach_id_to_json(&self, value: &mut Value) { + let Some(id) = self.id().filter(|id| !id.is_empty()) else { + return; + }; + let Some(obj) = value.as_object_mut() else { + return; + }; + obj.insert("id".to_string(), Value::String(id.to_string())); + } + /// Returns whether this item is an ordinary user-role message. pub fn is_user_message(&self) -> bool { matches!(self, Self::Message { role, .. } if role == "user") } } +pub(crate) fn attach_all_response_item_ids(values: &mut [Value], items: &[ResponseItem]) { + for (value, item) in values.iter_mut().zip(items) { + item.attach_id_to_json(value); + } +} + pub const BASE_INSTRUCTIONS_DEFAULT: &str = include_str!("prompts/base_instructions/default.md"); /// Base instructions for the model in a thread. Corresponds to the `instructions` field in the ResponsesAPI. @@ -1131,38 +1277,32 @@ impl From for ResponseItem { content, phase, } => Self::Message { + id: Some(ResponseItemIdKind::Message.new_id()), role, content, - id: None, phase, }, ResponseInputItem::FunctionCallOutput { call_id, output } => { - Self::FunctionCallOutput { call_id, output } + Self::new_function_call_output(call_id, output) } ResponseInputItem::McpToolCallOutput { call_id, output } => { let output = output.into_function_call_output_payload(); - Self::FunctionCallOutput { call_id, output } + Self::new_function_call_output(call_id, output) } ResponseInputItem::CustomToolCallOutput { call_id, name, output, - } => Self::CustomToolCallOutput { - call_id, - name, - output, + } => match name { + Some(name) => Self::new_named_custom_tool_call_output(call_id, name, output), + None => Self::new_custom_tool_call_output(call_id, output), }, ResponseInputItem::ToolSearchOutput { call_id, status, execution, tools, - } => Self::ToolSearchOutput { - call_id: Some(call_id), - status, - execution, - tools, - }, + } => Self::new_tool_search_output(call_id, status, execution, tools), } } } @@ -1672,17 +1812,24 @@ mod tests { phase: Some(MessagePhase::Commentary), }); + let ResponseItem::Message { + id: Some(id), + role, + content, + phase, + } = item + else { + panic!("expected assistant message with generated id"); + }; + assert!(id.starts_with("msg_")); + assert_eq!(role, "assistant"); assert_eq!( - item, - ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { - text: "still working".to_string(), - }], - phase: Some(MessagePhase::Commentary), - } + content, + vec![ContentItem::OutputText { + text: "still working".to_string(), + }] ); + assert_eq!(phase, Some(MessagePhase::Commentary)); } #[test] @@ -2503,6 +2650,7 @@ mod tests { assert_eq!( item, ResponseItem::Compaction { + id: None, encrypted_content: "abc".into(), } ); @@ -2767,9 +2915,15 @@ mod tests { } })], }; + let item = ResponseItem::from(input.clone()); + let ResponseItem::ToolSearchOutput { id: Some(id), .. } = &item else { + panic!("expected tool search output id"); + }; + assert!(id.starts_with("tso_")); assert_eq!( - ResponseItem::from(input.clone()), + item, ResponseItem::ToolSearchOutput { + id: Some(id.clone()), call_id: Some("search-1".to_string()), status: "completed".to_string(), execution: "client".to_string(), @@ -2855,6 +3009,7 @@ mod tests { assert_eq!( parsed_output, ResponseItem::ToolSearchOutput { + id: None, call_id: None, status: "completed".to_string(), execution: "server".to_string(), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 642f2e5dee..8230b07fee 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -43,6 +43,7 @@ use crate::models::ResponseInputItem; use crate::models::ResponseItem; use crate::models::SandboxEnforcement; use crate::models::WebSearchAction; +use crate::models::attach_all_response_item_ids; use crate::num_format::format_with_separators; use crate::openai_models::ReasoningEffort as ReasoningEffortConfig; use crate::parse_command::ParsedCommand; @@ -705,15 +706,22 @@ impl InterAgentCommunication { } } - pub fn to_model_input_item(&self) -> ResponseItem { + pub fn to_model_input_item(&self, source_message_id: Option<&str>) -> ResponseItem { match &self.encrypted_content { - Some(encrypted_content) => ResponseItem::AgentMessage { - author: self.author.to_string(), - recipient: self.recipient.to_string(), - content: vec![AgentMessageInputContent::EncryptedContent { - encrypted_content: encrypted_content.clone(), - }], - }, + Some(encrypted_content) => { + let id = source_message_id + .and_then(|id| id.strip_prefix("msg_")) + .filter(|suffix| !suffix.is_empty()) + .map(|suffix| format!("amsg_{suffix}")); + ResponseItem::AgentMessage { + id, + author: self.author.to_string(), + recipient: self.recipient.to_string(), + content: vec![AgentMessageInputContent::EncryptedContent { + encrypted_content: encrypted_content.clone(), + }], + } + } None => self.to_response_input_item().into(), } } @@ -2857,6 +2865,34 @@ pub enum RolloutItem { EventMsg(EventMsg), } +impl RolloutItem { + pub fn attach_response_item_ids_to_json(&self, value: &mut Value) { + let Some(payload) = value.get_mut("payload") else { + return; + }; + + match self { + Self::ResponseItem(item) => item.attach_id_to_json(payload), + Self::Compacted(CompactedItem { + replacement_history: Some(items), + .. + }) => { + let Some(Value::Array(values)) = payload.get_mut("replacement_history") else { + return; + }; + attach_all_response_item_ids(values, items); + } + Self::SessionMeta(_) + | Self::Compacted(CompactedItem { + replacement_history: None, + .. + }) + | Self::TurnContext(_) + | Self::EventMsg(_) => {} + } + } +} + #[derive(Serialize, Deserialize, Clone, Debug, JsonSchema, TS)] pub struct CompactedItem { pub message: String, @@ -4085,6 +4121,41 @@ mod tests { ); } + #[test] + fn encrypted_inter_agent_communication_model_input_uses_source_message_id() { + let communication = InterAgentCommunication::new_encrypted( + AgentPath::root(), + AgentPath::root().join("reviewer").expect("recipient path"), + Vec::new(), + "encrypted-payload".to_string(), + /*trigger_turn*/ true, + ); + let expected = ResponseItem::AgentMessage { + id: Some("amsg_stable".to_string()), + author: "/root".to_string(), + recipient: "/root/reviewer".to_string(), + content: vec![AgentMessageInputContent::EncryptedContent { + encrypted_content: "encrypted-payload".to_string(), + }], + }; + + assert_eq!( + communication.to_model_input_item(Some("msg_stable")), + expected + ); + assert_eq!( + communication.to_model_input_item(/*source_message_id*/ None), + ResponseItem::AgentMessage { + id: None, + author: "/root".to_string(), + recipient: "/root/reviewer".to_string(), + content: vec![AgentMessageInputContent::EncryptedContent { + encrypted_content: "encrypted-payload".to_string(), + }], + } + ); + } + #[test] fn session_source_from_startup_arg_normalizes_custom_values() { assert_eq!( diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index d102d5c540..08dbbce63d 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -1678,12 +1678,13 @@ impl JsonlWriter { let timestamp = OffsetDateTime::now_utc() .format(timestamp_format) .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; - let line = RolloutLineRef { timestamp, item: rollout_item, }; - self.write_line(&line).await + let mut value = serde_json::to_value(line)?; + rollout_item.attach_response_item_ids_to_json(&mut value); + self.write_line(&value).await } async fn write_line(&mut self, item: &impl serde::Serialize) -> std::io::Result<()> { let mut json = serde_json::to_string(item)?; diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 07029abe37..747c198485 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -7,6 +7,7 @@ use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; @@ -442,6 +443,80 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result< Ok(()) } +#[tokio::test] +async fn recorder_persists_response_item_ids_in_rollout_history() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let thread_id = ThreadId::new(); + let recorder = RolloutRecorder::new( + &config, + RolloutRecorderParams::new( + thread_id, + /*forked_from_id*/ None, + /*parent_thread_id*/ None, + SessionSource::Exec, + /*thread_source*/ None, + BaseInstructions::default(), + Vec::new(), + ), + ) + .await?; + let rollout_path = recorder.rollout_path().to_path_buf(); + let top_level = ResponseItem::Message { + id: Some("msg_top_level".to_string()), + role: "assistant".to_string(), + content: Vec::new(), + phase: None, + }; + let historical_user = ResponseItem::Message { + id: None, + role: "user".to_string(), + content: Vec::new(), + phase: None, + }; + let compacted = RolloutItem::Compacted(CompactedItem { + message: "summary".to_string(), + replacement_history: Some(vec![ + ResponseItem::Message { + id: Some("msg_replacement".to_string()), + role: "assistant".to_string(), + content: Vec::new(), + phase: None, + }, + ResponseItem::Compaction { + id: Some("cmp_replacement".to_string()), + encrypted_content: "opaque".to_string(), + }, + ]), + }); + + recorder + .record_canonical_items(&[ + RolloutItem::ResponseItem(top_level), + RolloutItem::ResponseItem(historical_user), + compacted, + ]) + .await?; + recorder.flush().await?; + + let lines = fs::read_to_string(&rollout_path)? + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert_eq!(lines[1]["payload"]["id"], "msg_top_level"); + assert_eq!(lines[2]["payload"].get("id"), None); + assert_eq!( + lines[3]["payload"]["replacement_history"][0]["id"], + "msg_replacement" + ); + assert_eq!( + lines[3]["payload"]["replacement_history"][1]["id"], + "cmp_replacement" + ); + + Ok(()) +} + #[tokio::test] async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir");