From 6574bcdba89a630257c659418e241a3d7a242f6b Mon Sep 17 00:00:00 2001 From: ioanbudea-OAI Date: Tue, 27 Oct 2026 17:07:18 +0000 Subject: [PATCH] Defer synchronous Guardian heartbeat compression to preparation (#57519) ## What changed Keep original scheduled messages and heartbeat provenance during synchronous transcript collection. Add `prepare_sync_context` to compress repeated heartbeat instructions across the complete collected transcript before selecting the requested suffix, preserving reference numbering in full and delta prompts. Route synchronous reviewer prompts through this preparation step and reject composition of unprepared scheduled messages. Move transcript profiles and REPL evidence rendering into the preparation module. Asynchronous heartbeat compression remains in collection. ## Testing Extend heartbeat transcript coverage to compare prepared full and delta prompts with the existing rendering and verify that direct composition rejects unprepared scheduled messages. GitOrigin-RevId: c9824b8862021301a1f537a74f83bc7eb20aa0ad --- codex-rs/core/src/guardian/prompt.rs | 3 +- codex-rs/ext/guardian-reviewer/src/prompt.rs | 28 ++----- codex-rs/guardian-context/src/composition.rs | 17 +++- codex-rs/guardian-context/src/entry.rs | 14 +++- codex-rs/guardian-context/src/lib.rs | 6 +- .../guardian-context/src/preparation/mod.rs | 44 +++++++++++ .../src/{node_repl.rs => preparation/repl.rs} | 0 .../{profile.rs => preparation/transcript.rs} | 78 ++++++++++++++++++- .../transcript}/window.rs | 0 .../transcript_tests.rs} | 2 + codex-rs/guardian-context/src/transcript.rs | 39 ++++------ .../guardian-context/src/transcript_tests.rs | 49 ++++++++++++ 12 files changed, 228 insertions(+), 52 deletions(-) rename codex-rs/guardian-context/src/{node_repl.rs => preparation/repl.rs} (100%) rename codex-rs/guardian-context/src/{profile.rs => preparation/transcript.rs} (83%) rename codex-rs/guardian-context/src/{profile => preparation/transcript}/window.rs (100%) rename codex-rs/guardian-context/src/{profile_tests.rs => preparation/transcript_tests.rs} (99%) diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 92bcb9b249..77ae4a197b 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -208,7 +208,8 @@ pub(crate) fn render_guardian_transcript_entries( .map(|item| match item.content { codex_guardian_context::TranscriptContent::Text(text) => text, codex_guardian_context::TranscriptContent::Record(record) => record.to_string(), - codex_guardian_context::TranscriptContent::NativeMessage(_) => { + codex_guardian_context::TranscriptContent::ScheduledMessage(_) + | codex_guardian_context::TranscriptContent::NativeMessage(_) => { panic!("expected text transcript") } }) diff --git a/codex-rs/ext/guardian-reviewer/src/prompt.rs b/codex-rs/ext/guardian-reviewer/src/prompt.rs index 6e0bc2c76a..788951ccb3 100644 --- a/codex-rs/ext/guardian-reviewer/src/prompt.rs +++ b/codex-rs/ext/guardian-reviewer/src/prompt.rs @@ -1,7 +1,6 @@ //! Composes synchronous review prompts from the host's captured context sections. //! Full/delta selection, empty-transcript text, and cursor accounting preserve the existing layout. -use codex_guardian_context::Budgeted; use codex_guardian_context::CollectedContext; use codex_guardian_context::ComposedContext; use codex_guardian_context::ContextPresentation; @@ -26,27 +25,14 @@ pub fn compose_guardian_prompt( ) -> anyhow::Result { let (selection, transcript_cursor) = mode.select(sections.transcript_entries(), history_version); - let (transcript_entries, offset, placeholder, presentation) = match selection { - TranscriptSelection::Full(entries) => ( - entries, - 0, - "", - ContextPresentation::SyncFull { session_id }, - ), - TranscriptSelection::Delta { entries, offset } => ( - entries, - offset, - "", - ContextPresentation::SyncDelta { session_id }, - ), + let (offset, presentation) = match selection { + TranscriptSelection::Full(_) => (0, ContextPresentation::SyncFull { session_id }), + TranscriptSelection::Delta { offset, .. } => { + (offset, ContextPresentation::SyncDelta { session_id }) + } }; - let mut transcript = profile.prepare_transcript(transcript_entries, offset); - if transcript_entries.is_empty() { - transcript.items.push(Budgeted::required( - codex_guardian_context::TranscriptContent::Text(placeholder.to_owned()), - )); - } - let context = sections.compose(presentation, transcript)?; + let context = + codex_guardian_context::prepare_sync_context(sections, profile, offset, presentation)?; Ok(GuardianPromptItems { context, transcript_cursor, diff --git a/codex-rs/guardian-context/src/composition.rs b/codex-rs/guardian-context/src/composition.rs index f45769342f..9b8acb68c1 100644 --- a/codex-rs/guardian-context/src/composition.rs +++ b/codex-rs/guardian-context/src/composition.rs @@ -148,7 +148,8 @@ impl CollectedContext { .unwrap_or_default() } - /// Compatibility entrypoint for existing synchronous and asynchronous callers. + /// Compatibility entrypoint for callers without unprepared scheduled messages. + /// Synchronous scheduler evidence requires `prepare_sync_context` before slicing. pub fn compose( mut self, presentation: ContextPresentation<'_>, @@ -156,6 +157,15 @@ impl CollectedContext { ) -> Result { match presentation { ContextPresentation::SyncFull { .. } | ContextPresentation::SyncDelta { .. } => { + if self + .transcript_entries() + .iter() + .any(|entry| matches!(entry.content, TranscriptContent::ScheduledMessage(_))) + { + return Err(SectionError::UnsupportedDelivery { + section: "sync_preparation", + }); + } crate::preparation::prepare_sync_sections(&mut self); self.compose_prepared(presentation, transcript) } @@ -312,6 +322,11 @@ impl CollectedContext { )); continue; } + TranscriptContent::ScheduledMessage(_) => { + return Err(SectionError::UnsupportedDelivery { + section: "sync_preparation", + }); + } TranscriptContent::Text(text) => ContentItem::InputText { text: if session_id.is_some() { format!("{}{text}\n", if index == 0 { "" } else { "\n" }) diff --git a/codex-rs/guardian-context/src/entry.rs b/codex-rs/guardian-context/src/entry.rs index ae6e6bbd0b..e5a3497b49 100644 --- a/codex-rs/guardian-context/src/entry.rs +++ b/codex-rs/guardian-context/src/entry.rs @@ -1,8 +1,8 @@ //! Structured transcript evidence shared by synchronous and asynchronous Guardian. //! //! Entry kinds preserve source attribution for consumer-specific retention and -//! rendering. User text and manual approvals stay complete for admission; other text is -//! capped during collection, with its original size retained for accounting. +//! rendering. Sync entries remain complete until request preparation; async collection +//! retains its existing caps. Original sizes remain available for accounting. /// Semantic role of one parent-conversation transcript entry. #[derive(Clone, Debug, Eq, PartialEq)] @@ -44,6 +44,8 @@ impl ConversationTranscriptEntryKind { #[derive(Clone, PartialEq)] pub enum TranscriptContent { Text(String), + /// Original scheduler input, compressed only during synchronous preparation. + ScheduledMessage(ScheduledTranscriptMessage), /// Host-attributed text prepared by the profile, rendered at delivery. Record(crate::TranscriptRecord), NativeMessage(Box), @@ -53,6 +55,7 @@ impl std::fmt::Debug for TranscriptContent { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str(match self { Self::Text(_) => "Text(..)", + Self::ScheduledMessage(_) => "ScheduledMessage(..)", Self::Record(_) => "Record(..)", Self::NativeMessage(_) => "NativeMessage(..)", }) @@ -80,3 +83,10 @@ pub struct RetainedTranscriptSource { pub order: String, pub source: codex_history::RetainedSource, } + +/// Original scheduler text and the provenance needed to compare heartbeat instructions. +#[derive(Clone, PartialEq)] +pub struct ScheduledTranscriptMessage { + pub(crate) text: String, + pub(crate) heartbeat: Option, +} diff --git a/codex-rs/guardian-context/src/lib.rs b/codex-rs/guardian-context/src/lib.rs index 7e0b9905b5..61e0fa9bbc 100644 --- a/codex-rs/guardian-context/src/lib.rs +++ b/codex-rs/guardian-context/src/lib.rs @@ -92,25 +92,25 @@ pub use cursor::TranscriptCursor; pub use cursor::TranscriptMode; pub use cursor::TranscriptSelection; mod preparation; -mod profile; pub use codex_protocol::TranscriptFormat; pub use composition::CollectedContext; pub use composition::ComposedContext; pub use composition::ContextPresentation; pub use composition::PreparedTranscript; +pub use preparation::ContextProfile; +pub use preparation::prepare_sync_context; pub use preparation::prepare_sync_images; pub use preparation::root::PreparedRootContext; pub use preparation::root::prepare_root_context; -pub use profile::ContextProfile; mod authorization; mod entry; mod history; mod images; -mod node_repl; pub use node_repl::NodeReplContext; pub use node_repl::NodeReplResponse; pub use node_repl::NodeReplReviewEvidenceMode; pub use node_repl::RenderedNodeReplEvidence; +use preparation::repl as node_repl; mod permissions; pub use images::TranscriptImageInput; pub use images::TranscriptImages; diff --git a/codex-rs/guardian-context/src/preparation/mod.rs b/codex-rs/guardian-context/src/preparation/mod.rs index 42f386cc4f..ee283c3cae 100644 --- a/codex-rs/guardian-context/src/preparation/mod.rs +++ b/codex-rs/guardian-context/src/preparation/mod.rs @@ -2,13 +2,20 @@ //! Collection keeps captured sources intact. This owner applies text and image policy; //! delivery checks repeat against current reviewer history after compaction. +pub(crate) mod repl; pub(crate) mod root; +mod transcript; +pub use transcript::ContextProfile; +pub(crate) use transcript::HeartbeatInstruction; +pub(crate) use transcript::HeartbeatReferences; use crate::CollectedContext; use crate::ComposedContext; +use crate::ContextPresentation; use crate::ContextSection; use crate::GuardianRootMessage; use crate::RetainedInstructionItem; +use crate::SectionError; use crate::retained_instructions::RetainedInstructionKind; use crate::retained_instructions::USER_OMISSION; use codex_protocol::models::ImageDetail; @@ -23,6 +30,43 @@ pub(crate) const MAX_INSTRUCTION_TOKENS: usize = 900; const MAX_APPROVAL_REASON_TOKENS: usize = 512; const MAX_DESCRIPTION_TOKENS: usize = 400; +/// Prepares the host-selected transcript suffix and all synchronous context sections. +/// Cursor selection remains with the host; its offset addresses the collected entries. +/// No storage, reviewer history or cursor state is changed by this operation. +pub fn prepare_sync_context( + mut context: CollectedContext, + profile: &ContextProfile, + entry_number_offset: usize, + presentation: ContextPresentation<'_>, +) -> Result { + let placeholder = match &presentation { + ContextPresentation::SyncFull { .. } => "", + ContextPresentation::SyncDelta { .. } => "", + ContextPresentation::Async | ContextPresentation::AsyncDelta => { + return Err(SectionError::UnsupportedDelivery { + section: "sync_preparation", + }); + } + }; + transcript::prepare_sync_heartbeats(&mut context); + let entries = context + .transcript_entries() + .get(entry_number_offset..) + .ok_or(SectionError::UnsupportedDelivery { + section: "conversation_transcript", + })?; + let mut transcript = profile.prepare_transcript(entries, entry_number_offset); + if entries.is_empty() { + transcript + .items + .push(crate::Budgeted::required(crate::TranscriptContent::Text( + placeholder.to_owned(), + ))); + } + prepare_sync_sections(&mut context); + context.compose_prepared(presentation, transcript) +} + /// Applies synchronous section policy to the already bounded collection snapshot. /// The host prepares its borrowed transcript selection before consuming the collection. pub(crate) fn prepare_sync_sections(context: &mut CollectedContext) { diff --git a/codex-rs/guardian-context/src/node_repl.rs b/codex-rs/guardian-context/src/preparation/repl.rs similarity index 100% rename from codex-rs/guardian-context/src/node_repl.rs rename to codex-rs/guardian-context/src/preparation/repl.rs diff --git a/codex-rs/guardian-context/src/profile.rs b/codex-rs/guardian-context/src/preparation/transcript.rs similarity index 83% rename from codex-rs/guardian-context/src/profile.rs rename to codex-rs/guardian-context/src/preparation/transcript.rs index 28b3fd805a..78a5641331 100644 --- a/codex-rs/guardian-context/src/profile.rs +++ b/codex-rs/guardian-context/src/preparation/transcript.rs @@ -6,6 +6,7 @@ //! The host selects line or JSON rendering; JSON keeps embedded headers inside text. use codex_protocol::protocol::TruncationPolicy; +use std::collections::HashMap; use crate::BudgetPriority; use crate::Budgeted; @@ -175,6 +176,7 @@ impl ContextProfile { }; let (content, tokens, retained_bytes) = match content.as_ref() { TranscriptContent::Text(text) => prepare_text(text), + TranscriptContent::ScheduledMessage(message) => prepare_text(&message.text), TranscriptContent::Record(record) => prepare_text(record.text()), TranscriptContent::NativeMessage(message) => ( TranscriptContent::NativeMessage(message.clone()), @@ -320,6 +322,7 @@ fn prepare_sync_content<'a>( ) -> std::borrow::Cow<'a, TranscriptContent> { let text = match &entry.content { TranscriptContent::Text(text) => text.as_str(), + TranscriptContent::ScheduledMessage(message) => message.text.as_str(), TranscriptContent::Record(record) => record.text(), TranscriptContent::NativeMessage(_) => { // Native ciphertext stays whole, or leaves an omission at the same position. @@ -348,6 +351,79 @@ fn prepare_sync_content<'a>( std::borrow::Cow::Owned(TranscriptContent::Text(crate::truncate_text(text, tokens))) } +#[derive(Clone, PartialEq)] +pub(crate) struct HeartbeatInstruction { + automation_id: String, + timestamp: String, + instructions: String, +} + +impl From> for HeartbeatInstruction { + fn from(value: codex_history::Heartbeat<'_>) -> Self { + Self { + automation_id: value.automation_id.to_owned(), + timestamp: value.timestamp.to_owned(), + instructions: value.instructions.to_owned(), + } + } +} + +#[derive(Default)] +pub(crate) struct HeartbeatReferences { + versions: HashMap, +} + +impl HeartbeatReferences { + /// Returns whether the representation no longer proves complete source delivery. + pub(crate) fn prepare( + &mut self, + heartbeat: Option<&HeartbeatInstruction>, + text: &mut String, + number: usize, + ) -> bool { + let Some(heartbeat) = heartbeat else { + self.versions.clear(); + return false; + }; + match self.versions.get(&heartbeat.automation_id) { + Some((instructions, previous)) if *instructions == heartbeat.instructions => { + let automation_id = &heartbeat.automation_id; + let timestamp = &heartbeat.timestamp; + *text = format!( + "Scheduled automation {automation_id} ran at {timestamp}. Instructions unchanged from transcript entry [{previous}]; this is a replay of that earlier instruction, not a new human instruction. If the referenced instructions are unavailable, do not infer authorization from this reference." + ); + true + } + _ => { + self.versions.insert( + heartbeat.automation_id.clone(), + (heartbeat.instructions.clone(), number), + ); + false + } + } + } +} + +/// Runs over the complete collected transcript so delta references keep their old targets. +pub(super) fn prepare_sync_heartbeats(context: &mut crate::CollectedContext) { + for section in &mut context.sections { + let crate::ContextSection::ConversationTranscript { items } = section else { + continue; + }; + let mut references = HeartbeatReferences::default(); + for (index, entry) in items.iter_mut().enumerate() { + let crate::TranscriptContent::ScheduledMessage(message) = &mut entry.content else { + continue; + }; + if references.prepare(message.heartbeat.as_ref(), &mut message.text, index + 1) { + entry.retained_source = None; + } + entry.content = crate::TranscriptContent::Text(std::mem::take(&mut message.text)); + } + } +} + #[cfg(test)] -#[path = "profile_tests.rs"] +#[path = "transcript_tests.rs"] mod tests; diff --git a/codex-rs/guardian-context/src/profile/window.rs b/codex-rs/guardian-context/src/preparation/transcript/window.rs similarity index 100% rename from codex-rs/guardian-context/src/profile/window.rs rename to codex-rs/guardian-context/src/preparation/transcript/window.rs diff --git a/codex-rs/guardian-context/src/profile_tests.rs b/codex-rs/guardian-context/src/preparation/transcript_tests.rs similarity index 99% rename from codex-rs/guardian-context/src/profile_tests.rs rename to codex-rs/guardian-context/src/preparation/transcript_tests.rs index 8ce8cec864..6ebf92d76b 100644 --- a/codex-rs/guardian-context/src/profile_tests.rs +++ b/codex-rs/guardian-context/src/preparation/transcript_tests.rs @@ -1,3 +1,5 @@ +#![cfg(test)] + //! Profile retention keeps the existing sync and async evidence priorities. use super::*; diff --git a/codex-rs/guardian-context/src/transcript.rs b/codex-rs/guardian-context/src/transcript.rs index edc0ca9336..12cc688b19 100644 --- a/codex-rs/guardian-context/src/transcript.rs +++ b/codex-rs/guardian-context/src/transcript.rs @@ -171,7 +171,7 @@ fn collect_transcript_for_review( ) -> Vec { let mut entries = Vec::new(); let mut tool_names_by_call_id = HashMap::new(); - let mut heartbeat_versions = HashMap::new(); + let mut heartbeat_references = crate::preparation::HeartbeatReferences::default(); // Positional legacy labels cannot establish reusable delivery proof, including // through transcript copies. The separate retained section stays authoritative. let retained_context = history @@ -451,28 +451,14 @@ fn collect_transcript_for_review( continue; } let original_bytes = text.len(); - if let Some(heartbeat) = codex_history::Heartbeat::from_message(item) { - match heartbeat_versions.get(heartbeat.automation_id) { - Some((instructions, number)) if *instructions == heartbeat.instructions => { - // A reference no longer delivers the complete source instruction. - source = None; - text = format!( - "Scheduled automation {} ran at {}. Instructions unchanged from transcript entry [{}]; this is a replay of that earlier instruction, not a new human instruction. If the referenced instructions are unavailable, do not infer authorization from this reference.", - heartbeat.automation_id, heartbeat.timestamp, number - ); - } - _ => { - heartbeat_versions.insert( - heartbeat.automation_id, - (heartbeat.instructions, entries.len() + 1), - ); - } - } - } else if codex_history::UserInputOrigin::from_message(item) - == codex_history::UserInputOrigin::Heartbeat + let scheduled = codex_history::UserInputOrigin::from_message(item) + == codex_history::UserInputOrigin::Heartbeat; + let heartbeat = codex_history::Heartbeat::from_message(item).map(Into::into); + if target == crate::ContextTarget::Async + && scheduled + && heartbeat_references.prepare(heartbeat.as_ref(), &mut text, entries.len() + 1) { - // Unknown scheduler envelopes may change instructions; do not bridge them. - heartbeat_versions.clear(); + source = None; } let text = match &kind { ConversationTranscriptEntryKind::User | ConversationTranscriptEntryKind::Developer => { @@ -546,7 +532,14 @@ fn collect_transcript_for_review( ) }), kind, - content: TranscriptContent::Text(text), + content: if target == crate::ContextTarget::Sync && scheduled { + TranscriptContent::ScheduledMessage(crate::entry::ScheduledTranscriptMessage { + text, + heartbeat, + }) + } else { + TranscriptContent::Text(text) + }, original_bytes, }); } diff --git a/codex-rs/guardian-context/src/transcript_tests.rs b/codex-rs/guardian-context/src/transcript_tests.rs index 785cbe4e2d..b19df61f27 100644 --- a/codex-rs/guardian-context/src/transcript_tests.rs +++ b/codex-rs/guardian-context/src/transcript_tests.rs @@ -72,6 +72,55 @@ fn heartbeat_references_preserve_positions_changes_and_human_messages() { assert!(text.contains("unchanged from transcript entry [1]")); assert!(!text.contains("Monitor only.")); } + // Sync collection keeps originals. Preparation compresses before slicing, so + // full and delta requests retain exactly the existing reference text and numbering. + let raw = super::collect_transcript_for_target( + &history, + &transcript_config(), + crate::ContextTarget::Sync, + ); + let collected = crate::CollectedContext { + sections: vec![ContextSection::ConversationTranscript { items: raw.clone() }], + }; + let prepared = crate::ContextProfile::synchronous().prepare_transcript( + collected.transcript_entries(), + /*entry_number_offset*/ 0, + ); + assert!(matches!( + collected.compose( + crate::ContextPresentation::SyncFull { + session_id: "heartbeat" + }, + prepared, + ), + Err(crate::SectionError::UnsupportedDelivery { + section: "sync_preparation" + }) + )); + for offset in [0, 1, 3] { + let [expected, actual] = [entries.clone(), raw.clone()].map(|items| { + let presentation = if offset == 0 { + crate::ContextPresentation::SyncFull { + session_id: "heartbeat", + } + } else { + crate::ContextPresentation::SyncDelta { + session_id: "heartbeat", + } + }; + crate::prepare_sync_context( + crate::CollectedContext { + sections: vec![crate::ContextSection::ConversationTranscript { items }], + }, + &crate::ContextProfile::synchronous(), + offset, + presentation, + ) + .unwrap() + .into_messages() + }); + assert_eq!(actual, expected); + } // A rebuilt window starts with a full body, never a dangling old reference. let rebuilt = collect_transcript(&history[3..].to_vec(), &transcript_config()); assert!(