Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-08 22:13:59 +01:00
Code Issues Packages Projects Releases Wiki Activity

test(runner): bound Codex provider event waits by a hang ceiling, not 5s (#19306)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The Rust runner drives provider processes such as Codex, and CI
tests it on Linux, macOS and Windows
> - The "Runner Rust (Windows x64)" job fails at random on pull requests
that do not touch the runner
> - The failures come from Codex provider integration tests that wait
for fake-provider events under a fixed five-second deadline
> - On a loaded Windows runner, process spawn, antivirus scans and
fsyncs of durable state make these exchanges take longer than five
seconds, so the wait stops before the event arrives
> - This pull request keeps each wait on its condition and changes the
deadline to a shared hang ceiling
> - The benefit is that unrelated pull requests do not fail on this job,
and a real hang still fails the test

## Linked Issues or Issue Description

**What happened?**
The "Runner Rust (Windows x64)" job failed on pull requests that do not
change the runner. Examples:
- Run 42073331106 (PR #19273) and run 42058757139:
`rejected_declared_completion_stays_failed_when_the_provider_completes`
panicked with `called Option::unwrap() on a None value`. The loop ended
at its 5 s deadline before `session.failed` and `run.terminal` arrived.
- Run 41892735530 (master): the same test failed in the same way.
- Runs 41718881566 and 41670873220:
`durable_descendant_lineage_survives_capacity_and_provider_restoration`
failed with "capacity exhaustion must become a specific durable recovery
reason". Its third phase waited only 5 s. An earlier phase of this test
already had a 30 s deadline with a comment about this contention.

**What did you expect?**
The tests wait until the fake provider produces the event. They fail
only when the provider hangs.

**Steps to reproduce**
Run `cargo test --release --workspace` for the runner on a contended
Windows x64 runner. The suite takes about 380 s there, and some tests
run for more than 60 s. The failures are intermittent.

## What Changed

- `tests/codex_provider.rs`: add `EVENT_WAIT_TIMEOUT` (120 s) and
`event_deadline()`. Replace all 50 fixed 5 s and 30 s wait deadlines
with `event_deadline()`. Each wait already stops when its condition
holds, so passing tests do not get slower.
-
`rejected_declared_completion_stays_failed_when_the_provider_completes`:
assert that the run reached `run.terminal`, and print the observed event
types if it did not. Replace the bare `unwrap()` on `session.failed`
with a message that names the scenario.
- `src/bin/fake-codex-app-server.rs`: the two background gate threads
used 5 s and 10 s bounds. A test releases a gate only after its own
waits succeed, so these bounds had the same problem. They now use one
`TEST_GATE_TIMEOUT` of 120 s.
- No production code changes. The two remaining `from_secs(5)` values
are the unused `max_runtime` in the test config and a Unix-only latency
assertion. Both stay the same.

## Verification

- `cargo test --release -p paperclip-runner-core --test codex_provider
--locked`: 109 passed, 0 failed, 2 ignored.
- I ran the two flaky tests 48 times, with 6 copies in parallel per
round to load the machine. All 48 runs passed.
- `cargo fmt --all --check`: clean.
- `cargo clippy -p paperclip-runner-core --test codex_provider --bin
fake-codex-app-server`: no new warnings in the changed lines. The lib
target already has clippy findings from the local toolchain (Rust 1.97).
They are not in the changed files, and CI does not run clippy.
- I did not reproduce the original failure on macOS. The slow condition
occurs only on the loaded Windows runner. The Windows CI job on this
pull request is the main check.

## Risks

- Low risk. The change affects only tests and the test fixture binary.
- If a test hangs for real, it now fails after 120 s and not after 5 s.
This makes a broken run slower, but the failure message stays the same.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file edits, GitHub CLI).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed 2026-11-05 13:00:37 -06:00
1 parent b37188ce42
commit 03eb49053f
2 files changed
+69 -56
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
+4 -3
View File
@@ -8,6 +8,8 @@ use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::{json, Value}; use serde_json::{json, Value};
const TEST_GATE_TIMEOUT: Duration = Duration::from_secs(120);
#[derive(Clone, Debug, Default, Deserialize, Serialize)] #[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
struct FakeState { struct FakeState {
@@ -2026,7 +2028,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
.iter() .iter()
.any(|arg| arg == "--late-accounting-duplicate-burst"); .any(|arg| arg == "--late-accounting-duplicate-burst");
thread::spawn(move || { thread::spawn(move || {
let deadline = std::time::Instant::now() + Duration::from_secs(10); let deadline = std::time::Instant::now() + TEST_GATE_TIMEOUT;
while !gate.is_file() && std::time::Instant::now() < deadline { while !gate.is_file() && std::time::Instant::now() < deadline {
thread::sleep(Duration::from_millis(1)); thread::sleep(Duration::from_millis(1));
} }
@@ -2105,8 +2107,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
thread::spawn(move || { thread::spawn(move || {
let result = (|| -> io::Result<()> { let result = (|| -> io::Result<()> {
if let Some(gate) = gate.as_ref() { if let Some(gate) = gate.as_ref() {
let deadline = let deadline = std::time::Instant::now() + TEST_GATE_TIMEOUT;
std::time::Instant::now() + Duration::from_secs(5);
while !gate.is_file() { while !gate.is_file() {
if std::time::Instant::now() >= deadline { if std::time::Instant::now() >= deadline {
return Err(io::Error::new( return Err(io::Error::new(
packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
+65 -53
View File
@@ -27,6 +27,12 @@ mod human_input_validation;
static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1); static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1);
static RECEIPT_LIMIT_TEST_LOCK: Mutex<()> = Mutex::new(()); static RECEIPT_LIMIT_TEST_LOCK: Mutex<()> = Mutex::new(());
const EVENT_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
fn event_deadline() -> std::time::Instant {
std::time::Instant::now() + EVENT_WAIT_TIMEOUT
}
fn lock_receipt_limit_test() -> MutexGuard<'static, ()> { fn lock_receipt_limit_test() -> MutexGuard<'static, ()> {
RECEIPT_LIMIT_TEST_LOCK RECEIPT_LIMIT_TEST_LOCK
.lock() .lock()
@@ -754,7 +760,7 @@ fn retained_and_ack(
} }
fn wait_for_notification(provider: &mut CodexProvider, expected_method: &str) -> Value { fn wait_for_notification(provider: &mut CodexProvider, expected_method: &str) -> Value {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
match provider.poll().expect("poll provider notification") { match provider.poll().expect("poll provider notification") {
Some(CodexProviderEvent::Notification { method, params }) => { Some(CodexProviderEvent::Notification { method, params }) => {
@@ -770,7 +776,7 @@ fn wait_for_notification(provider: &mut CodexProvider, expected_method: &str) ->
} }
fn wait_for_provider_error(provider: &mut CodexProvider) -> String { fn wait_for_provider_error(provider: &mut CodexProvider) -> String {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
match provider.poll() { match provider.poll() {
Err(error) => return error.to_string(), Err(error) => return error.to_string(),
@@ -785,7 +791,7 @@ fn wait_for_executor_event(
executor: &mut impl CommandExecutor, executor: &mut impl CommandExecutor,
expected_event_type: &str, expected_event_type: &str,
) -> PolledEvent { ) -> PolledEvent {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(executor).expect("poll Codex executor event"); let events = poll_and_ack(executor).expect("poll Codex executor event");
if let Some(event) = events if let Some(event) = events
@@ -800,7 +806,7 @@ fn wait_for_executor_event(
} }
fn wait_for_provider_exit(provider: &mut CodexProvider) { fn wait_for_provider_exit(provider: &mut CodexProvider) {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
if matches!( if matches!(
provider.poll().expect("poll terminated provider"), provider.poll().expect("poll terminated provider"),
@@ -814,7 +820,7 @@ fn wait_for_provider_exit(provider: &mut CodexProvider) {
} }
fn wait_for_reused_identity_reap(provider: &mut CodexProvider) { fn wait_for_reused_identity_reap(provider: &mut CodexProvider) {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
match provider.poll() { match provider.poll() {
Err(error) => assert!( Err(error) => assert!(
@@ -840,7 +846,7 @@ fn wait_for_reused_identity_reap(provider: &mut CodexProvider) {
fn wait_for_fake_provider_idle(directory: &Path) { fn wait_for_fake_provider_idle(directory: &Path) {
let state_path = directory.join("fake-state.json"); let state_path = directory.join("fake-state.json");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let active_turn_id = fs::read(&state_path) let active_turn_id = fs::read(&state_path)
.ok() .ok()
@@ -960,7 +966,7 @@ fn wait_for_executor_event_batch(
executor: &mut CodexCommandExecutor, executor: &mut CodexCommandExecutor,
expected_event_type: &str, expected_event_type: &str,
) -> Vec<PolledEvent> { ) -> Vec<PolledEvent> {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let mut retained = Vec::new(); let mut retained = Vec::new();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
retained.extend(poll_and_ack(executor).expect("poll and retain the entire event batch")); retained.extend(poll_and_ack(executor).expect("poll and retain the entire event batch"));
@@ -995,7 +1001,7 @@ fn codex_transport_buffers_notifications_while_waiting_for_responses() {
.start_turn("Complete the fake task.", &config.cwd) .start_turn("Complete the fake task.", &config.cwd)
.expect("start provider turn"); .expect("start provider turn");
let mut event_types = Vec::new(); let mut event_types = Vec::new();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
if let Some(CodexProviderEvent::Notification { method, params }) = if let Some(CodexProviderEvent::Notification { method, params }) =
provider.poll().expect("poll provider event") provider.poll().expect("poll provider event")
@@ -1030,7 +1036,7 @@ fn codex_account_updates_do_not_interrupt_turns_or_publish_account_details() {
.expect("start provider turn"); .expect("start provider turn");
let mut completed = false; let mut completed = false;
let mut account_notices = 0; let mut account_notices = 0;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
match provider.poll().expect("poll provider event") { match provider.poll().expect("poll provider event") {
Some(CodexProviderEvent::ProtocolFailure { diagnostic }) => { Some(CodexProviderEvent::ProtocolFailure { diagnostic }) => {
@@ -1155,7 +1161,7 @@ fn helper_tool_requests_do_not_terminate_or_borrow_root_authority() {
provider provider
.start_turn("Hire a persistent teammate.", &config.cwd) .start_turn("Hire a persistent teammate.", &config.cwd)
.unwrap(); .unwrap();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let mut completed = false; let mut completed = false;
let mut root_tool_delivered = false; let mut root_tool_delivered = false;
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
@@ -1222,7 +1228,7 @@ fn codex_dynamic_tool_round_trips_through_the_provider_boundary() {
let mut delivered = false; let mut delivered = false;
let mut completed = false; let mut completed = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
match provider.poll().expect("poll semantic tool event") { match provider.poll().expect("poll semantic tool event") {
Some(CodexProviderEvent::ToolCall { Some(CodexProviderEvent::ToolCall {
@@ -1289,7 +1295,7 @@ fn codex_rejects_replay_of_a_completed_tool_call_id_in_the_same_turn() {
.start_turn("Inspect the fake task once.", &config.cwd) .start_turn("Inspect the fake task once.", &config.cwd)
.expect("start provider turn"); .expect("start provider turn");
let first_call_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let first_call_deadline = event_deadline();
let first_call = let first_call =
std::iter::from_fn(|| (std::time::Instant::now() < first_call_deadline).then_some(())) std::iter::from_fn(|| (std::time::Instant::now() < first_call_deadline).then_some(()))
.find_map(|_| match provider.poll().expect("poll first tool call") { .find_map(|_| match provider.poll().expect("poll first tool call") {
@@ -1313,7 +1319,7 @@ fn codex_rejects_replay_of_a_completed_tool_call_id_in_the_same_turn() {
}) })
.expect("deliver the first semantic result"); .expect("deliver the first semantic result");
let replay_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let replay_deadline = event_deadline();
let replay_error = let replay_error =
std::iter::from_fn(|| (std::time::Instant::now() < replay_deadline).then_some(())) std::iter::from_fn(|| (std::time::Instant::now() < replay_deadline).then_some(()))
.find_map(|_| { .find_map(|_| {
@@ -1423,7 +1429,7 @@ fn codex_completion_survives_failed_pending_request_cancellation() {
.start_turn("Complete and exit with a tool call pending.", &config.cwd) .start_turn("Complete and exit with a tool call pending.", &config.cwd)
.expect("start provider turn"); .expect("start provider turn");
let call_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let call_deadline = event_deadline();
let mut call = None; let mut call = None;
while std::time::Instant::now() < call_deadline { while std::time::Instant::now() < call_deadline {
if let Some(CodexProviderEvent::ToolCall { if let Some(CodexProviderEvent::ToolCall {
@@ -1440,7 +1446,7 @@ fn codex_completion_survives_failed_pending_request_cancellation() {
let call = call.expect("observe the pending semantic tool call"); let call = call.expect("observe the pending semantic tool call");
std::thread::sleep(std::time::Duration::from_millis(50)); std::thread::sleep(std::time::Duration::from_millis(50));
let completed_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let completed_deadline = event_deadline();
let mut completed = false; let mut completed = false;
while std::time::Instant::now() < completed_deadline { while std::time::Instant::now() < completed_deadline {
completed = matches!( completed = matches!(
@@ -1486,7 +1492,7 @@ fn clean_idle_provider_exit_preserves_completed_turn_success() {
let mut completion_seen = false; let mut completion_seen = false;
let mut post_completion_output_seen = false; let mut post_completion_output_seen = false;
let mut clean_exit = None; let mut clean_exit = None;
let exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let exit_deadline = event_deadline();
while std::time::Instant::now() < exit_deadline { while std::time::Instant::now() < exit_deadline {
match provider.poll().expect("poll completion and clean exit") { match provider.poll().expect("poll completion and clean exit") {
Some(CodexProviderEvent::Notification { method, .. }) => { Some(CodexProviderEvent::Notification { method, .. }) => {
@@ -1550,7 +1556,7 @@ fn clean_provider_exit_does_not_refail_a_completed_turn() {
.expect("start provider turn"); .expect("start provider turn");
let mut event_types = Vec::new(); let mut event_types = Vec::new();
let exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let exit_deadline = event_deadline();
while std::time::Instant::now() < exit_deadline { while std::time::Instant::now() < exit_deadline {
event_types.extend( event_types.extend(
poll_and_ack(&mut executor) poll_and_ack(&mut executor)
@@ -1679,7 +1685,7 @@ fn durable_backend_reaps_before_rotating_full_provider_identity_epochs() {
assert!(rolled["toolBridge"]["durableRunReceiptLimitReached"].is_null()); assert!(rolled["toolBridge"]["durableRunReceiptLimitReached"].is_null());
let mut event_types = Vec::new(); let mut event_types = Vec::new();
let semantic_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let semantic_deadline = event_deadline();
while std::time::Instant::now() < semantic_deadline { while std::time::Instant::now() < semantic_deadline {
event_types.extend( event_types.extend(
poll_and_ack(&mut recovered) poll_and_ack(&mut recovered)
@@ -1923,7 +1929,7 @@ fn post_completion_observation_does_not_hide_same_or_resumed_process_failure() {
.expect("start provider turn"); .expect("start provider turn");
let mut event_types = Vec::new(); let mut event_types = Vec::new();
let first_exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let first_exit_deadline = event_deadline();
while std::time::Instant::now() < first_exit_deadline { while std::time::Instant::now() < first_exit_deadline {
event_types.extend( event_types.extend(
poll_and_ack(&mut executor) poll_and_ack(&mut executor)
@@ -1966,7 +1972,7 @@ fn post_completion_observation_does_not_hide_same_or_resumed_process_failure() {
// run outcome, while the later idle provider failure remains visible. // run outcome, while the later idle provider failure remains visible.
let mut recovered = CodexCommandExecutor::new(&directory); let mut recovered = CodexCommandExecutor::new(&directory);
let mut recovered_events = Vec::new(); let mut recovered_events = Vec::new();
let recovered_exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let recovered_exit_deadline = event_deadline();
while std::time::Instant::now() < recovered_exit_deadline { while std::time::Instant::now() < recovered_exit_deadline {
recovered_events recovered_events
.extend(poll_and_ack(&mut recovered).expect("poll restored provider after idle crash")); .extend(poll_and_ack(&mut recovered).expect("poll restored provider after idle crash"));
@@ -2058,7 +2064,7 @@ fn rejected_replacement_turn_start_preserves_result_and_exit_authority() {
.start_turn("Reject replacement work.", &config.cwd) .start_turn("Reject replacement work.", &config.cwd)
.expect_err("the replacement turn/start returns a definite rejection"); .expect_err("the replacement turn/start returns a definite rejection");
let mut buffered_notification_seen = false; let mut buffered_notification_seen = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let rejected_start_exit = loop { let rejected_start_exit = loop {
if std::time::Instant::now() >= deadline { if std::time::Instant::now() >= deadline {
break None; break None;
@@ -2135,7 +2141,7 @@ fn rejected_replacement_turn_start_does_not_hide_contradictory_turn_evidence() {
"unexpected duplicate-start error: {duplicate_error}" "unexpected duplicate-start error: {duplicate_error}"
); );
let mut contradictory_turn_seen = false; let mut contradictory_turn_seen = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let rejected_start_exit = loop { let rejected_start_exit = loop {
if std::time::Instant::now() >= deadline { if std::time::Instant::now() >= deadline {
break None; break None;
@@ -2201,7 +2207,7 @@ fn ambiguous_or_dead_replacement_start_preserves_result_not_exit_authority() {
provider provider
.start_turn("Accept replacement work before failing.", &config.cwd) .start_turn("Accept replacement work before failing.", &config.cwd)
.expect_err("the accepted replacement turn has no valid response"); .expect_err("the accepted replacement turn has no valid response");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let ambiguous_start_exit = (0..) let ambiguous_start_exit = (0..)
.take_while(|_| std::time::Instant::now() < deadline) .take_while(|_| std::time::Instant::now() < deadline)
.find_map(|_| { .find_map(|_| {
@@ -2347,7 +2353,7 @@ fn ambiguous_replacement_turn_adopts_one_later_completion_identity() {
let mut replacement_started = false; let mut replacement_started = false;
let mut replacement_completed = false; let mut replacement_completed = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let replacement_exit = (0..) let replacement_exit = (0..)
.take_while(|_| std::time::Instant::now() < deadline) .take_while(|_| std::time::Instant::now() < deadline)
.find_map(|_| { .find_map(|_| {
@@ -2438,7 +2444,7 @@ fn ambiguous_replacement_turn_rejects_conflicting_later_identity() {
); );
assert_eq!(provider.active_provider_turn_id(), Some("provider-turn-2")); assert_eq!(provider.active_provider_turn_id(), Some("provider-turn-2"));
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let diagnostic = loop { let diagnostic = loop {
assert!( assert!(
std::time::Instant::now() < deadline, std::time::Instant::now() < deadline,
@@ -2479,7 +2485,7 @@ fn ambiguous_replacement_turn_rejects_an_older_settled_identity() {
"unexpected older-identity error: {error}" "unexpected older-identity error: {error}"
); );
assert_eq!(provider.active_provider_turn_id(), None); assert_eq!(provider.active_provider_turn_id(), None);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let completed_turn_authoritative = (0..) let completed_turn_authoritative = (0..)
.take_while(|_| std::time::Instant::now() < deadline) .take_while(|_| std::time::Instant::now() < deadline)
.find_map(|_| { .find_map(|_| {
@@ -2873,7 +2879,7 @@ fn ambiguous_replacement_completion_replaces_durable_turn_authority() {
.expect_err("accepted replacement response is transport-ambiguous"); .expect_err("accepted replacement response is transport-ambiguous");
let mut replacement_events = Vec::new(); let mut replacement_events = Vec::new();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
replacement_events.extend( replacement_events.extend(
poll_and_ack(&mut executor) poll_and_ack(&mut executor)
@@ -2929,7 +2935,7 @@ fn accepted_replacement_turn_revokes_prior_authority_before_idle_crash() {
provider provider
.start_turn("Start genuinely new provider work.", &config.cwd) .start_turn("Start genuinely new provider work.", &config.cwd)
.expect("start second provider turn"); .expect("start second provider turn");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let second_exit = loop { let second_exit = loop {
if std::time::Instant::now() >= deadline { if std::time::Instant::now() >= deadline {
break None; break None;
@@ -3645,7 +3651,7 @@ fn durable_backend_routes_a_semantic_tool_result_back_to_codex() {
let mut result_seen = false; let mut result_seen = false;
let mut terminal_seen = false; let mut terminal_seen = false;
let completion_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let completion_deadline = event_deadline();
while std::time::Instant::now() < completion_deadline { while std::time::Instant::now() < completion_deadline {
let events = poll_and_ack(&mut executor).expect("poll result and completion"); let events = poll_and_ack(&mut executor).expect("poll result and completion");
result_seen |= events result_seen |= events
@@ -3740,7 +3746,7 @@ fn durable_backend_replays_completed_results_without_mutating_the_event_queue()
let state_path = directory.join("codex-provider-state.json"); let state_path = directory.join("codex-provider-state.json");
let state_before_replays = fs::read(&state_path).expect("read state before exact replays"); let state_before_replays = fs::read(&state_path).expect("read state before exact replays");
let expected_response = r#"{"ok":true,"task":{"id":"task-1"}}"#; let expected_response = r#"{"ok":true,"task":{"id":"task-1"}}"#;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while recorded_tool_responses(&directory).len() < 5 && std::time::Instant::now() < deadline { while recorded_tool_responses(&directory).len() < 5 && std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut executor).expect("service completed tool replay"); let events = poll_and_ack(&mut executor).expect("service completed tool replay");
assert!( assert!(
@@ -4156,7 +4162,7 @@ fn durable_backend_rotates_tool_authority_for_fresh_run_attach() {
.expect("settle the first run's semantic tool call"); .expect("settle the first run's semantic tool call");
let mut turn_completed = false; let mut turn_completed = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut executor).expect("drain the first run"); let events = poll_and_ack(&mut executor).expect("drain the first run");
turn_completed |= events turn_completed |= events
@@ -4234,7 +4240,7 @@ fn durable_backend_rotates_tool_authority_for_fresh_run_attach() {
json!({"text": "Attempt to replay the old tool call."}), json!({"text": "Attempt to replay the old tool call."}),
)) ))
.expect("start the second provider turn"); .expect("start the second provider turn");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let rejection = loop { let rejection = loop {
assert!( assert!(
std::time::Instant::now() < deadline, std::time::Instant::now() < deadline,
@@ -4305,7 +4311,7 @@ fn durable_backend_drains_completed_turn_usage_and_passive_tail_during_warm_atta
.expect("start the first turn"); .expect("start the first turn");
let mut old_usage = Vec::new(); let mut old_usage = Vec::new();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
loop { loop {
assert!( assert!(
std::time::Instant::now() < deadline, std::time::Instant::now() < deadline,
@@ -4420,7 +4426,7 @@ fn durable_backend_rejects_new_work_in_a_completed_turn_tail() {
)) ))
.expect("start the first turn"); .expect("start the first turn");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
loop { loop {
assert!( assert!(
std::time::Instant::now() < deadline, std::time::Instant::now() < deadline,
@@ -4437,7 +4443,7 @@ fn durable_backend_rejects_new_work_in_a_completed_turn_tail() {
} }
fs::write(&notification_gate, b"release").expect("release the foreign-turn barrier"); fs::write(&notification_gate, b"release").expect("release the foreign-turn barrier");
let emitted_gate = notification_gate.with_extension("emitted"); let emitted_gate = notification_gate.with_extension("emitted");
let emitted_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let emitted_deadline = event_deadline();
while !emitted_gate.is_file() { while !emitted_gate.is_file() {
assert!( assert!(
std::time::Instant::now() < emitted_deadline, std::time::Instant::now() < emitted_deadline,
@@ -4510,7 +4516,7 @@ fn durable_backend_attaches_after_a_settled_resume_usage_snapshot() {
.expect("start the provider turn"); .expect("start the provider turn");
let mut saw_terminal = false; let mut saw_terminal = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut first).expect("drain the settled provider turn"); let events = poll_and_ack(&mut first).expect("drain the settled provider turn");
saw_terminal |= events saw_terminal |= events
@@ -4673,7 +4679,7 @@ fn durable_backend_attaches_after_a_settled_restore_notice() {
.expect("start the provider turn"); .expect("start the provider turn");
let mut saw_terminal = false; let mut saw_terminal = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut first).expect("drain the settled provider turn"); let events = poll_and_ack(&mut first).expect("drain the settled provider turn");
saw_terminal |= events saw_terminal |= events
@@ -5027,7 +5033,7 @@ fn durable_stop_does_not_wait_for_a_repeated_interrupt_acknowledgement() {
assert_eq!(interrupted.result["status"], "interrupt_requested"); assert_eq!(interrupted.result["status"], "interrupt_requested");
// Mirror the actual producer ordering: the first RPC was acknowledged and // Mirror the actual producer ordering: the first RPC was acknowledged and
// the provider has aborted, but the runner has not polled its terminal. // the provider has aborted, but the runner has not polled its terminal.
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
loop { loop {
let state: Value = let state: Value =
serde_json::from_slice(&fs::read(directory.join("fake-state.json")).unwrap()).unwrap(); serde_json::from_slice(&fs::read(directory.join("fake-state.json")).unwrap()).unwrap();
@@ -5140,7 +5146,7 @@ fn assert_durable_stop_prepares_resumed_provider(ended_before_resume: bool) {
// The real provider persists completion, but runnerd never polls its // The real provider persists completion, but runnerd never polls its
// terminal notification before losing the process. Do not manufacture // terminal notification before losing the process. Do not manufacture
// or clear the runner's durable active-turn identity in this fixture. // or clear the runner's durable active-turn identity in this fixture.
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
loop { loop {
let state: Value = serde_json::from_slice( let state: Value = serde_json::from_slice(
&fs::read(directory.join("fake-state.json")).expect("read provider-owned state"), &fs::read(directory.join("fake-state.json")).expect("read provider-owned state"),
@@ -6119,7 +6125,7 @@ fn receipt_limit_polls_an_authoritative_terminal_with_unacknowledged_events() {
saturate_provider_tool_receipts(&directory); saturate_provider_tool_receipts(&directory);
let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config); let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let pending = loop { let pending = loop {
let events = recovered let events = recovered
.poll_events() .poll_events()
@@ -6202,7 +6208,7 @@ fn pending_runtime_request_count_limit_rejects_the_overflowing_request() {
.interrupt_turn() .interrupt_turn()
.expect("request the runtime-request flood"); .expect("request the runtime-request flood");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let mut accepted = 0; let mut accepted = 0;
let mut rejected_at_capacity = false; let mut rejected_at_capacity = false;
while !rejected_at_capacity && std::time::Instant::now() < deadline { while !rejected_at_capacity && std::time::Instant::now() < deadline {
@@ -6274,7 +6280,7 @@ fn receipt_limit_polling_bounds_and_rejects_runtime_request_floods() {
saturate_provider_tool_receipts(&directory); saturate_provider_tool_receipts(&directory);
let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config); let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); let deadline = event_deadline();
let mut rejected_at_capacity = false; let mut rejected_at_capacity = false;
while !rejected_at_capacity && std::time::Instant::now() < deadline { while !rejected_at_capacity && std::time::Instant::now() < deadline {
let events = recovered let events = recovered
@@ -6574,7 +6580,7 @@ fn structured_question_round_trips_through_the_normalized_backend() {
let mut question_set = None; let mut question_set = None;
let mut request_id = None; let mut request_id = None;
let mut provider_started_events = 0; let mut provider_started_events = 0;
let question_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let question_deadline = event_deadline();
while std::time::Instant::now() < question_deadline { while std::time::Instant::now() < question_deadline {
for event in poll_and_ack(&mut executor).expect("poll question") { for event in poll_and_ack(&mut executor).expect("poll question") {
provider_started_events += usize::from(event.event_type == "turn.started"); provider_started_events += usize::from(event.event_type == "turn.started");
@@ -6807,7 +6813,7 @@ fn rejected_declared_completion_stays_failed_when_the_provider_completes() {
.unwrap(); .unwrap();
let mut emitted = Vec::new(); let mut emitted = Vec::new();
let mut input_calls = 0; let mut input_calls = 0;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut executor).unwrap(); let events = poll_and_ack(&mut executor).unwrap();
for event in &events { for event in &events {
@@ -6828,10 +6834,18 @@ fn rejected_declared_completion_stays_failed_when_the_provider_completes() {
} }
std::thread::sleep(std::time::Duration::from_millis(1)); std::thread::sleep(std::time::Duration::from_millis(1));
} }
let emitted_types = emitted
.iter()
.map(|event| event.event_type.as_str())
.collect::<Vec<_>>();
assert!(
emitted_types.contains(&"run.terminal"),
"{scenario}: run did not reach run.terminal; observed {emitted_types:?}"
);
let failure = emitted let failure = emitted
.iter() .iter()
.position(|event| event.event_type == "session.failed") .position(|event| event.event_type == "session.failed")
.unwrap(); .unwrap_or_else(|| panic!("{scenario}: no session.failed in {emitted_types:?}"));
let turn = emitted let turn = emitted
.iter() .iter()
.position(|event| event.event_type == "turn.completed") .position(|event| event.event_type == "turn.completed")
@@ -7065,9 +7079,7 @@ fn durable_descendant_lineage_survives_capacity_and_provider_restoration() {
json!({"text": "Read test context."}), json!({"text": "Read test context."}),
)) ))
.unwrap(); .unwrap();
// Persisting 300 descendant notifications can exceed five seconds while let deadline = event_deadline();
// the other provider tests contend for disk and CPU on a shared runner.
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
let mut completed = false; let mut completed = false;
let mut children = std::collections::BTreeSet::new(); let mut children = std::collections::BTreeSet::new();
while std::time::Instant::now() < deadline && !completed { while std::time::Instant::now() < deadline && !completed {
@@ -7095,7 +7107,7 @@ fn durable_descendant_lineage_survives_capacity_and_provider_restoration() {
drop(first); drop(first);
let mut restored = CodexCommandExecutor::with_runner_config(&directory, &runner_config); let mut restored = CodexCommandExecutor::with_runner_config(&directory, &runner_config);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let mut child_seen = false; let mut child_seen = false;
while std::time::Instant::now() < deadline && !child_seen { while std::time::Instant::now() < deadline && !child_seen {
for event in poll_and_ack(&mut restored).unwrap() { for event in poll_and_ack(&mut restored).unwrap() {
@@ -7137,7 +7149,7 @@ fn durable_descendant_lineage_survives_capacity_and_provider_restoration() {
json!({"text":"Continue bounded child work."}), json!({"text":"Continue bounded child work."}),
)) ))
.unwrap(); .unwrap();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let mut capacity_failed = false; let mut capacity_failed = false;
while std::time::Instant::now() < deadline && !capacity_failed { while std::time::Instant::now() < deadline && !capacity_failed {
for event in poll_and_ack(&mut bounded).unwrap() { for event in poll_and_ack(&mut bounded).unwrap() {
@@ -7245,7 +7257,7 @@ fn resume_usage_is_a_historical_diagnostic_not_a_warning_or_charge() {
.expect("start the provider turn"); .expect("start the provider turn");
let mut saw_terminal = false; let mut saw_terminal = false;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
while std::time::Instant::now() < deadline { while std::time::Instant::now() < deadline {
let events = poll_and_ack(&mut first).expect("drain the settled provider turn"); let events = poll_and_ack(&mut first).expect("drain the settled provider turn");
saw_terminal |= events saw_terminal |= events
@@ -7694,7 +7706,7 @@ fn late_codex_usage_reports_its_owner_without_blocking_the_session() {
json!({"text":"Start successor paid work."}), json!({"text":"Start successor paid work."}),
)) ))
.unwrap(); .unwrap();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
loop { loop {
assert!(std::time::Instant::now() < deadline); assert!(std::time::Instant::now() < deadline);
let events = poll_and_ack(&mut executor).unwrap(); let events = poll_and_ack(&mut executor).unwrap();
@@ -7711,7 +7723,7 @@ fn late_codex_usage_reports_its_owner_without_blocking_the_session() {
} }
} }
fs::write(&gate, "release").unwrap(); fs::write(&gate, "release").unwrap();
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); let deadline = event_deadline();
let retained = loop { let retained = loop {
assert!(std::time::Instant::now() < deadline); assert!(std::time::Instant::now() < deadline);
let events = if mode == "unsafe" { let events = if mode == "unsafe" {