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!(