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

Defer synchronous Guardian heartbeat compression to preparation (#57519)

Browse Source 
## 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
This commit is contained in:
ioanbudea-OAI
2026-10-27 17:19:51 +00:00
committed by copyberry
parent 4c54667b40
commit 6574bcdba8
12 changed files with 228 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/prompt.rs
+2 -1
View File
@@ -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")
}
})
codex-rs/ext/guardian-reviewer/src/prompt.rs
+7 -21
View File
@@ -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<GuardianPromptItems> {
let (selection, transcript_cursor) =
mode.select(sections.transcript_entries(), history_version);
let (transcript_entries, offset, placeholder, presentation) = match selection {
TranscriptSelection::Full(entries) => (
entries,
0,
"<no retained transcript entries>",
ContextPresentation::SyncFull { session_id },
),
TranscriptSelection::Delta { entries, offset } => (
entries,
offset,
"<no retained transcript delta entries>",
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,
codex-rs/guardian-context/src/composition.rs
+16 -1
View File
@@ -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<ComposedContext, SectionError> {
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" })
codex-rs/guardian-context/src/entry.rs
+12 -2
View File
@@ -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<codex_protocol::models::ResponseItem>),
@@ -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<crate::preparation::HeartbeatInstruction>,
}
codex-rs/guardian-context/src/lib.rs
+3 -3
View File
@@ -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;
codex-rs/guardian-context/src/preparation/mod.rs
+44
View File
@@ -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<ComposedContext, SectionError> {
let placeholder = match &presentation {
ContextPresentation::SyncFull { .. } => "<no retained transcript entries>",
ContextPresentation::SyncDelta { .. } => "<no retained transcript delta entries>",
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) {
codex-rs/guardian-context/src/node_repl.rs → codex-rs/guardian-context/src/preparation/repl.rs
View File
codex-rs/guardian-context/src/profile.rs → codex-rs/guardian-context/src/preparation/transcript.rs
+77 -1
View File
@@ -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<codex_history::Heartbeat<'_>> 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<String, (String, usize)>,
}
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;
codex-rs/guardian-context/src/profile/window.rs → codex-rs/guardian-context/src/preparation/transcript/window.rs
View File
codex-rs/guardian-context/src/profile_tests.rs → codex-rs/guardian-context/src/preparation/transcript_tests.rs
+2
View File
@@ -1,3 +1,5 @@
#![cfg(test)]
//! Profile retention keeps the existing sync and async evidence priorities.
use super::*;
codex-rs/guardian-context/src/transcript.rs
+16 -23
View File
@@ -171,7 +171,7 @@ fn collect_transcript_for_review(
) -> Vec<ConversationTranscriptEntry> {
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,
});
}
codex-rs/guardian-context/src/transcript_tests.rs
+49
View File
@@ -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!(