openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Store model history in response item envelopes (#38045)

Browse Source 
## What changed

- Add `ResponseItemEnvelope` as the history-owned wrapper for a `ResponseItem`.
- Store envelopes in `ContextManager` while keeping raw-item iteration and consuming APIs for existing callers.
- Preserve envelopes when compaction and history normalization rewrite retained items, and wrap newly synthesized items.

## Testing

- Cover envelope access, mutation, and conversion back to raw response items.
- Cover annotated history replacement and annotated user-message collection for compaction.

GitOrigin-RevId: 3fa2e7bdd981b08cc7aec410c80deccaecdabab4
This commit is contained in:
rka-oai
2026-08-11 17:58:11 +00:00
committed by copyberry
parent be751dd1df
commit 99915080b6
20 changed files with 382 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control_tests.rs
+27 -42
View File
@@ -273,8 +273,10 @@ async fn persisted_originator(thread: &CodexThread) -> String {
.expect("session metadata should be persisted")
}
fn has_subagent_notification(history_items: &[ResponseItem]) -> bool {
history_items.iter().any(|item| {
fn has_subagent_notification<'a>(
history_items: impl IntoIterator<Item = &'a ResponseItem>,
) -> bool {
history_items.into_iter().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
@@ -291,8 +293,11 @@ fn has_subagent_notification(history_items: &[ResponseItem]) -> bool {
}
/// Returns true when any message item contains `needle` in a text span.
fn history_contains_text(history_items: &[ResponseItem], needle: &str) -> bool {
history_items.iter().any(|item| {
fn history_contains_text<'a>(
history_items: impl IntoIterator<Item = &'a ResponseItem>,
needle: &str,
) -> bool {
history_items.into_iter().any(|item| {
let ResponseItem::Message { content, .. } = item else {
return false;
};
@@ -305,11 +310,11 @@ fn history_contains_text(history_items: &[ResponseItem], needle: &str) -> bool {
})
}
fn history_contains_assistant_inter_agent_communication(
history_items: &[ResponseItem],
fn history_contains_assistant_inter_agent_communication<'a>(
history_items: impl IntoIterator<Item = &'a ResponseItem>,
expected: &InterAgentCommunication,
) -> bool {
history_items.iter().any(|item| {
history_items.into_iter().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
@@ -333,13 +338,8 @@ fn history_contains_assistant_inter_agent_communication(
async fn wait_for_subagent_notification(parent_thread: &Arc<CodexThread>) -> bool {
let wait = async {
loop {
let history_items = parent_thread
.session
.clone_history()
.await
.raw_items()
.to_vec();
if has_subagent_notification(&history_items) {
let history = parent_thread.session.clone_history().await;
if has_subagent_notification(history.raw_items()) {
return true;
}
sleep(Duration::from_millis(25)).await;
@@ -680,9 +680,9 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
.await
.expect("inter-agent communication should stay pending");
let history_items = thread.session.clone_history().await.raw_items().to_vec();
let history = thread.session.clone_history().await;
assert!(!history_contains_assistant_inter_agent_communication(
&history_items,
history.raw_items(),
&communication
));
}
@@ -1351,7 +1351,7 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
.clone_history()
.await
.raw_items()
.first()
.next()
.cloned()
.expect("parent seed should be recorded");
let turn_context = parent_thread.session.new_default_turn().await;
@@ -1498,8 +1498,9 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
internal_chat_message_metadata_passthrough: None,
},
];
let history_items = history.raw_items().cloned().collect::<Vec<_>>();
assert_eq!(
strip_response_item_ids(history.raw_items()),
strip_response_item_ids(&history_items),
strip_response_item_ids(&expected_history),
"full-history forked child history should replace parent usage hints with the child subagent hint while filtering non-final assistant/tool chatter"
);
@@ -1735,7 +1736,6 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
assert!(
!history
.raw_items()
.iter()
.any(|item| matches!(item, ResponseItem::AgentMessage { .. })),
"forked child history should not inherit compacted parent agent messages"
);
@@ -2873,14 +2873,9 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
})
);
let root_history_items = root_thread
.session
.clone_history()
.await
.raw_items()
.to_vec();
let root_history = root_thread.session.clone_history().await;
assert!(!history_contains_assistant_inter_agent_communication(
&root_history_items,
root_history.raw_items(),
&InterAgentCommunication::new(
tester_path,
AgentPath::root(),
@@ -2889,7 +2884,7 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
/*trigger_turn*/ true,
)
));
assert!(!has_subagent_notification(&root_history_items));
assert!(!has_subagent_notification(root_history.raw_items()));
}
#[tokio::test]
@@ -2976,14 +2971,9 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
.await
.expect("completion watcher should queue a direct-parent message");
let root_history_items = root_thread
.session
.clone_history()
.await
.raw_items()
.to_vec();
let root_history = root_thread.session.clone_history().await;
assert!(!history_contains_assistant_inter_agent_communication(
&root_history_items,
root_history.raw_items(),
&InterAgentCommunication::new(
tester_path,
AgentPath::root(),
@@ -3015,21 +3005,16 @@ async fn completion_watcher_notifies_parent_when_child_is_missing() {
assert_eq!(wait_for_subagent_notification(&parent_thread).await, true);
let history_items = parent_thread
.session
.clone_history()
.await
.raw_items()
.to_vec();
let history = parent_thread.session.clone_history().await;
assert_eq!(
history_contains_text(
&history_items,
history.raw_items(),
&format!("\"agent_path\":\"{child_thread_id}\"")
),
true
);
assert_eq!(
history_contains_text(&history_items, "\"status\":\"not_found\""),
history_contains_text(history.raw_items(), "\"status\":\"not_found\""),
true
);
}
codex-rs/core/src/compact.rs
+32 -22
View File
@@ -28,6 +28,7 @@ use codex_analytics::CompactionStatus;
use codex_analytics::CompactionStrategy;
use codex_analytics::CompactionTrigger;
use codex_analytics::now_unix_seconds;
use codex_history::ResponseItemEnvelope;
use codex_protocol::error::CodexErr;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::error::Result as CodexResult;
@@ -340,10 +341,11 @@ async fn run_compact_task_inner_impl(
}
let history_snapshot = sess.clone_history().await;
let history_items = history_snapshot.raw_items();
let summary_suffix = get_last_assistant_message_from_turn(history_items).unwrap_or_default();
let history_items = history_snapshot.annotated_items();
let summary_suffix =
get_last_assistant_message_from_turn(history_snapshot.raw_items()).unwrap_or_default();
let summary_text = format!("{SUMMARY_PREFIX}\n{summary_suffix}");
let user_messages = collect_user_messages(history_items);
let user_messages = collect_annotated_user_messages(history_items);
let mut new_history = build_compacted_history(Vec::new(), &user_messages, &summary_text);
if let Some(summary_item) = new_history.last_mut() {
@@ -517,31 +519,39 @@ pub(crate) struct CompactedUserMessage {
internal_chat_message_metadata_passthrough: Option<InternalChatMessageMetadataPassthrough>,
}
#[cfg(test)]
pub(crate) fn collect_user_messages(items: &[ResponseItem]) -> Vec<CompactedUserMessage> {
items.iter().filter_map(compacted_user_message).collect()
}
pub(crate) fn collect_annotated_user_messages(
items: &[ResponseItemEnvelope],
) -> Vec<CompactedUserMessage> {
items
.iter()
.filter_map(|item| match crate::event_mapping::parse_turn_item(item) {
Some(TurnItem::UserMessage(user)) => {
if is_summary_message(&user.message()) {
None
} else {
Some(CompactedUserMessage {
message: user.message(),
internal_chat_message_metadata_passthrough: match item {
ResponseItem::Message {
internal_chat_message_metadata_passthrough,
..
} => internal_chat_message_metadata_passthrough.clone(),
_ => None,
},
})
}
}
_ => None,
})
.filter_map(|envelope| compacted_user_message(&envelope.item))
.collect()
}
fn compacted_user_message(item: &ResponseItem) -> Option<CompactedUserMessage> {
let Some(TurnItem::UserMessage(user)) = crate::event_mapping::parse_turn_item(item) else {
return None;
};
if is_summary_message(&user.message()) {
return None;
}
Some(CompactedUserMessage {
message: user.message(),
internal_chat_message_metadata_passthrough: match item {
ResponseItem::Message {
internal_chat_message_metadata_passthrough,
..
} => internal_chat_message_metadata_passthrough.clone(),
_ => None,
},
})
}
pub(crate) fn is_summary_message(message: &str) -> bool {
message.starts_with(format!("{SUMMARY_PREFIX}\n").as_str())
}
codex-rs/core/src/compact_remote.rs
+11 -5
View File
@@ -27,6 +27,7 @@ use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionTrigger;
use codex_history::ResponseItemEnvelope;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::ContextCompactionItem;
@@ -379,15 +380,15 @@ pub(crate) fn trim_function_call_history_to_fit_context_window(
// saturation in the normal history estimator.
let base_tokens =
i128::try_from(approx_token_count(&base_instructions.text)).unwrap_or(i128::MAX);
let original_items = history.raw_items();
let mut estimated_tokens = history_item_groups(original_items)
let original_items = history.annotated_items();
let mut estimated_tokens = history_item_groups(original_items.iter().map(|item| &item.item))
.map(|group| group.estimated_token_count())
.fold(base_tokens, i128::saturating_add);
let initial_estimated_tokens = i64::try_from(estimated_tokens).unwrap_or(i64::MAX);
let mut rewritten_items = Vec::new();
let mut consumed_items = 0;
for group in history_item_groups(original_items)
for group in history_item_groups(original_items.iter().map(|item| &item.item))
.collect::<Vec<_>>()
.into_iter()
.rev()
@@ -409,8 +410,13 @@ pub(crate) fn trim_function_call_history_to_fit_context_window(
if rewritten_outputs > 0 {
let retained_len = original_items.len() - consumed_items;
let mut items = original_items[..retained_len].to_vec();
items.extend(rewritten_items.into_iter().rev());
history.replace(items);
items.extend(
rewritten_items
.into_iter()
.rev()
.map(ResponseItemEnvelope::new),
);
history.replace_annotated(items);
}
let final_estimated_tokens = i64::try_from(estimated_tokens).unwrap_or(i64::MAX);
codex-rs/core/src/compact_remote_request.rs
+1 -1
View File
@@ -57,7 +57,7 @@ pub(super) async fn run_remote_compact_attempt(
}
let trace_input_history = compaction_trace
.is_enabled()
.then(|| history.raw_items().to_vec());
.then(|| history.raw_items().cloned().collect());
let prompt_input = history.for_prompt(&turn_context.model_info.input_modalities);
let tool_router = &step_context.tool_router;
let prompt = Prompt {
codex-rs/core/src/compact_remote_v2_attempt.rs
+1 -1
View File
@@ -65,7 +65,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
let trace_input_history = compaction_trace
.is_enabled()
.then(|| history.raw_items().to_vec());
.then(|| history.raw_items().cloned().collect());
let mut input = history.for_prompt(&turn_context.model_info.input_modalities);
let tool_router = &step_context.tool_router;
input.push(ResponseItem::CompactionTrigger {});
codex-rs/core/src/compact_tests.rs
+13
View File
@@ -1,4 +1,5 @@
use super::*;
use codex_history::ResponseItemEnvelope;
use codex_protocol::ResponseItemId;
use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
use codex_protocol::models::InternalChatMessageMetadataPassthrough;
@@ -117,6 +118,18 @@ fn collect_user_messages_extracts_user_text_only() {
assert_eq!(vec![compacted_user_message("first")], collected);
}
#[test]
fn collect_annotated_user_messages_extracts_user_text_only() {
let items = vec![
ResponseItemEnvelope::new(user_message("first")),
ResponseItemEnvelope::new(ResponseItem::Other),
];
let collected = collect_annotated_user_messages(&items);
assert_eq!(vec![compacted_user_message("first")], collected);
}
#[test]
fn collect_user_messages_filters_session_prefix_entries() {
let items = vec![
codex-rs/core/src/context/world_state/mod.rs
+15 -8
View File
@@ -361,20 +361,21 @@ impl WorldState {
}
/// Falls back to retained model history when no exact persisted snapshot is available.
pub(crate) fn render_history_diff(
pub(crate) fn render_history_diff<'a>(
&self,
previous: Option<&WorldStateSnapshot>,
items: &[ResponseItem],
items: impl IntoIterator<Item = &'a ResponseItem> + Clone,
) -> Vec<Box<dyn ContextualUserFragment>> {
self.render_with(|id, section| {
if let Some(previous) = previous.and_then(|previous| previous.sections.get(id)) {
if section.has_retained_fragment_matcher() && !has_retained_fragment(items, section)
if section.has_retained_fragment_matcher()
&& !has_retained_fragment(items.clone(), section)
{
PreviousSectionState::Absent
} else {
PreviousSectionState::Known(previous)
}
} else if has_legacy_fragment(items, section) {
} else if has_legacy_fragment(items.clone(), section) {
PreviousSectionState::Unknown
} else {
PreviousSectionState::Absent
@@ -393,8 +394,11 @@ impl WorldState {
}
}
fn has_retained_fragment(items: &[ResponseItem], section: &dyn ErasedWorldStateSection) -> bool {
items.iter().any(|item| {
fn has_retained_fragment<'a>(
items: impl IntoIterator<Item = &'a ResponseItem>,
section: &dyn ErasedWorldStateSection,
) -> bool {
items.into_iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
@@ -409,8 +413,11 @@ fn has_retained_fragment(items: &[ResponseItem], section: &dyn ErasedWorldStateS
})
}
fn has_legacy_fragment(items: &[ResponseItem], section: &dyn ErasedWorldStateSection) -> bool {
items.iter().any(|item| {
fn has_legacy_fragment<'a>(
items: impl IntoIterator<Item = &'a ResponseItem>,
section: &dyn ErasedWorldStateSection,
) -> bool {
items.into_iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
codex-rs/core/src/context_manager/history.rs
+46 -24
View File
@@ -9,6 +9,7 @@ use crate::event_mapping::is_contextual_user_message_content;
use crate::session::turn_context::TurnContext;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_history::ResponseItemEnvelope;
use codex_protocol::models::AgentMessageInputContent;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::ContentItem;
@@ -42,7 +43,7 @@ use std::sync::LazyLock;
pub(crate) struct ContextManager {
/// The oldest items are at the beginning of the vector. Snapshots share the vector until a
/// caller needs to mutate it, avoiding deep copies for read-only history consumers.
items: Arc<Vec<ResponseItem>>,
items: Arc<Vec<ResponseItemEnvelope>>,
/// Bumped whenever history is rewritten, such as compaction or rollback.
history_version: u64,
token_info: Option<TokenUsageInfo>,
@@ -96,7 +97,7 @@ impl ContextManager {
) -> (Vec<Box<dyn ContextualUserFragment>>, Option<WorldStateItem>) {
let snapshot = world_state.snapshot();
let fragments =
world_state.render_history_diff(self.world_state_baseline.as_ref(), &self.items);
world_state.render_history_diff(self.world_state_baseline.as_ref(), self.raw_items());
let rollout_item = self.world_state_baseline.as_ref().map_or_else(
|| Some(WorldStateItem::full(snapshot.clone().into_value())),
|previous| {
@@ -135,7 +136,7 @@ impl ContextManager {
}
let processed = Self::process_item(item_ref, policy);
Arc::make_mut(&mut self.items).push(processed);
Arc::make_mut(&mut self.items).push(ResponseItemEnvelope::new(processed));
}
}
@@ -145,16 +146,29 @@ impl ContextManager {
pub(crate) fn for_prompt(mut self, input_modalities: &[InputModality]) -> Vec<ResponseItem> {
self.normalize_history(input_modalities);
Arc::unwrap_or_clone(self.items)
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
}
/// Returns raw items in the history.
pub(crate) fn raw_items(&self) -> &[ResponseItem] {
/// Iterates over raw response items without exposing their history envelopes.
pub(crate) fn raw_items(
&self,
) -> impl Clone + ExactSizeIterator<Item = &ResponseItem> + DoubleEndedIterator {
self.items.iter().map(|envelope| &envelope.item)
}
/// Returns annotated history items without cloning their response payloads.
pub(crate) fn annotated_items(&self) -> &[ResponseItemEnvelope] {
&self.items
}
/// Returns raw items in the history and consumes the snapshot.
pub(crate) fn into_raw_items(self) -> Vec<ResponseItem> {
Arc::unwrap_or_clone(self.items)
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
}
pub(crate) fn history_version(&self) -> u64 {
@@ -183,7 +197,7 @@ impl ContextManager {
let items_tokens = self
.items
.iter()
.map(estimate_item_token_count)
.map(|envelope| estimate_item_token_count(&envelope.item))
.fold(0i64, i64::saturating_add);
Some(base_tokens.saturating_add(items_tokens))
@@ -198,12 +212,16 @@ impl ContextManager {
// If the removed item participates in a call/output pair, also remove
// its corresponding counterpart to keep the invariants intact without
// running a full normalization pass.
normalize::remove_corresponding_for(items, &removed);
normalize::remove_corresponding_for(items, &removed.item);
self.world_state_baseline = None;
}
}
pub(crate) fn replace(&mut self, items: Vec<ResponseItem>) {
self.replace_annotated(items.into_iter().map(ResponseItemEnvelope::new).collect());
}
pub(crate) fn replace_annotated(&mut self, items: Vec<ResponseItemEnvelope>) {
self.items = Arc::new(items);
self.history_version = self.history_version.saturating_add(1);
self.world_state_baseline = None;
@@ -233,7 +251,7 @@ impl ContextManager {
let snapshot = self.items.clone();
let user_positions = user_message_positions(&snapshot);
let Some(&first_instruction_turn_idx) = user_positions.first() else {
self.replace(Arc::unwrap_or_clone(snapshot));
self.replace_annotated(Arc::unwrap_or_clone(snapshot));
return;
};
@@ -253,7 +271,7 @@ impl ContextManager {
{
retained_items.retain_mut(|item| {
if item.turn_id() == Some(first_turn_id)
&& let ResponseItem::Message { role, content, .. } = item
&& let ResponseItem::Message { role, content, .. } = &mut item.item
&& role == "developer"
{
content.retain(|content| {
@@ -270,7 +288,7 @@ impl ContextManager {
});
}
self.replace(retained_items);
self.replace_annotated(retained_items);
}
pub(crate) fn update_token_info(
@@ -287,35 +305,41 @@ impl ContextManager {
fn get_non_last_reasoning_items_tokens(&self) -> i64 {
// Get reasoning items excluding all the ones after the last instruction boundary.
let Some(last_user_index) = self.items.iter().rposition(is_user_turn_boundary) else {
let Some(last_user_index) = self
.items
.iter()
.rposition(|envelope| is_user_turn_boundary(&envelope.item))
else {
return 0;
};
self.items
.iter()
.take(last_user_index)
.filter(|item| {
.filter(|envelope| {
matches!(
item,
&envelope.item,
ResponseItem::Reasoning {
encrypted_content: Some(_),
..
}
)
})
.map(estimate_item_token_count)
.map(|envelope| estimate_item_token_count(&envelope.item))
.fold(0i64, i64::saturating_add)
}
// These are local items added after the most recent model-emitted item.
// They are not reflected in `last_token_usage.total_tokens`.
fn items_after_last_model_generated_item(&self) -> &[ResponseItem] {
fn items_after_last_model_generated_item(
&self,
) -> impl Clone + ExactSizeIterator<Item = &ResponseItem> + DoubleEndedIterator {
let start = self
.items
.iter()
.rposition(is_model_generated_item)
.rposition(|envelope| is_model_generated_item(&envelope.item))
.map_or(self.items.len(), |index| index.saturating_add(1));
&self.items[start..]
self.items[start..].iter().map(|envelope| &envelope.item)
}
/// When true, the server already accounted for past reasoning tokens and
@@ -328,7 +352,6 @@ impl ContextManager {
.unwrap_or(0);
let items_after_last_model_generated_tokens = self
.items_after_last_model_generated_item()
.iter()
.map(estimate_item_token_count)
.fold(0i64, i64::saturating_add);
if server_reasoning_included {
@@ -342,7 +365,6 @@ impl ContextManager {
pub(crate) fn estimated_tokens_after_last_model_generated_item(&self) -> i64 {
self.items_after_last_model_generated_item()
.iter()
.map(estimate_item_token_count)
.fold(0i64, i64::saturating_add)
}
@@ -431,12 +453,12 @@ impl ContextManager {
/// reinjection.
fn trim_pre_turn_context_updates(
&mut self,
snapshot: &[ResponseItem],
snapshot: &[ResponseItemEnvelope],
first_instruction_turn_idx: usize,
mut cut_idx: usize,
) -> usize {
while cut_idx > first_instruction_turn_idx {
match &snapshot[cut_idx - 1] {
match &snapshot[cut_idx - 1].item {
ResponseItem::Message { role, content, .. }
if role == "developer" && is_contextual_dev_message_content(content) =>
{
@@ -833,10 +855,10 @@ fn is_inter_agent_instruction_content(content: &[ContentItem]) -> bool {
InterAgentCommunication::is_message_content(content)
}
fn user_message_positions(items: &[ResponseItem]) -> Vec<usize> {
fn user_message_positions(items: &[ResponseItemEnvelope]) -> Vec<usize> {
let mut positions = Vec::new();
for (idx, item) in items.iter().enumerate() {
if is_user_turn_boundary(item) {
for (idx, envelope) in items.iter().enumerate() {
if is_user_turn_boundary(&envelope.item) {
positions.push(idx);
}
}
codex-rs/core/src/context_manager/history_tests.rs
+47 -29
View File
@@ -5,6 +5,7 @@ use crate::context::world_state::WorldState;
use crate::context::world_state::WorldStateSection;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_history::ResponseItemEnvelope;
use codex_protocol::AgentPath;
use codex_protocol::ResponseItemId;
use codex_protocol::models::BaseInstructions;
@@ -106,6 +107,10 @@ fn create_history_with_items(items: Vec<ResponseItem>) -> ContextManager {
h
}
fn raw_items(history: &ContextManager) -> Vec<ResponseItem> {
history.raw_items().cloned().collect()
}
struct TestWorldStateSection;
impl WorldStateSection for TestWorldStateSection {
@@ -334,7 +339,7 @@ fn filters_non_api_messages() {
let a = assistant_msg("hello");
h.record_items([&u, &a], policy);
let items = h.raw_items();
let items = raw_items(&h);
assert_eq!(
items,
vec![
@@ -408,7 +413,6 @@ fn items_after_last_model_generated_tokens_include_user_and_tool_output() {
assert_eq!(
history
.items_after_last_model_generated_item()
.iter()
.map(estimate_item_token_count)
.fold(0i64, i64::saturating_add),
expected_tokens
@@ -422,7 +426,6 @@ fn items_after_last_model_generated_tokens_are_zero_without_model_generated_item
assert_eq!(
history
.items_after_last_model_generated_item()
.iter()
.map(estimate_item_token_count)
.fold(0i64, i64::saturating_add),
0
@@ -441,7 +444,7 @@ fn for_prompt_preserves_inter_agent_assistant_messages() {
let item = inter_agent_assistant_msg("continue");
let history = create_history_with_items(vec![item.clone()]);
assert_eq!(history.raw_items(), std::slice::from_ref(&item));
assert_eq!(raw_items(&history), std::slice::from_ref(&item));
assert_eq!(history.for_prompt(&default_input_modalities()), vec![item]);
}
@@ -453,8 +456,8 @@ fn cloned_history_shares_items_until_mutated() {
let mut snapshot = history.clone();
assert!(std::ptr::eq(
history.raw_items().as_ptr(),
snapshot.raw_items().as_ptr()
history.annotated_items().as_ptr(),
snapshot.annotated_items().as_ptr()
));
snapshot.record_items(
@@ -463,11 +466,26 @@ fn cloned_history_shares_items_until_mutated() {
);
assert!(!std::ptr::eq(
history.raw_items().as_ptr(),
snapshot.raw_items().as_ptr()
history.annotated_items().as_ptr(),
snapshot.annotated_items().as_ptr()
));
assert_eq!(history.raw_items(), std::slice::from_ref(&first));
assert_eq!(snapshot.raw_items(), &[first, second]);
assert_eq!(raw_items(&history), std::slice::from_ref(&first));
assert_eq!(raw_items(&snapshot), &[first, second]);
}
#[test]
fn annotated_history_apis_preserve_envelopes() {
let first_item = assistant_msg("first");
let first_envelope = ResponseItemEnvelope::new(first_item.clone());
let mut history = ContextManager::new();
history.replace_annotated(vec![first_envelope.clone()]);
assert_eq!(
history.annotated_items(),
std::slice::from_ref(&first_envelope)
);
assert_eq!(history.into_raw_items(), vec![first_item]);
}
#[test]
@@ -485,7 +503,7 @@ fn drop_last_n_user_turns_treats_inter_agent_assistant_messages_as_instruction_t
history.drop_last_n_user_turns(/*num_turns*/ 1);
assert_eq!(history.raw_items(), &vec![first_turn, first_reply]);
assert_eq!(raw_items(&history), vec![first_turn, first_reply]);
}
#[test]
@@ -868,7 +886,7 @@ fn remove_first_item_removes_matching_output_for_function_call() {
];
let mut h = create_history_with_items(items);
h.remove_first_item();
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[test]
@@ -892,7 +910,7 @@ fn remove_first_item_removes_matching_call_for_output() {
];
let mut h = create_history_with_items(items);
h.remove_first_item();
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[test]
@@ -920,7 +938,7 @@ fn remove_first_item_handles_local_shell_pair() {
];
let mut h = create_history_with_items(items);
h.remove_first_item();
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[test]
@@ -1185,7 +1203,7 @@ fn remove_first_item_handles_custom_tool_pair() {
];
let mut h = create_history_with_items(items);
h.remove_first_item();
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[test]
@@ -1242,7 +1260,7 @@ fn record_items_truncates_function_call_output_content() {
history.record_items([&item], policy);
assert_eq!(history.items.len(), 1);
match &history.items[0] {
match &history.items[0].item {
ResponseItem::FunctionCallOutput { output, .. } => {
let content = output.text_content().unwrap_or_default();
assert_ne!(content, long_output);
@@ -1277,7 +1295,7 @@ fn record_items_truncates_custom_tool_call_output_content() {
history.record_items([&item], policy);
assert_eq!(history.items.len(), 1);
match &history.items[0] {
match &history.items[0].item {
ResponseItem::CustomToolCallOutput { output, .. } => {
let output = output.text_content().unwrap_or_default();
assert_ne!(output, long_output);
@@ -1311,7 +1329,7 @@ fn record_items_respects_custom_token_limit() {
history.record_items([&item], policy);
let stored = match &history.items[0] {
let stored = match &history.items[0].item {
ResponseItem::FunctionCallOutput { output, .. } => output,
other => panic!("unexpected history item: {other:?}"),
};
@@ -1434,7 +1452,7 @@ fn normalize_adds_missing_output_for_function_call() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::FunctionCall {
id: None,
@@ -1472,7 +1490,7 @@ fn normalize_adds_missing_output_for_custom_tool_call() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::CustomToolCall {
id: None,
@@ -1515,7 +1533,7 @@ fn normalize_adds_missing_output_for_local_shell_call_with_id() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::LocalShellCall {
id: None,
@@ -1553,7 +1571,7 @@ fn normalize_removes_orphan_function_call_output() {
h.normalize_history(&default_input_modalities());
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[cfg(not(debug_assertions))]
@@ -1570,7 +1588,7 @@ fn normalize_removes_orphan_custom_tool_call_output() {
h.normalize_history(&default_input_modalities());
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[cfg(not(debug_assertions))]
@@ -1624,7 +1642,7 @@ fn normalize_mixed_inserts_and_removals() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::FunctionCall {
id: None,
@@ -1694,7 +1712,7 @@ 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!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::FunctionCall {
id: None,
@@ -1774,7 +1792,7 @@ fn normalize_adds_missing_output_for_tool_search_call() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![
ResponseItem::ToolSearchCall {
id: None,
@@ -1878,7 +1896,7 @@ fn normalize_removes_orphan_client_tool_search_output() {
h.normalize_history(&default_input_modalities());
assert_eq!(h.raw_items(), vec![]);
assert_eq!(raw_items(&h), vec![]);
}
#[cfg(debug_assertions)]
@@ -1912,7 +1930,7 @@ fn normalize_keeps_server_tool_search_output_without_matching_call() {
h.normalize_history(&default_input_modalities());
assert_eq!(
h.raw_items(),
raw_items(&h),
vec![ResponseItem::ToolSearchOutput {
id: None,
call_id: Some("server-search".to_string()),
@@ -2163,7 +2181,7 @@ fn record_items_omits_audio_that_exceeds_the_output_budget() {
history.record_items([&item], TruncationPolicy::Tokens(50));
assert_eq!(
history.raw_items(),
raw_items(&history),
&[ResponseItem::FunctionCallOutput {
id: None,
call_id: "call-audio".to_string(),
codex-rs/core/src/context_manager/normalize.rs
+34 -33
View File
@@ -1,3 +1,4 @@
use codex_history::ResponseItemEnvelope;
use codex_protocol::ResponseItemId;
use codex_protocol::models::ContentItem;
use codex_protocol::models::FunctionCallOutputContentItem;
@@ -17,12 +18,12 @@ const AUDIO_CONTENT_OMITTED_PLACEHOLDER: &str =
// Changing this value would change model-visible IDs and invalidate prompt caches.
const SYNTHETIC_OUTPUT_ID_NAMESPACE: Uuid = Uuid::from_u128(0x90d38d3e_6a5b_4d52_bfe2_2f1e634bfac4);
pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItemEnvelope>) {
let mut function_output_ids = HashSet::new();
let mut tool_search_output_ids = HashSet::new();
let mut custom_tool_output_ids = HashSet::new();
for item in items.iter() {
match item {
for envelope in items.iter() {
match &envelope.item {
ResponseItem::FunctionCallOutput { call_id, .. } => {
function_output_ids.insert(call_id.as_str());
}
@@ -42,22 +43,22 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
// Collect synthetic outputs to insert immediately after their calls.
// Store the insertion position (index of call) alongside the item so
// we can insert in reverse order and avoid index shifting.
let mut missing_outputs_to_insert: Vec<(usize, ResponseItem)> = Vec::new();
let mut missing_outputs_to_insert: Vec<(usize, ResponseItemEnvelope)> = Vec::new();
for (idx, item) in items.iter().enumerate() {
match item {
for (idx, envelope) in items.iter().enumerate() {
match &envelope.item {
ResponseItem::FunctionCall { id, call_id, .. }
if !function_output_ids.contains(call_id.as_str()) =>
{
info!("Function call output is missing for call id: {call_id}");
missing_outputs_to_insert.push((
idx,
ResponseItem::FunctionCallOutput {
ResponseItemEnvelope::new(ResponseItem::FunctionCallOutput {
id: synthetic_output_id("fco", id.as_deref()),
call_id: call_id.clone(),
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
internal_chat_message_metadata_passthrough: None,
},
}),
));
}
ResponseItem::ToolSearchCall {
@@ -68,14 +69,14 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
info!("Tool search output is missing for call id: {call_id}");
missing_outputs_to_insert.push((
idx,
ResponseItem::ToolSearchOutput {
ResponseItemEnvelope::new(ResponseItem::ToolSearchOutput {
id: synthetic_output_id("tso", id.as_deref()),
call_id: Some(call_id.clone()),
status: "completed".to_string(),
execution: "client".to_string(),
tools: Vec::new(),
internal_chat_message_metadata_passthrough: None,
},
}),
));
}
ResponseItem::CustomToolCall { id, call_id, .. }
@@ -86,13 +87,13 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
));
missing_outputs_to_insert.push((
idx,
ResponseItem::CustomToolCallOutput {
ResponseItemEnvelope::new(ResponseItem::CustomToolCallOutput {
id: synthetic_output_id("ctco", id.as_deref()),
call_id: call_id.clone(),
name: None,
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
internal_chat_message_metadata_passthrough: None,
},
}),
));
}
// LocalShellCall is represented in upstream streams by a FunctionCallOutput
@@ -106,12 +107,12 @@ pub(crate) fn ensure_call_outputs_present(items: &mut Vec<ResponseItem>) {
));
missing_outputs_to_insert.push((
idx,
ResponseItem::FunctionCallOutput {
ResponseItemEnvelope::new(ResponseItem::FunctionCallOutput {
id: synthetic_output_id("fco", id.as_deref()),
call_id: call_id.clone(),
output: FunctionCallOutputPayload::from_text("aborted".to_string()),
internal_chat_message_metadata_passthrough: None,
},
}),
));
}
_ => {}
@@ -144,10 +145,10 @@ fn synthetic_output_id(prefix: &str, item_id: Option<&str>) -> Option<ResponseIt
))
}
pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItemEnvelope>) {
let function_call_ids: HashSet<String> = items
.iter()
.filter_map(|i| match i {
.filter_map(|envelope| match &envelope.item {
ResponseItem::FunctionCall { call_id, .. } => Some(call_id.clone()),
_ => None,
})
@@ -155,7 +156,7 @@ pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
let tool_search_call_ids: HashSet<String> = items
.iter()
.filter_map(|i| match i {
.filter_map(|envelope| match &envelope.item {
ResponseItem::ToolSearchCall {
call_id: Some(call_id),
..
@@ -166,7 +167,7 @@ pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
let local_shell_call_ids: HashSet<String> = items
.iter()
.filter_map(|i| match i {
.filter_map(|envelope| match &envelope.item {
ResponseItem::LocalShellCall {
call_id: Some(call_id),
..
@@ -177,13 +178,13 @@ pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
let custom_tool_call_ids: HashSet<String> = items
.iter()
.filter_map(|i| match i {
.filter_map(|envelope| match &envelope.item {
ResponseItem::CustomToolCall { call_id, .. } => Some(call_id.clone()),
_ => None,
})
.collect();
items.retain(|item| match item {
items.retain(|envelope| match &envelope.item {
ResponseItem::FunctionCallOutput { call_id, .. } => {
let has_match =
function_call_ids.contains(call_id) || local_shell_call_ids.contains(call_id);
@@ -219,7 +220,7 @@ pub(crate) fn remove_orphan_outputs(items: &mut Vec<ResponseItem>) {
});
}
pub(crate) fn remove_corresponding_for(items: &mut Vec<ResponseItem>, item: &ResponseItem) {
pub(crate) fn remove_corresponding_for(items: &mut Vec<ResponseItemEnvelope>, item: &ResponseItem) {
match item {
ResponseItem::FunctionCall { call_id, .. } => {
remove_first_matching(items, |i| {
@@ -232,12 +233,12 @@ pub(crate) fn remove_corresponding_for(items: &mut Vec<ResponseItem>, item: &Res
});
}
ResponseItem::FunctionCallOutput { call_id, .. } => {
if let Some(pos) = items.iter().position(|i| {
matches!(i, ResponseItem::FunctionCall { call_id: existing, .. } if existing == call_id)
if let Some(pos) = items.iter().position(|envelope| {
matches!(&envelope.item, ResponseItem::FunctionCall { call_id: existing, .. } if existing == call_id)
}) {
items.remove(pos);
} else if let Some(pos) = items.iter().position(|i| {
matches!(i, ResponseItem::LocalShellCall { call_id: Some(existing), .. } if existing == call_id)
} else if let Some(pos) = items.iter().position(|envelope| {
matches!(&envelope.item, ResponseItem::LocalShellCall { call_id: Some(existing), .. } if existing == call_id)
}) {
items.remove(pos);
}
@@ -306,11 +307,11 @@ pub(crate) fn remove_corresponding_for(items: &mut Vec<ResponseItem>, item: &Res
}
}
fn remove_first_matching<F>(items: &mut Vec<ResponseItem>, predicate: F)
fn remove_first_matching<F>(items: &mut Vec<ResponseItemEnvelope>, predicate: F)
where
F: Fn(&ResponseItem) -> bool,
{
if let Some(pos) = items.iter().position(predicate) {
if let Some(pos) = items.iter().position(|envelope| predicate(&envelope.item)) {
items.remove(pos);
}
}
@@ -319,15 +320,15 @@ where
/// When `input_modalities` contains `InputModality::Image`, no stripping is performed.
pub(crate) fn strip_images_when_unsupported(
input_modalities: &[InputModality],
items: &mut [ResponseItem],
items: &mut [ResponseItemEnvelope],
) {
let supports_images = input_modalities.contains(&InputModality::Image);
if supports_images {
return;
}
for item in items.iter_mut() {
match item {
for envelope in items.iter_mut() {
match &mut envelope.item {
ResponseItem::Message { content, .. } => {
let mut normalized_content = Vec::with_capacity(content.len());
for content_item in content.iter() {
@@ -373,14 +374,14 @@ pub(crate) fn strip_images_when_unsupported(
/// When `input_modalities` contains `InputModality::Audio`, no stripping is performed.
pub(crate) fn strip_audio_when_unsupported(
input_modalities: &[InputModality],
items: &mut [ResponseItem],
items: &mut [ResponseItemEnvelope],
) {
if input_modalities.contains(&InputModality::Audio) {
return;
}
for item in items.iter_mut() {
match item {
for envelope in items.iter_mut() {
match &mut envelope.item {
ResponseItem::Message { content, .. } => {
for content_item in content.iter_mut() {
if matches!(content_item, ContentItem::InputAudio { .. }) {
codex-rs/core/src/guardian/prompt.rs
+2 -2
View File
@@ -436,8 +436,8 @@ fn render_guardian_transcript_entries_with_offset(
/// Keep both tool calls and tool results here. The reviewer often needs the
/// agent's exact queried path / arguments as well as the returned evidence to
/// decide whether the pending approval is justified.
pub(crate) fn collect_guardian_transcript_entries(
items: &[ResponseItem],
pub(crate) fn collect_guardian_transcript_entries<'a>(
items: impl IntoIterator<Item = &'a ResponseItem>,
) -> Vec<GuardianTranscriptEntry> {
let mut entries = Vec::new();
let mut tool_names_by_call_id = HashMap::new();
codex-rs/core/src/guardian/review_session.rs
+6 -2
View File
@@ -221,8 +221,12 @@ impl GuardianReviewSessionReuseKey {
}
}
fn encrypted_parent_compaction(items: &[ResponseItem]) -> Option<ResponseItem> {
let item = items.iter().rev().find(|item| {
fn encrypted_parent_compaction<'a, I>(items: I) -> Option<ResponseItem>
where
I: IntoIterator<Item = &'a ResponseItem>,
I::IntoIter: DoubleEndedIterator,
{
let item = items.into_iter().rev().find(|item| {
matches!(
item,
ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. }
codex-rs/core/src/realtime_context.rs
+3 -1
View File
@@ -209,7 +209,9 @@ async fn build_recent_work_section(
(!sections.is_empty()).then(|| sections.join("\n\n"))
}
fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
fn build_current_thread_section<'a>(
items: impl IntoIterator<Item = &'a ResponseItem>,
) -> Option<String> {
let mut turns = Vec::new();
let mut current_user = Vec::new();
let mut current_assistant = Vec::new();
codex-rs/core/src/session/handlers.rs
-1
View File
@@ -629,7 +629,6 @@ pub async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool {
let history = sess.clone_history().await;
let turn_count = history
.raw_items()
.iter()
.filter(|item| is_user_turn_boundary(item))
.count();
sess.services.session_telemetry.counter(
codex-rs/core/src/session/rollout_reconstruction.rs
+2 -1
View File
@@ -353,7 +353,8 @@ impl Session {
// prompt shape.
// 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 = compact::collect_user_messages(history.raw_items());
let user_messages =
compact::collect_annotated_user_messages(history.annotated_items());
let rebuilt = compact::build_compacted_history(
Vec::new(),
&user_messages,
codex-rs/core/src/session/rollout_reconstruction_tests.rs
+5 -4
View File
@@ -2,6 +2,7 @@ use super::*;
use super::tests::build_world_state_from_turn_context;
use super::tests::make_session_and_context;
use super::tests::raw_history_items;
use codex_history::CompactedItem;
use codex_history::InitialHistory;
use codex_history::ResumedHistory;
@@ -129,8 +130,8 @@ async fn record_initial_history_reconstructs_typed_inter_agent_message() {
.await;
assert_eq!(
session.state.lock().await.clone_history().raw_items(),
&[communication.to_model_input_item()]
raw_history_items(&session.state.lock().await.clone_history()),
vec![communication.to_model_input_item()]
);
}
@@ -169,8 +170,8 @@ async fn record_initial_history_restores_world_state_baseline() {
.expect("world state should build");
assert_eq!(
session.clone_history().await.raw_items(),
expected_history.as_slice(),
raw_history_items(&session.clone_history().await),
expected_history,
);
}
codex-rs/core/src/session/tests.rs
+64 -46
View File
@@ -656,6 +656,10 @@ fn test_model_client_session() -> crate::client::ModelClientSession {
.new_session()
}
pub(super) fn raw_history_items(history: &ContextManager) -> Vec<ResponseItem> {
history.raw_items().cloned().collect()
}
fn developer_input_texts(items: &[ResponseItem]) -> Vec<&str> {
items
.iter()
@@ -2058,7 +2062,7 @@ async fn record_initial_history_reconstructs_resumed_transcript() {
.await;
let history = session.state.lock().await.clone_history();
assert_eq!(expected, history.raw_items());
assert_eq!(expected, raw_history_items(&history));
}
#[tokio::test]
@@ -2078,8 +2082,8 @@ async fn record_conversation_items_stamps_missing_turn_id_and_preserves_existing
expected_fresh_item.set_turn_id_if_missing(&turn_context.sub_id);
let expected_items = vec![expected_fresh_item, existing_item];
assert_eq!(
session.clone_history().await.raw_items(),
expected_items.as_slice()
raw_history_items(&session.clone_history().await),
expected_items
);
}
@@ -2181,7 +2185,7 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() {
.await;
assert_eq!(
strip_response_item_ids(session.clone_history().await.raw_items()),
strip_response_item_ids(&raw_history_items(&session.clone_history().await)),
strip_response_item_ids(std::slice::from_ref(&expected_item))
);
@@ -2221,7 +2225,7 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() {
.record_initial_history(InitialHistory::Resumed(resumed))
.await;
assert_eq!(
strip_response_item_ids(resumed_session.clone_history().await.raw_items()),
strip_response_item_ids(&raw_history_items(&resumed_session.clone_history().await)),
strip_response_item_ids(std::slice::from_ref(&expected_item))
);
}
@@ -2249,7 +2253,8 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum
.await;
let live_history = session.clone_history().await;
let [live_item] = live_history.raw_items() else {
let live_items = raw_history_items(&live_history);
let [live_item] = live_items.as_slice() else {
panic!("expected exactly one live history item");
};
let live_item_id = live_item
@@ -2285,7 +2290,8 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum
.record_initial_history(InitialHistory::Resumed(resumed))
.await;
let resumed_history = resumed_session.clone_history().await;
let [resumed_item] = resumed_history.raw_items() else {
let resumed_items = raw_history_items(&resumed_history);
let [resumed_item] = resumed_items.as_slice() else {
panic!("expected exactly one resumed history item");
};
assert_eq!(
@@ -2329,7 +2335,10 @@ async fn prepares_image_failures_before_history_insertion() {
.await;
let history = session.state.lock().await.clone_history();
let id = history.raw_items()[0]
let id = history
.raw_items()
.next()
.expect("history should contain one item")
.id()
.expect("history item should have an ID");
let uuid = id
@@ -2357,7 +2366,10 @@ async fn prepares_image_failures_before_history_insertion() {
},
internal_chat_message_metadata_passthrough: None,
}];
assert_eq!(strip_metadata_from_items(history.raw_items()), expected);
assert_eq!(
strip_metadata_from_items(&raw_history_items(&history)),
expected
);
}
#[tokio::test]
@@ -2392,8 +2404,8 @@ async fn prepares_resumed_history_before_installing_it() {
.await;
assert_eq!(
session.state.lock().await.clone_history().raw_items(),
&[ResponseItem::Message {
raw_history_items(&session.state.lock().await.clone_history()),
vec![ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![
@@ -2487,7 +2499,7 @@ async fn record_initial_history_new_defers_initial_context_until_first_turn() {
session.record_initial_history(InitialHistory::New).await;
let history = session.clone_history().await;
assert_eq!(history.raw_items().to_vec(), Vec::<ResponseItem>::new());
assert_eq!(raw_history_items(&history), Vec::<ResponseItem>::new());
assert!(session.reference_context_item().await.is_none());
assert_eq!(session.previous_turn_settings().await, None);
}
@@ -2522,7 +2534,7 @@ 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_eq!(expected, raw_history_items(&history_before_seed));
let step_context = StepContext::for_test(Arc::clone(&turn_context));
session
@@ -2534,7 +2546,7 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only()
let history_after_seed = session.clone_history().await;
assert_eq!(
strip_response_item_ids(&expected),
strip_response_item_ids(history_after_seed.raw_items())
strip_response_item_ids(&raw_history_items(&history_after_seed))
);
session
@@ -2543,8 +2555,8 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only()
.expect("world state should build");
let history_after_second_seed = session.clone_history().await;
assert_eq!(
history_after_seed.raw_items(),
history_after_second_seed.raw_items()
raw_history_items(&history_after_seed),
raw_history_items(&history_after_second_seed)
);
}
@@ -3152,7 +3164,7 @@ async fn record_initial_history_reconstructs_forked_transcript() {
let history = session.state.lock().await.clone_history();
assert_eq!(
strip_response_item_ids(&expected),
strip_response_item_ids(history.raw_items())
strip_response_item_ids(&raw_history_items(&history))
);
}
@@ -3182,13 +3194,8 @@ async fn start_new_context_window_assigns_and_persists_item_ids() {
.await;
let live_history = session.clone_history().await;
assert!(!live_history.raw_items().is_empty());
assert!(
live_history
.raw_items()
.iter()
.all(|item| item.id().is_some())
);
assert!(live_history.raw_items().next().is_some());
assert!(live_history.raw_items().all(|item| item.id().is_some()));
session.flush_rollout().await.expect("rollout should flush");
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
@@ -3208,8 +3215,8 @@ async fn start_new_context_window_assigns_and_persists_item_ids() {
| RolloutItem::EventMsg(_) => None,
});
assert_eq!(
persisted_replacement_history.map(Vec::as_slice),
Some(live_history.raw_items())
persisted_replacement_history.cloned(),
Some(raw_history_items(&live_history))
);
}
@@ -3236,7 +3243,8 @@ async fn record_initial_history_assigns_and_persists_id_for_forked_response_item
.await;
let live_history = session.clone_history().await;
let [live_item] = live_history.raw_items() else {
let live_items = raw_history_items(&live_history);
let [live_item] = live_items.as_slice() else {
panic!("expected one forked response item");
};
let live_item_id = live_item
@@ -3245,7 +3253,7 @@ async fn record_initial_history_assigns_and_persists_id_for_forked_response_item
.to_string();
assert!(live_item_id.starts_with("msg_"));
expected_item.set_id(live_item.id().cloned());
assert_eq!(live_history.raw_items(), &[expected_item]);
assert_eq!(raw_history_items(&live_history), vec![expected_item]);
session.flush_rollout().await.expect("rollout should flush");
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
@@ -3490,7 +3498,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
realtime_active: Some(turn_context.realtime_active),
})
);
assert_eq!(history.raw_items(), &[]);
assert_eq!(raw_history_items(&history), Vec::<ResponseItem>::new());
assert_eq!(
serde_json::to_value(session.reference_context_item().await)
.expect("serialize fork reference context item"),
@@ -3548,7 +3556,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_eq!(expected, raw_history_items(&history));
assert_eq!(sess.previous_turn_settings().await, None);
assert!(sess.reference_context_item().await.is_none());
@@ -3594,7 +3602,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_eq!(initial_context, raw_history_items(&history));
}
#[tokio::test]
@@ -3618,8 +3626,8 @@ async fn thread_rollback_fails_without_persisted_thread_history() {
Some(CodexErrorInfo::ThreadRollbackFailed)
);
assert_eq!(
sess.clone_history().await.raw_items(),
history_before_rollback.raw_items()
raw_history_items(&sess.clone_history().await),
raw_history_items(&history_before_rollback)
);
}
@@ -3730,7 +3738,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context
assert_eq!(rollback_event.num_turns, 1);
assert_eq!(
sess.clone_history().await.raw_items(),
raw_history_items(&sess.clone_history().await),
vec![turn_one_user, turn_one_assistant]
);
assert_eq!(
@@ -3885,7 +3893,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_eq!(
raw_history_items(&sess.clone_history().await),
compacted_history
);
assert!(sess.reference_context_item().await.is_none());
assert_eq!(
sess.state.lock().await.auto_compact_window_ids(),
@@ -4006,7 +4017,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
assert_eq!(second_rollback.num_turns, 1);
assert_eq!(
sess.clone_history().await.raw_items(),
raw_history_items(&sess.clone_history().await),
vec![
user_message("turn 1 user"),
assistant_message("turn 1 assistant")
@@ -4046,7 +4057,10 @@ async fn thread_rollback_fails_when_turn_in_progress() {
);
let history = sess.clone_history().await;
assert_eq!(history_before_rollback.raw_items(), history.raw_items());
assert_eq!(
raw_history_items(&history_before_rollback),
raw_history_items(&history)
);
}
#[tokio::test]
@@ -4068,7 +4082,10 @@ async fn thread_rollback_fails_when_num_turns_is_zero() {
);
let history = sess.clone_history().await;
assert_eq!(history_before_rollback.raw_items(), history.raw_items());
assert_eq!(
raw_history_items(&history_before_rollback),
raw_history_items(&history)
);
}
#[tokio::test]
@@ -9051,7 +9068,7 @@ async fn record_context_update_items(
.await
.expect("world state should build");
let history = session.clone_history().await;
history.raw_items()[previous_len..].to_vec()
history.raw_items().skip(previous_len).cloned().collect()
}
#[tokio::test]
@@ -9279,7 +9296,8 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_
.expect("world state should build");
let history = session.clone_history().await;
let developer_messages = developer_message_texts(history.raw_items());
let history_items = raw_history_items(&history);
let developer_messages = developer_message_texts(&history_items);
assert!(
developer_messages
.iter()
@@ -9543,7 +9561,7 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont
let history = session.clone_history().await;
let initial_context = build_initial_context(&session, &turn_context).await;
assert_eq!(
strip_response_item_ids(history.raw_items()),
strip_response_item_ids(&raw_history_items(&history)),
strip_response_item_ids(&initial_context)
);
@@ -9598,7 +9616,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co
let initial_context = build_initial_context(&session, &turn_context).await;
expected_history.extend(initial_context);
assert_eq!(
strip_response_item_ids(history.raw_items()),
strip_response_item_ids(&raw_history_items(&history)),
strip_response_item_ids(&expected_history)
);
}
@@ -9640,7 +9658,7 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
.expect("world state should build");
assert_eq!(
session.clone_history().await.raw_items().to_vec(),
raw_history_items(&session.clone_history().await),
retained_world_state
);
assert_eq!(
@@ -10031,7 +10049,7 @@ async fn failed_user_message_persistence_stops_turn_processing() {
item.set_turn_id_if_missing(&turn_context.sub_id);
}
let history = session.clone_history().await;
let history_items = strip_response_item_ids(history.raw_items());
let history_items = strip_response_item_ids(&raw_history_items(&history));
assert!(
history_items.ends_with(&strip_response_item_ids(&expected_suffix)),
"later steer, response, and mailbox inputs should survive the failed queued message"
@@ -10478,7 +10496,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input()
internal_chat_message_metadata_passthrough: None,
};
assert!(
strip_response_item_ids(&strip_metadata_from_items(history.raw_items()))
strip_response_item_ids(&strip_metadata_from_items(&raw_history_items(&history)))
.contains(&expected),
"expected pending input to be persisted into history on turn completion"
);
@@ -11315,7 +11333,7 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
let history = sess.clone_history().await;
// Verify the `<turn_aborted>` marker is still recorded in history for the model.
assert!(
history.raw_items().iter().any(|item| {
history.raw_items().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
codex-rs/core/src/session/turn.rs
+4 -2
View File
@@ -2753,8 +2753,10 @@ async fn try_run_sampling_request(
outcome
}
pub(crate) fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -> Option<String> {
for item in responses.iter().rev() {
pub(crate) fn get_last_assistant_message_from_turn<'a>(
responses: impl DoubleEndedIterator<Item = &'a ResponseItem>,
) -> Option<String> {
for item in responses.rev() {
if let Some(message) = last_assistant_message_from_item(item, /*plan_mode*/ false) {
return Some(message);
}
codex-rs/history/src/lib.rs
+39
View File
@@ -1,5 +1,8 @@
//! Model-history and persisted-rollout domain types.
use std::borrow::Borrow;
use std::ops::Deref;
use std::ops::DerefMut;
use std::path::PathBuf;
use std::sync::Arc;
@@ -24,6 +27,42 @@ use serde::Deserialize;
use serde::Deserializer;
use serde::Serialize;
/// A history-owned, in-memory wrapper around a raw response item.
#[derive(Debug, Clone, PartialEq)]
pub struct ResponseItemEnvelope {
pub item: ResponseItem,
}
impl ResponseItemEnvelope {
pub fn new(item: ResponseItem) -> Self {
Self { item }
}
pub fn into_item(self) -> ResponseItem {
self.item
}
}
impl Deref for ResponseItemEnvelope {
type Target = ResponseItem;
fn deref(&self) -> &Self::Target {
&self.item
}
}
impl DerefMut for ResponseItemEnvelope {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.item
}
}
impl Borrow<ResponseItem> for ResponseItemEnvelope {
fn borrow(&self) -> &ResponseItem {
&self.item
}
}
/// Persisted rollout item used by core history and rollout storage.
#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema)]
#[serde(tag = "type", content = "payload", rename_all = "snake_case")]
codex-rs/history/src/tests.rs
+30
View File
@@ -4,6 +4,36 @@ use serde_json::json;
use super::*;
#[test]
fn response_item_envelope_accessors_preserve_item() {
let expected_item = ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "hello".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
};
let mut envelope = ResponseItemEnvelope::new(expected_item.clone());
assert_eq!(&*envelope, &expected_item);
let borrowed: &ResponseItem = envelope.borrow();
assert_eq!(borrowed, &expected_item);
let replacement_item = ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: "goodbye".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
};
*envelope = replacement_item.clone();
assert_eq!(envelope.into_item(), replacement_item);
}
#[test]
fn response_item_rollout_line_preserves_shape() -> Result<()> {
let legacy_line = json!({