diff --git a/codex-rs/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index 7d0826e8f5..708c673723 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/codex-rs/core/src/agent/control/user_authorization.rs @@ -1,7 +1,7 @@ -//! Projects bounded root evidence for worker reviewers using the session's capture policy. +//! Projects bounded retained root evidence for worker reviewers. //! Retained root instructions stay authoritative while old checkpoints use legacy review. //! User evidence wins, then assistant context preceding ordinary replies, then other live evidence. -//! Thread-owned recovery includes commentary; legacy projection keeps its phase filter. +//! Recovery includes commentary and confirmed messaging context. //! Projection limits do not change authorization completeness; unavailable source text does. //! Retained-history reconciliation owns recovery order and missing-instruction provenance. //! Known positions preserve host order, not delivery order or inferred question-answer pairs. @@ -15,7 +15,6 @@ use super::LocalAgentControl; use crate::codex_thread::GuardianRootMessage; use crate::codex_thread::GuardianRootSnapshot; use crate::compact::is_summary_message; -use crate::context::GuardianContextMode; use crate::context::GuardianReviewEvidence; use crate::context::is_contextual_user_fragment; use crate::event_mapping::parse_turn_item; @@ -30,7 +29,6 @@ use codex_protocol::ThreadId; use codex_protocol::items::AgentMessageContent; use codex_protocol::items::TurnItem; use codex_protocol::models::ContentItem; -use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::MultiAgentVersion; @@ -85,78 +83,70 @@ impl LocalAgentControl { .services .thread_extension_data .get_or_init(GuardianReviewEvidence::default); - let context_mode = root_thread.session.guardian_context_mode; let mut latest_user_turn_id = None; - let (messages, authorization_version) = if context_mode == GuardianContextMode::ThreadOwned - { - let retained_context = root_history.retained_context(); - let reconciled = ReconciledRetainedContext::new( - Some(retained_context), - root_history - .annotated_items() - .iter() - .filter_map(|envelope| { - let item = &envelope.item; - let Some(TurnItem::UserMessage(message)) = parse_turn_item(item) else { + let retained_context = root_history.retained_context(); + let reconciled = ReconciledRetainedContext::new( + Some(retained_context), + root_history + .annotated_items() + .iter() + .filter_map(|envelope| { + let item = &envelope.item; + let Some(TurnItem::UserMessage(message)) = parse_turn_item(item) else { + return None; + }; + let text = message.message(); + if is_summary_message(&text) || text.trim_start().starts_with("") { + return None; + } + let order = envelope + .metadata + .as_ref() + .filter(|metadata| !metadata.inherited_user_message) + .and_then(|metadata| metadata.user_input_order); + Some(( + order, + RetainedUserMessage { + turn_id: item.turn_id().unwrap_or_default().to_owned(), + message_id: item.id().map(|id| id.as_str().to_owned()), + text, + complete: false, + }, + )) + }), + ); + let mut missing_root_instructions = reconciled.missing_user_messages; + let mut messages = reconciled + .ordered_entries() + .filter_map(|(order, entry)| match entry { + RetainedContextEntry::UserMessage(message) => { + let text = if message.text.is_empty() && !message.complete { + // Older records may omit a large instruction. Recover that exact + // source while it remains available in the parent context. + let original = message.message_id.as_deref().and_then(|id| { + root_history + .raw_items() + .chain(root_history.guardian_history_items().into_iter().flatten()) + .find(|item| { + item.id().is_some_and(|item_id| item_id.as_str() == id) + }) + }); + let Some(TurnItem::UserMessage(original)) = + original.and_then(parse_turn_item) + else { + missing_root_instructions = true; return None; }; - let text = message.message(); - if is_summary_message(&text) - || text.trim_start().starts_with("") - { - return None; - } - let order = envelope - .metadata - .as_ref() - .filter(|metadata| !metadata.inherited_user_message) - .and_then(|metadata| metadata.user_input_order); - Some(( - order, - RetainedUserMessage { - turn_id: item.turn_id().unwrap_or_default().to_owned(), - message_id: item.id().map(|id| id.as_str().to_owned()), - text, - complete: false, - }, - )) - }), - ); - let mut missing_root_instructions = reconciled.missing_user_messages; - let mut messages = reconciled - .ordered_entries() - .filter_map(|(order, entry)| match entry { - RetainedContextEntry::UserMessage(message) => { - let text = if message.text.is_empty() && !message.complete { - // Older records may omit a large instruction. Recover that exact - // source while it remains available in the parent context. - let original = message.message_id.as_deref().and_then(|id| { - root_history - .raw_items() - .chain( - root_history.guardian_history_items().into_iter().flatten(), - ) - .find(|item| { - item.id().is_some_and(|item_id| item_id.as_str() == id) - }) - }); - let Some(TurnItem::UserMessage(original)) = - original.and_then(parse_turn_item) - else { - missing_root_instructions = true; - return None; - }; - Cow::Owned(original.message()) - } else { - Cow::Borrowed(message.text.as_str()) - }; - if is_contextual_user_fragment(&ContentItem::InputText { - text: text.to_string(), - }) { - return None; - } - (!is_summary_message(&text) - && !text.trim_start().starts_with("")) + Cow::Owned(original.message()) + } else { + Cow::Borrowed(message.text.as_str()) + }; + if is_contextual_user_fragment(&ContentItem::InputText { + text: text.to_string(), + }) { + return None; + } + (!is_summary_message(&text) && !text.trim_start().starts_with("")) .then(|| { latest_user_turn_id = Some(message.turn_id.clone()); ( @@ -167,197 +157,154 @@ impl LocalAgentControl { ), ) }) - } - RetainedContextEntry::VerifiedAnswer(answer) => { - codex_guardian_context::render_verified_answer(answer) - .map(|text| (Some(order), GuardianRootMessage::UserInput(text))) - } - RetainedContextEntry::AssistantMessage(_) => None, - }) - .collect::>(); - messages.drain(..messages.len().saturating_sub(MAX_ROOT_MESSAGES)); - // Old opt-out checkpoints have genuine instructions but no retained source order. - // Keep them separately bounded and explicitly unordered; never replace newer facts. - let mut legacy_messages = VecDeque::new(); - let legacy_capacity = MAX_ROOT_MESSAGES.saturating_sub(messages.len()); - if missing_root_instructions && legacy_capacity > 0 { - let mut latest_legacy_turn_id = None; - for message in - reconciled.unmatched_user_messages(root_history.legacy_user_messages()) - { - latest_legacy_turn_id = - (!message.turn_id.is_empty()).then_some(message.turn_id); - legacy_messages.push_back(GuardianRootMessage::User(message.text)); - if legacy_messages.len() > legacy_capacity { - legacy_messages.pop_front(); - } } - if latest_user_turn_id.is_none() { - latest_user_turn_id = latest_legacy_turn_id; + RetainedContextEntry::VerifiedAnswer(answer) => { + codex_guardian_context::render_verified_answer(answer) + .map(|text| (Some(order), GuardianRootMessage::UserInput(text))) + } + RetainedContextEntry::AssistantMessage(_) => None, + }) + .collect::>(); + messages.drain(..messages.len().saturating_sub(MAX_ROOT_MESSAGES)); + // Old opt-out checkpoints have genuine instructions but no retained source order. + // Keep them separately bounded and explicitly unordered; never replace newer facts. + let mut legacy_messages = VecDeque::new(); + let legacy_capacity = MAX_ROOT_MESSAGES.saturating_sub(messages.len()); + if missing_root_instructions && legacy_capacity > 0 { + let mut latest_legacy_turn_id = None; + for message in reconciled.unmatched_user_messages(root_history.legacy_user_messages()) { + latest_legacy_turn_id = (!message.turn_id.is_empty()).then_some(message.turn_id); + legacy_messages.push_back(GuardianRootMessage::User(message.text)); + if legacy_messages.len() > legacy_capacity { + legacy_messages.pop_front(); } } - let mut missing_assistant_context = retained_context.has_omitted_assistant_messages(); - // Prefer a renderable retained original over a shortened checkpoint copy. - // Otherwise preserve the bounded live evidence, including confirmed messaging sends. - let live_assistant_messages = root_history - .annotated_items() - .iter() - .filter(|envelope| { - !envelope - .metadata - .as_ref() - .is_some_and(|metadata| metadata.compaction_output) - }) - .filter_map(|envelope| { - let text = root_assistant_text(&envelope.item, &delivered_messages)?; - let id = envelope - .item - .id() - .map(codex_protocol::ResponseItemId::as_str); - let retained = reconciled.ordered_entries().find_map(|(order, entry)| { - let RetainedContextEntry::AssistantMessage(message) = entry else { - return None; - }; - id.filter(|id| message.message_id.as_deref() == Some(*id)) - .map(|_| (order, message)) - }); - if let Some((order, message)) = retained - && let Some(message) = - codex_guardian_context::retained_assistant_message(message) - { - return Some((id, Some(order), message)); - } - let order = envelope - .metadata - .as_ref() - .filter(|metadata| !metadata.inherited_user_message) - .and_then(|metadata| metadata.user_input_order) - .map(RetainedContextOrder::Local); - let order = retained.map(|(order, _)| order).or(order); - let message = if order.is_some() { - GuardianRootMessage::Assistant(text) - } else { - GuardianRootMessage::UnorderedAssistant(text) - }; - Some((id, order, message)) - }) - .collect::>(); - let mut assistant_messages = reconciled - .ordered_entries() - .filter_map(|(order, entry)| { + if latest_user_turn_id.is_none() { + latest_user_turn_id = latest_legacy_turn_id; + } + } + let mut missing_assistant_context = retained_context.has_omitted_assistant_messages(); + // Prefer a renderable retained original over a shortened checkpoint copy. + // Otherwise preserve the bounded live evidence, including confirmed messaging sends. + let live_assistant_messages = root_history + .annotated_items() + .iter() + .filter(|envelope| { + !envelope + .metadata + .as_ref() + .is_some_and(|metadata| metadata.compaction_output) + }) + .filter_map(|envelope| { + let text = root_assistant_text(&envelope.item, &delivered_messages)?; + let id = envelope + .item + .id() + .map(codex_protocol::ResponseItemId::as_str); + let retained = reconciled.ordered_entries().find_map(|(order, entry)| { let RetainedContextEntry::AssistantMessage(message) = entry else { return None; }; - if message.message_id.as_deref().is_some_and(|id| { - live_assistant_messages - .iter() - .any(|(live_id, _, _)| *live_id == Some(id)) - }) { - return None; - } - let rendered = codex_guardian_context::retained_assistant_message(message); - missing_assistant_context |= rendered.is_none(); - rendered.map(|message| (Some(order), message)) - }) - .collect::>(); - assistant_messages.extend( - live_assistant_messages - .into_iter() - .map(|(_, order, message)| (order, message)), - ); - // Keep the nearest known assistant context for each ordinary reply. Verified - // answers already include their questions; unsequenced sources cannot be paired. - let reply_context_orders = messages - .iter() - .filter_map(|(user_order, message)| { - let (Some(user_order), GuardianRootMessage::User(_)) = (user_order, message) - else { - return None; - }; - assistant_messages + id.filter(|id| message.message_id.as_deref() == Some(*id)) + .map(|_| (order, message)) + }); + if let Some((order, message)) = retained + && let Some(message) = + codex_guardian_context::retained_assistant_message(message) + { + return Some((id, Some(order), message)); + } + let order = envelope + .metadata + .as_ref() + .filter(|metadata| !metadata.inherited_user_message) + .and_then(|metadata| metadata.user_input_order) + .map(RetainedContextOrder::Local); + let order = retained.map(|(order, _)| order).or(order); + let message = if order.is_some() { + GuardianRootMessage::Assistant(text) + } else { + GuardianRootMessage::UnorderedAssistant(text) + }; + Some((id, order, message)) + }) + .collect::>(); + let mut assistant_messages = reconciled + .ordered_entries() + .filter_map(|(order, entry)| { + let RetainedContextEntry::AssistantMessage(message) = entry else { + return None; + }; + if message.message_id.as_deref().is_some_and(|id| { + live_assistant_messages .iter() - .filter_map(|(order, _)| *order) - .filter(|order| order < user_order) - .max() - }) - .collect::>(); - // Stable sorting keeps other live evidence ahead of retained-only extras. - assistant_messages.sort_by_key(|(order, _)| { - order.filter(|order| reply_context_orders.contains(order)) - }); - let available = - MAX_ROOT_MESSAGES.saturating_sub(messages.len() + legacy_messages.len()); - missing_assistant_context |= assistant_messages.len() > available; - assistant_messages.drain(..assistant_messages.len().saturating_sub(available)); - messages.extend(assistant_messages); - messages.sort_by_key(|(order, _)| (order.is_none(), *order)); - let mut messages = messages + .any(|(live_id, _, _)| *live_id == Some(id)) + }) { + return None; + } + let rendered = codex_guardian_context::retained_assistant_message(message); + missing_assistant_context |= rendered.is_none(); + rendered.map(|message| (Some(order), message)) + }) + .collect::>(); + assistant_messages.extend( + live_assistant_messages .into_iter() - .map(|(_, message)| message) - .collect::>(); - if missing_assistant_context { - messages.insert( - /*index*/ 0, - GuardianRootMessage::IncompleteAssistantContext, - ); - } - let mut authorization_version = root_evidence.authorization_version(history.as_ref()); - if !authorization_version.retained_context_complete { - messages.insert( - /*index*/ 0, - GuardianRootMessage::IncompleteVerifiedAnswers, - ); - } - if missing_root_instructions { - authorization_version.retained_context_complete = false; - messages.insert( - /*index*/ 0, - GuardianRootMessage::IncompleteRootInstructions, - ); - } - messages.insert(/*index*/ 0, GuardianRootMessage::RetainedContextScope); - if !legacy_messages.is_empty() { - legacy_messages.push_front(GuardianRootMessage::LegacyContextScope); - legacy_messages.extend(messages); - messages = legacy_messages.into_iter().collect(); - } - (messages, authorization_version) - } else { - let mut messages = history - .review_items() - .filter_map(|item| match (parse_turn_item(item), item) { - (Some(TurnItem::UserMessage(message)), _) => { - let message = message.message(); - (!is_summary_message(&message) - && !message.trim_start().starts_with("")) - .then(|| { - latest_user_turn_id = item.turn_id().map(str::to_owned); - GuardianRootMessage::User( - guardian_truncate_text(&message, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS) - .0, - ) - }) - } - (_, ResponseItem::FunctionCall { call_id, .. }) => root_evidence - .user_input_for_call(history.as_ref(), call_id) - .map(GuardianRootMessage::UserInput), - (Some(TurnItem::AgentMessage(message)), _) - if !matches!(message.phase, None | Some(MessagePhase::FinalAnswer)) => - { - None - } - _ => root_assistant_text(item, &delivered_messages) - .map(GuardianRootMessage::Assistant), - }) - .collect::>(); - let authorization_version = root_evidence.authorization_version(history.as_ref()); - if !authorization_version.retained_context_complete { - // Keep the host warning even when the root-message cap evicts older evidence. - messages.push(GuardianRootMessage::IncompleteVerifiedAnswers); - } - messages.drain(..messages.len().saturating_sub(MAX_ROOT_MESSAGES)); - (messages, authorization_version) - }; + .map(|(_, order, message)| (order, message)), + ); + // Keep the nearest known assistant context for each ordinary reply. Verified + // answers already include their questions; unsequenced sources cannot be paired. + let reply_context_orders = messages + .iter() + .filter_map(|(user_order, message)| { + let (Some(user_order), GuardianRootMessage::User(_)) = (user_order, message) else { + return None; + }; + assistant_messages + .iter() + .filter_map(|(order, _)| *order) + .filter(|order| order < user_order) + .max() + }) + .collect::>(); + // Stable sorting keeps other live evidence ahead of retained-only extras. + assistant_messages + .sort_by_key(|(order, _)| order.filter(|order| reply_context_orders.contains(order))); + let available = MAX_ROOT_MESSAGES.saturating_sub(messages.len() + legacy_messages.len()); + missing_assistant_context |= assistant_messages.len() > available; + assistant_messages.drain(..assistant_messages.len().saturating_sub(available)); + messages.extend(assistant_messages); + messages.sort_by_key(|(order, _)| (order.is_none(), *order)); + let mut messages = messages + .into_iter() + .map(|(_, message)| message) + .collect::>(); + if missing_assistant_context { + messages.insert( + /*index*/ 0, + GuardianRootMessage::IncompleteAssistantContext, + ); + } + let mut authorization_version = root_evidence.authorization_version(history.as_ref()); + if !authorization_version.retained_context_complete { + messages.insert( + /*index*/ 0, + GuardianRootMessage::IncompleteVerifiedAnswers, + ); + } + if missing_root_instructions { + authorization_version.retained_context_complete = false; + messages.insert( + /*index*/ 0, + GuardianRootMessage::IncompleteRootInstructions, + ); + } + messages.insert(/*index*/ 0, GuardianRootMessage::RetainedContextScope); + if !legacy_messages.is_empty() { + legacy_messages.push_front(GuardianRootMessage::LegacyContextScope); + legacy_messages.extend(messages); + messages = legacy_messages.into_iter().collect(); + } let trusted_skill_paths = latest_user_turn_id .as_deref() .map(|turn_id| root_evidence.trusted_skill_paths(turn_id)) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index bedbbac682..f7e10cfb33 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -166,8 +166,6 @@ pub use codex_guardian_context::GuardianRootMessage; pub struct GuardianAuthorizationVersion { /// User-message/reset revision, preserved across compaction and internal context. pub user_message_revision: u64, - /// Successful host answers captured by the temporary legacy path. - pub user_input_response_count: usize, /// False when required retained answers or root instructions are unavailable. pub retained_context_complete: bool, } diff --git a/codex-rs/core/src/context/guardian_review_evidence.rs b/codex-rs/core/src/context/guardian_review_evidence.rs index 6067ff72ca..9363b9c589 100644 --- a/codex-rs/core/src/context/guardian_review_evidence.rs +++ b/codex-rs/core/src/context/guardian_review_evidence.rs @@ -1,6 +1,5 @@ //! Selects Guardian answer evidence from the review's history snapshot and retains reviews. //! Retained answers survive restart even while incompatible checkpoints use legacy review. -//! Sessions without retained capture continue using the bounded runtime buffer. use std::collections::BTreeSet; use std::collections::VecDeque; @@ -11,22 +10,14 @@ use std::sync::PoisonError; use codex_extension_api::ConversationHistorySnapshot; use codex_guardian_context::MAX_PREVIOUS_REVIEWS; use codex_protocol::models::ContentItemKind; -use codex_protocol::models::ResponseItem; use codex_protocol::protocol::GuardianAssessmentEvent; -use codex_protocol::request_user_input::RequestUserInputQuestion; -use codex_protocol::request_user_input::RequestUserInputResponse; use serde_json::json; use super::ContextualUserFragment; use crate::codex_thread::GuardianAuthorizationVersion; -use crate::codex_thread::GuardianRootMessage; -use crate::guardian::guardian_truncate_text; -const MAX_RETAINED_USER_INPUTS: usize = 8; const MAX_TRUSTED_SKILLS: usize = 16; const MAX_TRUSTED_SKILL_PATHS_BYTES: usize = 2_048; -const MAX_GUARDIAN_USER_INPUT_ANSWERS: usize = 8; -const MAX_GUARDIAN_USER_INPUT_TOKENS: usize = 900; /// Selected answer fragments and the authorization state they describe. pub struct GuardianUserInputSnapshot { @@ -47,68 +38,11 @@ pub struct GuardianReviewEvidence { #[derive(Debug, Default)] struct GuardianReviewEvidenceState { reviews: VecDeque>, - user_inputs: VecDeque<(String, String)>, - user_input_response_count: usize, trusted_skill_turn_id: Option, trusted_skill_paths: BTreeSet, } impl GuardianReviewEvidence { - /// Preserves the legacy capture limits before hooks can replace the tool output. - pub(crate) fn record_user_input( - &self, - history: &dyn ConversationHistorySnapshot, - call_id: &str, - questions: &[RequestUserInputQuestion], - response: &RequestUserInputResponse, - ) { - if history.retained_context().is_some() { - return; - } - let fragment = questions - .iter() - .filter_map(|question| { - let response = response.answers.get(&question.id)?; - let answers = response - .answers - .iter() - .filter(|answer| !answer.trim().is_empty()) - .take(MAX_GUARDIAN_USER_INPUT_ANSWERS) - .cloned() - .collect::>(); - if answers.is_empty() { - return None; - } - let mut question_text = question.question.clone(); - for option in question - .options - .iter() - .flatten() - .filter(|option| response.answers.contains(&option.label)) - .take(MAX_GUARDIAN_USER_INPUT_ANSWERS) - { - question_text.push_str(&format!("\n{}: {}", option.label, option.description)); - } - Some(format!( - "{}{}", - GuardianRootMessage::Assistant(question_text).render(), - GuardianRootMessage::User(answers.join("\n")).render() - )) - }) - .take(MAX_GUARDIAN_USER_INPUT_ANSWERS) - .collect::(); - if fragment.is_empty() { - return; - } - let fragment = guardian_truncate_text(&fragment, MAX_GUARDIAN_USER_INPUT_TOKENS).0; - let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); - state.user_input_response_count = state.user_input_response_count.saturating_add(1); - state.user_inputs.push_back((call_id.to_owned(), fragment)); - while state.user_inputs.len() > MAX_RETAINED_USER_INPUTS { - state.user_inputs.pop_front(); - } - } - /// Reads the selected answer path against the caller's action-time history snapshot. pub fn user_input_snapshot( &self, @@ -119,7 +53,6 @@ impl GuardianReviewEvidence { let answers = codex_guardian_context::render_verified_answers(context); let authorization_version = GuardianAuthorizationVersion { user_message_revision: history.user_message_revision(), - user_input_response_count: 0, retained_context_complete: answers.complete, }; GuardianUserInputSnapshot { @@ -127,27 +60,13 @@ impl GuardianReviewEvidence { authorization_version, } } - None => { - let state = self.state.lock().unwrap_or_else(PoisonError::into_inner); - let fragments = state - .user_inputs - .iter() - .filter(|(call_id, _)| { - history.items().chain(history.review_items()).any(|item| { - matches!(item, ResponseItem::FunctionCall { call_id: id, .. } if id == call_id) - }) - }) - .map(|(_, fragment)| fragment.clone()) - .collect(); - GuardianUserInputSnapshot { - fragments, - authorization_version: GuardianAuthorizationVersion { - user_message_revision: history.user_message_revision(), - user_input_response_count: state.user_input_response_count, - retained_context_complete: true, - }, - } - } + None => GuardianUserInputSnapshot { + fragments: Vec::new(), + authorization_version: GuardianAuthorizationVersion { + user_message_revision: history.user_message_revision(), + retained_context_complete: true, + }, + }, } } @@ -158,26 +77,6 @@ impl GuardianReviewEvidence { self.user_input_snapshot(history).authorization_version } - pub(crate) fn user_input_for_call( - &self, - history: &dyn ConversationHistorySnapshot, - call_id: &str, - ) -> Option { - match history.retained_context() { - Some(context) => context - .verified_answers() - .find(|answer| answer.call_id == call_id) - .and_then(codex_guardian_context::render_verified_answer), - None => self - .state - .lock() - .unwrap_or_else(PoisonError::into_inner) - .user_inputs - .iter() - .find_map(|(id, fragment)| (id == call_id).then(|| fragment.clone())), - } - } - /// 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() { diff --git a/codex-rs/core/src/tools/handlers/request_user_input.rs b/codex-rs/core/src/tools/handlers/request_user_input.rs index b6f4356980..0f4e7d501f 100644 --- a/codex-rs/core/src/tools/handlers/request_user_input.rs +++ b/codex-rs/core/src/tools/handlers/request_user_input.rs @@ -1,5 +1,4 @@ use crate::context::GuardianContextMode; -use crate::context::GuardianReviewEvidence; use crate::function_tool::FunctionCallError; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolInvocation; @@ -102,14 +101,6 @@ impl RequestUserInputHandler { "failed to serialize {REQUEST_USER_INPUT_TOOL_NAME} response: {err}" )) })?; - if turn.config.features.enabled(Feature::GuardianApproval) { - let history = session.conversation_history_snapshot().await; - session - .services - .thread_extension_data - .get_or_init(GuardianReviewEvidence::default) - .record_user_input(history.as_ref(), &call_id, &questions, &response); - } // Capture follows the session flag even while an older checkpoint uses legacy review. if turn.config.features.enabled(Feature::GuardianApproval) && session.guardian_context_mode == GuardianContextMode::ThreadOwned diff --git a/codex-rs/core/src/tools/handlers/request_user_input_tests.rs b/codex-rs/core/src/tools/handlers/request_user_input_tests.rs index dc368eb8b2..1c1dbfca16 100644 --- a/codex-rs/core/src/tools/handlers/request_user_input_tests.rs +++ b/codex-rs/core/src/tools/handlers/request_user_input_tests.rs @@ -1,8 +1,8 @@ use super::*; use crate::GuardianAuthorizationVersion; +use crate::context::GuardianReviewEvidence; use crate::session::step_context::StepContext; use crate::session::tests::make_session_and_context; -use crate::session::tests::make_session_and_context_with_auth_and_config_and_rx; use crate::session::tests::make_session_and_context_with_rx; use crate::session::tests::update_selected_settings_for_test; use crate::session::tests::update_turn_settings_for_test; @@ -102,160 +102,120 @@ async fn request_user_input_sets_non_blocking_outside_plan_mode( answer: Option<(&str, String)>, expected: RenderedVerifiedAnswers, ) { - for thread_context_enabled in [false, true] { - let (session, turn, events) = make_session_and_context_with_auth_and_config_and_rx( - codex_login::CodexAuth::from_api_key("Test API Key"), - Vec::new(), - |config| { - config - .features - .set_enabled(Feature::GuardianThreadContext, thread_context_enabled) - .expect("test context mode"); + let (session, turn, events) = make_session_and_context_with_rx().await; + session + .services + .thread_extension_data + .insert(GuardianReviewEvidence::default()); + let original_history = session.conversation_history_snapshot().await; + *session.active_turn.lock().await = Some(ActiveTurn::default()); + + let request = tokio::spawn({ + let session = Arc::clone(&session); + let turn = Arc::clone(&turn); + async move { + RequestUserInputHandler { + available_modes: vec![ModeKind::Default], + } + .handle(ToolInvocation { + session, + step_context: StepContext::for_test(Arc::clone(&turn)), + turn, + cancellation_token: tokio_util::sync::CancellationToken::new(), + tracker: Arc::new(Mutex::new(TurnDiffTracker::default())), + call_id: "call-1".to_string(), + tool_name: codex_tools::ToolName::plain(REQUEST_USER_INPUT_TOOL_NAME), + source: crate::tools::context::ToolCallSource::Direct, + payload: ToolPayload::Function { + arguments: json!({ + "questions": [{ + "header": "Hdr", + "question": "Pick one", + "id": "pick_one", + "options": [ + { + "label": "A", + "description": "A" + }, + { + "label": "B", + "description": "B" + } + ] + }] + }) + .to_string(), + }, + }) + .await + } + }); + + let event = events.recv().await.expect("request_user_input event"); + let EventMsg::RequestUserInput(request_event) = event.msg else { + panic!("expected request_user_input event"); + }; + assert_eq!(request_event.call_id, "call-1"); + assert!(!request_event.is_blocking); + + session + .notify_user_input_response( + &request_event.turn_id, + RequestUserInputResponse { + answers: answer + .iter() + .map(|(question_id, answer)| { + ( + (*question_id).to_owned(), + RequestUserInputAnswer { + answers: vec![answer.clone()], + }, + ) + }) + .collect(), }, ) .await; - session - .services - .thread_extension_data - .insert(GuardianReviewEvidence::default()); - let original_history = session.conversation_history_snapshot().await; - *session.active_turn.lock().await = Some(ActiveTurn::default()); - let request = tokio::spawn({ - let session = Arc::clone(&session); - let turn = Arc::clone(&turn); - async move { - RequestUserInputHandler { - available_modes: vec![ModeKind::Default], - } - .handle(ToolInvocation { - session, - step_context: StepContext::for_test(Arc::clone(&turn)), - turn, - cancellation_token: tokio_util::sync::CancellationToken::new(), - tracker: Arc::new(Mutex::new(TurnDiffTracker::default())), - call_id: "call-1".to_string(), - tool_name: codex_tools::ToolName::plain(REQUEST_USER_INPUT_TOOL_NAME), - source: crate::tools::context::ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: json!({ - "questions": [{ - "header": "Hdr", - "question": "Pick one", - "id": "pick_one", - "options": [ - { - "label": "A", - "description": "A" - }, - { - "label": "B", - "description": "B" - } - ] - }] - }) - .to_string(), - }, - }) - .await + let output = request + .await + .expect("request_user_input handler task should finish") + .expect("request_user_input handler should succeed"); + assert!(output.success_for_logging()); + let history = session.conversation_history_snapshot().await; + let RenderedVerifiedAnswers { + fragments, + complete, + } = codex_guardian_context::render_verified_answers( + history.retained_context().expect("host context snapshot"), + ); + let expected_fragments = expected + .fragments + .iter() + .map(|fragment| { + if expected.complete { + format!("Retained source order: 0\n{fragment}") + } else { + fragment.clone() } - }); - - let event = events.recv().await.expect("request_user_input event"); - let EventMsg::RequestUserInput(request_event) = event.msg else { - panic!("expected request_user_input event"); - }; - assert_eq!(request_event.call_id, "call-1"); - assert!(!request_event.is_blocking); - - session - .notify_user_input_response( - &request_event.turn_id, - RequestUserInputResponse { - answers: answer - .iter() - .map(|(question_id, answer)| { - ( - (*question_id).to_owned(), - RequestUserInputAnswer { - answers: vec![answer.clone()], - }, - ) - }) - .collect(), - }, - ) - .await; - - let output = request - .await - .expect("request_user_input handler task should finish") - .expect("request_user_input handler should succeed"); - assert!(output.success_for_logging()); - let history = session.conversation_history_snapshot().await; - if thread_context_enabled { - let RenderedVerifiedAnswers { - fragments, - complete, - } = codex_guardian_context::render_verified_answers( - history.retained_context().expect("host context snapshot"), - ); - let expected_fragments = expected - .fragments - .iter() - .map(|fragment| { - if expected.complete { - format!("Retained source order: 0\n{fragment}") - } else { - fragment.clone() - } - }) - .collect::>(); - assert_eq!( - (&fragments, complete), - (&expected_fragments, expected.complete) - ); - } else { - assert_eq!( - history.retained_context(), - original_history.retained_context() - ); - } - let evidence = session - .services - .thread_extension_data - .get_or_init(GuardianReviewEvidence::default); - let legacy_answer = evidence.user_input_for_call(history.as_ref(), "call-1"); - let has_answer = matches!(&answer, Some(("pick_one", answer)) if !answer.trim().is_empty()); - assert_eq!( - evidence.authorization_version(history.as_ref()), - GuardianAuthorizationVersion { - user_message_revision: original_history.user_message_revision() - + u64::from(thread_context_enabled && has_answer), - user_input_response_count: usize::from(!thread_context_enabled && has_answer), - retained_context_complete: !thread_context_enabled || expected.complete, - }, - ); - match &answer { - Some(("pick_one", answer)) if !thread_context_enabled && !answer.trim().is_empty() => { - let fragment = legacy_answer.expect("legacy answer"); - assert!(codex_utils_output_truncation::approx_token_count(&fragment) <= 900); - assert_eq!( - fragment.contains(" Result<()> { skip_if_no_network!(Ok(())); const PARENT_GRANT: &str = "You may publish the private release. Delegate its inspection."; @@ -1129,10 +1126,6 @@ async fn forked_parent_instructions_do_not_become_local_authorization( .enable(feature) .expect("enable test feature"); } - config - .features - .set_enabled(Feature::GuardianThreadContext, thread_context_enabled) - .expect("test context mode"); }) .build_with_auto_env(&server) .await?; @@ -1234,7 +1227,7 @@ async fn forked_parent_instructions_do_not_become_local_authorization( context.ordered_entries().count(), context.user_messages_complete() )), - thread_context_enabled.then_some((0, true)), + Some((0, true)), "inherited conversation must not populate child-local authorization", ); let root_snapshot = child.guardian_root_snapshot().await.context("live root")?; @@ -1280,7 +1273,7 @@ async fn forked_parent_instructions_do_not_become_local_authorization( panic!("unexpected answer"), }) .collect::>()), - thread_context_enabled.then_some(vec![LOCAL_INSTRUCTION]), + Some(vec![LOCAL_INSTRUCTION]), "genuine child-local instructions must still be captured", ); let child = resume(&test, &child).await?; diff --git a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs index 0e90d188e2..27afc2c42a 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -87,13 +87,13 @@ enum MessagingOutcome { #[derive(Clone, Copy)] enum MessagingTool { Plural, + PluralConnector, SingularConnector, SingularFlat, } #[derive(Clone, Copy)] enum RootContext { - Legacy, Retained, Migrating, RetainedAtMessageLimit, @@ -160,8 +160,6 @@ async fn mount_completion( .await } -#[test_case(RootAnswer::Complete, RootContext::Legacy, MessagingOutcome::Complete, MessagingTool::Plural; "legacy_complete_answer")] -#[test_case(RootAnswer::Oversized, RootContext::Legacy, MessagingOutcome::Complete, MessagingTool::Plural; "legacy_oversized_answer")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Complete, MessagingTool::Plural; "retained_complete_answer")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Block, MessagingTool::Plural; "retained_blocked_post_hook")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::CancelPostHook, MessagingTool::Plural; "retained_cancelled_post_hook")] @@ -170,11 +168,11 @@ async fn mount_completion( #[test_case(RootAnswer::Complete, RootContext::Migrating, MessagingOutcome::Complete, MessagingTool::Plural; "migrating_complete_answer")] #[test_case(RootAnswer::Oversized, RootContext::Migrating, MessagingOutcome::Complete, MessagingTool::Plural; "migrating_oversized_answer")] #[test_case(RootAnswer::Complete, RootContext::RetainedAtMessageLimit, MessagingOutcome::Complete, MessagingTool::Plural; "bounded_retained_root_messages")] -#[test_case(RootAnswer::Complete, RootContext::Legacy, MessagingOutcome::Complete, MessagingTool::SingularConnector; "legacy_user_message_connector")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Complete, MessagingTool::SingularConnector; "retained_user_message_connector")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Complete, MessagingTool::SingularFlat; "retained_user_message_flat")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Block, MessagingTool::SingularConnector; "retained_user_message_connector_blocked_post_hook")] #[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::CancelBeforeConfirmation, MessagingTool::SingularFlat; "retained_user_message_flat_cancelled_before_confirmation")] +#[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Complete, MessagingTool::PluralConnector; "retained_user_messaging_connector")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_subagent_review_preserves_late_root_user_authorization( root_answer: RootAnswer, @@ -188,7 +186,6 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( "Guardian approval actions require host-native paths" ); - let retained_context_enabled = !matches!(root_context, RootContext::Legacy); let block_post_hook = matches!(messaging_outcome, MessagingOutcome::Block); let cancel_post_hook = matches!(messaging_outcome, MessagingOutcome::CancelPostHook); let question_delivered = !matches!( @@ -196,27 +193,21 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( MessagingOutcome::CancelBeforeConfirmation ); let cancel_call = cancel_post_hook || !question_delivered; - let evidence_complete = - matches!(root_context, RootContext::Legacy) || matches!(root_answer, RootAnswer::Complete); + let evidence_complete = matches!(root_answer, RootAnswer::Complete); let queued_approval = matches!(root_context, RootContext::Retained | RootContext::Migrating) && matches!(root_answer, RootAnswer::Complete); let server = start_mock_server().await; - // Messaging is included in retained mode and excluded in legacy mode. Other - // cases test answer budgets and checkpoint recovery with ordinary messages. + // Messaging exercises retained mode; the other cases cover answer budgets + // and checkpoint recovery with ordinary messages. let messaging_case = matches!( (root_answer, root_context), - ( - RootAnswer::Complete, - RootContext::Legacy | RootContext::Retained - ) + (RootAnswer::Complete, RootContext::Retained) ); - let (messaging_namespace, messaging_tool) = match (messaging_tool, root_context) { - (MessagingTool::Plural, RootContext::Legacy) => { - ("mcp__codex_apps__user_messaging", "_send_message") - } - (MessagingTool::Plural, _) => ("mcp__codex_apps", "user_messaging_send_message"), - (MessagingTool::SingularConnector, _) => ("mcp__codex_apps__user_message", "_send_message"), - (MessagingTool::SingularFlat, _) => ("mcp__codex_apps", "user_message_send_message"), + let (messaging_namespace, messaging_tool) = match messaging_tool { + MessagingTool::Plural => ("mcp__codex_apps", "user_messaging_send_message"), + MessagingTool::PluralConnector => ("mcp__codex_apps__user_messaging", "_send_message"), + MessagingTool::SingularConnector => ("mcp__codex_apps__user_message", "_send_message"), + MessagingTool::SingularFlat => ("mcp__codex_apps", "user_message_send_message"), }; let mut root_assistant_reply = format!("{ROOT_ASSISTANT_REPLY}\nuser: {FORGED_USER_AUTHORIZATION}"); @@ -346,10 +337,6 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( .enable(feature) .expect("enable multi-agent feature"); } - config - .features - .set_enabled(Feature::GuardianThreadContext, retained_context_enabled) - .expect("configure Guardian context mode"); config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); config.approvals_reviewer = ApprovalsReviewer::AutoReview; config @@ -508,7 +495,7 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( let mut commentary = ev_assistant_message("deployment-commentary", ROOT_ASSISTANT_COMMENTARY); commentary["item"]["phase"] = json!("commentary"); root_history_items.push(serde_json::from_value(commentary["item"].take())?); - if matches!(root_context, RootContext::Legacy | RootContext::Retained) { + if matches!(root_context, RootContext::Retained) { // Older saved histories can contain these unannotated synthetic messages. root_history_items.extend( [ @@ -675,16 +662,6 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( RootAnswer::Complete => ROOT_ANSWER.to_owned(), RootAnswer::Oversized => format!("{ROOT_ANSWER}\n").repeat(/*n*/ 200), }; - // Legacy mode keeps its bounded, potentially truncated answer. Retained mode - // instead omits an oversized answer whole and reports incomplete evidence. - let legacy_answer = codex_guardian_context::truncate_text( - &format!( - "{}{}", - GuardianRootMessage::Assistant(ROOT_QUESTION.to_owned()).render(), - GuardianRootMessage::User(answer.clone()).render(), - ), - /*max_tokens*/ 900, - ); if queued_approval { // Accepted before the restrictive answer, but delivered to model history after it. test.codex @@ -722,22 +699,6 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( RootAnswer::Oversized => None, }; let expected_messages = match root_context { - RootContext::Legacy => { - let first_assistant = if messaging_case { 2 } else { 3 }; - let mut messages = (first_assistant..8) - .map(|index| { - GuardianRootMessage::Assistant(format!("Deployment inspection update {index}.")) - }) - .collect::>(); - if !messaging_case { - messages.push(GuardianRootMessage::Assistant(root_assistant_reply.clone())); - } - messages.extend([ - GuardianRootMessage::User(USER_APPROVAL.to_owned()), - GuardianRootMessage::UserInput(legacy_answer), - ]); - messages - } RootContext::RetainedAtMessageLimit => { let mut messages = vec![ GuardianRootMessage::RetainedContextScope, @@ -830,21 +791,18 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( guardian_transcript .matches(&format!("user: {INITIAL_PROMPT}")) .count(), - 1 + usize::from( - retained_context_enabled - && !matches!(root_context, RootContext::RetainedAtMessageLimit) - ), + 1 + usize::from(!matches!(root_context, RootContext::RetainedAtMessageLimit)), "the worker transcript keeps the original instructions; the root projection selects bounded retained evidence" ); assert_eq!( guardian_transcript.contains("some verified user answers are unavailable"), - retained_context_enabled && !evidence_complete, + !evidence_complete, ); for text in [ROOT_QUESTION, ROOT_ANSWER] { assert_eq!( guardian_transcript.contains(text), - !retained_context_enabled || matches!(root_answer, RootAnswer::Complete), - "retained mode omits oversized answers whole; legacy mode keeps its truncated answer" + matches!(root_answer, RootAnswer::Complete), + "oversized retained answers are omitted whole" ); } assert!(guardian_transcript.contains(&format!("user: {USER_APPROVAL}"))); @@ -854,14 +812,12 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( ] { assert_eq!( guardian_transcript.contains(&format!("assistant: {text}")), - !matches!(root_context, RootContext::RetainedAtMessageLimit) - && (!messaging_case || retained_context_enabled) - && question_delivered, + !matches!(root_context, RootContext::RetainedAtMessageLimit) && question_delivered, ); } assert_eq!( guardian_transcript.contains(ROOT_ASSISTANT_COMMENTARY), - retained_context_enabled && !matches!(root_context, RootContext::RetainedAtMessageLimit), + !matches!(root_context, RootContext::RetainedAtMessageLimit), ); assert!(!guardian_transcript.contains(ORIGINAL_QUESTION)); assert!(!guardian_transcript.contains(SYNTHETIC_AUTHORIZATION));