diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index f2a9d61071..95133ec108 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -402,6 +402,7 @@ fn submit_nested_tool( runtime_tool_call_id, }, cancellation_token, + Arc::default(), ); Ok(async move { Ok(result.await?.code_mode_result()) }) } diff --git a/codex-rs/core/src/tools/context.rs b/codex-rs/core/src/tools/context.rs index 0b73875bd8..1b02ab7abf 100644 --- a/codex-rs/core/src/tools/context.rs +++ b/codex-rs/core/src/tools/context.rs @@ -26,6 +26,8 @@ use serde::Serialize; use serde_json::Value as JsonValue; use std::num::NonZeroUsize; use std::sync::Arc; +use std::sync::OnceLock; +use std::sync::atomic::AtomicBool; use std::time::Duration; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; @@ -42,6 +44,14 @@ where pub type SharedTurnDiffTracker = Arc>; +/// Host-observed state for one call, owned outside its abortable dispatch task. +/// Delivery does not finish the call: post-tool hooks can still be cancelled. +#[derive(Default)] +pub(crate) struct ToolCallState { + pub(crate) terminal_outcome_reached: AtomicBool, + pub(crate) delivered_assistant_message: OnceLock, +} + #[derive(Clone, Debug, Eq, PartialEq)] pub enum ToolCallSource { Direct, diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 12d3e8dfdf..53ddadc403 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -1,6 +1,5 @@ use std::sync::Arc; use std::sync::OnceLock; -use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Instant; @@ -20,6 +19,7 @@ use crate::session::step_context::StepContext; use crate::tools::call_trace; use crate::tools::context::AbortedToolOutput; use crate::tools::context::SharedTurnDiffTracker; +use crate::tools::context::ToolCallState; use crate::tools::context::ToolPayload; use crate::tools::lifecycle::notify_tool_aborted; use crate::tools::registry::AnyToolResult; @@ -83,8 +83,14 @@ impl ToolCallRuntime { let recorder = self.session.services.executed_tool_calls.clone(); let recorded_call = recorder.prepare_direct_call(&call, &source, &self.step_context); let step_context = Arc::clone(&self.step_context); - let future = - self.handle_tool_call_with_source(step_context, call, source, cancellation_token); + let call_state = Arc::new(ToolCallState::default()); + let future = self.handle_tool_call_with_source( + step_context, + call, + source, + cancellation_token, + Arc::clone(&call_state), + ); async move { let result = future.await; let mut recorded_call = @@ -103,6 +109,12 @@ impl ToolCallRuntime { ResponseItemEnvelope::new(Self::failure_response(error_call, other).into()) } }; + if let Some(text) = call_state.delivered_assistant_message.get() { + response + .metadata + .get_or_insert_default() + .delivered_assistant_message = Some(text.clone()); + } recorder.attach_direct_call_to_output(&mut response.item, recorded_call); Ok(response) } @@ -115,6 +127,7 @@ impl ToolCallRuntime { call: ToolCall, source: ToolCallSource, cancellation_token: CancellationToken, + call_state: Arc, ) -> impl std::future::Future> { self.session .services @@ -144,8 +157,7 @@ impl ToolCallRuntime { let abort_session = Arc::clone(&session); let abort_source = source.clone(); let abort_turn = Arc::clone(&turn); - let terminal_outcome_reached = Arc::new(AtomicBool::new(false)); - let dispatch_terminal_outcome_reached = Arc::clone(&terminal_outcome_reached); + let dispatch_call_state = Arc::clone(&call_state); let dispatch_call = call.clone(); let thread_id = session.thread_id; let trace_source = match &source { @@ -188,14 +200,14 @@ impl ToolCallRuntime { } let result = router - .dispatch_tool_call_with_terminal_outcome( + .dispatch_tool_call_with_state( session, step_context, invocation_cancellation_token, tracker, dispatch_call, source, - dispatch_terminal_outcome_reached, + dispatch_call_state, ) .instrument(dispatch_span.clone()) .await; @@ -223,7 +235,7 @@ impl ToolCallRuntime { let mut result = tokio::select! { res = &mut dispatch_handle => res.map_err(Self::tool_task_join_error)?, _ = cancellation_token.cancelled() => { - if terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() { + if call_state.terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() { dispatch_handle.await.map_err(Self::tool_task_join_error)? } else { let secs = started.elapsed().as_secs_f32().max(0.1); diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index e9f7b8c35e..4d3c50c0f9 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -1,6 +1,5 @@ use std::collections::BTreeMap; use std::sync::Arc; -use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Duration; @@ -14,6 +13,7 @@ use crate::memory_usage::shell_script_for_invocation; use crate::session::session::Session; use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; +use crate::tools::context::ToolCallState; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; @@ -193,7 +193,6 @@ pub(crate) struct AnyToolResult { impl AnyToolResult { pub(crate) fn into_response(self) -> ResponseItemEnvelope { - let delivered_assistant_message = self.delivered_assistant_message(); let Self { call_id, payload, @@ -203,11 +202,8 @@ impl AnyToolResult { let history_truncation_token_limit = result.fallback_token_limit_override(); ResponseItemEnvelope { item: result.to_response_item(&call_id, &payload).into(), - metadata: (history_truncation_token_limit.is_some() - || delivered_assistant_message.is_some()) - .then_some(CodexHarnessMetadata { - history_truncation_token_limit, - delivered_assistant_message, + metadata: history_truncation_token_limit.map(|limit| CodexHarnessMetadata { + history_truncation_token_limit: Some(limit), ..Default::default() }), } @@ -517,10 +513,10 @@ impl ToolRegistry { clippy::await_holding_invalid_type, reason = "tool dispatch must keep active-turn accounting atomic" )] - pub(crate) async fn dispatch_any_with_terminal_outcome( + pub(crate) async fn dispatch_any_with_state( &self, mut invocation: ToolInvocation, - terminal_outcome_reached: Option>, + call_state: Option>, ) -> Result { let tool_name = invocation.tool_name.clone(); let call_id_owned = invocation.call_id.clone(); @@ -608,7 +604,7 @@ impl ToolRegistry { dispatch_trace.record_failed(&err); notify_tool_finish_if_unclaimed( &invocation, - terminal_outcome_reached.as_deref(), + call_state.as_deref(), ToolCallOutcome::Blocked, ) .await; @@ -628,7 +624,7 @@ impl ToolRegistry { dispatch_trace.record_failed(&err); notify_tool_finish_if_unclaimed( &invocation, - terminal_outcome_reached.as_deref(), + call_state.as_deref(), ToolCallOutcome::Failed { handler_executed: false, }, @@ -675,7 +671,7 @@ impl ToolRegistry { log_payload.as_ref(), &tool_result_tags, &extra_trace_fields, - || handle_any_tool(tool.as_ref(), invocation.clone()), + || handle_any_tool(tool.as_ref(), invocation.clone(), call_state.as_deref()), |result| { ( result.result.log_output(), @@ -736,12 +732,8 @@ impl ToolRegistry { handler_executed: true, }, }; - notify_tool_finish_if_unclaimed( - &invocation, - terminal_outcome_reached.as_deref(), - lifecycle_outcome, - ) - .await; + notify_tool_finish_if_unclaimed(&invocation, call_state.as_deref(), lifecycle_outcome) + .await; match result { Ok(mut result) => { @@ -783,10 +775,10 @@ impl ToolRegistry { async fn notify_tool_finish_if_unclaimed( invocation: &ToolInvocation, - terminal_outcome_reached: Option<&AtomicBool>, + call_state: Option<&ToolCallState>, outcome: ToolCallOutcome, ) -> bool { - if terminal_outcome_reached.is_some_and(|reached| reached.swap(true, Ordering::AcqRel)) { + if call_state.is_some_and(|state| state.terminal_outcome_reached.swap(true, Ordering::AcqRel)) { return false; } @@ -797,11 +789,26 @@ async fn notify_tool_finish_if_unclaimed( async fn handle_any_tool( tool: &dyn CoreToolRuntime, invocation: ToolInvocation, + call_state: Option<&ToolCallState>, ) -> Result { let call_id = invocation.call_id.clone(); let payload = invocation.payload.clone(); let output = tool.handle(invocation.clone()).await?; - if output.contains_external_context() + let post_tool_use_payload = + CoreToolRuntime::post_tool_use_payload(tool, &invocation, output.as_ref()); + let result = AnyToolResult { + call_id, + payload, + result: output, + post_tool_use_payload, + }; + // Capture confirmed delivery before any further await, including post-tool hooks. + if let Some(call_state) = call_state + && let Some(text) = result.delivered_assistant_message() + { + let _ = call_state.delivered_assistant_message.set(text); + } + if result.result.contains_external_context() && invocation.turn.config.memories.disable_on_external_context { state_db::mark_thread_memory_mode_polluted( @@ -811,14 +818,7 @@ async fn handle_any_tool( ) .await; } - let post_tool_use_payload = - CoreToolRuntime::post_tool_use_payload(tool, &invocation, output.as_ref()); - Ok(AnyToolResult { - call_id, - payload, - result: output, - post_tool_use_payload, - }) + Ok(result) } fn function_hook_tool_name(invocation: &ToolInvocation) -> HookToolName { diff --git a/codex-rs/core/src/tools/registry_tests.rs b/codex-rs/core/src/tools/registry_tests.rs index b09d12ba15..f1992670d4 100644 --- a/codex-rs/core/src/tools/registry_tests.rs +++ b/codex-rs/core/src/tools/registry_tests.rs @@ -756,25 +756,25 @@ async fn dispatch_uses_canonical_tool_names_for_lifecycle_contributors() -> anyh let turn = Arc::new(turn); registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation( Arc::clone(&session), Arc::clone(&turn), "ok-call", codex_tools::ToolName::namespaced(DEFAULT_FUNCTION_NAMESPACE, "ok_tool"), ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await?; let err = match registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation( Arc::clone(&session), Arc::clone(&turn), "failing-call", failing_tool.clone(), ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await { diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index c3526fee10..7111483348 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -5,6 +5,7 @@ use crate::session::step_context::StepContext; #[cfg(test)] use crate::session::turn_context::TurnContext; use crate::tools::context::SharedTurnDiffTracker; +use crate::tools::context::ToolCallState; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolPayload; #[cfg(test)] @@ -27,7 +28,6 @@ use codex_tools::ToolSpec; use std::borrow::Cow; use std::collections::BTreeMap; use std::sync::Arc; -use std::sync::atomic::AtomicBool; use tokio_util::sync::CancellationToken; use tracing::instrument; @@ -313,14 +313,14 @@ impl ToolRouter { tracker, call, source, - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await } #[instrument(level = "trace", skip_all, err)] #[allow(clippy::too_many_arguments)] - pub(crate) async fn dispatch_tool_call_with_terminal_outcome( + pub(crate) async fn dispatch_tool_call_with_state( &self, session: Arc, step_context: Arc, @@ -328,7 +328,7 @@ impl ToolRouter { tracker: SharedTurnDiffTracker, call: ToolCall, source: ToolCallSource, - terminal_outcome_reached: Arc, + call_state: Arc, ) -> Result { self.dispatch_tool_call_with_code_mode_result_inner( session, @@ -337,7 +337,7 @@ impl ToolRouter { tracker, call, source, - Some(terminal_outcome_reached), + Some(call_state), ) .await } @@ -351,7 +351,7 @@ impl ToolRouter { tracker: SharedTurnDiffTracker, call: ToolCall, source: ToolCallSource, - terminal_outcome_reached: Option>, + call_state: Option>, ) -> Result { let ToolCall { tool_name, @@ -375,7 +375,7 @@ impl ToolRouter { }; self.registry - .dispatch_any_with_terminal_outcome(invocation, terminal_outcome_reached) + .dispatch_any_with_state(invocation, call_state) .await } } diff --git a/codex-rs/core/src/tools/tool_dispatch_trace_tests.rs b/codex-rs/core/src/tools/tool_dispatch_trace_tests.rs index 692d1704e8..5c841f56a9 100644 --- a/codex-rs/core/src/tools/tool_dispatch_trace_tests.rs +++ b/codex-rs/core/src/tools/tool_dispatch_trace_tests.rs @@ -131,7 +131,7 @@ async fn dispatch_lifecycle_trace_records_direct_and_code_mode_requesters() -> a let turn = Arc::new(turn); registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation( Arc::clone(&session), Arc::clone(&turn), @@ -140,11 +140,11 @@ async fn dispatch_lifecycle_trace_records_direct_and_code_mode_requesters() -> a ToolCallSource::Direct, "{}", ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await?; registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation( session, turn, @@ -156,7 +156,7 @@ async fn dispatch_lifecycle_trace_records_direct_and_code_mode_requesters() -> a }, "{}", ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await?; @@ -216,7 +216,7 @@ async fn dispatch_lifecycle_trace_records_unsupported_tool_failures() -> anyhow: let turn = Arc::new(turn); let result = registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation( session, turn, @@ -225,7 +225,7 @@ async fn dispatch_lifecycle_trace_records_unsupported_tool_failures() -> anyhow: ToolCallSource::Direct, "{}", ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await; @@ -251,7 +251,7 @@ async fn dispatch_lifecycle_trace_records_incompatible_payload_failures() -> any let turn = Arc::new(turn); let result = registry - .dispatch_any_with_terminal_outcome( + .dispatch_any_with_state( test_invocation_with_payload( session, turn, @@ -262,7 +262,7 @@ async fn dispatch_lifecycle_trace_records_incompatible_payload_failures() -> any input: "{}".to_string(), }, ), - /*terminal_outcome_reached*/ None, + /*call_state*/ None, ) .await; @@ -308,7 +308,7 @@ async fn missing_code_mode_wait_traces_only_the_wait_tool_call() -> anyhow::Resu ); registry - .dispatch_any_with_terminal_outcome(invocation, /*terminal_outcome_reached*/ None) + .dispatch_any_with_state(invocation, /*call_state*/ None) .await?; let replayed = codex_rollout_trace::replay_bundle(single_bundle_dir(temp.path())?)?; diff --git a/codex-rs/core/src/tools/user_messaging.rs b/codex-rs/core/src/tools/user_messaging.rs index 0cbb219158..d404e29487 100644 --- a/codex-rs/core/src/tools/user_messaging.rs +++ b/codex-rs/core/src/tools/user_messaging.rs @@ -1,5 +1,5 @@ -//! Captures confirmed, post-hook User Messaging text for Guardian's parent context. -//! The completed output owns the evidence; attempted arguments are not delivery proof. +//! Extracts confirmed User Messaging text after input-rewriting hooks. +//! The per-call runtime state retains this evidence independently of post-tool hooks. use crate::guardian::GUARDIAN_MAX_ROOT_MESSAGE_TOKENS; use crate::guardian::guardian_truncate_text; diff --git a/codex-rs/core/src/tools/user_messaging_tests.rs b/codex-rs/core/src/tools/user_messaging_tests.rs index 039139b9c9..5bde9903dc 100644 --- a/codex-rs/core/src/tools/user_messaging_tests.rs +++ b/codex-rs/core/src/tools/user_messaging_tests.rs @@ -27,12 +27,6 @@ fn undelivered_or_unrelated_messages_have_no_evidence(name: &str, input: Value, tool_input: input, tool_response: json!({}), }), - } - .into_response(); - assert_eq!( - output - .metadata - .and_then(|metadata| metadata.delivered_assistant_message), - None, - ); + }; + assert_eq!(output.delivered_assistant_message(), None); } diff --git a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs index 712e6f1516..0a20d67c42 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -47,6 +47,7 @@ use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use test_case::test_case; +use tokio::sync::Notify; const INITIAL_PROMPT: &str = "Spawn a worker to inspect the deployment. Do not delete production data."; @@ -67,6 +68,7 @@ const ROOT_QUESTION: &str = "May the worker deploy the reviewed change?"; const ROOT_ANSWER: &str = "Only deploy privately."; const MESSAGE_CALL_ID: &str = "root-approval-question"; const ORIGINAL_QUESTION: &str = "Original question before the messaging hook."; +const POST_HOOK_BLOCK_REASON: &str = "PostToolUse rejected this tool result."; #[derive(Clone, Copy)] enum RootAnswer { @@ -74,6 +76,14 @@ enum RootAnswer { Oversized, } +#[derive(Clone, Copy)] +enum MessagingOutcome { + Complete, + Block, + CancelBeforeConfirmation, + CancelPostHook, +} + #[derive(Clone, Copy)] enum RootContext { Legacy, @@ -143,17 +153,21 @@ async fn mount_completion( .await } -#[test_case(RootAnswer::Complete, RootContext::Legacy; "legacy_complete_answer")] -#[test_case(RootAnswer::Oversized, RootContext::Legacy; "legacy_oversized_answer")] -#[test_case(RootAnswer::Complete, RootContext::Retained; "retained_complete_answer")] -#[test_case(RootAnswer::Oversized, RootContext::Retained; "retained_oversized_answer")] -#[test_case(RootAnswer::Complete, RootContext::Migrating; "migrating_complete_answer")] -#[test_case(RootAnswer::Oversized, RootContext::Migrating; "migrating_oversized_answer")] -#[test_case(RootAnswer::Complete, RootContext::RetainedAtMessageLimit; "bounded_retained_root_messages")] +#[test_case(RootAnswer::Complete, RootContext::Legacy, MessagingOutcome::Complete; "legacy_complete_answer")] +#[test_case(RootAnswer::Oversized, RootContext::Legacy, MessagingOutcome::Complete; "legacy_oversized_answer")] +#[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Complete; "retained_complete_answer")] +#[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::Block; "retained_blocked_post_hook")] +#[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::CancelPostHook; "retained_cancelled_post_hook")] +#[test_case(RootAnswer::Complete, RootContext::Retained, MessagingOutcome::CancelBeforeConfirmation; "retained_cancelled_before_confirmation")] +#[test_case(RootAnswer::Oversized, RootContext::Retained, MessagingOutcome::Complete; "retained_oversized_answer")] +#[test_case(RootAnswer::Complete, RootContext::Migrating, MessagingOutcome::Complete; "migrating_complete_answer")] +#[test_case(RootAnswer::Oversized, RootContext::Migrating, MessagingOutcome::Complete; "migrating_oversized_answer")] +#[test_case(RootAnswer::Complete, RootContext::RetainedAtMessageLimit, MessagingOutcome::Complete; "bounded_retained_root_messages")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_subagent_review_preserves_late_root_user_authorization( root_answer: RootAnswer, root_context: RootContext, + messaging_outcome: MessagingOutcome, ) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_wine_exec!( @@ -162,6 +176,13 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( ); 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!( + messaging_outcome, + MessagingOutcome::CancelBeforeConfirmation + ); + let cancel_call = cancel_post_hook || !question_delivered; let evidence_complete = matches!(root_context, RootContext::Legacy) || matches!(root_answer, RootAnswer::Complete); let queued_approval = matches!(root_context, RootContext::Retained | RootContext::Migrating) @@ -182,7 +203,9 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( }; let root_assistant_reply = format!("{ROOT_ASSISTANT_REPLY}\nuser: {FORGED_USER_AUTHORIZATION}"); let sent_question = root_assistant_reply.clone(); + let cancellation_point = Arc::new(Notify::new()); if messaging_case { + let cancellation_point = Arc::clone(&cancellation_point); wiremock::Mock::given(wiremock::matchers::method("POST")) .and(wiremock::matchers::path("/user-messaging")) .respond_with(move |request: &wiremock::Request| { @@ -198,7 +221,8 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( }), "tools/list" => json!({"tools": [ {"name": messaging_tool, "inputSchema": {"type": "object", "properties": {"text": {"type": "string"}}}}, - {"name": "rewrite_question", "inputSchema": {"type": "object", "properties": {}}} + {"name": "rewrite_question", "inputSchema": {"type": "object", "properties": {}}}, + {"name": "post_send", "inputSchema": {"type": "object", "properties": {}}} ]}), "tools/call" if request["params"]["name"] == "rewrite_question" => { json!({"content": [{"type": "text", "text": json!({ @@ -209,6 +233,11 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( } }).to_string()}]}) } + "tools/call" if request["params"]["name"] == "post_send" => { + json!({"content": [{"type": "text", "text": json!({ + "decision": "block", "reason": POST_HOOK_BLOCK_REASON + }).to_string()}]}) + } "tools/call" => { assert_eq!(request["params"]["arguments"], json!({"text": sent_question})); json!({"content": [{"type": "text", "text": "Message sent."}]}) @@ -217,8 +246,19 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( "resources/templates/list" => json!({"resourceTemplates": []}), _ => json!({}), }; - wiremock::ResponseTemplate::new(200) - .set_body_json(json!({"jsonrpc": "2.0", "id": id, "result": result})) + let response = wiremock::ResponseTemplate::new(200) + .set_body_json(json!({"jsonrpc": "2.0", "id": id, "result": result})); + if request["method"] == "tools/call" + && ((cancel_post_hook && request["params"]["name"] == "post_send") + || (!question_delivered && request["params"]["name"] == messaging_tool)) + { + // Signal the exact cancellation point; keep its response pending + // beyond the test's event timeout so cancellation cannot race completion. + cancellation_point.notify_one(); + response.set_delay(Duration::from_secs(/*secs*/ 60)) + } else { + response + } }) .mount(&server) .await; @@ -229,7 +269,7 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( if !messaging_case { return; } - let hooks = json!({"hooks": {"PreToolUse": [{ + let mut hooks = json!({"hooks": {"PreToolUse": [{ "matcher": "mcp__codex_apps__user_messaging.*send_message", "hooks": [{ "type": "mcp_tool", @@ -238,6 +278,15 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( "input": {} }] }]}}); + if block_post_hook || cancel_post_hook { + hooks["hooks"]["PostToolUse"] = json!([{ + "matcher": "mcp__codex_apps__user_messaging.*send_message", + "hooks": [{ + "type": "mcp_tool", "server": messaging_namespace, + "tool": "post_send", "input": {} + }] + }]); + } fs::write(home.join("hooks.json"), hooks.to_string()) .expect("write messaging rewrite hook"); }) @@ -360,9 +409,11 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( sse(question_events), ) .await; - if messaging_case { - mount_completion(&server, root_thread_id, MESSAGE_CALL_ID).await; - } + let messaging_completion = if messaging_case && !cancel_call { + Some(mount_completion(&server, root_thread_id, MESSAGE_CALL_ID).await) + } else { + None + }; mount_sse_once_match( &server, move |request: &wiremock::Request| { @@ -377,7 +428,40 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( ) .await; - test.submit_text_turn(INITIAL_PROMPT).await?; + if cancel_call { + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: INITIAL_PROMPT.to_owned(), + text_elements: Vec::new(), + }])) + .await?; + tokio::time::timeout( + Duration::from_secs(/*secs*/ 10), + cancellation_point.notified(), + ) + .await?; + test.codex.submit(Op::Interrupt).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + assert!(test.codex.conversation_history_snapshot().await.items().any(|item| { + matches!(item, ResponseItem::FunctionCallOutput { call_id, output, .. } + if call_id.as_deref() == Some(MESSAGE_CALL_ID) + && output.body.to_text().is_some_and(|text| text.starts_with("aborted by user"))) + })); + } else { + test.submit_text_turn(INITIAL_PROMPT).await?; + } + if block_post_hook { + assert_eq!( + messaging_completion + .expect("messaging completion") + .function_call_output_text(MESSAGE_CALL_ID) + .as_deref(), + Some(POST_HOOK_BLOCK_REASON), + ); + } let worker_thread_id = created_threads.recv().await?; let worker_thread = test.thread_manager.get_thread(worker_thread_id).await?; wait_for_event(worker_thread.as_ref(), |event| { @@ -643,13 +727,21 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( } messages.push(GuardianRootMessage::User(INITIAL_PROMPT.to_owned())); let first_assistant = match root_answer { - RootAnswer::Complete => 5, + RootAnswer::Complete => { + if question_delivered { + 5 + } else { + 4 + } + } RootAnswer::Oversized => 3, }; messages.extend((first_assistant..8).map(|index| { GuardianRootMessage::Assistant(format!("Deployment inspection update {index}.")) })); - messages.push(GuardianRootMessage::Assistant(root_assistant_reply.clone())); + if question_delivered { + messages.push(GuardianRootMessage::Assistant(root_assistant_reply.clone())); + } messages.push(GuardianRootMessage::User(USER_APPROVAL.to_owned())); if queued_approval { messages.push(GuardianRootMessage::User(QUEUED_APPROVAL.to_owned())); @@ -723,7 +815,8 @@ 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), + && (!messaging_case || retained_context_enabled) + && question_delivered, ); } assert!(!guardian_transcript.contains(ROOT_ASSISTANT_COMMENTARY)); @@ -981,7 +1074,9 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( let approval = GuardianRootMessage::User(USER_APPROVAL.to_owned()); assert_eq!( exchange, - if preserve_acceptance_order { + if !question_delivered { + vec![approval] + } else if preserve_acceptance_order { vec![ GuardianRootMessage::Assistant(root_assistant_reply.clone()), approval,