diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index d5af9304a..9cd58e044 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -75,7 +75,7 @@ pub(crate) async fn run_codex_thread_interactive( let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_ops, rx_ops) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); - let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs { + let CodexSpawnOk { codex, .. } = Box::pin(Codex::spawn(CodexSpawnArgs { config, auth_manager, models_manager, @@ -97,7 +97,7 @@ pub(crate) async fn run_codex_thread_interactive( inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)), parent_trace: None, analytics_events_client: Some(parent_session.services.analytics_events_client.clone()), - }) + })) .await?; if parent_session.enabled(codex_features::Feature::GeneralAnalytics) { let thread_config = codex.thread_config_snapshot().await; @@ -172,7 +172,7 @@ pub(crate) async fn run_codex_thread_one_shot( // Use a child token so we can stop the delegate after completion without // requiring the caller to cancel the parent token. let child_cancel = cancel_token.child_token(); - let io = run_codex_thread_interactive( + let io = Box::pin(run_codex_thread_interactive( config, auth_manager, models_manager, @@ -181,7 +181,7 @@ pub(crate) async fn run_codex_thread_one_shot( child_cancel.clone(), subagent_source, initial_history, - ) + )) .await?; // Send the initial input to kick off the one-shot turn. diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 861d3576c..18107f737 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -163,14 +163,14 @@ async fn run_guardian_review( let schema = guardian_output_schema(); let terminal_action = action_summary.clone(); - let outcome = run_guardian_review_session( + let outcome = Box::pin(run_guardian_review_session( session.clone(), turn.clone(), request, retry_reason, schema, external_cancel, - ) + )) .await; let assessment = match outcome { @@ -303,14 +303,16 @@ pub(crate) async fn review_approval_request( request: GuardianApprovalRequest, retry_reason: Option, ) -> ReviewDecision { - run_guardian_review( + // Box the delegated review future so callers do not inline the entire + // guardian session state machine into their own async stack. + Box::pin(run_guardian_review( Arc::clone(session), Arc::clone(turn), review_id, request, retry_reason, /*external_cancel*/ None, - ) + )) .await } @@ -322,14 +324,14 @@ pub(crate) async fn review_approval_request_with_cancel( retry_reason: Option, cancel_token: CancellationToken, ) -> ReviewDecision { - run_guardian_review( + Box::pin(run_guardian_review( Arc::clone(session), Arc::clone(turn), review_id, request, retry_reason, Some(cancel_token), - ) + )) .await } @@ -411,22 +413,24 @@ pub(super) async fn run_guardian_review_session( Err(err) => return GuardianReviewOutcome::Completed(Err(err)), }; - match session - .guardian_review_session - .run_review(GuardianReviewSessionParams { - parent_session: Arc::clone(&session), - parent_turn: turn.clone(), - spawn_config: guardian_config, - request, - retry_reason, - schema, - model: guardian_model, - reasoning_effort: guardian_reasoning_effort, - reasoning_summary: turn.reasoning_summary, - personality: turn.personality, - external_cancel, - }) - .await + match Box::pin( + session + .guardian_review_session + .run_review(GuardianReviewSessionParams { + parent_session: Arc::clone(&session), + parent_turn: turn.clone(), + spawn_config: guardian_config, + request, + retry_reason, + schema, + model: guardian_model, + reasoning_effort: guardian_reasoning_effort, + reasoning_summary: turn.reasoning_summary, + personality: turn.personality, + external_cancel, + }), + ) + .await { GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => { GuardianReviewOutcome::Completed(parse_guardian_assessment( diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 08825cfde..239754c50 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -327,32 +327,30 @@ impl GuardianReviewSessionManager { }; if trunk.reuse_key != next_reuse_key { - return self - .run_ephemeral_review( - params, - next_reuse_key, - deadline, - /*fork_snapshot*/ None, - ) - .await; + return Box::pin(self.run_ephemeral_review( + params, + next_reuse_key, + deadline, + /*fork_snapshot*/ None, + )) + .await; } let trunk_guard = match trunk.review_lock.try_lock() { Ok(trunk_guard) => trunk_guard, Err(_) => { - return self - .run_ephemeral_review( - params, - next_reuse_key, - deadline, - trunk.fork_snapshot().await, - ) - .await; + return Box::pin(self.run_ephemeral_review( + params, + next_reuse_key, + deadline, + trunk.fork_snapshot().await, + )) + .await; } }; let (outcome, keep_review_session) = - run_review_on_session(trunk.as_ref(), ¶ms, deadline).await; + Box::pin(run_review_on_session(trunk.as_ref(), ¶ms, deadline)).await; if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) { trunk.refresh_last_committed_fork_snapshot().await; } @@ -488,7 +486,12 @@ impl GuardianReviewSessionManager { let mut cleanup = EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session)); - let (outcome, _) = run_review_on_session(review_session.as_ref(), ¶ms, deadline).await; + let (outcome, _) = Box::pin(run_review_on_session( + review_session.as_ref(), + ¶ms, + deadline, + )) + .await; if let Some(review_session) = self.take_active_ephemeral(&review_session).await { cleanup.disarm(); review_session.shutdown_in_background(); @@ -512,7 +515,7 @@ async fn spawn_guardian_review_session( ), None => (None, 0, None), }; - let codex = run_codex_thread_interactive( + let codex = Box::pin(run_codex_thread_interactive( spawn_config, params.parent_session.services.auth_manager.clone(), params.parent_session.services.models_manager.clone(), @@ -521,7 +524,7 @@ async fn spawn_guardian_review_session( cancel_token.clone(), SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()), initial_history, - ) + )) .await?; Ok(GuardianReviewSession { diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 5e44aa972..8940753d2 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -1332,222 +1332,242 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() -> Ok(()) } -#[tokio::test(flavor = "current_thread")] -async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> { - let first_assessment = serde_json::json!({ - "risk_level": "low", - "user_authorization": "high", - "outcome": "allow", - "rationale": "first guardian rationale", - }) - .to_string(); - let second_assessment = serde_json::json!({ - "risk_level": "low", - "user_authorization": "high", - "outcome": "allow", - "rationale": "second guardian rationale", - }) - .to_string(); - let third_assessment = serde_json::json!({ - "risk_level": "low", - "user_authorization": "high", - "outcome": "allow", - "rationale": "third guardian rationale", - }) - .to_string(); - let (gate_tx, gate_rx) = tokio::sync::oneshot::channel(); - let (server, _) = start_streaming_sse_server(vec![ - vec![StreamingSseChunk { - gate: None, - body: sse(vec![ - ev_response_created("resp-guardian-1"), - ev_assistant_message("msg-guardian-1", &first_assessment), - ev_completed("resp-guardian-1"), - ]), - }], - vec![ - StreamingSseChunk { +#[test] +fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> { + const TEST_STACK_SIZE_BYTES: usize = 2 * 1024 * 1024; + + let handle = + std::thread::Builder::new() + .name("guardian_parallel_reviews_fork_from_last_committed_trunk_history".to_string()) + .stack_size(TEST_STACK_SIZE_BYTES) + .spawn(|| -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + runtime.block_on(Box::pin(async { + let first_assessment = serde_json::json!({ + "risk_level": "low", + "user_authorization": "high", + "outcome": "allow", + "rationale": "first guardian rationale", + }) + .to_string(); + let second_assessment = serde_json::json!({ + "risk_level": "low", + "user_authorization": "high", + "outcome": "allow", + "rationale": "second guardian rationale", + }) + .to_string(); + let third_assessment = serde_json::json!({ + "risk_level": "low", + "user_authorization": "high", + "outcome": "allow", + "rationale": "third guardian rationale", + }) + .to_string(); + let (gate_tx, gate_rx) = tokio::sync::oneshot::channel(); + let (server, _) = start_streaming_sse_server(vec![ + vec![StreamingSseChunk { gate: None, - body: sse(vec![ev_response_created("resp-guardian-2")]), - }, - StreamingSseChunk { - gate: Some(gate_rx), body: sse(vec![ - ev_assistant_message("msg-guardian-2", &second_assessment), - ev_completed("resp-guardian-2"), + ev_response_created("resp-guardian-1"), + ev_assistant_message("msg-guardian-1", &first_assessment), + ev_completed("resp-guardian-1"), ]), - }, - ], - vec![StreamingSseChunk { - gate: None, - body: sse(vec![ - ev_response_created("resp-guardian-3"), - ev_assistant_message("msg-guardian-3", &third_assessment), - ev_completed("resp-guardian-3"), - ]), - }], - ]) - .await; + }], + vec![ + StreamingSseChunk { + gate: None, + body: sse(vec![ev_response_created("resp-guardian-2")]), + }, + StreamingSseChunk { + gate: Some(gate_rx), + body: sse(vec![ + ev_assistant_message("msg-guardian-2", &second_assessment), + ev_completed("resp-guardian-2"), + ]), + }, + ], + vec![StreamingSseChunk { + gate: None, + body: sse(vec![ + ev_response_created("resp-guardian-3"), + ev_assistant_message("msg-guardian-3", &third_assessment), + ev_completed("resp-guardian-3"), + ]), + }], + ]) + .await; - let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await; - seed_guardian_parent_history(&session, &turn).await; + let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await; + seed_guardian_parent_history(&session, &turn).await; - let initial_request = GuardianApprovalRequest::Shell { - id: "shell-guardian-1".to_string(), - command: vec!["git".to_string(), "status".to_string()], - cwd: test_path_buf("/repo/codex-rs/core").abs(), - sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, - additional_permissions: None, - justification: Some("Inspect repo state before proceeding.".to_string()), - }; - assert_eq!( - review_approval_request( + let initial_request = GuardianApprovalRequest::Shell { + id: "shell-guardian-1".to_string(), + command: vec!["git".to_string(), "status".to_string()], + cwd: test_path_buf("/repo/codex-rs/core").abs(), + sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, + additional_permissions: None, + justification: Some("Inspect repo state before proceeding.".to_string()), + }; + assert_eq!( + review_approval_request( + &session, + &turn, + "review-shell-guardian-1".to_string(), + initial_request, + /*retry_reason*/ None + ) + .await, + ReviewDecision::Approved + ); + session + .record_into_history( + &[ + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: "Please inspect pending changes before pushing.".to_string(), + }], + end_turn: None, + phase: None, + }, + ResponseItem::Message { + id: None, + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: "I need approval to run git diff.".to_string(), + }], + end_turn: None, + phase: None, + }, + ], + turn.as_ref(), + ) + .await; + + let second_request = GuardianApprovalRequest::Shell { + id: "shell-guardian-2".to_string(), + command: vec!["git".to_string(), "diff".to_string()], + cwd: test_path_buf("/repo/codex-rs/core").abs(), + sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, + additional_permissions: None, + justification: Some("Inspect pending changes before proceeding.".to_string()), + }; + let third_request = GuardianApprovalRequest::Shell { + id: "shell-guardian-3".to_string(), + command: vec!["git".to_string(), "push".to_string()], + cwd: test_path_buf("/repo/codex-rs/core").abs(), + sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, + additional_permissions: None, + justification: Some("Inspect whether pushing is safe before proceeding.".to_string()), + }; + + let session_for_second = Arc::clone(&session); + let turn_for_second = Arc::clone(&turn); + let mut second_review = tokio::spawn(async move { + review_approval_request( + &session_for_second, + &turn_for_second, + "review-shell-guardian-2".to_string(), + second_request, + Some("trunk follow-up".to_string()), + ) + .await + }); + + let second_request_observed = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if server.requests().await.len() >= 2 { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!( + second_request_observed.is_ok(), + "second guardian request was not observed" + ); + session + .record_into_history( + &[ + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: "Now inspect whether pushing is safe.".to_string(), + }], + end_turn: None, + phase: None, + }, + ResponseItem::Message { + id: None, + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: "I need approval to push after the diff check.".to_string(), + }], + end_turn: None, + phase: None, + }, + ], + turn.as_ref(), + ) + .await; + + let third_decision = review_approval_request( &session, &turn, - "review-shell-guardian-1".to_string(), - initial_request, - /*retry_reason*/ None - ) - .await, - ReviewDecision::Approved - ); - session - .record_into_history( - &[ - ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { - text: "Please inspect pending changes before pushing.".to_string(), - }], - end_turn: None, - phase: None, - }, - ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { - text: "I need approval to run git diff.".to_string(), - }], - end_turn: None, - phase: None, - }, - ], - turn.as_ref(), + "review-shell-guardian-3".to_string(), + third_request, + Some("parallel follow-up".to_string()), ) .await; + assert_eq!(third_decision, ReviewDecision::Approved); + let requests = server.requests().await; + assert_eq!(requests.len(), 3); + let third_request_body = serde_json::from_slice::(&requests[2])?; + let third_request_body_text = third_request_body.to_string(); + assert!( + third_request_body_text.contains("first guardian rationale"), + "forked guardian review should include the last committed trunk assessment" + ); + let third_user_message = last_user_message_text_from_body(&third_request_body); + assert!(third_user_message.contains(">>> TRANSCRIPT DELTA START\n")); + assert!( + third_user_message.contains("[5] user: Please inspect pending changes before pushing.") + ); + assert!(third_user_message.contains("[7] user: Now inspect whether pushing is safe.")); + assert!(!third_user_message.contains("[1] user: Please check the repo visibility")); + assert!( + !third_request_body_text.contains("second guardian rationale"), + "forked guardian review should not include the still in-flight trunk assessment" + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut second_review) + .await + .is_err(), + "the trunk guardian review should still be blocked on its gated response" + ); - let second_request = GuardianApprovalRequest::Shell { - id: "shell-guardian-2".to_string(), - command: vec!["git".to_string(), "diff".to_string()], - cwd: test_path_buf("/repo/codex-rs/core").abs(), - sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, - additional_permissions: None, - justification: Some("Inspect pending changes before proceeding.".to_string()), - }; - let third_request = GuardianApprovalRequest::Shell { - id: "shell-guardian-3".to_string(), - command: vec!["git".to_string(), "push".to_string()], - cwd: test_path_buf("/repo/codex-rs/core").abs(), - sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, - additional_permissions: None, - justification: Some("Inspect whether pushing is safe before proceeding.".to_string()), - }; + gate_tx + .send(()) + .expect("second guardian review gate should still be open"); + assert_eq!(second_review.await?, ReviewDecision::Approved); + server.shutdown().await; - let session_for_second = Arc::clone(&session); - let turn_for_second = Arc::clone(&turn); - let mut second_review = tokio::spawn(async move { - review_approval_request( - &session_for_second, - &turn_for_second, - "review-shell-guardian-2".to_string(), - second_request, - Some("trunk follow-up".to_string()), - ) - .await - }); + Ok(()) + })) + })?; - let second_request_observed = tokio::time::timeout(Duration::from_secs(5), async { - loop { - if server.requests().await.len() >= 2 { - break; - } - tokio::task::yield_now().await; - } - }) - .await; - assert!( - second_request_observed.is_ok(), - "second guardian request was not observed" - ); - session - .record_into_history( - &[ - ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { - text: "Now inspect whether pushing is safe.".to_string(), - }], - end_turn: None, - phase: None, - }, - ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { - text: "I need approval to push after the diff check.".to_string(), - }], - end_turn: None, - phase: None, - }, - ], - turn.as_ref(), - ) - .await; - - let third_decision = review_approval_request( - &session, - &turn, - "review-shell-guardian-3".to_string(), - third_request, - Some("parallel follow-up".to_string()), - ) - .await; - assert_eq!(third_decision, ReviewDecision::Approved); - let requests = server.requests().await; - assert_eq!(requests.len(), 3); - let third_request_body = serde_json::from_slice::(&requests[2])?; - let third_request_body_text = third_request_body.to_string(); - assert!( - third_request_body_text.contains("first guardian rationale"), - "forked guardian review should include the last committed trunk assessment" - ); - let third_user_message = last_user_message_text_from_body(&third_request_body); - assert!(third_user_message.contains(">>> TRANSCRIPT DELTA START\n")); - assert!( - third_user_message.contains("[5] user: Please inspect pending changes before pushing.") - ); - assert!(third_user_message.contains("[7] user: Now inspect whether pushing is safe.")); - assert!(!third_user_message.contains("[1] user: Please check the repo visibility")); - assert!( - !third_request_body_text.contains("second guardian rationale"), - "forked guardian review should not include the still in-flight trunk assessment" - ); - assert!( - tokio::time::timeout(Duration::from_millis(100), &mut second_review) - .await - .is_err(), - "the trunk guardian review should still be blocked on its gated response" - ); - - gate_tx - .send(()) - .expect("second guardian review gate should still be open"); - assert_eq!(second_review.await?, ReviewDecision::Approved); - server.shutdown().await; - - Ok(()) + match handle.join() { + Ok(result) => result, + Err(_) => Err(anyhow::anyhow!( + "guardian_parallel_reviews_fork_from_last_committed_trunk_history thread panicked" + )), + } } #[test] fn guardian_review_session_config_preserves_parent_network_proxy() {