diff --git a/codex-rs/rollout-trace/src/reducer/conversation.rs b/codex-rs/rollout-trace/src/reducer/conversation.rs index 2502e770b..db9752d24 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation.rs @@ -6,6 +6,7 @@ //! later model-facing payload carries their content. use anyhow::Context; +use anyhow::Ok; use anyhow::Result; use anyhow::bail; use serde_json::Value; @@ -209,7 +210,6 @@ impl TraceReducer { let index = context.start_index + offset; let tool_link_item = item.clone(); self.ensure_call_id_consistency(context.thread_id, &item)?; - self.ensure_reasoning_consistency(context.thread_id, &item)?; let item_id = if let Some(previous_item_id) = previous_snapshot.get(index) { if self.item_matches(previous_item_id, &item) { previous_item_id.clone() @@ -366,7 +366,6 @@ impl TraceReducer { for item in items { let tool_link_item = item.clone(); self.ensure_call_id_consistency(context.thread_id, &item)?; - self.ensure_reasoning_consistency(context.thread_id, &item)?; let item_id = self .find_matching_snapshot_item(&context.candidates, &item_ids, &item) .unwrap_or_else(|| { @@ -497,31 +496,6 @@ impl TraceReducer { Ok(()) } - fn ensure_reasoning_consistency( - &self, - thread_id: &str, - normalized: &NormalizedConversationItem, - ) -> Result<()> { - if normalized.kind != ConversationItemKind::Reasoning { - return Ok(()); - }; - let Some((label, value)) = reasoning_encoded_part(&normalized.body) else { - return Ok(()); - }; - - for item in self.rollout.conversation_items.values() { - if item.thread_id == thread_id - && item.kind == ConversationItemKind::Reasoning - && item.channel == normalized.channel - && reasoning_encoded_part(&item.body) == Some((label, value)) - && !reasoning_body_matches(&item.body, &normalized.body) - { - bail!("reasoning encrypted_content was reused with different readable content"); - } - } - Ok(()) - } - fn item_matches(&self, item_id: &str, normalized: &NormalizedConversationItem) -> bool { let Some(item) = self.rollout.conversation_items.get(item_id) else { return false; @@ -636,9 +610,10 @@ fn reasoning_body_matches(left: &ConversationBody, right: &ConversationBody) -> } // The Responses API may return readable reasoning on completion, but later - // request snapshots often replay only the encrypted blob. The blob is the - // stable model-visible identity; readable text/summary is extra evidence - // that must agree whenever both sides provide it. + // request snapshots often replay only the encrypted blob. Treat the blob as + // stable model-visible identity and merge readable text as best-effort + // evidence, because request/response serialization can observe different + // readable forms for the same encrypted reasoning item. let Some(left_encoded) = reasoning_encoded_part(left) else { return false; }; @@ -646,7 +621,7 @@ fn reasoning_body_matches(left: &ConversationBody, right: &ConversationBody) -> return false; }; - left_encoded == right_encoded && readable_reasoning_parts_match(left, right) + left_encoded == right_encoded } fn merge_reasoning_body( @@ -657,16 +632,64 @@ fn merge_reasoning_body( return Ok(()); } if !reasoning_body_matches(existing, incoming) { - bail!("reasoning encrypted_content was reused with different readable content"); + bail!("reasoning item merge attempted with different encrypted_content identity"); } - if readable_reasoning_parts(existing).is_empty() - && !readable_reasoning_parts(incoming).is_empty() - { - existing.parts = incoming.parts.clone(); + + let existing_text_parts = reasoning_text_parts(existing); + let existing_summary_parts = reasoning_summary_parts(existing); + if !existing_text_parts.is_empty() && !existing_summary_parts.is_empty() { + return Ok(()); } + + let incoming_text_parts = reasoning_text_parts(incoming); + let incoming_summary_parts = reasoning_summary_parts(incoming); + + let text_parts = if !existing_text_parts.is_empty() { + existing_text_parts + } else { + incoming_text_parts + }; + + let summary_parts = if !existing_summary_parts.is_empty() { + existing_summary_parts + } else { + incoming_summary_parts + }; + + // We already know that the encoded part exist (and matches). + let encoded_parts = reasoning_encoded_parts(existing); + + existing.parts = text_parts + .into_iter() + .cloned() + .chain(summary_parts.into_iter().cloned()) + .chain(encoded_parts.into_iter().cloned()) + .collect(); + Ok(()) } +fn reasoning_text_parts(body: &ConversationBody) -> Vec<&ConversationPart> { + body.parts + .iter() + .filter(|part| matches!(part, ConversationPart::Text { .. })) + .collect() +} + +fn reasoning_summary_parts(body: &ConversationBody) -> Vec<&ConversationPart> { + body.parts + .iter() + .filter(|part| matches!(part, ConversationPart::Summary { .. })) + .collect() +} + +fn reasoning_encoded_parts(body: &ConversationBody) -> Vec<&ConversationPart> { + body.parts + .iter() + .filter(|part| matches!(part, ConversationPart::Encoded { .. })) + .collect() +} + fn reasoning_encoded_part(body: &ConversationBody) -> Option<(&str, &str)> { body.parts.iter().find_map(|part| { if let ConversationPart::Encoded { label, value } = part { @@ -677,24 +700,6 @@ fn reasoning_encoded_part(body: &ConversationBody) -> Option<(&str, &str)> { }) } -fn readable_reasoning_parts_match(left: &ConversationBody, right: &ConversationBody) -> bool { - let left = readable_reasoning_parts(left); - let right = readable_reasoning_parts(right); - left.is_empty() || right.is_empty() || left == right -} - -fn readable_reasoning_parts(body: &ConversationBody) -> Vec<&ConversationPart> { - body.parts - .iter() - .filter(|part| { - matches!( - part, - ConversationPart::Text { .. } | ConversationPart::Summary { .. } - ) - }) - .collect() -} - #[cfg(test)] #[path = "conversation_tests.rs"] mod tests; diff --git a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs index aa8aee5ff..e6dd3922d 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs @@ -441,7 +441,74 @@ fn encrypted_reasoning_reuses_response_item_in_later_request() -> anyhow::Result } #[test] -fn same_encrypted_reasoning_with_different_text_is_reducer_error() -> anyhow::Result<()> { +fn encrypted_reasoning_upgrades_when_later_sighting_has_more_readable_body() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let writer = create_started_writer(&temp)?; + start_turn(&writer, "turn-1")?; + + let user = message("user", "count files"); + // Both sightings carry the same encrypted identity, but each has a + // different kind of readable evidence. The reducer should keep both + // observations because text and summary are complementary. + let text_only_reasoning = json!({ + "type": "reasoning", + "content": [{"type": "text", "text": "need count"}], + "summary": [], + "encrypted_content": "encoded-reasoning" + }); + let summary_only_reasoning = json!({ + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "counting files"}], + "encrypted_content": "encoded-reasoning" + }); + + let first_request = writer.write_json_payload( + RawPayloadKind::InferenceRequest, + &json!({ + "input": [user, text_only_reasoning] + }), + )?; + append_inference_start(&writer, "inference-1", "turn-1", first_request)?; + start_turn(&writer, "turn-2")?; + + let second_request = writer.write_json_payload( + RawPayloadKind::InferenceRequest, + &json!({ + "input": [user, summary_only_reasoning] + }), + )?; + append_inference_start(&writer, "inference-2", "turn-2", second_request)?; + + let rollout = replay_bundle(temp.path())?; + let first = &rollout.inference_calls["inference-1"]; + let second = &rollout.inference_calls["inference-2"]; + let reasoning_item_id = &first.request_item_ids[1]; + + // The reducer should keep one conversation item and merge the missing + // readable kind without treating the second sighting as a conflict. + assert_eq!(&second.request_item_ids[1], reasoning_item_id); + assert_eq!( + rollout.conversation_items[reasoning_item_id].body.parts, + vec![ + ConversationPart::Text { + text: "need count".to_string(), + }, + ConversationPart::Summary { + text: "counting files".to_string(), + }, + ConversationPart::Encoded { + label: "encrypted_content".to_string(), + value: "encoded-reasoning".to_string(), + }, + ], + ); + assert_eq!(rollout.conversation_items.len(), 2); + + Ok(()) +} + +#[test] +fn same_encrypted_reasoning_with_different_text_reuses_first_readable_body() -> anyhow::Result<()> { let temp = TempDir::new()?; let writer = create_started_writer(&temp)?; start_turn(&writer, "turn-1")?; @@ -455,6 +522,9 @@ fn same_encrypted_reasoning_with_different_text_is_reducer_error() -> anyhow::Re )?; append_inference_start(&writer, "inference-1", "turn-1", request)?; + // The response is the first readable observation for this encrypted + // reasoning blob, so it is the body later conflicting sightings must not + // overwrite. let response = writer.write_json_payload( RawPayloadKind::InferenceResponse, &json!({ @@ -486,10 +556,32 @@ fn same_encrypted_reasoning_with_different_text_is_reducer_error() -> anyhow::Re )?; append_inference_start(&writer, "inference-2", "turn-2", conflicting_request)?; - expect_replay_error( - &temp, - "reasoning encrypted_content was reused with different readable content", - ) + let rollout = replay_bundle(temp.path())?; + let first = &rollout.inference_calls["inference-1"]; + let second = &rollout.inference_calls["inference-2"]; + let reasoning_item_id = &first.response_item_ids[0]; + + // Same encrypted identity still reuses the response item, but conflicting + // readable text is not a safe upgrade. + assert_eq!( + second.request_item_ids, + vec![first.request_item_ids[0].clone(), reasoning_item_id.clone(),], + ); + assert_eq!( + rollout.conversation_items[reasoning_item_id].body.parts, + vec![ + ConversationPart::Text { + text: "first text".to_string(), + }, + ConversationPart::Encoded { + label: "encrypted_content".to_string(), + value: "encoded-reasoning".to_string(), + }, + ], + ); + assert_eq!(rollout.conversation_items.len(), 2); + + Ok(()) } #[test]