diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 59aa85abaa..1767ac344c 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3659,8 +3659,11 @@ dependencies = [ "codex-otel", "codex-prompts", "codex-protocol", + "codex-protocol-test-support", "codex-shell-command", + "codex-utils-output-truncation", "codex-utils-path-uri", + "codex-utils-string", "opentelemetry", "opentelemetry_sdk", "pretty_assertions", diff --git a/codex-rs/core/src/agent/control/sender_context.rs b/codex-rs/core/src/agent/control/sender_context.rs index b056b43bc7..9a0cd514f9 100644 --- a/codex-rs/core/src/agent/control/sender_context.rs +++ b/codex-rs/core/src/agent/control/sender_context.rs @@ -3,12 +3,8 @@ //! quoted delegation text cannot establish sender provenance. Unloaded senders are read //! through the host's authenticated thread store, without resuming their runtime. -use crate::context::GuardianSenderExchange; use crate::context::GuardianSenderMessages; use crate::context_manager::ContextManager; -use codex_guardian_context::GuardianRootMessage; -use codex_history::RetainedContextEntry; -use codex_history::RetainedContextOrder; use codex_history::RolloutItem; use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; @@ -111,38 +107,7 @@ impl LocalAgentRuntime { } }; if let Some(context) = context { - let mut exchanges = Vec::new(); - let mut assistant = None; - for (order, entry) in context.ordered_entries() { - match (order, entry) { - ( - RetainedContextOrder::Local(_), - RetainedContextEntry::UserMessage(message), - ) => exchanges.push((message, assistant.take())), - ( - RetainedContextOrder::Local(_), - RetainedContextEntry::AssistantMessage(message), - ) => assistant = Some(message), - (RetainedContextOrder::Inherited(_), _) - | (_, RetainedContextEntry::VerifiedAnswer(_)) => {} - } - } - fragment.messages = exchanges - .into_iter() - .rev() - .take(/*n*/ 3) - .map(|(user, assistant)| GuardianSenderExchange { - user: user.complete.then(|| user.text.clone()), - assistant: assistant.map(|message| { - if message.complete { - GuardianRootMessage::Assistant(message.text.clone()) - } else { - GuardianRootMessage::IncompleteAssistantContext - } - }), - }) - .collect(); - fragment.messages.reverse(); + fragment.messages = codex_guardian_reviewer::guardian_sender_exchanges(&context); } } Some(fragment) diff --git a/codex-rs/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index f1f9793a81..b1bbb9094e 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/codex-rs/core/src/agent/control/user_authorization.rs @@ -10,9 +10,6 @@ //! Handoff projection preserves saved heartbeat instructions and the active turn's skills. use std::borrow::Cow; -use std::collections::hash_map::DefaultHasher; -use std::hash::Hash; -use std::hash::Hasher; use super::LocalAgentControl; use crate::codex_thread::GuardianRootMessage; @@ -307,30 +304,12 @@ impl LocalAgentControl { .services .thread_extension_data .get_or_init(GuardianReviewEvidence::default); - let mut authorization_version = root_evidence.authorization_version(history.as_ref()); - if prepared.missing_root_instructions { - authorization_version.retained_context_complete = false; - } - let mut review_context_revision = history.guardian_review_context_revision(); - if let Some(selected) = &prepared.handoff_selection { - let mut revision = DefaultHasher::new(); - review_context_revision.hash(&mut revision); - selected.hash(&mut revision); - review_context_revision = revision.finish(); - } - let trusted_skill_paths = prepared - .latest_user_turn_id - .as_deref() - .map(|turn_id| root_evidence.trusted_skill_paths(turn_id)) - .unwrap_or_default(); - Some(GuardianRootSnapshot { + Some(root_evidence.root_snapshot( root_thread_id, history_reset_version, - authorization_version, - review_context_revision, - messages: prepared.messages, - trusted_skill_paths, - request_candidates: Some(request_candidates), - }) + history.as_ref(), + prepared, + request_candidates, + )) } } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index c7e27be989..95b0158045 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -187,30 +187,8 @@ pub enum ConfigRefreshOutcome { pub use codex_guardian_context::GuardianRootMessage; -/// Authorization state that changes on genuine user input or history resets. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub struct GuardianAuthorizationVersion { - /// User-message/reset revision, preserved across compaction and internal context. - pub user_message_revision: u64, - /// False when required retained answers or root instructions are unavailable. - pub retained_context_complete: bool, -} - -/// Root authorization state, selected messages and request-local originals from one snapshot. -#[derive(Debug, PartialEq)] -pub struct GuardianRootSnapshot { - /// Authoritative root from which this evidence was captured. - pub root_thread_id: ThreadId, - /// Distinguishes a history reset from additional authorization in the same history. - pub(crate) history_reset_version: u64, - pub authorization_version: GuardianAuthorizationVersion, - /// Confirmed assistant context is relevant to review but cannot authorize an action. - pub review_context_revision: u64, - pub messages: Vec, - /// Request-local originals retained before the bounded compatibility projection. - pub(crate) request_candidates: Option, - pub trusted_skill_paths: Vec, -} +pub use codex_guardian_reviewer::GuardianAuthorizationVersion; +pub use codex_guardian_reviewer::GuardianRootSnapshot; pub struct CodexThread { pub(crate) session: Arc, diff --git a/codex-rs/core/src/context/guardian_review_evidence.rs b/codex-rs/core/src/context/guardian_review_evidence.rs index e30de8f0d4..5cbe209886 100644 --- a/codex-rs/core/src/context/guardian_review_evidence.rs +++ b/codex-rs/core/src/context/guardian_review_evidence.rs @@ -1,190 +1,13 @@ -//! Selects Guardian answer evidence from the review's history snapshot and retains reviews. -//! Retained answers survive restart even while incompatible checkpoints use legacy review. +//! Typed context fragment for bounded, host-supplied synchronous review evidence. +//! Runtime evidence storage and selection belong to the reviewer extension. -use std::collections::BTreeSet; -use std::collections::VecDeque; -use std::sync::Arc; -use std::sync::Mutex; -use std::sync::PoisonError; - -use codex_extension_api::ConversationHistorySnapshot; -use codex_guardian_context::MAX_PREVIOUS_REVIEWS; +use super::ContextualUserFragment; +pub use codex_guardian_reviewer::GuardianReviewEvidence; +pub use codex_guardian_reviewer::GuardianReviewEvidenceRecord; +pub use codex_guardian_reviewer::GuardianUserInputSnapshot; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::input_text; -use codex_protocol::protocol::GuardianAssessmentEvent; -use serde_json::json; - -use super::ContextualUserFragment; -use crate::codex_thread::GuardianAuthorizationVersion; - -const MAX_TRUSTED_SKILLS: usize = 16; -const MAX_TRUSTED_SKILL_PATHS_BYTES: usize = 2_048; - -/// Selected answer fragments and the authorization state they describe. -pub struct GuardianUserInputSnapshot { - pub fragments: Vec, - pub authorization_version: GuardianAuthorizationVersion, -} - -/// Selected answer evidence, verified skill paths, and completed Guardian reviews. -/// -/// This runtime-only evidence is never inserted into the agent's conversation. -/// Only bounded, turn-matched skill paths are exposed to delegated workers; -/// completed reviews remain thread-local, and authorization changes invalidate stale records. -#[derive(Debug, Default)] -pub struct GuardianReviewEvidence { - state: Mutex, -} - -#[derive(Debug, Default)] -struct GuardianReviewEvidenceState { - reviews: VecDeque>, - trusted_skill_turn_id: Option, - trusted_skill_paths: BTreeSet, -} - -impl GuardianReviewEvidence { - /// Reads the selected answer path against the caller's action-time history snapshot. - pub fn user_input_snapshot( - &self, - history: &dyn ConversationHistorySnapshot, - ) -> GuardianUserInputSnapshot { - match history.retained_context() { - Some(context) => { - let answers = codex_guardian_context::render_verified_answers(context); - let authorization_version = GuardianAuthorizationVersion { - user_message_revision: history.user_message_revision(), - retained_context_complete: answers.complete, - }; - GuardianUserInputSnapshot { - fragments: answers.fragments, - authorization_version, - } - } - None => GuardianUserInputSnapshot { - fragments: Vec::new(), - authorization_version: GuardianAuthorizationVersion { - user_message_revision: history.user_message_revision(), - retained_context_complete: true, - }, - }, - } - } - - pub fn authorization_version( - &self, - history: &dyn ConversationHistorySnapshot, - ) -> GuardianAuthorizationVersion { - self.user_input_snapshot(history).authorization_version - } - - /// Records a bounded, verified user-owned skill path for one host-owned turn. - pub fn record_trusted_skill(&self, turn_id: &str, path: String) { - if turn_id.is_empty() { - return; - } - let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); - if state.trusted_skill_turn_id.as_deref() != Some(turn_id) { - state.trusted_skill_turn_id = Some(turn_id.to_owned()); - state.trusted_skill_paths.clear(); - } - if state.trusted_skill_paths.contains(&path) - || state.trusted_skill_paths.len() >= MAX_TRUSTED_SKILLS - || state - .trusted_skill_paths - .iter() - .map(String::len) - .sum::() - .saturating_add(path.len()) - > MAX_TRUSTED_SKILL_PATHS_BYTES - { - return; - } - state.trusted_skill_paths.insert(path); - } - - /// Returns verified skill paths only for their original host-owned turn. - pub fn trusted_skill_paths(&self, turn_id: &str) -> Vec { - let state = self.state.lock().unwrap_or_else(PoisonError::into_inner); - if state.trusted_skill_turn_id.as_deref() != Some(turn_id) { - return Vec::new(); - } - state.trusted_skill_paths.iter().cloned().collect() - } - - /// Records a genuine allow/deny assessment, not a timeout or fail-closed error. - pub(crate) fn record( - &self, - assessment: &GuardianAssessmentEvent, - action: &str, - authorization_version: GuardianAuthorizationVersion, - review_context_revision: u64, - root_authorization_version: Option, - root_review_context_revision: Option, - ) { - let Some(completed_at_ms) = assessment.completed_at_ms else { - return; - }; - let review = Arc::new(GuardianReviewEvidenceRecord { - delivery_id: codex_protocol::ResponseItemId::new("guardian_review"), - completed_at_ms, - authorization_version, - review_context_revision, - root_authorization_version, - root_review_context_revision, - correlation: json!({ - "review_id": assessment.id, - "turn_id": assessment.turn_id, - "target_item_id": assessment.target_item_id, - "completed_at_ms": completed_at_ms, - }), - decision: json!({ - "status": assessment.status, - "risk_level": assessment.risk_level, - "user_authorization": assessment.user_authorization, - }), - action: action.to_owned(), - rationale: assessment.rationale.clone(), - }); - let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); - state.reviews.push_back(review); - state - .reviews - .make_contiguous() - .sort_by_key(|review| review.completed_at_ms); - while state.reviews.len() > MAX_PREVIOUS_REVIEWS { - state.reviews.pop_front(); - } - } - - /// Freezes the latest completed reviews, oldest first, for one classifier sample. - pub fn snapshot(&self) -> Vec> { - self.state - .lock() - .unwrap_or_else(PoisonError::into_inner) - .reviews - .iter() - .cloned() - .collect() - } -} - -/// Structured synchronous-review evidence retained for Guardian V2 classification. -#[derive(Debug)] -pub struct GuardianReviewEvidenceRecord { - /// Identifies this immutable completion even if the same action is reviewed again. - pub delivery_id: codex_protocol::ResponseItemId, - pub authorization_version: GuardianAuthorizationVersion, - pub review_context_revision: u64, - pub root_authorization_version: Option, - pub root_review_context_revision: Option, - completed_at_ms: i64, - pub correlation: serde_json::Value, - pub decision: serde_json::Value, - pub action: String, - pub rationale: Option, -} /// A bounded, host-supplied sync-review record for async classifier input only. #[derive(Clone, Debug)] diff --git a/codex-rs/core/src/context/guardian_sender_messages.rs b/codex-rs/core/src/context/guardian_sender_messages.rs index 60cc562bf4..9f6054b03a 100644 --- a/codex-rs/core/src/context/guardian_sender_messages.rs +++ b/codex-rs/core/src/context/guardian_sender_messages.rs @@ -3,6 +3,7 @@ use super::ContextualUserFragment; use codex_guardian_context::GuardianRootMessage; +pub(crate) use codex_guardian_reviewer::GuardianSenderExchange; use codex_protocol::ThreadId; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; @@ -16,12 +17,6 @@ pub(crate) struct GuardianSenderMessages { pub messages: Vec, } -/// Recorded adjacency provides context, not a verified question/answer association. -pub(crate) struct GuardianSenderExchange { - pub user: Option, - pub assistant: Option, -} - impl ContextualUserFragment for GuardianSenderMessages { fn content_kind(&self) -> ContentItemKind { ContentItemKind("guardian.sender_messages".to_owned()) diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index 6e66f74239..0071ad8f6c 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/codex-rs/core/src/context/mod.rs @@ -93,7 +93,6 @@ pub use guardian_review_evidence::GuardianReviewEvidence; pub use guardian_review_evidence::GuardianReviewEvidenceFragment; pub use guardian_review_evidence::GuardianReviewEvidenceRecord; pub use guardian_review_evidence::GuardianUserInputSnapshot; -pub(crate) use guardian_sender_messages::GuardianSenderExchange; pub(crate) use guardian_sender_messages::GuardianSenderMessages; pub(crate) use hook_additional_context::HookAdditionalContext; pub(crate) use image_resize_notice::ImageResizeNotice; diff --git a/codex-rs/core/src/context/node_repl_review_evidence.rs b/codex-rs/core/src/context/node_repl_review_evidence.rs index 1558573a76..bb4aa0411a 100644 --- a/codex-rs/core/src/context/node_repl_review_evidence.rs +++ b/codex-rs/core/src/context/node_repl_review_evidence.rs @@ -1,31 +1,12 @@ -//! Thread-owned capture and bounded storage for completed REPL evidence. -//! Immutable snapshots feed shared rendering; sequence and eviction stay host-owned. -//! Collection shares frozen records; image eviction uses copy-on-write to preserve readers. - -use std::collections::HashSet; -use std::collections::VecDeque; -use std::sync::Arc; -use std::sync::Mutex; -use std::sync::PoisonError; +//! Chooses REPL evidence capture mode from the issuing turn. +//! The reviewer extension owns the bounded store and immutable snapshots. use codex_features::Feature; -use codex_guardian_context::NodeReplContext; -use codex_guardian_context::NodeReplResponse; pub(crate) use codex_guardian_context::NodeReplReviewEvidenceMode; -use codex_protocol::models::ContentItem; -use codex_protocol::models::ImageReference; -use codex_protocol::user_input::UserInput; -use codex_protocol::user_input::UserInput::Image; -use codex_protocol::user_input::UserInput::Text; -use codex_utils_output_truncation::approx_token_count; -use codex_utils_string::take_bytes_at_char_boundary; +pub use codex_guardian_reviewer::NodeReplReviewEvidence; -use crate::guardian::GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS; -use crate::guardian::guardian_truncate_text; use crate::session::turn_context::TurnContext; -const MAX_PROVENANCE_BYTES: usize = 128; - pub(crate) fn node_repl_review_evidence_mode(turn: &TurnContext) -> NodeReplReviewEvidenceMode { let features = &turn.config.features; if turn.model_info().computer_use_review_required() @@ -39,264 +20,3 @@ pub(crate) fn node_repl_review_evidence_mode(turn: &TurnContext) -> NodeReplRevi NodeReplReviewEvidenceMode::Disabled } } - -fn retained_bytes(response: &NodeReplResponse) -> usize { - std::mem::size_of::() - .saturating_add(std::mem::size_of::>()) - .saturating_add(response.provenance.len()) - .saturating_add( - response - .items - .capacity() - .saturating_mul(std::mem::size_of::()), - ) - .saturating_add(response.items.iter().fold(0_usize, |bytes, item| { - bytes.saturating_add(match item { - Text { text, .. } => text.len(), - Image { - image: ImageReference::Inline { image_url }, - .. - } => image_url.len(), - Image { - image: ImageReference::File { file_id }, - .. - } => file_id.len(), - _ => 0, - }) - })) -} - -fn discard_images(response: &mut NodeReplResponse) -> usize { - let before = retained_bytes(response); - response.items.retain(|item| !matches!(item, Image { .. })); - response.items.shrink_to_fit(); - before.saturating_sub(retained_bytes(response)) -} - -#[derive(Debug, Default)] -struct NodeReplReviewEvidenceState { - responses: VecDeque>, - next_sequence: u64, - retained_bytes: usize, - image_capture_enabled: bool, -} - -/// Bounded, thread-scoped evidence collected from nested `node_repl` or `cua_repl` calls. -#[derive(Debug, Default)] -pub struct NodeReplReviewEvidence(Mutex); - -impl NodeReplReviewEvidence { - pub(crate) const MAX_RETAINED_BYTES: usize = 8 * 1024 * 1024; - - /// Enables screenshot capture even when synchronous Guardian transcripts are disabled. - pub fn enable_image_capture(&self) { - self.0 - .lock() - .unwrap_or_else(PoisonError::into_inner) - .image_capture_enabled = true; - } - - pub(crate) fn image_capture_enabled(&self) -> bool { - self.0 - .lock() - .unwrap_or_else(PoisonError::into_inner) - .image_capture_enabled - } - - /// Returns the screenshots currently retained for this thread, oldest first. - pub fn images(&self) -> Vec { - let state = self.0.lock().unwrap_or_else(PoisonError::into_inner); - let mut seen_images = HashSet::new(); - state - .responses - .iter() - .flat_map(|response| &response.items) - .filter_map(|item| match item { - Image { - image: ImageReference::Inline { image_url }, - detail, - } if seen_images.insert(image_url) => Some(ContentItem::InputImage { - image: ImageReference::Inline { - image_url: image_url.clone(), - }, - detail: *detail, - }), - _ => None, - }) - .collect() - } - - pub(crate) fn record( - &self, - tool_name: &str, - cell_id: &str, - call_id: &str, - mut items: Vec, - ) { - if !items.iter().any(|item| matches!(item, Image { .. })) { - let text = items - .into_iter() - .filter_map(|item| match item { - Text { text, .. } => Some(text), - _ => None, - }) - .collect::>() - .join("\n"); - items = Vec::from_iter((!text.is_empty()).then(|| text_input(text))); - } - let mut bounded_items = Vec::with_capacity(items.len()); - let mut rendered_text_tokens = 0_usize; - let mut remaining_source_text_tokens = items - .iter() - .map(|item| match item { - Text { text, .. } => approx_token_count(text), - _ => 0, - }) - .fold(0_usize, usize::saturating_add); - for item in items { - let Text { text, .. } = item else { - if matches!(item, Image { .. }) { - bounded_items.push(item); - } - continue; - }; - if text.trim().is_empty() { - continue; - } - remaining_source_text_tokens = - remaining_source_text_tokens.saturating_sub(approx_token_count(&text)); - let remaining_tokens = - GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS.saturating_sub(rendered_text_tokens); - let reserved_tail_tokens = remaining_source_text_tokens.min(remaining_tokens / 2); - let (text, _) = guardian_truncate_text( - &text.replace(" Self::MAX_RETAINED_BYTES { - discard_images(Arc::make_mut(&mut response)); - if response.items.is_empty() { - return; - } - } - while state - .retained_bytes - .saturating_add(retained_bytes(&response)) - > Self::MAX_RETAINED_BYTES - { - if state.responses.front().is_some_and(|retained| { - retained - .items - .iter() - .any(|item| matches!(item, Text { .. })) - }) { - let reclaimed = state - .responses - .iter_mut() - .find(|retained| { - retained - .items - .iter() - .any(|item| matches!(item, Image { .. })) - }) - .map_or(0, |retained| discard_images(Arc::make_mut(retained))); - state.retained_bytes = state.retained_bytes.saturating_sub(reclaimed); - if reclaimed > 0 || discard_images(Arc::make_mut(&mut response)) > 0 { - if response.items.is_empty() { - return; - } - continue; - } - } - let Some(evicted) = state.responses.pop_front() else { - return; - }; - state.retained_bytes = state - .retained_bytes - .saturating_sub(retained_bytes(&evicted)); - } - state.retained_bytes = state - .retained_bytes - .saturating_add(retained_bytes(&response)); - state.responses.push_back(response); - } - - pub(crate) fn snapshot_since( - &self, - reviewed_sequence: u64, - ) -> Option { - let state = self.0.lock().unwrap_or_else(PoisonError::into_inner); - if state.next_sequence <= reviewed_sequence { - return None; - } - - let responses = state - .responses - .iter() - .filter(|response| response.sequence > reviewed_sequence) - .cloned() - .collect::>(); - let retained_since_review = u64::try_from(responses.len()).unwrap_or(u64::MAX); - - Some(NodeReplReviewEvidenceSnapshot { - omitted_responses: state - .next_sequence - .saturating_sub(reviewed_sequence) - .saturating_sub(retained_since_review), - sequence: state.next_sequence, - responses, - }) - } -} - -fn bounded_provenance(value: &str) -> String { - let sanitized = take_bytes_at_char_boundary(value, MAX_PROVENANCE_BYTES) - .replace(['\n', '\r', '[', ']', '='], "_") - .replace(" UserInput { - UserInput::Text { - text, - text_elements: Vec::new(), - } -} - -pub(crate) struct NodeReplReviewEvidenceSnapshot { - responses: Vec>, - omitted_responses: u64, - pub(crate) sequence: u64, -} - -impl NodeReplReviewEvidenceSnapshot { - pub(crate) fn context(&self, mode: NodeReplReviewEvidenceMode) -> NodeReplContext<'_> { - NodeReplContext { - responses: &self.responses, - omitted_responses: self.omitted_responses, - mode, - } - } -} - -#[cfg(test)] -#[path = "node_repl_review_evidence_tests.rs"] -mod tests; diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index 3db89b948b..9737236bc1 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -48,10 +48,10 @@ pub(crate) use approval_request::GuardianMcpAnnotations; pub(crate) use approval_request::GuardianNetworkAccessTrigger; #[cfg(test)] pub(crate) use approval_request::guardian_approval_request_to_json; +pub(crate) use codex_guardian_reviewer::guardian_truncate_text; pub(crate) use codex_guardian_reviewer::is_basic_session_source; pub(crate) use decision::decide_approval; pub(crate) use decision::spawn_approval_decision; -pub(crate) use prompt::guardian_truncate_text; pub(crate) use review::GuardianReviewOptions; pub(crate) use review::new_guardian_review_id; pub(crate) use review::routes_approval_policy_to_guardian; @@ -74,7 +74,7 @@ const GUARDIAN_MAX_TOOL_ENTRY_TOKENS: usize = { PROFILE.transcript.entry_limits.tool_tokens }; pub(crate) const GUARDIAN_MAX_ROOT_MESSAGE_TOKENS: usize = 900; -pub(crate) const GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS: usize = 6_000; +pub(crate) use codex_guardian_reviewer::GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS; /// Captures review inputs from the issuing step without retaining its MCP bindings or tool router. /// Background network approvals and Unix interception use the active task's resolved settings. diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 4c341ae1de..b2f0469099 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -24,7 +24,6 @@ use crate::context::NodeReplReviewEvidenceMode; use crate::context::is_guardian_context_message; use crate::context::node_repl_review_evidence_mode; use crate::session::session::Session; -use codex_utils_output_truncation::approx_bytes_for_tokens; use super::ApprovalRequestReasons; use super::GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS; @@ -334,10 +333,3 @@ impl SectionHistory for FilteredGuardianHistory<'_> { self.0.render_retained_assistant(message) } } - -pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> (String, bool) { - ( - codex_guardian_context::truncate_text(content, token_cap), - content.len() > approx_bytes_for_tokens(token_cap), - ) -} diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 8594964be4..a879ac16f8 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -1212,18 +1212,6 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() { } } -#[test] -fn guardian_truncate_text_keeps_prefix_suffix_and_xml_marker() { - let content = "prefix ".repeat(200) + &" suffix".repeat(200); - - let (truncated, was_truncated) = guardian_truncate_text(&content, /*token_cap*/ 20); - - assert!(truncated.starts_with("prefix")); - assert!(truncated.contains(" anyhow::Result<()> { let (session, _) = crate::session::tests::make_session_and_context().await; diff --git a/codex-rs/ext/guardian-reviewer/Cargo.toml b/codex-rs/ext/guardian-reviewer/Cargo.toml index e68d48e7ee..b7d6ff6479 100644 --- a/codex-rs/ext/guardian-reviewer/Cargo.toml +++ b/codex-rs/ext/guardian-reviewer/Cargo.toml @@ -27,6 +27,8 @@ codex-prompts = { workspace = true } codex-protocol = { workspace = true } codex-shell-command = { workspace = true } codex-utils-path-uri = { workspace = true } +codex-utils-output-truncation = { workspace = true } +codex-utils-string = { workspace = true } rand = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -35,6 +37,8 @@ tokio-util = { workspace = true, features = ["rt"] } tracing = { workspace = true } [dev-dependencies] +codex-context-fragments = { workspace = true } +codex-protocol-test-support = { workspace = true } opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true, features = ["experimental_metrics_custom_reader", "metrics"] } test-case = "3.3.1" diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index ebe83146d0..7c5d436d86 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -26,6 +26,7 @@ mod execution; mod feedback; mod metrics; mod model; +mod node_repl_review_evidence; mod outcome; mod planned_action; mod pool; @@ -33,6 +34,7 @@ mod prompt; mod reporting; mod retry; mod review; +mod review_evidence; mod review_state; pub use review_state::CheckpointRecovery; pub use review_state::GuardianReviewForkSnapshot; @@ -43,6 +45,7 @@ pub use review_state::PendingNodeReplEvidenceAdmission; pub use review_state::ReviewerSelection; pub use review_state::admit_node_repl_evidence; mod routing; +mod sender_evidence; mod session; mod settings; @@ -116,3 +119,15 @@ pub use reporting::ReviewReport; pub use routing::ReviewRequest; pub use routing::routes_approval_policy_to_guardian; + +pub use node_repl_review_evidence::GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS; +pub use node_repl_review_evidence::NodeReplReviewEvidence; +pub use node_repl_review_evidence::NodeReplReviewEvidenceSnapshot; +pub use prompt::guardian_truncate_text; +pub use review_evidence::GuardianAuthorizationVersion; +pub use review_evidence::GuardianReviewEvidence; +pub use review_evidence::GuardianReviewEvidenceRecord; +pub use review_evidence::GuardianRootSnapshot; +pub use review_evidence::GuardianUserInputSnapshot; +pub use sender_evidence::GuardianSenderExchange; +pub use sender_evidence::guardian_sender_exchanges; diff --git a/codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence.rs b/codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence.rs new file mode 100644 index 0000000000..cf4efa494f --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence.rs @@ -0,0 +1,276 @@ +//! Bounded thread-scoped storage and immutable snapshots of host-captured REPL evidence. +//! Image eviction uses copy-on-write so captured snapshots remain unchanged. + +use std::collections::HashSet; +use std::collections::VecDeque; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::PoisonError; + +use codex_guardian_context::NodeReplContext; +use codex_guardian_context::NodeReplResponse; +use codex_guardian_context::NodeReplReviewEvidenceMode; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ImageReference; +use codex_protocol::user_input::UserInput; +use codex_protocol::user_input::UserInput::Image; +use codex_protocol::user_input::UserInput::Text; +use codex_utils_output_truncation::approx_token_count; +use codex_utils_string::take_bytes_at_char_boundary; + +use crate::prompt::guardian_truncate_text; + +const MAX_PROVENANCE_BYTES: usize = 128; +pub const GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS: usize = 6_000; + +fn retained_bytes(response: &NodeReplResponse) -> usize { + std::mem::size_of::() + .saturating_add(std::mem::size_of::>()) + .saturating_add(response.provenance.len()) + .saturating_add( + response + .items + .capacity() + .saturating_mul(std::mem::size_of::()), + ) + .saturating_add(response.items.iter().fold(0_usize, |bytes, item| { + bytes.saturating_add(match item { + Text { text, .. } => text.len(), + Image { + image: ImageReference::Inline { image_url }, + .. + } => image_url.len(), + Image { + image: ImageReference::File { file_id }, + .. + } => file_id.len(), + _ => 0, + }) + })) +} + +fn discard_images(response: &mut NodeReplResponse) -> usize { + let before = retained_bytes(response); + response.items.retain(|item| !matches!(item, Image { .. })); + response.items.shrink_to_fit(); + before.saturating_sub(retained_bytes(response)) +} + +#[derive(Debug, Default)] +struct NodeReplReviewEvidenceState { + responses: VecDeque>, + next_sequence: u64, + retained_bytes: usize, + image_capture_enabled: bool, +} + +/// Bounded, thread-scoped evidence collected from nested `node_repl` or `cua_repl` calls. +#[derive(Debug, Default)] +pub struct NodeReplReviewEvidence(Mutex); + +impl NodeReplReviewEvidence { + pub const MAX_RETAINED_BYTES: usize = 8 * 1024 * 1024; + + /// Enables screenshot capture even when synchronous Guardian transcripts are disabled. + pub fn enable_image_capture(&self) { + self.0 + .lock() + .unwrap_or_else(PoisonError::into_inner) + .image_capture_enabled = true; + } + + pub fn image_capture_enabled(&self) -> bool { + self.0 + .lock() + .unwrap_or_else(PoisonError::into_inner) + .image_capture_enabled + } + + /// Returns the screenshots currently retained for this thread, oldest first. + pub fn images(&self) -> Vec { + let state = self.0.lock().unwrap_or_else(PoisonError::into_inner); + let mut seen_images = HashSet::new(); + state + .responses + .iter() + .flat_map(|response| &response.items) + .filter_map(|item| match item { + Image { + image: ImageReference::Inline { image_url }, + detail, + } if seen_images.insert(image_url) => Some(ContentItem::InputImage { + image: ImageReference::Inline { + image_url: image_url.clone(), + }, + detail: *detail, + }), + _ => None, + }) + .collect() + } + + pub fn record(&self, tool_name: &str, cell_id: &str, call_id: &str, mut items: Vec) { + if !items.iter().any(|item| matches!(item, Image { .. })) { + let text = items + .into_iter() + .filter_map(|item| match item { + Text { text, .. } => Some(text), + _ => None, + }) + .collect::>() + .join("\n"); + items = Vec::from_iter((!text.is_empty()).then(|| text_input(text))); + } + let mut bounded_items = Vec::with_capacity(items.len()); + let mut rendered_text_tokens = 0_usize; + let mut remaining_source_text_tokens = items + .iter() + .map(|item| match item { + Text { text, .. } => approx_token_count(text), + _ => 0, + }) + .fold(0_usize, usize::saturating_add); + for item in items { + let Text { text, .. } = item else { + if matches!(item, Image { .. }) { + bounded_items.push(item); + } + continue; + }; + if text.trim().is_empty() { + continue; + } + remaining_source_text_tokens = + remaining_source_text_tokens.saturating_sub(approx_token_count(&text)); + let remaining_tokens = + GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS.saturating_sub(rendered_text_tokens); + let reserved_tail_tokens = remaining_source_text_tokens.min(remaining_tokens / 2); + let (text, _) = guardian_truncate_text( + &text.replace(" Self::MAX_RETAINED_BYTES { + discard_images(Arc::make_mut(&mut response)); + if response.items.is_empty() { + return; + } + } + while state + .retained_bytes + .saturating_add(retained_bytes(&response)) + > Self::MAX_RETAINED_BYTES + { + if state.responses.front().is_some_and(|retained| { + retained + .items + .iter() + .any(|item| matches!(item, Text { .. })) + }) { + let reclaimed = state + .responses + .iter_mut() + .find(|retained| { + retained + .items + .iter() + .any(|item| matches!(item, Image { .. })) + }) + .map_or(0, |retained| discard_images(Arc::make_mut(retained))); + state.retained_bytes = state.retained_bytes.saturating_sub(reclaimed); + if reclaimed > 0 || discard_images(Arc::make_mut(&mut response)) > 0 { + if response.items.is_empty() { + return; + } + continue; + } + } + let Some(evicted) = state.responses.pop_front() else { + return; + }; + state.retained_bytes = state + .retained_bytes + .saturating_sub(retained_bytes(&evicted)); + } + state.retained_bytes = state + .retained_bytes + .saturating_add(retained_bytes(&response)); + state.responses.push_back(response); + } + + pub fn snapshot_since(&self, reviewed_sequence: u64) -> Option { + let state = self.0.lock().unwrap_or_else(PoisonError::into_inner); + if state.next_sequence <= reviewed_sequence { + return None; + } + + let responses = state + .responses + .iter() + .filter(|response| response.sequence > reviewed_sequence) + .cloned() + .collect::>(); + let retained_since_review = u64::try_from(responses.len()).unwrap_or(u64::MAX); + + Some(NodeReplReviewEvidenceSnapshot { + omitted_responses: state + .next_sequence + .saturating_sub(reviewed_sequence) + .saturating_sub(retained_since_review), + sequence: state.next_sequence, + responses, + }) + } +} + +fn bounded_provenance(value: &str) -> String { + let sanitized = take_bytes_at_char_boundary(value, MAX_PROVENANCE_BYTES) + .replace(['\n', '\r', '[', ']', '='], "_") + .replace(" UserInput { + UserInput::Text { + text, + text_elements: Vec::new(), + } +} + +pub struct NodeReplReviewEvidenceSnapshot { + responses: Vec>, + omitted_responses: u64, + pub sequence: u64, +} + +impl NodeReplReviewEvidenceSnapshot { + pub fn context(&self, mode: NodeReplReviewEvidenceMode) -> NodeReplContext<'_> { + NodeReplContext { + responses: &self.responses, + omitted_responses: self.omitted_responses, + mode, + } + } +} + +#[cfg(test)] +#[path = "node_repl_review_evidence_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/context/node_repl_review_evidence_tests.rs b/codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence_tests.rs similarity index 99% rename from codex-rs/core/src/context/node_repl_review_evidence_tests.rs rename to codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence_tests.rs index a3faba9870..2347023167 100644 --- a/codex-rs/core/src/context/node_repl_review_evidence_tests.rs +++ b/codex-rs/ext/guardian-reviewer/src/node_repl_review_evidence_tests.rs @@ -1,3 +1,5 @@ +#![cfg(test)] + use codex_protocol::models::ImageReference; use codex_protocol::user_input::UserInput; use codex_protocol_test_support::ContentItemTestExt; diff --git a/codex-rs/ext/guardian-reviewer/src/prompt.rs b/codex-rs/ext/guardian-reviewer/src/prompt.rs index 3cc0f99d76..7ea0ba3587 100644 --- a/codex-rs/ext/guardian-reviewer/src/prompt.rs +++ b/codex-rs/ext/guardian-reviewer/src/prompt.rs @@ -41,3 +41,14 @@ pub fn compose_guardian_prompt( node_repl_evidence_sequence, }) } + +pub fn guardian_truncate_text(content: &str, token_cap: usize) -> (String, bool) { + ( + codex_guardian_context::truncate_text(content, token_cap), + content.len() > codex_utils_output_truncation::approx_bytes_for_tokens(token_cap), + ) +} + +#[cfg(test)] +#[path = "prompt_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-reviewer/src/prompt_tests.rs b/codex-rs/ext/guardian-reviewer/src/prompt_tests.rs new file mode 100644 index 0000000000..63377f6b6e --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/prompt_tests.rs @@ -0,0 +1,15 @@ +#![cfg(test)] + +use super::guardian_truncate_text; + +#[test] +fn guardian_truncate_text_keeps_prefix_suffix_and_xml_marker() { + let content = "prefix ".repeat(200) + &" suffix".repeat(200); + + let (truncated, was_truncated) = guardian_truncate_text(&content, /*token_cap*/ 20); + + assert!(truncated.starts_with("prefix")); + assert!(truncated.contains(", + /// Request-local originals retained before the bounded compatibility projection. + pub request_candidates: Option, + pub trusted_skill_paths: Vec, +} + +/// Selected answer fragments and the authorization state they describe. +pub struct GuardianUserInputSnapshot { + pub fragments: Vec, + pub authorization_version: GuardianAuthorizationVersion, +} + +/// Selected answer evidence, verified skill paths, and completed Guardian reviews. +/// +/// This runtime-only evidence is never inserted into the agent's conversation. +/// Only bounded, turn-matched skill paths are exposed to delegated workers; +/// completed reviews remain thread-local, and authorization changes invalidate stale records. +#[derive(Debug, Default)] +pub struct GuardianReviewEvidence { + state: Mutex, +} + +#[derive(Debug, Default)] +struct GuardianReviewEvidenceState { + reviews: VecDeque>, + trusted_skill_turn_id: Option, + trusted_skill_paths: BTreeSet, +} + +impl GuardianReviewEvidence { + /// Projects prepared root evidence without accessing the host runtime. + pub fn root_snapshot( + &self, + root_thread_id: ThreadId, + history_reset_version: u64, + history: &dyn ConversationHistorySnapshot, + prepared: PreparedRootContext, + request_candidates: RootConversationInput, + ) -> GuardianRootSnapshot { + let mut authorization_version = self.authorization_version(history); + if prepared.missing_root_instructions { + authorization_version.retained_context_complete = false; + } + let mut review_context_revision = history.guardian_review_context_revision(); + if let Some(selected) = &prepared.handoff_selection { + let mut revision = DefaultHasher::new(); + review_context_revision.hash(&mut revision); + selected.hash(&mut revision); + review_context_revision = revision.finish(); + } + let trusted_skill_paths = prepared + .latest_user_turn_id + .as_deref() + .map(|turn_id| self.trusted_skill_paths(turn_id)) + .unwrap_or_default(); + GuardianRootSnapshot { + root_thread_id, + history_reset_version, + authorization_version, + review_context_revision, + messages: prepared.messages, + trusted_skill_paths, + request_candidates: Some(request_candidates), + } + } + + /// Reads the selected answer path against the caller's action-time history snapshot. + pub fn user_input_snapshot( + &self, + history: &dyn ConversationHistorySnapshot, + ) -> GuardianUserInputSnapshot { + match history.retained_context() { + Some(context) => { + let answers = codex_guardian_context::render_verified_answers(context); + let authorization_version = GuardianAuthorizationVersion { + user_message_revision: history.user_message_revision(), + retained_context_complete: answers.complete, + }; + GuardianUserInputSnapshot { + fragments: answers.fragments, + authorization_version, + } + } + None => GuardianUserInputSnapshot { + fragments: Vec::new(), + authorization_version: GuardianAuthorizationVersion { + user_message_revision: history.user_message_revision(), + retained_context_complete: true, + }, + }, + } + } + + pub fn authorization_version( + &self, + history: &dyn ConversationHistorySnapshot, + ) -> GuardianAuthorizationVersion { + self.user_input_snapshot(history).authorization_version + } + + /// Records a bounded, verified user-owned skill path for one host-owned turn. + pub fn record_trusted_skill(&self, turn_id: &str, path: String) { + if turn_id.is_empty() { + return; + } + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + if state.trusted_skill_turn_id.as_deref() != Some(turn_id) { + state.trusted_skill_turn_id = Some(turn_id.to_owned()); + state.trusted_skill_paths.clear(); + } + if state.trusted_skill_paths.contains(&path) + || state.trusted_skill_paths.len() >= MAX_TRUSTED_SKILLS + || state + .trusted_skill_paths + .iter() + .map(String::len) + .sum::() + .saturating_add(path.len()) + > MAX_TRUSTED_SKILL_PATHS_BYTES + { + return; + } + state.trusted_skill_paths.insert(path); + } + + /// Returns verified skill paths only for their original host-owned turn. + pub fn trusted_skill_paths(&self, turn_id: &str) -> Vec { + let state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + if state.trusted_skill_turn_id.as_deref() != Some(turn_id) { + return Vec::new(); + } + state.trusted_skill_paths.iter().cloned().collect() + } + + /// Records a genuine allow/deny assessment, not a timeout or fail-closed error. + pub fn record( + &self, + assessment: &GuardianAssessmentEvent, + action: &str, + authorization_version: GuardianAuthorizationVersion, + review_context_revision: u64, + root_authorization_version: Option, + root_review_context_revision: Option, + ) { + let Some(completed_at_ms) = assessment.completed_at_ms else { + return; + }; + let review = Arc::new(GuardianReviewEvidenceRecord { + delivery_id: codex_protocol::ResponseItemId::new("guardian_review"), + completed_at_ms, + authorization_version, + review_context_revision, + root_authorization_version, + root_review_context_revision, + correlation: json!({ + "review_id": assessment.id, + "turn_id": assessment.turn_id, + "target_item_id": assessment.target_item_id, + "completed_at_ms": completed_at_ms, + }), + decision: json!({ + "status": assessment.status, + "risk_level": assessment.risk_level, + "user_authorization": assessment.user_authorization, + }), + action: action.to_owned(), + rationale: assessment.rationale.clone(), + }); + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state.reviews.push_back(review); + state + .reviews + .make_contiguous() + .sort_by_key(|review| review.completed_at_ms); + while state.reviews.len() > MAX_PREVIOUS_REVIEWS { + state.reviews.pop_front(); + } + } + + /// Freezes the latest completed reviews, oldest first, for one classifier sample. + pub fn snapshot(&self) -> Vec> { + self.state + .lock() + .unwrap_or_else(PoisonError::into_inner) + .reviews + .iter() + .cloned() + .collect() + } +} + +/// Structured synchronous-review evidence retained for Guardian V2 classification. +#[derive(Debug)] +pub struct GuardianReviewEvidenceRecord { + /// Identifies this immutable completion even if the same action is reviewed again. + pub delivery_id: codex_protocol::ResponseItemId, + pub authorization_version: GuardianAuthorizationVersion, + pub review_context_revision: u64, + pub root_authorization_version: Option, + pub root_review_context_revision: Option, + completed_at_ms: i64, + pub correlation: serde_json::Value, + pub decision: serde_json::Value, + pub action: String, + pub rationale: Option, +} diff --git a/codex-rs/ext/guardian-reviewer/src/sender_evidence.rs b/codex-rs/ext/guardian-reviewer/src/sender_evidence.rs new file mode 100644 index 0000000000..a07b882a95 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/sender_evidence.rs @@ -0,0 +1,47 @@ +//! Projects recent sender exchanges from host-authenticated retained context. +//! Only local user messages and preceding assistant context are selected. + +use codex_guardian_context::GuardianRootMessage; +use codex_history::RetainedContext; +use codex_history::RetainedContextEntry; +use codex_history::RetainedContextOrder; + +/// Recorded adjacency provides context, not a verified question/answer association. +pub struct GuardianSenderExchange { + pub user: Option, + pub assistant: Option, +} + +pub fn guardian_sender_exchanges(context: &RetainedContext) -> Vec { + let mut exchanges = Vec::new(); + let mut assistant = None; + for (order, entry) in context.ordered_entries() { + match (order, entry) { + (RetainedContextOrder::Local(_), RetainedContextEntry::UserMessage(message)) => { + exchanges.push((message, assistant.take())) + } + (RetainedContextOrder::Local(_), RetainedContextEntry::AssistantMessage(message)) => { + assistant = Some(message) + } + (RetainedContextOrder::Inherited(_), _) + | (_, RetainedContextEntry::VerifiedAnswer(_)) => {} + } + } + let mut messages = exchanges + .into_iter() + .rev() + .take(/*n*/ 3) + .map(|(user, assistant)| GuardianSenderExchange { + user: user.complete.then(|| user.text.clone()), + assistant: assistant.map(|message| { + if message.complete { + GuardianRootMessage::Assistant(message.text.clone()) + } else { + GuardianRootMessage::IncompleteAssistantContext + } + }), + }) + .collect::>(); + messages.reverse(); + messages +}