mirror of
https://github.com/openai/codex.git
synced 2026-10-21 16:38:24 +00:00
Make Guardian thread context capture unconditional (#47689)
## What changed Always use `GuardianContextMode::ThreadOwned` for session capture and subagent history forks, regardless of `guardianv2.thread_context`. Continue selecting reviewer policy from saved checkpoints, preserving legacy review for incompatible checkpoints and removing the feature gate from reviewer compaction compatibility checks. ## Testing Update coverage for unconditional retained context across compaction, resume, and subagent forks. Exercise legacy review with saved checkpoints, and verify that local and managed opt-outs do not disable sender context capture or checkpoint migration. GitOrigin-RevId: cd74e2029c83f12bc655b6563242d06f29dd46af
This commit is contained in:
@@ -980,7 +980,7 @@ impl LocalAgentControl {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let context_mode = GuardianContextMode::from_features(&config.features);
|
||||
let context_mode = GuardianContextMode::ThreadOwned;
|
||||
let mut replaced_parent_developer_instructions = false;
|
||||
// Scrub inherited hints and replace only the parent's developer-instruction fragment.
|
||||
// Compaction stores response items separately, so sanitize both top-level messages and
|
||||
|
||||
@@ -2536,18 +2536,10 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
|
||||
.expect("parent shutdown should submit");
|
||||
}
|
||||
|
||||
#[test_case::test_case(true; "thread context enabled")]
|
||||
#[test_case::test_case(false; "thread context disabled")]
|
||||
#[tokio::test]
|
||||
async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(
|
||||
thread_context_enabled: bool,
|
||||
) {
|
||||
async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
|
||||
let harness = AgentControlHarness::new().await;
|
||||
let mut parent_config = harness.config.clone();
|
||||
parent_config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("test context mode");
|
||||
let _ = parent_config.features.enable(Feature::MultiAgentV2);
|
||||
parent_config.developer_instructions = Some("Parent developer instructions.".to_string());
|
||||
parent_config.multi_agent_v2.root_agent_usage_hint_text =
|
||||
@@ -2555,10 +2547,6 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(
|
||||
parent_config.multi_agent_v2.subagent_usage_hint_text =
|
||||
Some("Parent subagent guidance.".to_string());
|
||||
let mut child_config = harness.config.clone();
|
||||
child_config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("test context mode");
|
||||
let _ = child_config.features.enable(Feature::MultiAgentV2);
|
||||
child_config.developer_instructions = Some("Child developer instructions.".to_string());
|
||||
child_config.multi_agent_v2.subagent_developer_instructions =
|
||||
@@ -2736,17 +2724,12 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(
|
||||
),
|
||||
"a subagent must not inherit its parent review checkpoint",
|
||||
);
|
||||
assert_eq!(
|
||||
history_contains_text(history.raw_items(), "parent-private-release"),
|
||||
!thread_context_enabled,
|
||||
"only retained mode changes parent approval inheritance",
|
||||
assert!(
|
||||
!history_contains_text(history.raw_items(), "parent-private-release"),
|
||||
"a subagent must not inherit parent-local approval messages",
|
||||
);
|
||||
let mut inherited_context = codex_history::RetainedContext::default();
|
||||
if thread_context_enabled {
|
||||
inherited_context.reserve_order();
|
||||
} else {
|
||||
inherited_context.mark_user_messages_incomplete();
|
||||
}
|
||||
inherited_context.reserve_order();
|
||||
assert_eq!(history.retained_context(), &inherited_context);
|
||||
assert!(
|
||||
history_contains_text(history.raw_items(), "compacted parent summary"),
|
||||
|
||||
@@ -342,10 +342,9 @@ async fn run_remote_compact_task_inner_impl(
|
||||
replacement_history: &replacement_history,
|
||||
});
|
||||
}
|
||||
let reviewer_compaction_hash = if sess.enabled(Feature::GuardianThreadContext)
|
||||
&& crate::context::GuardianContextMode::from_history(
|
||||
sess.conversation_history_snapshot().await.as_ref(),
|
||||
) == crate::context::GuardianContextMode::Legacy
|
||||
let reviewer_compaction_hash = if crate::context::GuardianContextMode::from_history(
|
||||
sess.conversation_history_snapshot().await.as_ref(),
|
||||
) == crate::context::GuardianContextMode::Legacy
|
||||
&& let Some(review_turn) = sess.turn_context_for_sub_id(&turn_context.sub_id).await
|
||||
{
|
||||
// Previous-model compaction must remain compatible with the continuing turn's
|
||||
|
||||
@@ -2,8 +2,6 @@
|
||||
//! Unknown or incompatible checkpoints keep legacy review alongside retained user evidence.
|
||||
|
||||
use codex_extension_api::ConversationHistorySnapshot;
|
||||
use codex_features::Feature;
|
||||
use codex_features::Features;
|
||||
use codex_history::ResponseItemEnvelope;
|
||||
|
||||
/// Selects legacy compatibility or thread-owned evidence.
|
||||
@@ -24,14 +22,6 @@ impl GuardianContextMode {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn from_features(features: &Features) -> Self {
|
||||
if features.enabled(Feature::GuardianThreadContext) {
|
||||
Self::ThreadOwned
|
||||
} else {
|
||||
Self::Legacy
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn for_checkpoint(
|
||||
self,
|
||||
items: &[ResponseItemEnvelope],
|
||||
|
||||
@@ -27,10 +27,6 @@ async fn run_review_preserves_evidence_during_parent_compaction() {
|
||||
codex_login::CodexAuth::from_api_key("Test API Key"),
|
||||
Vec::new(),
|
||||
|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::GuardianThreadContext)
|
||||
.unwrap();
|
||||
config.features.disable(Feature::TokenBudget).unwrap();
|
||||
},
|
||||
)
|
||||
@@ -64,7 +60,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() {
|
||||
¶ms.spawn_config,
|
||||
parent.inherited_instructions().await,
|
||||
params.parent_history.history_version(),
|
||||
parent.guardian_context_mode,
|
||||
GuardianContextMode::ThreadOwned,
|
||||
)
|
||||
.with_environments(params.parent_context.environments())
|
||||
.with_node_repl_policy_eligibility(
|
||||
@@ -298,7 +294,7 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
|
||||
..Default::default()
|
||||
},
|
||||
/*parent_history_version*/ 0,
|
||||
parent.guardian_context_mode,
|
||||
GuardianContextMode::ThreadOwned,
|
||||
);
|
||||
let expected_key = GuardianReviewSessionReuseKey {
|
||||
user_instructions: latest_global,
|
||||
@@ -448,17 +444,14 @@ async fn guardian_review_session_config_change_invalidates_cached_session() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test_case::test_case(true; "thread owned")]
|
||||
#[test_case::test_case(false; "legacy")]
|
||||
#[test_case::test_case(GuardianContextMode::ThreadOwned; "thread owned")]
|
||||
#[test_case::test_case(GuardianContextMode::Legacy; "legacy checkpoint")]
|
||||
#[tokio::test]
|
||||
async fn encrypted_parent_compaction_requires_original_item_id(thread_context_enabled: bool) {
|
||||
async fn encrypted_parent_compaction_requires_original_item_id(mode: GuardianContextMode) {
|
||||
let (session, _) = crate::session::tests::make_session_and_context().await;
|
||||
let mut features = session.get_config().await.features.clone();
|
||||
features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("context mode");
|
||||
let policy =
|
||||
ReviewContextPolicy::for_context(GuardianContextMode::from_features(&features), &features);
|
||||
let features = session.get_config().await.features.clone();
|
||||
|
||||
let policy = ReviewContextPolicy::for_context(mode, &features);
|
||||
let item = ResponseItem::Compaction {
|
||||
id: Some(codex_protocol::ResponseItemId::from_server(
|
||||
"cmp_guardian_parent_summary".to_string(),
|
||||
@@ -493,7 +486,7 @@ async fn encrypted_parent_compaction_requires_original_item_id(thread_context_en
|
||||
);
|
||||
history.replace_annotated(items);
|
||||
let result = policy.parent_compaction(&history);
|
||||
if thread_context_enabled {
|
||||
if mode == GuardianContextMode::ThreadOwned {
|
||||
assert!(result.is_err());
|
||||
} else {
|
||||
assert_eq!(result.expect("legacy omission"), None);
|
||||
|
||||
@@ -988,8 +988,8 @@ impl Session {
|
||||
thread_id.to_string(),
|
||||
thread_extension_init,
|
||||
);
|
||||
// Capture follows the flag; replay selects reviewer policy from the saved checkpoint.
|
||||
let guardian_context_mode = GuardianContextMode::from_features(&config.features);
|
||||
// Capture is unconditional; replay selects reviewer policy from the saved checkpoint.
|
||||
let guardian_context_mode = GuardianContextMode::ThreadOwned;
|
||||
thread_extension_data.insert(crate::context::GuardianReviewEvidence::default());
|
||||
// Kick off independent async setup tasks in parallel to reduce startup latency.
|
||||
//
|
||||
|
||||
@@ -6403,7 +6403,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
|
||||
|
||||
let mut state = SessionState::new(session_configuration.clone());
|
||||
state.history = ContextManager::with_guardian_context_mode(
|
||||
GuardianContextMode::from_features(&config.features),
|
||||
GuardianContextMode::ThreadOwned,
|
||||
&session_configuration.session_source,
|
||||
);
|
||||
let (environment_manager, resolved_environments) =
|
||||
@@ -6551,7 +6551,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
|
||||
thread_settings_persistence: Semaphore::new(/*permits*/ 1),
|
||||
managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1),
|
||||
features: config.features.clone(),
|
||||
guardian_context_mode: GuardianContextMode::from_features(&config.features),
|
||||
guardian_context_mode: GuardianContextMode::ThreadOwned,
|
||||
isolation: codex_extension_api::SessionIsolation::Inherit,
|
||||
tool_policy: Arc::default(),
|
||||
windows_sandbox_proxy_settings_mode:
|
||||
@@ -8675,7 +8675,7 @@ where
|
||||
|
||||
let mut state = SessionState::new(session_configuration.clone());
|
||||
state.history = ContextManager::with_guardian_context_mode(
|
||||
GuardianContextMode::from_features(&config.features),
|
||||
GuardianContextMode::ThreadOwned,
|
||||
&session_configuration.session_source,
|
||||
);
|
||||
let (environment_manager, resolved_turn_environments) =
|
||||
@@ -8822,7 +8822,7 @@ where
|
||||
thread_settings_persistence: Semaphore::new(/*permits*/ 1),
|
||||
managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1),
|
||||
features: config.features.clone(),
|
||||
guardian_context_mode: GuardianContextMode::from_features(&config.features),
|
||||
guardian_context_mode: GuardianContextMode::ThreadOwned,
|
||||
isolation: codex_extension_api::SessionIsolation::Inherit,
|
||||
tool_policy: Arc::default(),
|
||||
windows_sandbox_proxy_settings_mode:
|
||||
|
||||
@@ -149,7 +149,6 @@ async fn guardian_revalidates_owning_session_before_allow(
|
||||
.set_legacy_sandbox_policy(SandboxPolicy::new_workspace_write_policy())
|
||||
.expect("set sandbox policy");
|
||||
for feature in [
|
||||
Feature::GuardianThreadContext,
|
||||
Feature::CodeMode,
|
||||
Feature::CodeModeInterrupt,
|
||||
Feature::DefaultModeRequestUserInput,
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
//! Old encrypted checkpoints keep legacy review while newly captured answers survive migration.
|
||||
//! A retired managed opt-out cannot disable capture or later checkpoint promotion.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Result;
|
||||
use codex_config::test_support::CloudConfigBundleFixture;
|
||||
use codex_core::CodexThread;
|
||||
use codex_core::StartThreadOptions;
|
||||
use codex_core::TurnInputRequest;
|
||||
@@ -63,7 +65,7 @@ pub(super) async fn resume(
|
||||
test.thread_manager.remove_thread(&thread_id).await;
|
||||
let mut config = test.config.clone();
|
||||
config.model = Some(model);
|
||||
config.features.enable(Feature::GuardianThreadContext)?;
|
||||
|
||||
Ok(test
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
@@ -84,6 +86,14 @@ pub(super) async fn migration_scenario() -> Result<Vec<responses::ResponsesReque
|
||||
let test = test_codex()
|
||||
.with_history_mode(ThreadHistoryMode::Paginated)
|
||||
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
|
||||
.with_cloud_config_bundle(
|
||||
CloudConfigBundleFixture::loader_with_enterprise_requirement(
|
||||
r#"
|
||||
[features]
|
||||
"guardianv2.thread_context" = false
|
||||
"#,
|
||||
),
|
||||
)
|
||||
.with_model_info_override("gpt-5.5", |model| {
|
||||
model.comp_hash = Some("previous-model".to_owned());
|
||||
model.auto_review_model_override = Some(model.slug.clone());
|
||||
|
||||
@@ -7,8 +7,10 @@ use codex_core::TurnInputRequest;
|
||||
use codex_core::config::Constrained;
|
||||
use codex_core::config::CurrentTimeReminderConfig;
|
||||
use codex_core::config::RolloutBudgetConfig;
|
||||
use codex_core::context::GuardianContextMode;
|
||||
use codex_extension_api::ExtensionRegistryBuilder;
|
||||
use codex_features::Feature;
|
||||
use codex_history::RolloutItem;
|
||||
use codex_protocol::config_types::ApprovalsReviewer;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
@@ -38,13 +40,7 @@ use test_case::test_case;
|
||||
use super::image_rollout::RecordingFileAttachmentStore;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
#[test_case(160_000, &[0, 0]; "complete_instructions_fit")]
|
||||
#[test_case(28_000, &[0, 1]; "oversized_history_compacts_then_truncates")]
|
||||
#[test_case(16_000, &[1]; "first_review_can_shorten_oversized_instructions")]
|
||||
async fn review_preserves_user_instructions_until_request_budgeting(
|
||||
window: i64,
|
||||
compactions_per_turn: &[usize],
|
||||
) -> Result<()> {
|
||||
async fn legacy_checkpoint_review_preserves_oversized_instruction_order() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
skip_if_wine_exec!(
|
||||
Ok(()),
|
||||
@@ -53,35 +49,52 @@ async fn review_preserves_user_instructions_until_request_budgeting(
|
||||
let server = responses::start_mock_server().await;
|
||||
let mut extensions = ExtensionRegistryBuilder::new();
|
||||
extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle));
|
||||
let test = test_codex()
|
||||
let mut test = test_codex()
|
||||
.with_extensions(Arc::new(extensions.build()))
|
||||
.with_model_info_override("gpt-5.6-luna", move |model| {
|
||||
model.context_window = Some(window);
|
||||
})
|
||||
.with_model_info_override("gpt-5.6-luna", |model| model.context_window = Some(28_000))
|
||||
.with_model_info_override("gpt-5.5", |model| {
|
||||
model.auto_review_model_override = Some("gpt-5.6-luna".to_owned());
|
||||
})
|
||||
.with_config(|config| {
|
||||
// This test covers the legacy full-transcript instruction budget.
|
||||
config
|
||||
.features
|
||||
.disable(Feature::GuardianThreadContext)
|
||||
.expect("disable thread context for the legacy budget test");
|
||||
.disable(Feature::GuardianReuseParentCompaction)
|
||||
.expect("review the legacy transcript rather than the opaque checkpoint");
|
||||
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
config.approvals_reviewer = ApprovalsReviewer::AutoReview;
|
||||
config.model_auto_compact_token_limit = Some(1_000_000);
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
let old_history: Vec<RolloutItem> = serde_json::from_value(json!([{
|
||||
"type": "compacted",
|
||||
"payload": {
|
||||
"message": "Historical checkpoint without producer metadata.",
|
||||
"replacement_history": [{
|
||||
"type": "compaction",
|
||||
"id": "old-checkpoint",
|
||||
"encrypted_content": "legacy opaque context"
|
||||
}]
|
||||
}
|
||||
}]))?;
|
||||
test.codex.ensure_rollout_materialized().await;
|
||||
test.codex.append_rollout_items(&old_history).await?;
|
||||
test.codex =
|
||||
super::guardian_checkpoint_migration::resume(&test, &test.codex, old_history).await?;
|
||||
assert_eq!(
|
||||
GuardianContextMode::from_history(
|
||||
test.codex.conversation_history_snapshot().await.as_ref()
|
||||
),
|
||||
GuardianContextMode::Legacy,
|
||||
);
|
||||
let command = json!({
|
||||
"cmd": "echo complete-instructions",
|
||||
"sandbox_permissions": "require_escalated",
|
||||
"justification": "Run the requested command."
|
||||
})
|
||||
.to_string();
|
||||
let events = compactions_per_turn
|
||||
.iter()
|
||||
.enumerate()
|
||||
.flat_map(|(turn, compactions)| {
|
||||
let events = (0..2)
|
||||
.flat_map(|turn| {
|
||||
let mut events = vec![
|
||||
sse(vec![
|
||||
ev_function_call(&format!("exec-{turn}"), "exec_command", &command),
|
||||
@@ -99,13 +112,17 @@ async fn review_preserves_user_instructions_until_request_budgeting(
|
||||
ev_completed(&format!("done-{turn}")),
|
||||
]),
|
||||
];
|
||||
for _ in 0..*compactions {
|
||||
if turn == 1 {
|
||||
events.insert(
|
||||
/*index*/ 1,
|
||||
sse(vec![
|
||||
json!({
|
||||
"type": "response.output_item.done",
|
||||
"item": {"type": "compaction", "encrypted_content": "Earlier reviewer context."},
|
||||
"item": {
|
||||
"type": "compaction",
|
||||
"id": "review-summary",
|
||||
"encrypted_content": "Earlier reviewer context."
|
||||
}
|
||||
}),
|
||||
ev_completed("review-compaction"),
|
||||
]),
|
||||
@@ -120,37 +137,30 @@ async fn review_preserves_user_instructions_until_request_budgeting(
|
||||
"{padding}Run the requested echo command. You may edit scratch files only.{padding}"
|
||||
);
|
||||
test.submit_text_turn(&initial).await?;
|
||||
// TurnComplete precedes active-turn cleanup; wait before injecting follow-up history.
|
||||
ThreadIdle::wait(&test.codex).await;
|
||||
// This whole message exceeds the old transcript allowance. A following
|
||||
// restriction must still reach the reviewer, with the original source order.
|
||||
let followup = format!("{padding}{padding}Keep all files private.{padding}{padding}");
|
||||
let approval = format!(
|
||||
"{}\nApproved action: {command}",
|
||||
codex_guardian_context::MANUAL_APPROVAL_DEVELOPER_PREFIX
|
||||
);
|
||||
let restriction = "Revoke permission to edit files. Only run the echo command.";
|
||||
// V2 retains the first review's user input. The smallest window exercises
|
||||
// first-review truncation only; the larger windows cover follow-up delivery.
|
||||
if compactions_per_turn.len() == 2 {
|
||||
test.codex
|
||||
.inject_response_items(vec![
|
||||
responses::user_message_item(&followup),
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "developer".to_owned(),
|
||||
content: vec![ContentItem::InputText {
|
||||
text: approval.clone(),
|
||||
}],
|
||||
phase: None,
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
},
|
||||
])
|
||||
.await?;
|
||||
test.submit_text_turn(restriction).await?;
|
||||
ThreadIdle::wait(&test.codex).await;
|
||||
}
|
||||
let (compact_requests, requests): (Vec<_>, Vec<_>) = responses
|
||||
test.codex
|
||||
.inject_response_items(vec![
|
||||
responses::user_message_item(&followup),
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "developer".to_owned(),
|
||||
content: vec![ContentItem::InputText {
|
||||
text: approval.clone(),
|
||||
}],
|
||||
phase: None,
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
},
|
||||
])
|
||||
.await?;
|
||||
test.submit_text_turn(restriction).await?;
|
||||
ThreadIdle::wait(&test.codex).await;
|
||||
let (compactions, requests): (Vec<_>, Vec<_>) = responses
|
||||
.requests()
|
||||
.into_iter()
|
||||
.partition(|request| !request.inputs_of_type("compaction_trigger").is_empty());
|
||||
@@ -158,53 +168,35 @@ async fn review_preserves_user_instructions_until_request_budgeting(
|
||||
.iter()
|
||||
.filter(|request| request.body_json()["client_metadata"]["x-openai-subagent"] == "guardian")
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(reviews.len(), compactions_per_turn.len());
|
||||
let initial_inputs = reviews[0].message_input_text_groups("user");
|
||||
let initial_context = initial_inputs.last().expect("first review input").concat();
|
||||
if compactions_per_turn[0] == 0 {
|
||||
assert!(initial_context.contains(&initial));
|
||||
} else {
|
||||
assert!(initial_context.contains("<truncated omitted_approx_tokens="));
|
||||
}
|
||||
assert_eq!(
|
||||
compact_requests.len(),
|
||||
compactions_per_turn.iter().sum::<usize>()
|
||||
);
|
||||
if let Some(review) = reviews.get(1) {
|
||||
let delta_inputs = review.message_input_text_groups("user");
|
||||
let delta = delta_inputs.last().expect("delta review input");
|
||||
let delta_context = delta.concat();
|
||||
if window == 160_000 {
|
||||
assert!(delta_context.contains(&followup));
|
||||
} else {
|
||||
assert!(delta_context.contains("<truncated omitted_approx_tokens="));
|
||||
assert!(
|
||||
delta_context.contains(
|
||||
"User instructions and prior approvals may be incomplete where marked."
|
||||
)
|
||||
);
|
||||
}
|
||||
let approval_start = delta_context
|
||||
.find(&approval)
|
||||
.expect("complete prior approval");
|
||||
let restriction_start = delta_context.find(restriction).expect("later restriction");
|
||||
assert!(approval_start + approval.len() < restriction_start);
|
||||
assert!(
|
||||
!delta_context.contains(&initial),
|
||||
"old entries are not resent"
|
||||
);
|
||||
assert_eq!(
|
||||
reviews[0].body_json()["client_metadata"]["thread_id"],
|
||||
review.body_json()["client_metadata"]["thread_id"]
|
||||
);
|
||||
}
|
||||
assert_eq!((compactions.len(), reviews.len()), (1, 2));
|
||||
assert!(
|
||||
requests
|
||||
reviews[0]
|
||||
.message_input_text_groups("user")
|
||||
.last()
|
||||
.expect("parent resumes after review")
|
||||
.function_call_output(&format!("exec-{}", reviews.len() - 1))
|
||||
.to_string()
|
||||
.contains("complete-instructions")
|
||||
.expect("first review")
|
||||
.concat()
|
||||
.contains(&initial)
|
||||
);
|
||||
let groups = reviews[1].message_input_text_groups("user");
|
||||
let delta = groups.last().expect("follow-up review").concat();
|
||||
assert!(delta.contains("<truncated omitted_approx_tokens="));
|
||||
assert!(
|
||||
delta.contains("User instructions and prior approvals may be incomplete where marked.")
|
||||
);
|
||||
let transcript = delta
|
||||
.split_once(">>> TRANSCRIPT DELTA START\n")
|
||||
.expect("legacy transcript delta")
|
||||
.1
|
||||
.split_once(">>> TRANSCRIPT DELTA END\n")
|
||||
.expect("legacy transcript delta end")
|
||||
.0;
|
||||
let approval_start = transcript.find(&approval).expect("complete prior approval");
|
||||
let restriction_start = transcript.find(restriction).expect("later restriction");
|
||||
assert!(approval_start + approval.len() < restriction_start);
|
||||
assert!(!delta.contains(&initial), "old entries are not resent");
|
||||
assert_eq!(
|
||||
reviews[0].body_json()["client_metadata"]["thread_id"],
|
||||
reviews[1].body_json()["client_metadata"]["thread_id"]
|
||||
);
|
||||
test.codex.shutdown_and_wait().await?;
|
||||
Ok(())
|
||||
|
||||
@@ -208,13 +208,10 @@ async fn guardian_history_uses_deltas_between_eviction_batches() -> Result<()> {
|
||||
"Guardian approval actions require host-native paths"
|
||||
);
|
||||
let server = start_mock_server().await;
|
||||
let test = test_codex()
|
||||
let mut test = test_codex()
|
||||
.with_config(|config| {
|
||||
config.features.enable(Feature::TokenBudget).unwrap();
|
||||
config
|
||||
.features
|
||||
.disable(Feature::GuardianThreadContext)
|
||||
.expect("use the retained legacy history");
|
||||
|
||||
config.update_plan_enabled = true;
|
||||
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
config.approvals_reviewer = ApprovalsReviewer::AutoReview;
|
||||
@@ -233,11 +230,19 @@ async fn guardian_history_uses_deltas_between_eviction_batches() -> Result<()> {
|
||||
.await;
|
||||
let restriction = "Only inspect the repository; do not publish it.";
|
||||
test.submit_text_turn(restriction).await?;
|
||||
test.codex.submit(Op::Compact).await?;
|
||||
wait_for_event(&test.codex, |event| {
|
||||
matches!(event, EventMsg::TurnComplete(_))
|
||||
})
|
||||
.await;
|
||||
// Restore a pre-rollout encrypted checkpoint with its legacy review transcript.
|
||||
let history = test.codex.conversation_history_snapshot().await;
|
||||
let checkpoint: RolloutItem = serde_json::from_value(json!({
|
||||
"type": "compacted", "payload": {
|
||||
"message": "old checkpoint",
|
||||
"replacement_history": [{
|
||||
"type": "compaction", "id": "old", "encrypted_content": "opaque checkpoint"
|
||||
}],
|
||||
"guardian_history": history.review_items().cloned().collect::<Vec<_>>()
|
||||
}
|
||||
}))?;
|
||||
test.codex =
|
||||
super::guardian_checkpoint_migration::resume(&test, &test.codex, vec![checkpoint]).await?;
|
||||
|
||||
let mut responses = Vec::new();
|
||||
for index in 0..3 {
|
||||
|
||||
@@ -411,12 +411,10 @@ enum InstructionSize {
|
||||
Oversized,
|
||||
}
|
||||
|
||||
#[test_case(true, InstructionSize::Normal; "enabled paginated resume")]
|
||||
#[test_case(false, InstructionSize::Normal; "disabled paginated resume")]
|
||||
#[test_case(true, InstructionSize::Oversized; "oversized instruction resume")]
|
||||
#[test_case(InstructionSize::Normal; "enabled paginated resume")]
|
||||
#[test_case(InstructionSize::Oversized; "oversized instruction resume")]
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn retained_instructions_keep_identity_across_compaction_and_resume(
|
||||
thread_context_enabled: bool,
|
||||
instruction_size: InstructionSize,
|
||||
) -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
@@ -435,10 +433,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
|
||||
.with_history_mode(ThreadHistoryMode::Paginated)
|
||||
.with_config(move |config| {
|
||||
config.experimental_thread_store = ThreadStoreConfig::Local;
|
||||
config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("test context mode");
|
||||
|
||||
// Exercise local compaction's rebuilt user messages, not an opaque checkpoint.
|
||||
config.model_provider.name = "Local compaction test provider".to_owned();
|
||||
config
|
||||
@@ -594,40 +589,17 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
|
||||
value
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let next_order = if thread_context_enabled { 7_u64 } else { 0 };
|
||||
let mut expected = json!({
|
||||
let next_order = 7_u64;
|
||||
let expected = json!({
|
||||
"user_messages": user_messages, "user_messages_incomplete": false,
|
||||
"assistant_messages": [{"order": 1, "turn_id": answers[0].turn_id,
|
||||
"message_id": "ordinary-question", "text": QUESTION, "complete": true}],
|
||||
"assistant_messages_incomplete": false,
|
||||
"verified_answers": ordered_answers, "incomplete": false, "next_order": next_order,
|
||||
});
|
||||
if !thread_context_enabled {
|
||||
expected = json!({
|
||||
"user_messages": [], "user_messages_incomplete": true,
|
||||
"assistant_messages": [], "assistant_messages_incomplete": false,
|
||||
"verified_answers": [], "incomplete": false, "next_order": 0,
|
||||
});
|
||||
answers.clear();
|
||||
for item in load_context(&test, &thread).await? {
|
||||
assert!(!matches!(item, RolloutItem::RetainedContext(_)));
|
||||
if let RolloutItem::ResponseItem(envelope) = item {
|
||||
assert!(
|
||||
envelope
|
||||
.metadata
|
||||
.as_ref()
|
||||
.is_none_or(|metadata| metadata.user_input_order.is_none())
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
serde_json::to_value(history.retained_context())?,
|
||||
if thread_context_enabled {
|
||||
expected.clone()
|
||||
} else {
|
||||
serde_json::Value::Null
|
||||
}
|
||||
expected.clone()
|
||||
);
|
||||
assert_eq!(
|
||||
serde_json::to_value(compact_and_assert_answers(&test, &thread, &answers).await?)?,
|
||||
@@ -643,8 +615,8 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
|
||||
compacted
|
||||
.items()
|
||||
.any(|item| item.id().map(ResponseItemId::as_str) == message["message_id"].as_str()),
|
||||
thread_context_enabled,
|
||||
"only thread-owned context preserves original user-message identity"
|
||||
true,
|
||||
"compaction preserves original user-message identity"
|
||||
);
|
||||
}
|
||||
let thread = resume(&test, &thread).await?;
|
||||
@@ -655,11 +627,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
|
||||
.await
|
||||
.retained_context()
|
||||
)?,
|
||||
if thread_context_enabled {
|
||||
expected.clone()
|
||||
} else {
|
||||
serde_json::Value::Null
|
||||
}
|
||||
expected.clone()
|
||||
);
|
||||
thread.shutdown_and_wait().await?;
|
||||
let requests = response_mock.requests();
|
||||
@@ -692,10 +660,7 @@ async fn legacy_checkpoint_recovers_root_excerpt_before_discarding_backup(
|
||||
.with_history_mode(history_mode)
|
||||
.with_config(|config| {
|
||||
config.experimental_thread_store = ThreadStoreConfig::Local;
|
||||
config
|
||||
.features
|
||||
.enable(Feature::GuardianThreadContext)
|
||||
.expect("enable retained instructions");
|
||||
|
||||
config
|
||||
.features
|
||||
.disable(Feature::TokenBudget)
|
||||
@@ -803,98 +768,6 @@ async fn legacy_checkpoint_recovers_root_excerpt_before_discarding_backup(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn disabled_capture_stays_incomplete_after_compaction_and_enabled_resume() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
let server = start_mock_server().await;
|
||||
let test = test_codex()
|
||||
.with_config(|config| {
|
||||
config.experimental_thread_store = ThreadStoreConfig::Local;
|
||||
config
|
||||
.features
|
||||
.disable(Feature::GuardianThreadContext)
|
||||
.expect("disable instruction capture");
|
||||
config
|
||||
.features
|
||||
.disable(Feature::TokenBudget)
|
||||
.expect("use local compaction");
|
||||
config.model_provider.name = "Local compaction test provider".to_owned();
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
mount_sse_sequence(
|
||||
&server,
|
||||
vec![
|
||||
sse(vec![ev_completed("initial-turn")]),
|
||||
sse(vec![
|
||||
ev_assistant_message("summary", "Compacted context."),
|
||||
ev_completed("compact"),
|
||||
]),
|
||||
sse(vec![
|
||||
ev_assistant_message("summary-again", "Compacted context."),
|
||||
ev_completed("compact-again"),
|
||||
]),
|
||||
],
|
||||
)
|
||||
.await;
|
||||
test.codex
|
||||
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
|
||||
text: "Never publish publicly.".to_owned(),
|
||||
text_elements: Vec::new(),
|
||||
}]))
|
||||
.await?;
|
||||
wait_for_event(&test.codex, |event| {
|
||||
matches!(event, EventMsg::TurnComplete(_))
|
||||
})
|
||||
.await;
|
||||
|
||||
let checkpoint = compact_and_assert_answers(&test, &test.codex, &[]).await?;
|
||||
assert!(!checkpoint.user_messages_complete());
|
||||
assert_eq!(checkpoint.ordered_entries().count(), 0);
|
||||
|
||||
let thread_id = test.codex.startup_metadata().thread_id;
|
||||
test.codex.shutdown_and_wait().await?;
|
||||
test.thread_manager.remove_thread(&thread_id).await;
|
||||
let items: Vec<RolloutItem> = serde_json::from_value(serde_json::to_value(
|
||||
load_context(&test, &test.codex).await?,
|
||||
)?)?;
|
||||
let mut config = test.config.clone();
|
||||
config
|
||||
.features
|
||||
.enable(Feature::GuardianThreadContext)
|
||||
.expect("enable instruction capture on resume");
|
||||
let resumed = test
|
||||
.thread_manager
|
||||
.resume_thread_with_history(
|
||||
config,
|
||||
InitialHistory::Resumed(ResumedHistory {
|
||||
conversation_id: thread_id,
|
||||
history: Arc::new(items),
|
||||
rollout_path: None,
|
||||
}),
|
||||
test.thread_manager.auth_manager(),
|
||||
/*parent_trace*/ None,
|
||||
ClientMcpExtensions::default(),
|
||||
)
|
||||
.await?
|
||||
.thread;
|
||||
assert!(
|
||||
!resumed
|
||||
.conversation_history_snapshot()
|
||||
.await
|
||||
.retained_context()
|
||||
.expect("resumed retained context")
|
||||
.user_messages_complete()
|
||||
);
|
||||
assert!(
|
||||
!compact_and_assert_answers(&test, &resumed, &[])
|
||||
.await?
|
||||
.user_messages_complete()
|
||||
);
|
||||
resumed.shutdown_and_wait().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Cover live copied history and a truncated referenced checkpoint. Parent-answer
|
||||
// omission is already exercised by retained_answers_cross_real_session_boundaries.
|
||||
#[test_case(ThreadHistoryMode::Legacy; "copied worker history")]
|
||||
@@ -917,7 +790,6 @@ async fn standalone_fork_retains_inherited_user_instructions(
|
||||
config.model_provider.name = "Local compaction test provider".to_owned();
|
||||
for feature in [
|
||||
Feature::GuardianApproval,
|
||||
Feature::GuardianThreadContext,
|
||||
Feature::Collab,
|
||||
Feature::MultiAgentV2,
|
||||
Feature::DefaultModeRequestUserInput,
|
||||
@@ -1316,7 +1188,6 @@ async fn retained_answers_cross_real_session_boundaries(
|
||||
Feature::DefaultModeRequestUserInput,
|
||||
Feature::Collab,
|
||||
Feature::MultiAgentV2,
|
||||
Feature::GuardianThreadContext,
|
||||
] {
|
||||
config
|
||||
.features
|
||||
|
||||
@@ -500,11 +500,7 @@ for (const phase of ["before", "after"]) {
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
#[test_case(false; "legacy_transcript")]
|
||||
#[test_case(true; "thread_owned_transcript")]
|
||||
async fn guardian_review_compacts_with_summary_despite_parent_token_budget(
|
||||
thread_owned: bool,
|
||||
) -> Result<()> {
|
||||
async fn guardian_review_compacts_with_summary_despite_parent_token_budget() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
skip_if_wine_exec!(
|
||||
Ok(()),
|
||||
@@ -535,10 +531,6 @@ async fn guardian_review_compacts_with_summary_despite_parent_token_budget(
|
||||
});
|
||||
})
|
||||
.with_config(move |config| {
|
||||
config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_owned)
|
||||
.expect("configure Guardian context mode");
|
||||
config.model_context_window = Some(100_000);
|
||||
config.model_auto_compact_token_limit = Some(50_000);
|
||||
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
@@ -658,9 +650,7 @@ async fn guardian_review_compacts_with_summary_despite_parent_token_budget(
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
#[test_case(false; "legacy_transcript")]
|
||||
#[test_case(true; "thread_owned_transcript")]
|
||||
async fn guardian_requests_record_only_their_own_tool_calls(thread_owned: bool) -> Result<()> {
|
||||
async fn guardian_requests_record_only_their_own_tool_calls() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
skip_if_wine_exec!(
|
||||
Ok(()),
|
||||
@@ -669,10 +659,6 @@ async fn guardian_requests_record_only_their_own_tool_calls(thread_owned: bool)
|
||||
|
||||
let server = start_mock_server().await;
|
||||
let mut builder = test_codex().with_config(move |config| {
|
||||
config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_owned)
|
||||
.expect("configure Guardian context mode");
|
||||
config
|
||||
.features
|
||||
.enable(Feature::ExecutedToolCallMetadata)
|
||||
|
||||
@@ -16,11 +16,24 @@ use core_test_support::wait_for_event;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
|
||||
#[test_case::test_case(None; "default")]
|
||||
#[test_case::test_case(Some(false); "retired opt out")]
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn guardian_receives_sender_user_messages_by_default() -> Result<()> {
|
||||
async fn guardian_receives_sender_user_messages_by_default(
|
||||
thread_context: Option<bool>,
|
||||
) -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
let server = responses::start_mock_server().await;
|
||||
let test = test_codex()
|
||||
.with_pre_build_hook(move |home| {
|
||||
if let Some(enabled) = thread_context {
|
||||
std::fs::write(
|
||||
home.join("config.toml"),
|
||||
format!("[features.guardianv2]\nthread_context = {enabled}\n"),
|
||||
)
|
||||
.expect("write compatibility configuration");
|
||||
}
|
||||
})
|
||||
.with_model_info_override("gpt-5.5", |model| {
|
||||
model.auto_review_model_override = Some("gpt-5.6-luna".to_owned());
|
||||
})
|
||||
@@ -120,7 +133,7 @@ async fn guardian_receives_sender_user_messages_by_default() -> Result<()> {
|
||||
let review_body = captured[1].body_json();
|
||||
let review = review_body["input"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.expect("review input")
|
||||
.iter()
|
||||
.filter_map(|item| item["content"].as_array())
|
||||
.flatten()
|
||||
@@ -130,9 +143,9 @@ async fn guardian_receives_sender_user_messages_by_default() -> Result<()> {
|
||||
let history = receiver.conversation_history_snapshot().await;
|
||||
let snapshot = history
|
||||
.retained_context()
|
||||
.unwrap()
|
||||
.expect("thread-owned context")
|
||||
.sender_user_messages()
|
||||
.unwrap();
|
||||
.expect("sender user messages");
|
||||
assert_eq!(
|
||||
snapshot
|
||||
.text
|
||||
@@ -145,11 +158,11 @@ async fn guardian_receives_sender_user_messages_by_default() -> Result<()> {
|
||||
let end = ">>> SENDER USER MESSAGES END\n";
|
||||
let current = review
|
||||
.rsplit_once(start)
|
||||
.unwrap()
|
||||
.expect("sender context start")
|
||||
.1
|
||||
.split(end)
|
||||
.next()
|
||||
.unwrap();
|
||||
.expect("sender context body");
|
||||
assert_eq!(format!("{start}{current}{end}"), snapshot.text);
|
||||
requests.extend(captured);
|
||||
}
|
||||
|
||||
@@ -2812,12 +2812,8 @@ async fn blocked_user_prompt_submit_persists_additional_context_for_next_turn()
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test_case::test_case(/*thread_context_enabled*/ true; "retained context enabled")]
|
||||
#[test_case::test_case(/*thread_context_enabled*/ false; "retained context disabled")]
|
||||
#[tokio::test]
|
||||
async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt(
|
||||
thread_context_enabled: bool,
|
||||
) -> Result<()> {
|
||||
async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
|
||||
@@ -2862,10 +2858,6 @@ async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt(
|
||||
})
|
||||
.with_config(move |config| {
|
||||
trust_discovered_hooks(config);
|
||||
config
|
||||
.features
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("test context mode");
|
||||
});
|
||||
let test = builder.build_with_streaming_server(&server).await?;
|
||||
|
||||
@@ -2925,7 +2917,7 @@ async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt(
|
||||
);
|
||||
|
||||
let history = test.codex.conversation_history_snapshot().await;
|
||||
assert_eq!(history.retained_context().is_some(), thread_context_enabled);
|
||||
assert!(history.retained_context().is_some());
|
||||
let retained = serde_json::to_value(history.retained_context().cloned().unwrap_or_default())?;
|
||||
let messages = retained["user_messages"]
|
||||
.as_array()
|
||||
@@ -2935,23 +2927,16 @@ async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt(
|
||||
.iter()
|
||||
.map(|message| message["text"].clone())
|
||||
.collect::<Vec<_>>(),
|
||||
if thread_context_enabled {
|
||||
vec![json!("initial prompt"), json!("accepted queued prompt")]
|
||||
} else {
|
||||
Vec::new()
|
||||
},
|
||||
vec![json!("initial prompt"), json!("accepted queued prompt")],
|
||||
);
|
||||
assert!(
|
||||
messages
|
||||
.windows(2)
|
||||
.all(|pair| pair[0]["order"].as_u64() < pair[1]["order"].as_u64())
|
||||
);
|
||||
assert_eq!(
|
||||
retained["next_order"],
|
||||
// Three accepted input positions (including the blocked prompt), then
|
||||
// the two completed assistant messages.
|
||||
json!(if thread_context_enabled { 5 } else { 0 })
|
||||
);
|
||||
// Three accepted input positions (including the blocked prompt), then
|
||||
// the two completed assistant messages.
|
||||
assert_eq!(retained["next_order"], json!(5));
|
||||
|
||||
let hook_inputs = read_user_prompt_submit_hook_inputs(test.codex_home_path())?;
|
||||
assert_eq!(hook_inputs.len(), 3);
|
||||
|
||||
@@ -3147,7 +3147,7 @@ async fn assert_parent_compaction_reuse(thread_context_enabled: bool) -> Result<
|
||||
/*metrics*/ None,
|
||||
)
|
||||
.await,
|
||||
(!thread_context_enabled).then_some(ReviewDecision::Approved),
|
||||
None,
|
||||
);
|
||||
|
||||
let oversized_compaction = ResponseItem::Compaction {
|
||||
|
||||
Reference in New Issue
Block a user