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

Remove legacy Guardian authorization evidence paths (#47688)

Browse Source 
## What changed

Use retained root context for subagent authorization reviews, removing the separate legacy history path. Remove the runtime `request_user_input` answer buffer and its response counter; authorization versions now track the user-message revision and retained-context completeness.

## Testing

Update authorization and input-handler tests to exercise retained context, and add coverage for the `user_messaging` connector in subagent reviews.

GitOrigin-RevId: 00ad151cf71b843af0e82ae90cbcf559d0c65f89
This commit is contained in:
felixxia-oai
2026-09-23 22:50:23 +00:00
committed by copyberry
parent a6df7385e6
commit 1d516d8b3d
7 changed files with 342 additions and 598 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control/user_authorization.rs
+203 -256
View File
@@ -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("<user_action>") {
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("<user_action>")
{
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("<user_action>"))
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("<user_action>"))
.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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
// 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::<Vec<_>>();
assistant_messages.extend(
live_assistant_messages
.into_iter()
.map(|(_, message)| message)
.collect::<Vec<_>>();
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("<user_action>"))
.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::<Vec<_>>();
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::<Vec<_>>();
// 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::<Vec<_>>();
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))
codex-rs/core/src/codex_thread.rs
-2
View File
@@ -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,
}
codex-rs/core/src/context/guardian_review_evidence.rs
+7 -108
View File
@@ -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<Arc<GuardianReviewEvidenceRecord>>,
user_inputs: VecDeque<(String, String)>,
user_input_response_count: usize,
trusted_skill_turn_id: Option<String>,
trusted_skill_paths: BTreeSet<String>,
}
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::<Vec<_>>();
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::<String>();
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<String> {
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() {
codex-rs/core/src/tools/handlers/request_user_input.rs
-9
View File
@@ -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
codex-rs/core/src/tools/handlers/request_user_input_tests.rs
+110 -150
View File
@@ -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::<Vec<_>>();
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("<truncated omitted_approx_tokens="),
!expected.complete,
);
}
_ => assert_eq!(
legacy_answer,
expected
.fragments
.first()
.cloned()
.filter(|_| expected.complete)
),
}
}
})
.collect::<Vec<_>>();
assert_eq!(
(&fragments, complete),
(&expected_fragments, expected.complete)
);
let evidence = session
.services
.thread_extension_data
.get_or_init(GuardianReviewEvidence::default);
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(has_answer),
retained_context_complete: expected.complete,
},
);
}
#[tokio::test]
codex-rs/core/tests/suite/guardian_retained_context.rs
+4 -11
View File
@@ -1100,14 +1100,11 @@ async fn standalone_fork_retains_inherited_user_instructions(
Ok(())
}
#[test_case(false, true; "enabled without checkpoint")]
#[test_case(true, true; "enabled after checkpoint")]
#[test_case(false, false; "legacy without checkpoint")]
#[test_case(true, false; "legacy after checkpoint")]
#[test_case(false; "without checkpoint")]
#[test_case(true; "after checkpoint")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn forked_parent_instructions_do_not_become_local_authorization(
compact_parent: bool,
thread_context_enabled: bool,
) -> 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::<Vec<_>>()),
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?;
codex-rs/core/tests/suite/guardian_subagent_authorization.rs
+18 -62
View File
@@ -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::<Vec<_>>();
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));