diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index aaed74c052..0b19013af9 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs b/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs index 2230e6f2da..95ad84b2c4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_inject_items.rs @@ -253,7 +253,7 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history( .collect::>(); assert_eq!( persisted_additional_context, - vec![("developer", Some(true)), ("user", None)] + vec![("developer", Some(true)), ("user", Some(false))] ); let injected_value = serde_json::to_value(&injected_item)?; @@ -616,7 +616,13 @@ async fn thread_inject_items_adds_raw_response_items_after_a_turn() -> Result<() _ => None, }) .context("injected developer item should be persisted")?; - assert_eq!(persisted_developer_item.metadata, None); + assert_eq!( + persisted_developer_item.metadata, + Some(serde_json::from_value(json!({ + "history_sequence": 5, + "sampled_through_sequence": 3, + }))?) + ); let second_turn_req = mcp .send_turn_start_request(TurnStartParams { diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 7203b70f16..91e7571270 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -6,6 +6,7 @@ use std::time::Instant; use crate::Prompt; use crate::client::ModelClientSession; use crate::client_common::ResponseEvent; +use crate::compact_pending_input::is_retained_pending_mail; use crate::context::BaseInstructionsFragment; use crate::context::CompactionSummary; use crate::context::ContextualUserFragment; @@ -414,6 +415,34 @@ async fn run_compact_task_inner_impl( // belongs to this compaction turn. summary_item.set_turn_id_if_missing(&turn_context.sub_id); } + // Use the final compactor input, so an overflow retry cannot restore discarded input. + let pending_sequence = + crate::compact_pending_input::pending_mail_sequence(history.annotated_items()); + if let Some(first) = pending_sequence { + let summary = new_history + .pop() + .ok_or_else(|| CodexErr::Fatal("local compaction summary is missing".to_string()))?; + new_history.extend( + history + .annotated_items() + .iter() + .filter(|envelope| is_retained_pending_mail(envelope, first)) + .cloned(), + ); + new_history.sort_by_key(|envelope| { + envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.history_sequence) + }); + new_history = crate::compact_remote_v2::truncate_retained_messages_for_remote_compaction( + new_history, + COMPACT_USER_MESSAGE_MAX_TOKENS, + &history.file_audio_durations, + ); + new_history = + crate::compact_pending_input::insert_summary(new_history, summary, pending_sequence); + } let (window_number, window_ids) = sess.advance_auto_compact_window().await; let (new_history, world_state_baseline) = build_compaction_replacement_history( diff --git a/codex-rs/core/src/compact_pending_input.rs b/codex-rs/core/src/compact_pending_input.rs new file mode 100644 index 0000000000..6d5149828b --- /dev/null +++ b/codex-rs/core/src/compact_pending_input.rs @@ -0,0 +1,49 @@ +//! Retains bounded unsampled mail and positions summaries before it and subsequent caller input. +//! Delivery is proven only by a regular inference checkpoint, never by compactor output. + +use codex_history::ResponseItemEnvelope; +use codex_protocol::models::ResponseItem; + +use crate::context_manager::estimate_item_token_count; +use crate::context_manager::sampled_through; + +pub(crate) const MAX_RETAINED_AGENT_MESSAGE_TOKENS: i64 = 10_000; + +pub(crate) fn pending_mail_sequence(history: &[ResponseItemEnvelope]) -> Option { + let sampled_through = sampled_through(history); + history + .iter() + .filter(|envelope| matches!(envelope.item, ResponseItem::AgentMessage { .. })) + .filter_map(|envelope| envelope.metadata.as_ref()?.history_sequence) + .find(|sequence| *sequence > sampled_through) +} + +pub(crate) fn is_retained_pending_mail(envelope: &ResponseItemEnvelope, first: u64) -> bool { + matches!(envelope.item, ResponseItem::AgentMessage { .. }) + && envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.history_sequence) + .is_some_and(|sequence| sequence >= first) + && estimate_item_token_count(&envelope.item) <= MAX_RETAINED_AGENT_MESSAGE_TOKENS +} + +pub(crate) fn insert_summary( + mut retained: Vec, + summary: ResponseItemEnvelope, + pending_sequence: Option, +) -> Vec { + let position = pending_sequence + .and_then(|first| { + retained.iter().position(|envelope| { + envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.history_sequence) + .is_some_and(|sequence| sequence >= first) + }) + }) + .unwrap_or(retained.len()); + retained.insert(position, summary); + retained +} diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 7e1fa4668d..dd9b289780 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -15,6 +15,8 @@ use crate::compact::CompactionAnalyticsDetails; use crate::compact::compaction_status_from_result; use crate::compact_model_fallback::record_model_fallback; use crate::compact_model_fallback::should_retry_with_current_model; +use crate::compact_pending_input::MAX_RETAINED_AGENT_MESSAGE_TOKENS; +use crate::compact_pending_input::is_retained_pending_mail; use crate::compact_remote_history::HistoryItemGroup; use crate::compact_remote_history::history_item_groups; use crate::context::world_state::WorldState; @@ -88,7 +90,6 @@ enum RetainedImageBudget { } pub(crate) const RETAINED_MESSAGE_TOKEN_BUDGET: usize = 64_000; -const MAX_RETAINED_AGENT_MESSAGE_TOKENS: i64 = 10_000; // Compact attempts can run much longer than normal turns, so keep the per-transport // retry budget smaller than the general Responses stream retry budget. const MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES: u64 = 2; @@ -507,11 +508,16 @@ fn build_v2_compacted_history( crate::context_manager::rejected_audio::checkpoint_input_has_no_file_error_sources( &prompt_input, ); + let pending_sequence = crate::compact_pending_input::pending_mail_sequence(&prompt_input); let retained = v2_history_item_groups(prompt_input) - .filter(|group| is_retained_for_remote_compaction_v2(&group.source, features)) + .filter(|group| { + is_retained_for_remote_compaction_v2(&group.source, features) + || pending_sequence + .is_some_and(|first| is_retained_pending_mail(&group.source, first)) + }) .flat_map(HistoryItemGroup::into_items) .collect::>(); - let mut retained = truncate_retained_messages( + let retained = truncate_retained_messages( retained, RETAINED_MESSAGE_TOKEN_BUDGET, image_budget, @@ -528,7 +534,8 @@ fn build_v2_compacted_history( .get_or_insert_default() .compaction_without_file_error_sources = true; } - retained.push(checkpoint); + let retained = + crate::compact_pending_input::insert_summary(retained, checkpoint, pending_sequence); (retained, retained_image_count) } diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 9ee9da0869..d1c8edf070 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -102,6 +102,7 @@ fn next_guardian_review_context_revision() -> u64 { #[derive(Clone)] pub(crate) struct ContextManager { item_id_generator: ItemIdGenerator, + sampling: super::sampling::SamplingProgress, /// 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>, @@ -313,6 +314,7 @@ impl ContextManager { pub(crate) fn new(item_id_generator: ItemIdGenerator) -> Self { Self { item_id_generator, + sampling: Default::default(), items: Arc::new(Vec::new()), review_history: None, retained_context: Arc::default(), @@ -622,6 +624,7 @@ impl ContextManager { policy: TruncationPolicy, ) { for envelope in items { + self.sampling.record(envelope); if let Some(source) = self.record_item_with_metadata(&envelope.item, envelope.metadata.as_ref(), policy) { @@ -665,6 +668,7 @@ impl ContextManager { item: item.clone(), metadata: metadata.cloned(), }; + self.sampling.record(&mut processed); if let ResponseItem::FunctionCallOutput { output, .. } | ResponseItem::CustomToolCallOutput { output, .. } = &mut processed.item { @@ -808,6 +812,14 @@ impl ContextManager { self.history_version } + pub(crate) fn input_sequence(&self) -> u64 { + self.sampling.last_sequence + } + + pub(crate) fn checkpoint_sampling(&mut self, items: &mut [ResponseItemEnvelope]) { + self.sampling.checkpoint(items); + } + // Estimate token usage using byte-based heuristics from the truncation helpers. // This is a coarse lower bound, not a tokenizer-accurate count. pub(crate) fn estimate_token_count(&self, turn_context: &TurnContext) -> Option { @@ -865,7 +877,8 @@ impl ContextManager { self.replace_annotated(items.into_iter().map(ResponseItemEnvelope::new).collect()); } - pub(crate) fn replace_annotated(&mut self, items: Vec) { + pub(crate) fn replace_annotated(&mut self, mut items: Vec) { + self.sampling = super::sampling::SamplingProgress::restore(&mut items); self.retained_context = Arc::default(); self.user_message_revision = self.user_message_revision.saturating_add(1); if let Some(review_history) = &mut self.review_history { diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index cdf79822a7..e667b587c9 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -61,6 +61,30 @@ const EXEC_FORMAT_MAX_BYTES: usize = 10_000; const EXEC_FORMAT_MAX_TOKENS: usize = 2_500; const TEST_WAV_SAMPLE_RATE: u32 = 8_000; +#[test] +fn sampling_checkpoint_excludes_input_appended_during_request() { + let mut history = create_history_with_items(vec![user_msg("included in the request")]); + let request_sequence = history.input_sequence(); + history.record_items( + [&user_msg("arrived while sampling")], + TruncationPolicy::Tokens(10_000), + ); + history.record_annotated_items( + &mut [ResponseItemEnvelope { + item: assistant_msg("response to the earlier request"), + metadata: Some(CodexHarnessMetadata { + sampled_through_sequence: Some(request_sequence), + ..Default::default() + }), + }], + TruncationPolicy::Tokens(10_000), + ); + assert_eq!( + crate::context_manager::sampled_through(history.annotated_items()), + request_sequence, + ); +} + fn unknown_content_metadata() -> InternalChatMessageMetadataPassthrough { InternalChatMessageMetadataPassthrough { content_item_kinds: Some(vec![ContentItemKind("unknown".to_string())]), @@ -926,6 +950,8 @@ fn retains_only_harness_authored_configuration_updates() { let trusted = ResponseItemEnvelope { item: update.clone(), metadata: Some(CodexHarnessMetadata { + history_sequence: Some(4), + sampled_through_sequence: Some(0), harness_authored_configuration: true, ..Default::default() }), @@ -991,6 +1017,7 @@ fn drop_last_n_user_turns_removes_post_input_configuration_update_with_its_turn( ResponseItemEnvelope::new(assistant_msg("removed answer")), ]); history.record_annotated_items(&mut items, TruncationPolicy::Tokens(10_000)); + let surviving = &items[..surviving.len()]; history.drop_last_n_user_turns(/*num_turns*/ 1); @@ -1148,7 +1175,11 @@ fn cloned_history_shares_items_until_mutated() { fn annotated_history_apis_preserve_envelopes() { let first_envelope = ResponseItemEnvelope { item: assistant_msg("first"), - metadata: Some(CodexHarnessMetadata::default()), + metadata: Some(CodexHarnessMetadata { + history_sequence: Some(7), + sampled_through_sequence: Some(4), + ..Default::default() + }), }; let mut history = ContextManager::default(); @@ -1238,7 +1269,11 @@ async fn for_prompt_annotated_preserves_metadata_while_normalizing_item() { phase: None, internal_chat_message_metadata_passthrough: None, }, - metadata: Some(CodexHarnessMetadata::default()), + metadata: Some(CodexHarnessMetadata { + history_sequence: Some(7), + sampled_through_sequence: Some(4), + ..Default::default() + }), }; let mut history = ContextManager::default(); history.replace_annotated(vec![envelope.clone()]); @@ -1395,7 +1430,9 @@ async fn for_prompt_replaces_unavailable_message_images() { } })) .unwrap(); - assert_eq!(projected, vec![ResponseItemEnvelope::new(expected)]); + let mut expected_envelope = history.annotated_items()[0].clone(); + expected_envelope.item = expected; + assert_eq!(projected, vec![expected_envelope]); assert_eq!(raw_items(&history), vec![original]); } diff --git a/codex-rs/core/src/context_manager/mod.rs b/codex-rs/core/src/context_manager/mod.rs index 9149702d9e..67be01956c 100644 --- a/codex-rs/core/src/context_manager/mod.rs +++ b/codex-rs/core/src/context_manager/mod.rs @@ -1,6 +1,7 @@ mod history; mod image_availability; mod normalize; +mod sampling; pub(crate) mod updates; pub(crate) use history::ContextManager; @@ -11,3 +12,4 @@ pub(crate) use history::estimate_item_token_count_with_audio; pub(crate) use history::is_user_turn_boundary; pub(crate) use history::rejected_audio; pub(crate) use image_availability::replace_unresolvable_image_files; +pub(crate) use sampling::sampled_through; diff --git a/codex-rs/core/src/context_manager/sampling.rs b/codex-rs/core/src/context_manager/sampling.rs new file mode 100644 index 0000000000..a576e91b6e --- /dev/null +++ b/codex-rs/core/src/context_manager/sampling.rs @@ -0,0 +1,76 @@ +//! Tracks the prefix included in normal inference, independently of later history appends. +//! These checkpoints are harness metadata; compaction and replay never acknowledge new input. + +use codex_history::ResponseItemEnvelope; + +#[derive(Debug, Clone, Default)] +pub(super) struct SamplingProgress { + pub(super) last_sequence: u64, + sampled_through: u64, +} + +impl SamplingProgress { + pub(super) fn record(&mut self, envelope: &mut ResponseItemEnvelope) { + let metadata = envelope.metadata.get_or_insert_default(); + let sequence = metadata + .history_sequence + .get_or_insert_with(|| self.last_sequence.strict_add(/* rhs */ 1)); + self.last_sequence = self.last_sequence.max(*sequence); + self.sampled_through = self + .sampled_through + .max(metadata.sampled_through_sequence.unwrap_or_default()); + metadata.sampled_through_sequence = Some(self.sampled_through); + } + + pub(super) fn restore(items: &mut [ResponseItemEnvelope]) -> Self { + let mut progress = Self::default(); + for metadata in items.iter().filter_map(|item| item.metadata.as_ref()) { + progress.last_sequence = progress + .last_sequence + .max(metadata.history_sequence.unwrap_or_default()); + progress.sampled_through = progress + .sampled_through + .max(metadata.sampled_through_sequence.unwrap_or_default()); + } + progress.last_sequence = progress.last_sequence.max(progress.sampled_through); + // Legacy inputs have no delivery proof. Give them distinct identities and keep + // them pending until a normal request includes them; never guess from output order. + progress.assign_missing_sequences(items); + progress + } + + pub(super) fn checkpoint(&mut self, items: &mut [ResponseItemEnvelope]) { + self.assign_missing_sequences(items); + if let Some(first) = items.first_mut() { + first + .metadata + .get_or_insert_default() + .sampled_through_sequence = Some(self.sampled_through); + } + } + + fn assign_missing_sequences(&mut self, items: &mut [ResponseItemEnvelope]) { + for envelope in items { + if envelope + .metadata + .as_ref() + .and_then(|m| m.history_sequence) + .is_none() + { + self.record(envelope); + } + } + } +} + +pub(crate) fn sampled_through(items: &[ResponseItemEnvelope]) -> u64 { + items + .iter() + .filter_map(|item| { + item.metadata + .as_ref() + .and_then(|metadata| metadata.sampled_through_sequence) + }) + .max() + .unwrap_or_default() +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 26940e692d..ea0c5c00bb 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -36,6 +36,7 @@ pub use thread_input::InputWaitOutcome; mod turn_extension_data; pub use turn_extension_data::WithTurnExtensionData; mod compact_model_fallback; +mod compact_pending_input; mod compact_remote_history; mod compact_remote_v2; mod compact_retained_tool_outputs; diff --git a/codex-rs/core/src/session/compaction_replay_tests.rs b/codex-rs/core/src/session/compaction_replay_tests.rs index 8a0e6c5902..86ac0a17fd 100644 --- a/codex-rs/core/src/session/compaction_replay_tests.rs +++ b/codex-rs/core/src/session/compaction_replay_tests.rs @@ -43,7 +43,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { ..Default::default() }), }; - let replacement_history = vec![ + let mut replacement_history = vec![ ResponseItemEnvelope { item: summary_item.clone(), metadata: Some(CodexHarnessMetadata::default()), @@ -82,6 +82,11 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { .reconstruct_history_from_rollout(&turn_context, &rollout_items) .await; + for (index, envelope) in replacement_history.iter_mut().enumerate() { + let metadata = envelope.metadata.get_or_insert_default(); + metadata.history_sequence = Some(index as u64 + 1); + metadata.sampled_through_sequence = Some(0); + } assert_eq!(reconstructed.history, replacement_history); assert_eq!(42, reconstructed.window_number); assert_eq!(Some(first_window_id), reconstructed.first_window_id); diff --git a/codex-rs/core/src/session/inject_tests.rs b/codex-rs/core/src/session/inject_tests.rs index b8a67a986a..af12838332 100644 --- a/codex-rs/core/src/session/inject_tests.rs +++ b/codex-rs/core/src/session/inject_tests.rs @@ -33,7 +33,10 @@ async fn harness_authored_configuration_updates_preserve_metadata_and_resume() { ) .await; - expected.metadata.as_mut().unwrap().mcp_attribution = Some( + let metadata = expected.metadata.as_mut().unwrap(); + metadata.history_sequence = Some(1); + metadata.sampled_through_sequence = Some(0); + metadata.mcp_attribution = Some( session .services .executed_tool_calls diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index e2b43fec18..a88383e738 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4461,6 +4461,7 @@ impl Session { } let (executor_hook_sources, mcp, tool_router, extension_data) = prepared_tools??; Ok(Arc::new(StepContext { + sampled_through_sequence: None, serving_assessment: None, preempt: turn_context .config @@ -4773,6 +4774,7 @@ impl Session { .cloned(), ); } + state.history.checkpoint_sampling(&mut items); let replacement_history = items.clone(); state .history diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 3e20cf7301..4cd1b144f1 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -235,7 +235,19 @@ fn assistant_message(text: &str) -> ResponseItem { } fn annotated(items: Vec) -> Vec { - items.into_iter().map(ResponseItemEnvelope::new).collect() + // Legacy replay assigns append identities without claiming prior sampling. + items + .into_iter() + .enumerate() + .map(|(index, item)| ResponseItemEnvelope { + item, + metadata: Some(codex_history::CodexHarnessMetadata { + history_sequence: Some(index as u64 + 1), + sampled_through_sequence: Some(0), + ..Default::default() + }), + }) + .collect() } fn inter_agent_assistant_message(text: &str) -> ResponseItem { @@ -1614,11 +1626,11 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( vec![ RolloutItem::Compacted(Box::new(CompactedItem { message: String::new(), - replacement_history: Some(annotated(vec![object!({ + replacement_history: Some(vec![ResponseItemEnvelope::new(object!({ "type": "compaction", "id": format!("checkpoint-{window_number}"), "encrypted_content": format!("summary-{window_number}"), - })])), + }))]), retained_context: None, guardian_history: Some(codex_history::GuardianHistoryCheckpoint(vec![ user_message("original task").into(), @@ -1701,7 +1713,14 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( bounded.guardian_history.as_ref(), Some(&codex_history::GuardianHistoryCheckpoint(vec![ user_message("original task").into(), - assistant_message("continued working").into() + ResponseItemEnvelope { + item: assistant_message("continued working"), + metadata: Some(codex_history::CodexHarnessMetadata { + history_sequence: Some(2), + sampled_through_sequence: Some(0), + ..Default::default() + }), + } ])), ); if current { diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index 8dc6bdb0fb..58aafb5623 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/codex-rs/core/src/session/step_context.rs @@ -22,6 +22,8 @@ use tokio_util::sync::CancellationToken; /// Request-scoped state that may change between model sampling requests. #[derive(Clone)] pub(crate) struct StepContext { + /// Input prefix captured for this normal inference attempt, acknowledged with its output. + pub(crate) sampled_through_sequence: Option, /// Final serving value for this invocation; retained by its code cell across waits. pub(crate) serving_assessment: Option>, pub(crate) turn: Arc, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index aaf6e5f8e9..d9474d1f44 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -292,6 +292,7 @@ impl StepContext { }); settings.service_tier = turn.config.service_tier.clone(); Arc::new(Self { + sampled_through_sequence: None, preempt: turn .config .features @@ -2816,6 +2817,8 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() { item: expected_item.clone(), metadata: Some(CodexHarnessMetadata { user_input_order: Some(0), + history_sequence: Some(1), + sampled_through_sequence: Some(0), ..Default::default() }), }), @@ -2988,11 +2991,11 @@ async fn annotated_history_uses_explicit_model_without_a_step( .executed_tool_calls .mcp_attribution_snapshot(), ); - for envelope in &mut expected { - envelope - .metadata - .get_or_insert_default() - .history_truncation_token_limit = Some(expected_budget); + for (index, envelope) in expected.iter_mut().enumerate() { + let metadata = envelope.metadata.get_or_insert_default(); + metadata.history_sequence = Some(index as u64 + 1); + metadata.sampled_through_sequence = Some(0); + metadata.history_truncation_token_limit = Some(expected_budget); let (ResponseItem::FunctionCallOutput { output, .. } | ResponseItem::CustomToolCallOutput { output, .. }) = &mut envelope.item else { @@ -3171,7 +3174,11 @@ async fn prepares_resumed_history_before_installing_it() { ); assert_eq!( history.annotated_items()[0].metadata, - Some(CodexHarnessMetadata::default()) + Some(CodexHarnessMetadata { + history_sequence: Some(1), + sampled_through_sequence: Some(0), + ..Default::default() + }) ); } @@ -4254,7 +4261,11 @@ async fn record_initial_history_assigns_and_persists_id_for_forked_response_item assert_eq!(raw_history_items(&live_history), vec![expected_item]); assert_eq!( live_history.annotated_items()[0].metadata, - Some(CodexHarnessMetadata::default()) + Some(CodexHarnessMetadata { + history_sequence: Some(1), + sampled_through_sequence: Some(0), + ..Default::default() + }) ); session.flush_rollout().await.expect("rollout should flush"); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 97e5ba96ed..7261e3378b 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -616,6 +616,7 @@ pub(crate) async fn run_turn( guardian_budget_compacted = false; let SamplingRequestResult { needs_follow_up: model_needs_follow_up, + mailbox_preempted, last_agent_message: sampling_request_last_agent_message, } = sampling_request_output; if model_needs_follow_up { @@ -729,9 +730,10 @@ pub(crate) async fn run_turn( if run_pending_session_start_hooks(&sess, &turn_context).await { return Ok(None); } - // After a signaled step, drain the steer even if compaction + // After interrupted sampling, drain pending input even if compaction // leaves the estimated context above the token limit. can_drain_pending_input = !model_needs_follow_up + || mailbox_preempted || step_context .preempt .as_ref() @@ -1885,6 +1887,10 @@ async fn run_sampling_request( sess.clone_history().await }; let input_item_count = history.raw_items().len(); + let sampling_step_context = Arc::new(StepContext { + sampled_through_sequence: Some(history.input_sequence()), + ..step_context.as_ref().clone() + }); let mut prompt_input = history .for_prompt( &step_context.settings.model_info.input_modalities, @@ -1948,7 +1954,7 @@ async fn run_sampling_request( let err = match try_run_sampling_request( tool_runtime.clone(), Arc::clone(&sess), - Arc::clone(&step_context), + sampling_step_context, Arc::clone(&turn_store), client_session, &responses_metadata, @@ -2017,6 +2023,7 @@ async fn run_sampling_request( return Ok(( SamplingRequestResult { needs_follow_up: true, + mailbox_preempted: draining_for_mailbox, last_agent_message: None, }, notify_messages, @@ -2166,6 +2173,8 @@ pub(crate) async fn built_tools( #[derive(Debug)] struct SamplingRequestResult { needs_follow_up: bool, + /// Queued mail interrupted sampling and must be drained before the next request. + mailbox_preempted: bool, last_agent_message: Option, } @@ -3035,6 +3044,7 @@ async fn try_run_sampling_request( // TODO: Reconcile any response item already being presented to the client. break Ok(SamplingRequestResult { needs_follow_up: true, + mailbox_preempted: false, last_agent_message, }); } @@ -3070,6 +3080,7 @@ async fn try_run_sampling_request( // receiving the interrupt. Do not retry without that pending input. break Ok(SamplingRequestResult { needs_follow_up: true, + mailbox_preempted: true, last_agent_message, }); } @@ -3269,6 +3280,7 @@ async fn try_run_sampling_request( } break Ok(SamplingRequestResult { needs_follow_up: true, + mailbox_preempted: true, last_agent_message, }); } @@ -3478,6 +3490,7 @@ async fn try_run_sampling_request( } break Ok(SamplingRequestResult { needs_follow_up, + mailbox_preempted: *draining_for_mailbox, last_agent_message, }); } diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 36396b5bec..4478f53747 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -2,6 +2,7 @@ use std::pin::Pin; use std::sync::Arc; use codex_extension_api::ExtensionData; +use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_protocol::ResponseItemId; use codex_protocol::config_types::ModeKind; @@ -96,10 +97,18 @@ pub(crate) async fn record_completed_response_item_with_finalized_facts( finalized_facts: Option<&FinalizedTurnItemFacts>, ) { let turn_context = &step_context.turn; - sess.record_conversation_items( + sess.record_annotated_conversation_items( turn_context, &step_context.settings.model_info, - std::slice::from_ref(item), + vec![ResponseItemEnvelope { + item: item.clone(), + metadata: step_context + .sampled_through_sequence + .map(|sequence| CodexHarnessMetadata { + sampled_through_sequence: Some(sequence), + ..Default::default() + }), + }], ) .await; if turn_context.config.otel.agent_response_logging_enabled() { diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index b522e2dedb..ac71c9637f 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -4595,6 +4595,16 @@ async fn manual_compact_twice_preserves_latest_user_messages() { .unwrap(); wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + let mail = json!({ + "type": "agent_message", "id": "agent_pending-result", + "author": "/root/worker", "recipient": "/root", + "content": [{"type": "input_text", "text": "Message Type: FINAL_ANSWER\nThe worker found a release blocker."}], + }); + codex + .inject_response_items(vec![serde_json::from_value(mail.clone()).unwrap()]) + .await + .unwrap(); + codex.submit(Op::Compact).await.unwrap(); wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; @@ -4620,6 +4630,11 @@ async fn manual_compact_twice_preserves_latest_user_messages() { wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; let requests = responses_mock.requests(); + assert_eq!( + responses::strip_metadata_from_json(json!(requests[2].inputs_of_type("agent_message"))), + json!([mail]), + ); + assert!(requests[4].inputs_of_type("agent_message").is_empty()); assert_eq!( requests.len(), 5, diff --git a/codex-rs/core/tests/suite/image_rollout.rs b/codex-rs/core/tests/suite/image_rollout.rs index cc2330bce5..668445aef5 100644 --- a/codex-rs/core/tests/suite/image_rollout.rs +++ b/codex-rs/core/tests/suite/image_rollout.rs @@ -881,7 +881,7 @@ async fn resumed_history_only_emits_resize_notices_for_new_images() -> anyhow::R .then(|| envelope.metadata.map(|metadata| metadata.client_authored)) }) .collect::>(); - assert_eq!(persisted_developer_metadata, vec![None, Some(true)]); + assert_eq!(persisted_developer_metadata, vec![Some(false), Some(true)]); Ok(()) } diff --git a/codex-rs/core/tests/suite/scenarios_compaction_tests.rs b/codex-rs/core/tests/suite/scenarios_compaction_tests.rs index dd8e6a9ee2..ddd1ddaa1f 100644 --- a/codex-rs/core/tests/suite/scenarios_compaction_tests.rs +++ b/codex-rs/core/tests/suite/scenarios_compaction_tests.rs @@ -1,7 +1,7 @@ -//! Snapshots remote `/compact`, its persisted replacement, and the next user turn. +//! Snapshots pending mail across remote compaction, restart, and normal inference. use super::*; -use codex_history::RolloutItem; +use core_test_support::responses::strip_metadata_from_json; use pretty_assertions::assert_eq; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -17,75 +17,91 @@ async fn remote_slash_compact_preserves_context_across_windows() -> Result<()> { ev_completed("plan-response"), ]), sse(vec![ - json!({ - "type": "response.output_item.done", - "item": { - "type": "compaction", - "encrypted_content": "REMOTE_SLASH_COMPACT_CHECKPOINT", - }, - }), + compaction_output("REMOTE_SLASH_COMPACT_CHECKPOINT"), ev_completed("compact-response"), ]), + sse(vec![ + compaction_output("RECOMPACTED_CHECKPOINT"), + ev_completed("recompact-response"), + ]), sse(vec![ ev_assistant_message("continue", "Start by validating the release candidate."), ev_completed("continue-response"), ]), + sse(vec![ + compaction_output("AFTER_MAIL_WAS_SAMPLED"), + ev_completed("sampled-compact-response"), + ]), ], ) .await; - let test = test_codex() + let configure = |config: &mut Config| { + configure_scenario_catalog(config); + config.workspace_roots = vec![config.cwd.clone()]; + config.model_auto_compact_token_limit = Some(i64::MAX); + }; + let mut builder = test_codex() .with_model("gpt-6-astra") .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) - .with_config(|config| { - configure_scenario_catalog(config); - config.workspace_roots = vec![config.cwd.clone()]; - }) - .build_with_auto_env(&server) - .await?; + .with_config(configure); + let mut test = builder.build_with_auto_env(&server).await?; test.submit_turn("Plan the release.").await?; + let mail = json!({ + "type": "agent_message", "id": "agent_pending-result", + "author": "/root/worker", "recipient": "/root", + "content": [{"type": "input_text", "text": "Message Type: FINAL_ANSWER\nThe worker found a release blocker."}], + }); + test.codex + .inject_response_items(vec![serde_json::from_value(mail.clone())?]) + .await?; + for compaction in 0..2 { + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + assert!(!matches!(event, EventMsg::Error(_)), "{event:?}"); + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + if compaction == 0 { + builder = builder.with_model("gpt-6-astra").with_config(configure); + test = builder.restart_with_auto_env(&server, &test).await?; + } + } + + test.submit_turn("Continue with the first step.").await?; test.codex.submit(Op::Compact).await?; wait_for_event(&test.codex, |event| { - assert!(!matches!(event, EventMsg::Error(_)), "{event:?}"); matches!(event, EventMsg::TurnComplete(_)) }) .await; - - // Observe the checkpoint before a later turn can initialize or update its context. - test.codex.flush_rollout().await?; - let rollout = fs::read_to_string(test.codex.rollout_path().expect("rollout path"))?; - let replacement = rollout - .lines() - .map(codex_rollout::parse_rollout_line) - .collect::, _>>()? - .into_iter() - .find_map(|line| match line.item { - RolloutItem::Compacted(compacted) => compacted.replacement_history, - _ => None, - }) - .expect("persisted compaction replacement") - .into_iter() - .map(|envelope| serde_json::to_value(envelope.item)) - .collect::, _>>()?; - - test.submit_turn("Continue with the first step.").await?; + let history = test.codex.conversation_history_snapshot().await; + assert!( + !history + .items() + .any(|item| matches!(item, ResponseItem::AgentMessage { .. })) + ); let requests = mock.requests(); - assert_eq!(requests.len(), 3); + assert_eq!(requests.len(), 5); assert_eq!(requests[1].inputs_of_type("compaction_trigger").len(), 1); assert_eq!( - requests[2].inputs_of_type("compaction")[0]["encrypted_content"], - "REMOTE_SLASH_COMPACT_CHECKPOINT" + requests[3].inputs_of_type("compaction")[0]["encrypted_content"], + "RECOMPACTED_CHECKPOINT" + ); + assert_eq!( + strip_metadata_from_json(json!(requests[3].inputs_of_type("agent_message"))), + json!([mail]), ); insta::assert_snapshot!( "remote_slash_compact", context_snapshot::format_context_snapshot( - "A user invokes /compact through remote compaction, persists the replacement history, then continues in the new window.", + "Pending mail survives repeated /compact and cold resume, then normal inference acknowledges it for subsequent compaction.", &[ SnapshotEntry::captured(&requests[0]).labeled("Before /compact"), SnapshotEntry::captured(&requests[1]).labeled("Remote /compact request"), - SnapshotEntry::items(&replacement).labeled("Stored replacement before next turn"), - SnapshotEntry::captured(&requests[2]).labeled("First turn after /compact"), + SnapshotEntry::captured(&requests[2]).labeled("Repeated /compact after resume"), + SnapshotEntry::captured(&requests[3]).labeled("First turn after /compact"), + SnapshotEntry::captured(&requests[4]).labeled("Compaction after mail was sampled"), ], &ContextSnapshotOptions::default().include_request_settings(), ) diff --git a/codex-rs/core/tests/suite/scenarios_injected_agent_mail.rs b/codex-rs/core/tests/suite/scenarios_injected_agent_mail.rs index 1590892853..c2ded73902 100644 --- a/codex-rs/core/tests/suite/scenarios_injected_agent_mail.rs +++ b/codex-rs/core/tests/suite/scenarios_injected_agent_mail.rs @@ -233,3 +233,129 @@ async fn injected_mail_respects_delivery_boundaries() -> anyhow::Result<()> { server.shutdown().await; Ok(()) } + +#[test_case::test_case(true; "instant interrupt enabled")] +#[test_case::test_case(false; "instant interrupt disabled")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn injected_agent_message_interrupting_sampling_survives_mid_turn_compaction( + instant_interrupt: bool, +) -> anyhow::Result<()> { + let (release_boundary, boundary_gate) = oneshot::channel(); + let chunk = |gate, events| StreamingSseChunk { + gate, + body: responses::sse(events), + }; + let long_reasoning = "reasoning ".repeat(400); + let (server, _completions) = start_streaming_sse_server(vec![ + vec![ + chunk( + None, + vec![ + responses::ev_response_created("first"), + responses::ev_reasoning_item_added("boundary", &[long_reasoning.as_str()]), + ], + ), + chunk( + Some(boundary_gate), + vec![responses::ev_reasoning_item( + "boundary", + &[long_reasoning.as_str()], + &[], + )], + ), + ], + vec![chunk( + None, + vec![ + responses::ev_response_created("compact"), + responses::ev_assistant_message("summary", "AUTO_COMPACT_SUMMARY"), + responses::ev_completed_with_tokens("compact", /*total_tokens*/ 50), + ], + )], + vec![chunk( + None, + vec![ + responses::ev_response_created("after-compact"), + responses::ev_assistant_message("answer", "The worker update is included."), + responses::ev_completed_with_tokens("after-compact", /*total_tokens*/ 60), + ], + )], + ]) + .await; + let config_server = responses::start_mock_server().await; + let base_url = format!("{}/v1", server.uri()); + let test = test_codex() + .with_model("gpt-5.4") + .with_config(move |config| { + config.model_provider.name = "OpenAI (test)".to_string(); + config.model_provider.base_url = Some(base_url); + config.model_provider.supports_websockets = false; + config.model_auto_compact_token_limit = Some(200); + config + .features + .set_enabled(Feature::InstantInterrupt, instant_interrupt) + .expect("set instant interrupt"); + config + .features + .disable(Feature::DeferMailboxPreemption) + .expect("enable preemption"); + config + .features + .disable(Feature::EnableRequestCompression) + .expect("capture request bodies"); + }) + .build_with_auto_env(&config_server) + .await?; + let message: ResponseItem = serde_json::from_value(json!({ + "type": "agent_message", "id": "agent_active-before-compaction", + "author": "/root/worker", "recipient": "/root", + "content": [{"type": "input_text", "text": "Use the completed worker result."}], + }))?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![super::text( + "Plan the next action.", + )])) + .await?; + wait_for_event( + &test.codex, + |event| matches!(event, EventMsg::ItemStarted(item) if item.item.id() == "boundary"), + ) + .await; + test.codex + .inject_response_items(vec![message.clone()]) + .await?; + release_boundary + .send(()) + .expect("release sampling boundary"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex.shutdown_and_wait().await?; + + let requests = server + .requests() + .await + .iter() + .map(|body| serde_json::from_slice::(body)) + .collect::>>()?; + assert_eq!(requests.len(), 3); + let after_compact_input = requests[2]["input"].as_array().expect("request input"); + let actual = after_compact_input + .iter() + .filter(|item| item["type"] == "agent_message") + .cloned() + .collect::>(); + assert_eq!( + responses::strip_metadata_from_json(json!(actual)), + responses::strip_metadata_from_json(serde_json::to_value([message])?) + ); + let entries = requests.iter().map(SnapshotEntry::body).collect::>(); + insta::assert_snapshot!(context_snapshot::format_context_snapshot( + "An injected agent message that interrupts sampling is included in the first request after mid-turn compaction.", + &entries, + &ContextSnapshotOptions::default().rewrite_known_segments(), + )); + server.shutdown().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact__manual_compact_with_history_shapes.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact__manual_compact_with_history_shapes.snap index f7ea8a443c..724b037da9 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact__manual_compact_with_history_shapes.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact__manual_compact_with_history_shapes.snap @@ -35,7 +35,8 @@ Scenario: Manual /compact with prior user history compacts existing history and first manual turn 04:message/assistant: FIRST_REPLY -05:message/user: +05:agent_message +06:message/user: ## Window 3 (after request 2: input diverged at item 04, settings changed (parallel_tool_calls, tools)) @@ -55,7 +56,8 @@ Scenario: Manual /compact with prior user history compacts existing history and 04:message/user: FIRST_MANUAL_SUMMARY -05:message/user: +05:agent_message +06:message/user: second manual turn ## Window 4 (after request 3: settings changed (parallel_tool_calls, tools)) @@ -75,11 +77,12 @@ Scenario: Manual /compact with prior user history compacts existing history and 04:message/user: FIRST_MANUAL_SUMMARY -05:message/user: +05:agent_message +06:message/user: second manual turn -06:message/assistant: +07:message/assistant: SECOND_LARGE_REPLY -07:message/user: +08:message/user: ## Window 5 (after request 4: input diverged at item 01, settings changed (parallel_tool_calls, tools)) diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__compaction__remote_slash_compact.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__compaction__remote_slash_compact.snap index 9fc3765404..2efbf44dad 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__compaction__remote_slash_compact.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__compaction__remote_slash_compact.snap @@ -1,8 +1,8 @@ --- source: core/tests/suite/scenarios_compaction_tests.rs -expression: "context_snapshot::format_context_snapshot(\"A user invokes /compact through remote compaction, persists the replacement history, then continues in the new window.\",\n&[SnapshotEntry::captured(&requests[0]).labeled(\"Before /compact\"),\nSnapshotEntry::captured(&requests[1]).labeled(\"Remote /compact request\"),\nSnapshotEntry::items(&replacement).labeled(\"Stored replacement before next turn\"),\nSnapshotEntry::captured(&requests[2]).labeled(\"First turn after /compact\"),],\n&ContextSnapshotOptions::default().include_request_settings(),)" +expression: "context_snapshot::format_context_snapshot(\"Pending mail survives repeated /compact and cold resume, then normal inference acknowledges it for subsequent compaction.\",\n&[SnapshotEntry::captured(&requests[0]).labeled(\"Before /compact\"),\nSnapshotEntry::captured(&requests[1]).labeled(\"Remote /compact request\"),\nSnapshotEntry::captured(&requests[2]).labeled(\"Repeated /compact after resume\"),\nSnapshotEntry::captured(&requests[3]).labeled(\"First turn after /compact\"),\nSnapshotEntry::captured(&requests[4]).labeled(\"Compaction after mail was sampled\"),],\n&ContextSnapshotOptions::default().include_request_settings(),)" --- -Scenario: A user invokes /compact through remote compaction, persists the replacement history, then continues in the new window. +Scenario: Pending mail survives repeated /compact and cold resume, then normal inference acknowledges it for subsequent compaction. ## Window 1 Settings: @@ -104,11 +104,12 @@ Settings: -- request 2 (compaction; Remote /compact request) -- 08:message/assistant: The release plan has three steps. -09:compaction_trigger +09:agent_message +10:compaction_trigger ## Window 2 (after request 2: input diverged at item 07) -Settings: unavailable (items only) --- request 3 (items; Stored replacement before next turn) -- +Settings: same as window 1 +-- request 3 (compaction; Repeated /compact after resume) -- 00:additional_tools/developer (3; hash=e019c949fc483f83): - namespace/functions; hash=03340277aa23936f - custom/exec @@ -194,8 +195,10 @@ Settings: unavailable (items only) 06:message/user: Plan the release. 07:compaction:encrypted=true; chars=31; hash=8255caf138edede8 +08:agent_message +09:compaction_trigger -## Window 3 (after request 3) +## Window 3 (after request 3: input diverged at item 02) Settings: same as window 1 -- request 4 (turn; First turn after /compact) -- 00:additional_tools/developer (3; hash=e019c949fc483f83): @@ -233,8 +236,21 @@ Settings: same as window 1 02:message/developer[2]: [01] - Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `danger-full-access`...ll commands are permitted. Network access is enabled. [hash=c134061cebb94a1b] - Approval policy is currently never. Do not provide the `sandbox_permissions` for any reason, commands will be rejected. + Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `read-only`: The san... permits reading files. Network access is restricted. [hash=be7da4425f89053c] + # Escalation Requests + + Commands are run outside the sandbox if they are approved by the user, or match an existing rule that al...hell control operators, including but not limited to: [hash=d24ebf5fa4e4b9bb] + + - Pipes: | + - Logical operators: &&, || + + NEVER provide a prefix_rule if your command uses a heredoc or herestring. + + ### Examples + Good examples of prefixes: + - ["npm", "run", "dev"] + - ["gh", "pr", "check"] + - ["cargo", "test"] [02] # Collaboration Mode: Default @@ -274,14 +290,30 @@ Settings: same as window 1 Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] 05:message/user: - + - <...e="unrestricted" /> [hash=d40b2f45cb028888] + ...ntry> [hash=8fc9d186c1133657] 06:message/user: Plan the release. -07:compaction:encrypted=true; chars=31; hash=8255caf138edede8 -08:message/user: +07:compaction:encrypted=true; chars=22; hash=5180312a53382d23 +08:agent_message +09:message/developer: + + Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `danger-full-access`...ll commands are permitted. Network access is enabled. [hash=c134061cebb94a1b] + Approval policy is currently never. Do not provide the `sandbox_permissions` for any reason, commands will be rejected. + +10:message/user: + + + + [hash=46e2a5c52f0891de] + +11:message/user: Continue with the first step. +-- request 5 (compaction; Compaction after mail was sampled) -- +12:message/assistant: + Start by validating the release candidate. +13:compaction_trigger diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__injected_agent_mail__injected_agent_message_interrupting_sampling_survives_mid_turn_compaction.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__injected_agent_mail__injected_agent_message_interrupting_sampling_survives_mid_turn_compaction.snap new file mode 100644 index 0000000000..18d0e6c7b8 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__injected_agent_mail__injected_agent_message_interrupting_sampling_survives_mid_turn_compaction.snap @@ -0,0 +1,93 @@ +--- +source: core/tests/suite/scenarios_injected_agent_mail.rs +expression: "context_snapshot::format_context_snapshot(\"An injected agent message that interrupts sampling is included in the first request after mid-turn compaction.\",\n&entries, &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: An injected agent message that interrupts sampling is included in the first request after mid-turn compaction. + +## Window 1 +-- request 1 (request) -- +00:message/developer: + You are a coding agent running in the Codex CLI, a terminal-based coding assistant. Codex CLI is an open...I. You are expected to be precise, safe, and helpful. [hash=5125daec1ed78d99] + + Your capabilities: + + - Receive user prompts and other context provided by the harness, such as files in the workspace. + - Communicate with the user by streaming thinking & responses, and by making & updating plans. + - Emit function calls to run terminal commands and apply patches. Depending on how this specific run is ... More on this in the "Sandbox and approvals" section. [hash=31cf026fa7b25d6d] + + + ## Shell commands + + When using the shell, you must adhere to the following guidelines: + + - When searching for text or files, prefer using `rg` or `rg --files` respectively because `rg` is much ...he `rg` command is not found, then use alternatives.) [hash=67af3ff3b546e946] + - Do not use python scripts to attempt to output larger chunks of a file. + + +01:message/developer: + +02:message/user: + +03:message/user: + Plan the next action. + +## Window 2 (after request 1: settings changed (parallel_tool_calls, tools)) +-- request 2 (request) -- +00:message/developer: + You are a coding agent running in the Codex CLI, a terminal-based coding assistant. Codex CLI is an open...I. You are expected to be precise, safe, and helpful. [hash=5125daec1ed78d99] + + Your capabilities: + + - Receive user prompts and other context provided by the harness, such as files in the workspace. + - Communicate with the user by streaming thinking & responses, and by making & updating plans. + - Emit function calls to run terminal commands and apply patches. Depending on how this specific run is ... More on this in the "Sandbox and approvals" section. [hash=31cf026fa7b25d6d] + + + ## Shell commands + + When using the shell, you must adhere to the following guidelines: + + - When searching for text or files, prefer using `rg` or `rg --files` respectively because `rg` is much ...he `rg` command is not found, then use alternatives.) [hash=67af3ff3b546e946] + - Do not use python scripts to attempt to output larger chunks of a file. + + +01:message/developer: + +02:message/user: + +03:message/user: + Plan the next action. +04:reasoning:summary=reasoning reasoning reasoning reasoning reasoning reasoning reasoning reasoning reasoning reasoning reas...ing reasoning reasoning reasoning reasoning reasoning [hash=ca3a0b1992c1437f]:encrypted=true +05:message/user: + + +## Window 3 (after request 2: input diverged at item 04, settings changed (parallel_tool_calls, tools)) +-- request 3 (request) -- +00:message/developer: + You are a coding agent running in the Codex CLI, a terminal-based coding assistant. Codex CLI is an open...I. You are expected to be precise, safe, and helpful. [hash=5125daec1ed78d99] + + Your capabilities: + + - Receive user prompts and other context provided by the harness, such as files in the workspace. + - Communicate with the user by streaming thinking & responses, and by making & updating plans. + - Emit function calls to run terminal commands and apply patches. Depending on how this specific run is ... More on this in the "Sandbox and approvals" section. [hash=31cf026fa7b25d6d] + + + ## Shell commands + + When using the shell, you must adhere to the following guidelines: + + - When searching for text or files, prefer using `rg` or `rg --files` respectively because `rg` is much ...he `rg` command is not found, then use alternatives.) [hash=67af3ff3b546e946] + - Do not use python scripts to attempt to output larger chunks of a file. + + +01:message/developer: + +02:message/user: + +03:message/user: + Plan the next action. +04:message/user: + + AUTO_COMPACT_SUMMARY +05:agent_message diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 6acdf082fd..27c81a3cec 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -66,6 +66,15 @@ pub struct ResponseItemEnvelope { /// #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)] pub struct CodexHarnessMetadata { + /// Thread-local append sequence. Compaction keeps the identities of retained inputs. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub history_sequence: Option, + + /// Highest input sequence included in a regular request that recorded model output. + /// Compaction carries this checkpoint without advancing it. Never model-visible. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sampled_through_sequence: Option, + /// Complete retained records actually delivered by this Guardian message. Host-only. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub guardian_sources: Vec, diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index 6c8784fe5f..39f6d7535b 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -359,6 +359,8 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()> item: RolloutItem::ResponseItem(ResponseItemEnvelope { item: response_item.clone(), metadata: Some(CodexHarnessMetadata { + history_sequence: Some(9), + sampled_through_sequence: Some(7), client_authored: true, history_truncation_token_limit: Some(20_000), inherited_user_message: true, @@ -376,6 +378,8 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()> "type": "response_item", "payload": response_item, "metadata": { + "history_sequence": 9, + "sampled_through_sequence": 7, "client_authored": true, "fallback_token_limit_override": 20_000, "inherited_user_message": true, @@ -391,6 +395,8 @@ fn response_item_envelope_stores_metadata_beside_rollout_payload() -> Result<()> assert_eq!( envelope.metadata, Some(CodexHarnessMetadata { + history_sequence: Some(9), + sampled_through_sequence: Some(7), client_authored: true, history_truncation_token_limit: Some(20_000), inherited_user_message: true,