Fix for CI Tests failing from stack overflow (#17846)

### **Issue**
guardian_parallel_reviews_fork_from_last_committed_trunk_history was
failing on Windows/Bazel with a stack overflow:

`thread
'guardian::tests::guardian_parallel_reviews_fork_from_last_committed_trunk_history'
has overflowed its stack`

- This problem was a stack-headroom problem

### **Solution**

Reduced stack pressure in the guardian async path by boxing thin wrapper
futures, and run the affected test on a dedicated 2 MiB thread stack.

Concretely:
- added Box::pin(...) around thin async wrapper hops in the guardian
review/delegate path
- changed
guardian_parallel_reviews_fork_from_last_committed_trunk_history to run
inside an explicitly sized thread stack so it has enough headroom in
low-stack environments
This commit is contained in:
Won Park authored and GitHub committed 2026-04-14 18:04:35 -07:00
1 parent 3cc689fb23
commit 2bfa627613
4 files changed
+277 -250

No files matched your search

+4 -4
View File
@@ -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_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_ops, rx_ops) = 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, config,
auth_manager, auth_manager,
models_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)), inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)),
parent_trace: None, parent_trace: None,
analytics_events_client: Some(parent_session.services.analytics_events_client.clone()), analytics_events_client: Some(parent_session.services.analytics_events_client.clone()),
}) }))
.await?; .await?;
if parent_session.enabled(codex_features::Feature::GeneralAnalytics) { if parent_session.enabled(codex_features::Feature::GeneralAnalytics) {
let thread_config = codex.thread_config_snapshot().await; 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 // Use a child token so we can stop the delegate after completion without
// requiring the caller to cancel the parent token. // requiring the caller to cancel the parent token.
let child_cancel = cancel_token.child_token(); let child_cancel = cancel_token.child_token();
let io = run_codex_thread_interactive( let io = Box::pin(run_codex_thread_interactive(
config, config,
auth_manager, auth_manager,
models_manager, models_manager,
@@ -181,7 +181,7 @@ pub(crate) async fn run_codex_thread_one_shot(
child_cancel.clone(), child_cancel.clone(),
subagent_source, subagent_source,
initial_history, initial_history,
) ))
.await?; .await?;
// Send the initial input to kick off the one-shot turn. // Send the initial input to kick off the one-shot turn.
+26 -22
View File
@@ -163,14 +163,14 @@ async fn run_guardian_review(
let schema = guardian_output_schema(); let schema = guardian_output_schema();
let terminal_action = action_summary.clone(); let terminal_action = action_summary.clone();
let outcome = run_guardian_review_session( let outcome = Box::pin(run_guardian_review_session(
session.clone(), session.clone(),
turn.clone(), turn.clone(),
request, request,
retry_reason, retry_reason,
schema, schema,
external_cancel, external_cancel,
) ))
.await; .await;
let assessment = match outcome { let assessment = match outcome {
@@ -303,14 +303,16 @@ pub(crate) async fn review_approval_request(
request: GuardianApprovalRequest, request: GuardianApprovalRequest,
retry_reason: Option<String>, retry_reason: Option<String>,
) -> ReviewDecision { ) -> 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(session),
Arc::clone(turn), Arc::clone(turn),
review_id, review_id,
request, request,
retry_reason, retry_reason,
/*external_cancel*/ None, /*external_cancel*/ None,
) ))
.await .await
} }
@@ -322,14 +324,14 @@ pub(crate) async fn review_approval_request_with_cancel(
retry_reason: Option<String>, retry_reason: Option<String>,
cancel_token: CancellationToken, cancel_token: CancellationToken,
) -> ReviewDecision { ) -> ReviewDecision {
run_guardian_review( Box::pin(run_guardian_review(
Arc::clone(session), Arc::clone(session),
Arc::clone(turn), Arc::clone(turn),
review_id, review_id,
request, request,
retry_reason, retry_reason,
Some(cancel_token), Some(cancel_token),
) ))
.await .await
} }
@@ -411,22 +413,24 @@ pub(super) async fn run_guardian_review_session(
Err(err) => return GuardianReviewOutcome::Completed(Err(err)), Err(err) => return GuardianReviewOutcome::Completed(Err(err)),
}; };
match session match Box::pin(
.guardian_review_session session
.run_review(GuardianReviewSessionParams { .guardian_review_session
parent_session: Arc::clone(&session), .run_review(GuardianReviewSessionParams {
parent_turn: turn.clone(), parent_session: Arc::clone(&session),
spawn_config: guardian_config, parent_turn: turn.clone(),
request, spawn_config: guardian_config,
retry_reason, request,
schema, retry_reason,
model: guardian_model, schema,
reasoning_effort: guardian_reasoning_effort, model: guardian_model,
reasoning_summary: turn.reasoning_summary, reasoning_effort: guardian_reasoning_effort,
personality: turn.personality, reasoning_summary: turn.reasoning_summary,
external_cancel, personality: turn.personality,
}) external_cancel,
.await }),
)
.await
{ {
GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => { GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => {
GuardianReviewOutcome::Completed(parse_guardian_assessment( GuardianReviewOutcome::Completed(parse_guardian_assessment(
+23 -20
View File
@@ -327,32 +327,30 @@ impl GuardianReviewSessionManager {
}; };
if trunk.reuse_key != next_reuse_key { if trunk.reuse_key != next_reuse_key {
return self return Box::pin(self.run_ephemeral_review(
.run_ephemeral_review( params,
params, next_reuse_key,
next_reuse_key, deadline,
deadline, /*fork_snapshot*/ None,
/*fork_snapshot*/ None, ))
) .await;
.await;
} }
let trunk_guard = match trunk.review_lock.try_lock() { let trunk_guard = match trunk.review_lock.try_lock() {
Ok(trunk_guard) => trunk_guard, Ok(trunk_guard) => trunk_guard,
Err(_) => { Err(_) => {
return self return Box::pin(self.run_ephemeral_review(
.run_ephemeral_review( params,
params, next_reuse_key,
next_reuse_key, deadline,
deadline, trunk.fork_snapshot().await,
trunk.fork_snapshot().await, ))
) .await;
.await;
} }
}; };
let (outcome, keep_review_session) = let (outcome, keep_review_session) =
run_review_on_session(trunk.as_ref(), &params, deadline).await; Box::pin(run_review_on_session(trunk.as_ref(), &params, deadline)).await;
if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) { if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) {
trunk.refresh_last_committed_fork_snapshot().await; trunk.refresh_last_committed_fork_snapshot().await;
} }
@@ -488,7 +486,12 @@ impl GuardianReviewSessionManager {
let mut cleanup = let mut cleanup =
EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session)); EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session));
let (outcome, _) = run_review_on_session(review_session.as_ref(), &params, deadline).await; let (outcome, _) = Box::pin(run_review_on_session(
review_session.as_ref(),
&params,
deadline,
))
.await;
if let Some(review_session) = self.take_active_ephemeral(&review_session).await { if let Some(review_session) = self.take_active_ephemeral(&review_session).await {
cleanup.disarm(); cleanup.disarm();
review_session.shutdown_in_background(); review_session.shutdown_in_background();
@@ -512,7 +515,7 @@ async fn spawn_guardian_review_session(
), ),
None => (None, 0, None), None => (None, 0, None),
}; };
let codex = run_codex_thread_interactive( let codex = Box::pin(run_codex_thread_interactive(
spawn_config, spawn_config,
params.parent_session.services.auth_manager.clone(), params.parent_session.services.auth_manager.clone(),
params.parent_session.services.models_manager.clone(), params.parent_session.services.models_manager.clone(),
@@ -521,7 +524,7 @@ async fn spawn_guardian_review_session(
cancel_token.clone(), cancel_token.clone(),
SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()), SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()),
initial_history, initial_history,
) ))
.await?; .await?;
Ok(GuardianReviewSession { Ok(GuardianReviewSession {
+224 -204
View File
@@ -1332,222 +1332,242 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() ->
Ok(()) Ok(())
} }
#[tokio::test(flavor = "current_thread")] #[test]
async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> { fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> anyhow::Result<()> {
let first_assessment = serde_json::json!({ const TEST_STACK_SIZE_BYTES: usize = 2 * 1024 * 1024;
"risk_level": "low",
"user_authorization": "high", let handle =
"outcome": "allow", std::thread::Builder::new()
"rationale": "first guardian rationale", .name("guardian_parallel_reviews_fork_from_last_committed_trunk_history".to_string())
}) .stack_size(TEST_STACK_SIZE_BYTES)
.to_string(); .spawn(|| -> anyhow::Result<()> {
let second_assessment = serde_json::json!({ let runtime = tokio::runtime::Builder::new_current_thread()
"risk_level": "low", .enable_all()
"user_authorization": "high", .build()?;
"outcome": "allow", runtime.block_on(Box::pin(async {
"rationale": "second guardian rationale", let first_assessment = serde_json::json!({
}) "risk_level": "low",
.to_string(); "user_authorization": "high",
let third_assessment = serde_json::json!({ "outcome": "allow",
"risk_level": "low", "rationale": "first guardian rationale",
"user_authorization": "high", })
"outcome": "allow", .to_string();
"rationale": "third guardian rationale", let second_assessment = serde_json::json!({
}) "risk_level": "low",
.to_string(); "user_authorization": "high",
let (gate_tx, gate_rx) = tokio::sync::oneshot::channel(); "outcome": "allow",
let (server, _) = start_streaming_sse_server(vec![ "rationale": "second guardian rationale",
vec![StreamingSseChunk { })
gate: None, .to_string();
body: sse(vec![ let third_assessment = serde_json::json!({
ev_response_created("resp-guardian-1"), "risk_level": "low",
ev_assistant_message("msg-guardian-1", &first_assessment), "user_authorization": "high",
ev_completed("resp-guardian-1"), "outcome": "allow",
]), "rationale": "third guardian rationale",
}], })
vec![ .to_string();
StreamingSseChunk { let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
let (server, _) = start_streaming_sse_server(vec![
vec![StreamingSseChunk {
gate: None, gate: None,
body: sse(vec![ev_response_created("resp-guardian-2")]),
},
StreamingSseChunk {
gate: Some(gate_rx),
body: sse(vec![ body: sse(vec![
ev_assistant_message("msg-guardian-2", &second_assessment), ev_response_created("resp-guardian-1"),
ev_completed("resp-guardian-2"), ev_assistant_message("msg-guardian-1", &first_assessment),
ev_completed("resp-guardian-1"),
]), ]),
}, }],
], vec![
vec![StreamingSseChunk { StreamingSseChunk {
gate: None, gate: None,
body: sse(vec![ body: sse(vec![ev_response_created("resp-guardian-2")]),
ev_response_created("resp-guardian-3"), },
ev_assistant_message("msg-guardian-3", &third_assessment), StreamingSseChunk {
ev_completed("resp-guardian-3"), gate: Some(gate_rx),
]), body: sse(vec![
}], ev_assistant_message("msg-guardian-2", &second_assessment),
]) ev_completed("resp-guardian-2"),
.await; ]),
},
],
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; let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
seed_guardian_parent_history(&session, &turn).await; seed_guardian_parent_history(&session, &turn).await;
let initial_request = GuardianApprovalRequest::Shell { let initial_request = GuardianApprovalRequest::Shell {
id: "shell-guardian-1".to_string(), id: "shell-guardian-1".to_string(),
command: vec!["git".to_string(), "status".to_string()], command: vec!["git".to_string(), "status".to_string()],
cwd: test_path_buf("/repo/codex-rs/core").abs(), cwd: test_path_buf("/repo/codex-rs/core").abs(),
sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
additional_permissions: None, additional_permissions: None,
justification: Some("Inspect repo state before proceeding.".to_string()), justification: Some("Inspect repo state before proceeding.".to_string()),
}; };
assert_eq!( assert_eq!(
review_approval_request( 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, &session,
&turn, &turn,
"review-shell-guardian-1".to_string(), "review-shell-guardian-3".to_string(),
initial_request, third_request,
/*retry_reason*/ None Some("parallel follow-up".to_string()),
)
.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; .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::<serde_json::Value>(&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 { gate_tx
id: "shell-guardian-2".to_string(), .send(())
command: vec!["git".to_string(), "diff".to_string()], .expect("second guardian review gate should still be open");
cwd: test_path_buf("/repo/codex-rs/core").abs(), assert_eq!(second_review.await?, ReviewDecision::Approved);
sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault, server.shutdown().await;
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); Ok(())
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 { match handle.join() {
loop { Ok(result) => result,
if server.requests().await.len() >= 2 { Err(_) => Err(anyhow::anyhow!(
break; "guardian_parallel_reviews_fork_from_last_committed_trunk_history thread panicked"
} )),
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::<serde_json::Value>(&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(())
} }
#[test] #[test]
fn guardian_review_session_config_preserves_parent_network_proxy() { fn guardian_review_session_config_preserves_parent_network_proxy() {