diff --git a/codex-rs/core/src/agent/api.rs b/codex-rs/core/src/agent/api.rs index e2ef8de4b5..e4d434667f 100644 --- a/codex-rs/core/src/agent/api.rs +++ b/codex-rs/core/src/agent/api.rs @@ -19,6 +19,7 @@ use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::error::Result; use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TokenUsage; @@ -225,6 +226,8 @@ pub struct AgentTurnOutcome { pub parent_turn_id: Option, pub initiating_agent_path: Option, pub status: AgentStatus, + /// Typed reason used to choose guidance in the parent notification. + pub error_info: Option, } /// Settings shared by the tree. A service tier of `None` restores the default tier. diff --git a/codex-rs/core/src/agent/control/completion.rs b/codex-rs/core/src/agent/control/completion.rs index 0f09257a97..8e041dcf70 100644 --- a/codex-rs/core/src/agent/control/completion.rs +++ b/codex-rs/core/src/agent/control/completion.rs @@ -8,10 +8,12 @@ use crate::TurnStartOptions; use crate::agent::api::AgentTurnOutcome; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; +use crate::session_prefix::format_guardian_interruption_message; use crate::session_prefix::format_inter_agent_completion_message; use codex_protocol::AgentPath; use codex_protocol::items::SubAgentActivityItem; use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentActivityKind; @@ -85,11 +87,21 @@ impl LocalAgentControl { } } - let Some(message) = format_inter_agent_completion_message( - parent_agent_path.clone(), - child_agent_path.clone(), - &status, - ) else { + let message = match (&status, outcome.error_info) { + (AgentStatus::Errored(error), Some(CodexErrorInfo::TooManyDenials)) => { + Some(format_guardian_interruption_message( + parent_agent_path.clone(), + child_agent_path.clone(), + error, + )) + } + _ => format_inter_agent_completion_message( + parent_agent_path.clone(), + child_agent_path.clone(), + &status, + ), + }; + let Some(message) = message else { return; }; // `communication` owns the message. Keep a second copy only when the diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index fbeea01823..99066080af 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2319,6 +2319,7 @@ impl Session { return; }; + let mut error_info = None; let status = match turn_context.terminal_error.lock().await.take() { Some(error) => { let status = AgentStatus::Errored(error.message); @@ -2326,10 +2327,21 @@ impl Session { status } None => { - let Some(status) = agent_status_from_event(msg) else { - return; - }; - status + if let EventMsg::TurnAborted(event) = msg + && event.reason == TurnAbortReason::Interrupted + && let Some(error) = &event.error + && error.codex_error_info == Some(CodexErrorInfo::TooManyDenials) + { + // Report the safety stop to the parent without changing the child's + // interrupted status or notifying for ordinary interruptions. + error_info = error.codex_error_info.clone(); + AgentStatus::Errored(error.message.clone()) + } else { + let Some(status) = agent_status_from_event(msg) else { + return; + }; + status + } } }; if !is_final(&status) { @@ -2349,6 +2361,7 @@ impl Session { .initiating_agent_path() .cloned(), status, + error_info, }, &self.services.rollout_thread_trace, ) diff --git a/codex-rs/core/src/session_prefix.rs b/codex-rs/core/src/session_prefix.rs index 315ef1cabf..8e0bbf5a06 100644 --- a/codex-rs/core/src/session_prefix.rs +++ b/codex-rs/core/src/session_prefix.rs @@ -11,6 +11,7 @@ const COMPLETION_MESSAGE_ENVELOPE_TOKEN_RESERVE: usize = 100; const ERROR_MAX_TOKENS: usize = COMPLETION_MESSAGE_MAX_TOKENS - COMPLETION_MESSAGE_ENVELOPE_TOKEN_RESERVE; const ERROR_NEXT_ACTION: &str = "This agent's turn failed. If you still need this agent, use the available collaboration tools to give it another task."; +const GUARDIAN_NEXT_ACTION: &str = "Tell the user this agent stopped after repeated Guardian denials. Do not resume this agent or retry its blocked work until the user explicitly confirms that it should continue."; // Helpers for model-visible session state markers that are stored in user-role // messages but are not user intent. @@ -35,6 +36,16 @@ pub(crate) fn format_inter_agent_completion_message( Some(InterAgentCompletionMessage::new(task_name, sender, payload).render()) } +pub(crate) fn format_guardian_interruption_message( + task_name: AgentPath, + sender: AgentPath, + error: &str, +) -> String { + let error = truncate_text(error, TruncationPolicy::Tokens(ERROR_MAX_TOKENS)); + let payload = format!("Agent interrupted by Guardian: {error}\n\n{GUARDIAN_NEXT_ACTION}"); + InterAgentCompletionMessage::new(task_name, sender, payload).render() +} + #[cfg(test)] #[path = "session_prefix_tests.rs"] mod tests; diff --git a/codex-rs/core/src/session_prefix_tests.rs b/codex-rs/core/src/session_prefix_tests.rs index 6a658ce852..227f0aa54a 100644 --- a/codex-rs/core/src/session_prefix_tests.rs +++ b/codex-rs/core/src/session_prefix_tests.rs @@ -4,6 +4,8 @@ use codex_utils_output_truncation::approx_token_count; use super::COMPLETION_MESSAGE_MAX_TOKENS; use super::ERROR_NEXT_ACTION; +use super::GUARDIAN_NEXT_ACTION; +use super::format_guardian_interruption_message; use super::format_inter_agent_completion_message; #[test] @@ -18,3 +20,15 @@ fn error_completion_message_stays_below_manual_review_threshold() { assert!(approx_token_count(&message) < COMPLETION_MESSAGE_MAX_TOKENS); assert!(message.contains(ERROR_NEXT_ACTION)); } + +#[test] +fn guardian_interruption_message_stays_below_manual_review_threshold() { + let message = format_guardian_interruption_message( + AgentPath::root(), + AgentPath::try_from("/root/worker").expect("valid agent path"), + &"Action is not authorized. ".repeat(1_000), + ); + + assert!(approx_token_count(&message) < COMPLETION_MESSAGE_MAX_TOKENS); + assert!(message.contains(GUARDIAN_NEXT_ACTION)); +} diff --git a/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs b/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs new file mode 100644 index 0000000000..495ae98065 --- /dev/null +++ b/codex-rs/core/tests/suite/guardian_subagent_notification_tests.rs @@ -0,0 +1,271 @@ +//! Guardian circuit breakers notify the parent without changing the child's interrupted state. + +use super::*; +use codex_config::config_toml::CircuitBreakAction; +use codex_core::config::Constrained; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::protocol::CodexErrorInfo; +use codex_protocol::protocol::ErrorEvent; +use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::TurnAbortReason; +use core_test_support::ThreadIdle; +use core_test_support::streaming_sse::StreamingSseChunk; +use core_test_support::streaming_sse::start_streaming_sse_server; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use test_case::test_case; +use tokio::sync::oneshot; + +#[test_case(CircuitBreakAction::Strict; "strict_notifies_parent")] +#[test_case(CircuitBreakAction::Default; "default_stays_silent")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_circuit_breaker_notifies_parent(action: CircuitBreakAction) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle)); + let test = test_codex() + .with_model_info_override("test-gpt-5.1-codex", |model| { + model.multi_agent_version = Some(MultiAgentVersion::V2); + model.auto_review_model_override = Some(model.slug.clone()); + }) + .with_extensions(Arc::new(extensions.build())) + .with_config(move |config| { + for feature in [ + Feature::Collab, + Feature::MultiAgentV2, + Feature::GuardianApproval, + ] { + config + .features + .enable(feature) + .expect("enable test feature"); + } + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config.guardian_circuit_break_action = action; + config.model_provider.stream_max_retries = Some(0); + }) + .build_with_auto_env(&server) + .await?; + let root = test.session_configured.thread_id; + let is_root = move |request: &wiremock::Request| { + decoded_body(request) + .and_then(|body| serde_json::from_slice::(&body).ok()) + .is_some_and(|body| body["client_metadata"]["thread_id"] == json!(root)) + }; + let is_worker = move |request: &wiremock::Request| { + decoded_body(request) + .and_then(|body| serde_json::from_slice::(&body).ok()) + .is_some_and(|body| { + body["client_metadata"]["x-codex-parent-thread-id"] == json!(root) + && body["client_metadata"]["x-openai-subagent"] != "guardian" + }) + }; + let spawn = mount_sse_once_match( + &server, + is_root, + sse(vec![ + ev_function_call_with_namespace( + SPAWN_CALL_ID, + MULTI_AGENT_V2_NAMESPACE, + "spawn_agent", + &json!({ + "task_name": "worker", "fork_turns": "none", "message": "Run the checks." + }) + .to_string(), + ), + ev_completed("spawn-worker"), + ]), + ) + .await; + let parent_idle = + mount_sse_once_match(&server, is_root, sse(vec![ev_completed("parent-idle")])).await; + let mut calls = Vec::new(); + let mut reviews = Vec::new(); + for call in ["first", "second", "third"] { + calls.push( + mount_sse_once_match( + &server, + is_worker, + sse(vec![ + ev_function_call( + call, + "exec_command", + &json!({ + "cmd": format!("echo {call}"), + "sandbox_permissions": "require_escalated", + "justification": "Exercise Guardian review." + }) + .to_string(), + ), + ev_completed(call), + ]), + ) + .await, + ); + reviews.push(mount_sse_once_match( + &server, + |request: &wiremock::Request| { + decoded_body(request) + .and_then(|body| serde_json::from_slice::(&body).ok()) + .is_some_and(|body| body["client_metadata"]["x-openai-subagent"] == "guardian") + }, + sse(vec![ + ev_assistant_message( + "denied", + r#"{"risk_level":"high","user_authorization":"low","outcome":"deny","rationale":"Action is not authorized."}"#, + ), + ev_completed("review-denied"), + ]), + ).await); + } + // A continuation can race the third denial's interruption. Keep it pending so + // only the actual Guardian circuit breaker can end the child's turn. + let (_release, gate) = oneshot::channel(); + let (pending, _) = start_streaming_sse_server(vec![vec![StreamingSseChunk { + gate: Some(gate), + body: sse(vec![ev_completed("unexpected-worker-completion")]), + }]]) + .await; + Mock::given(method("POST")) + .and(path("/v1/responses")) + .and(is_worker) + .respond_with( + wiremock::ResponseTemplate::new(/*s*/ 307) + .insert_header("location", format!("{}/v1/responses", pending.uri())), + ) + .with_priority(/*priority*/ 20) + .mount(&server) + .await; + + let mut created = test.thread_manager.subscribe_thread_created(); + test.submit_text_turn("Spawn a worker to run the checks.") + .await?; + ThreadIdle::wait(&test.codex).await; + let worker = test + .thread_manager + .get_thread(created.recv().await?) + .await?; + let warning = wait_for_event_match(&worker, |event| match event { + EventMsg::GuardianWarning(warning) if warning.message.contains("3 consecutive") => { + Some(warning.message.clone()) + } + _ => None, + }) + .await; + let aborted = wait_for_event_match(&worker, |event| match event { + EventMsg::TurnAborted(aborted) => Some(aborted.clone()), + _ => None, + }) + .await; + ThreadIdle::wait(&worker).await; + assert_eq!( + (aborted.reason, aborted.error, worker.agent_status().await), + ( + TurnAbortReason::Interrupted, + (action == CircuitBreakAction::Strict).then_some(ErrorEvent { + message: warning.clone(), + codex_error_info: Some(CodexErrorInfo::TooManyDenials), + misalignment: None, + }), + AgentStatus::Interrupted, + ), + ); + + // Delivery is queued. If the parent samples before draining its mailbox, + // wait_agent must observe the notification before the next model request. + let _wait = if action == CircuitBreakAction::Strict { + Some( + mount_sse_once_match( + &server, + move |request: &wiremock::Request| { + is_root(request) && !body_contains(request, "Agent interrupted by Guardian:") + }, + sse(vec![ + ev_function_call_with_namespace( + "wait-worker", + MULTI_AGENT_V2_NAMESPACE, + "wait_agent", + "{}", + ), + ev_completed("parent-wait"), + ]), + ) + .await, + ) + } else { + None + }; + let continued = mount_sse_once_match( + &server, + move |request: &wiremock::Request| { + is_root(request) + && (action == CircuitBreakAction::Default + || body_contains(request, "Agent interrupted by Guardian:")) + }, + sse(vec![ev_completed("parent-continued")]), + ) + .await; + test.submit_text_turn("Report the worker's status.").await?; + let continued_request = continued.single_request(); + let messages = continued_request.inputs_of_type("agent_message"); + let expected = if action == CircuitBreakAction::Strict { + vec![json!({ + "type": "agent_message", + "author": "/root/worker", + "recipient": "/root", + "content": [{ + "type": "input_text", + "text": format!( + "Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/worker\nPayload:\nAgent interrupted by Guardian: {warning}\n\nTell the user this agent stopped after repeated Guardian denials. Do not resume this agent or retry its blocked work until the user explicitly confirms that it should continue." + ), + }], + })] + } else { + Vec::new() + }; + assert_eq!( + strip_response_item_ids_from_json(strip_metadata_from_json(json!(messages))), + json!(expected), + ); + assert_eq!( + spawn.single_request().body_json()["client_metadata"]["thread_id"], + json!(root) + ); + assert!(parent_idle.requests().iter().any(|request| { + request.body_json()["client_metadata"]["thread_id"] == json!(root) + && request.function_call_output(SPAWN_CALL_ID)["output"].is_string() + })); + assert_eq!( + reviews + .iter() + .map(|review| review + .requests() + .into_iter() + .filter(|request| { + request.body_json()["client_metadata"]["x-openai-subagent"] == "guardian" + }) + .count()) + .collect::>(), + vec![1, 1, 1], + ); + assert_eq!( + calls + .iter() + .map(|call| call + .requests() + .into_iter() + .filter(|request| { + request.body_json()["client_metadata"]["x-codex-parent-thread-id"] + == json!(root) + && request.body_json()["client_metadata"]["x-openai-subagent"] != "guardian" + }) + .count()) + .collect::>(), + vec![1, 1, 1] + ); + pending.shutdown().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index ccd44bb92d..89deaae707 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -80,6 +80,9 @@ use super::direct_tool_metadata::tool_call_metadata; #[path = "spawn_settings_tests.rs"] mod spawn_settings_tests; +#[path = "guardian_subagent_notification_tests.rs"] +mod guardian_subagent_notification_tests; + const SPAWN_CALL_ID: &str = "spawn-call-1"; const MULTI_AGENT_V1_NAMESPACE: &str = "multi_agent_v1"; const MULTI_AGENT_V2_NAMESPACE: &str = "collaboration";