diff --git a/benchmarks/aft-search/slice-descriptors/land-flake-batch-6.json b/benchmarks/aft-search/slice-descriptors/land-flake-batch-6.json new file mode 100644 index 0000000000..e96a30f421 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-flake-batch-6.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "flake-batch-6" + ], + "note": "Test-only: timing and race fixes for merge-queue flakes (generous positive waits, waits on real signals). No product search behavior changed and no changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/crates/aft/src/commands/bash_watch.rs b/crates/aft/src/commands/bash_watch.rs index 14a5848b9a..0eaf77ac89 100644 --- a/crates/aft/src/commands/bash_watch.rs +++ b/crates/aft/src/commands/bash_watch.rs @@ -1489,8 +1489,11 @@ mod tests { let waited = job.wait(&crate::executor::JobCancellation::new()); let _ = tx.send((job, waited, started.elapsed())); }); + // The hard deadline must outlast the longest reply budget a test gives + // its watch (20 s), or a slow runner trips this guard while the watch + // is still inside its own budget. let (job, waited, elapsed) = rx - .recv_timeout(Duration::from_secs(10)) + .recv_timeout(Duration::from_secs(60)) .expect("watch did not answer within the test's hard deadline"); (job, waited.expect("watch was not cancelled"), elapsed) } diff --git a/crates/aft/src/subc/scope_workdir_tests.rs b/crates/aft/src/subc/scope_workdir_tests.rs index 13a80b217b..f4a710dea3 100644 --- a/crates/aft/src/subc/scope_workdir_tests.rs +++ b/crates/aft/src/subc/scope_workdir_tests.rs @@ -178,7 +178,7 @@ fn bind_observed( ) .await .unwrap(); - let completion = tokio::time::timeout(Duration::from_secs(5), completion_rx.recv()) + let completion = tokio::time::timeout(Duration::from_secs(30), completion_rx.recv()) .await .unwrap() .unwrap(); diff --git a/crates/aft/tests/integration/scope_revocation_test.rs b/crates/aft/tests/integration/scope_revocation_test.rs index 7b89968dbd..5183b44525 100644 --- a/crates/aft/tests/integration/scope_revocation_test.rs +++ b/crates/aft/tests/integration/scope_revocation_test.rs @@ -10,34 +10,79 @@ async fn open_scope_daemon_session(input: FakeDaemonInput) -> FakeDaemonSession .0 } -static OBSERVED_TASK: OnceLock>> = - OnceLock::new(); +/// A bash task that a spawn dispatch reported, with the command that started it. +#[derive(Clone)] +struct ObservedTask { + command: String, + registry: aft::bash_background::BgTaskRegistry, + task: String, +} + +/// Every bash task `observing_dispatch` has seen in this test process. +/// +/// The tests in one process share this list, and an earlier test's entry is +/// still here when a later test runs, so a reader looks up the entry for the +/// command it sent. Each test's command names a file inside its own temporary +/// root, which keeps the entries apart. +static OBSERVED_TASKS: Mutex> = Mutex::new(Vec::new()); fn observing_dispatch(req: RawRequest, ctx: &AppContext) -> Response { - let bash = req.command == "bash"; + let command = (req.command == "bash").then(|| { + req.params + .get("command") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() + }); let response = bridge_dispatch(req, ctx); - if bash { + if let Some(command) = command { if let Some(task) = response.data.get("task_id").and_then(Value::as_str) { - *OBSERVED_TASK - .get_or_init(|| Mutex::new(None)) + OBSERVED_TASKS .lock() - .unwrap() = Some((ctx.bash_background().clone(), task.into())); + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(ObservedTask { + command, + registry: ctx.bash_background().clone(), + task: task.into(), + }); } } response } -fn assert_run_not_killing() { - let (registry, task) = OBSERVED_TASK - .get() - .unwrap() - .lock() - .unwrap() - .clone() - .unwrap(); +/// The registry and id of the task whose bash command contains `marker`. +/// +/// Waits for it, because the child can publish its pid file before the spawn +/// dispatch returns and reports the task: the dispatch still has its +/// foreground wait window and hand-off to finish, and a loaded machine can +/// stretch that past the test's next steps. +async fn observed_task(marker: &Path) -> (aft::bash_background::BgTaskRegistry, String) { + let marker = marker.display().to_string(); + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let found = OBSERVED_TASKS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .rev() + .find(|observed| observed.command.contains(&marker)) + .cloned(); + if let Some(observed) = found { + return (observed.registry, observed.task); + } + assert!( + Instant::now() < deadline, + "bash dispatch never reported a task for a command naming {marker}" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +async fn assert_run_not_killing(marker: &Path) { + let (registry, task) = observed_task(marker).await; let snapshot = registry .status(&task, "run-session", None, None, 0) - .unwrap(); + .expect("the run's task is readable in its session"); assert_eq!( snapshot.info.status, BgTaskStatus::Running, @@ -45,23 +90,23 @@ fn assert_run_not_killing() { ); } -async fn assert_scope_kill_settled() { - assert_scope_kill_settled_as("run-session", "scope_revoked", "killed (scope revoked)").await; +async fn assert_scope_kill_settled(marker: &Path) { + assert_scope_kill_settled_as( + marker, + "run-session", + "scope_revoked", + "killed (scope revoked)", + ) + .await; } /// A codemode program's task killed at its run's end says so. -async fn assert_run_kill_settled(session: &str) { - assert_scope_kill_settled_as(session, "run_ended", "killed (run ended)").await; +async fn assert_run_kill_settled(marker: &Path, session: &str) { + assert_scope_kill_settled_as(marker, session, "run_ended", "killed (run ended)").await; } -async fn assert_scope_kill_settled_as(session: &str, reason: &str, rendered: &str) { - let (registry, task) = OBSERVED_TASK - .get() - .unwrap() - .lock() - .unwrap() - .clone() - .unwrap(); +async fn assert_scope_kill_settled_as(marker: &Path, session: &str, reason: &str, rendered: &str) { + let (registry, task) = observed_task(marker).await; let deadline = Instant::now() + Duration::from_secs(5); loop { let snapshot = registry.status(&task, session, None, None, 0).unwrap(); @@ -550,27 +595,6 @@ async fn wait_test_marker(path: &Path) { } } -#[cfg(unix)] -async fn await_registered_run_task() { - let until = Instant::now() + Duration::from_secs(5); - loop { - let registered = OBSERVED_TASK - .get_or_init(|| Mutex::new(None)) - .lock() - .unwrap() - .is_some(); - if registered { - assert_run_not_killing(); - return; - } - assert!( - Instant::now() < until, - "bash dispatch never acknowledged task registration" - ); - tokio::time::sleep(Duration::from_millis(10)).await; - } -} - #[cfg(unix)] async fn no_extra_frames_until_pong(stream: &mut tokio::net::TcpStream, corr: u64, phase: &str) { send_frame( @@ -638,7 +662,12 @@ enum CloseCase { } #[cfg(unix)] -async fn assert_hold_or_retry(stream: &mut tokio::net::TcpStream, pid: u32, reason: &str) { +async fn assert_hold_or_retry( + stream: &mut tokio::net::TcpStream, + pid: u32, + marker: &Path, + reason: &str, +) { #[cfg(feature = "test-timing-hooks")] let retry = describe(stream).await; #[cfg(not(feature = "test-timing-hooks"))] @@ -649,7 +678,7 @@ async fn assert_hold_or_retry(stream: &mut tokio::net::TcpStream, pid: u32, reas "production retries must wait for the regular poll" ); assert!(alive(pid), "{reason} must hold work"); - assert_run_not_killing(); + assert_run_not_killing(marker).await; #[cfg(feature = "test-timing-hooks")] answer( stream, @@ -724,12 +753,6 @@ fn close_case(name: &'static str, case: CloseCase) { } else { None }; - if matches!(case, CloseCase::Duplicate) { - *OBSERVED_TASK - .get_or_init(|| Mutex::new(None)) - .lock() - .unwrap() = None; - } let pid_path = daemon.root1.join("grandchild.pid"); let command = format!( "(trap '' TERM; sleep 30) & echo $! > '{}'; wait", @@ -782,7 +805,7 @@ fn close_case(name: &'static str, case: CloseCase) { // The command can start before its task joins scope monitoring. // Wait for spawn dispatch to finish so closing the sibling route // finds the task and asks whether its scope is still live. - await_registered_run_task().await; + assert_run_not_killing(&pid_path).await; } // Close the unused sibling first: it must already share the run's // lifetime state even though it has never made a tool call. @@ -838,7 +861,8 @@ fn close_case(name: &'static str, case: CloseCase) { None, ) .await; - assert_hold_or_retry(&mut daemon.stream, pid, "not-synced").await; + assert_hold_or_retry(&mut daemon.stream, pid, &pid_path, "not-synced") + .await; } #[cfg(feature = "test-timing-hooks")] CloseCase::SlowResync => { @@ -864,7 +888,7 @@ fn close_case(name: &'static str, case: CloseCase) { let body: Value = serde_json::from_slice(&request.body).unwrap(); assert_eq!(body["op"], "scope.describe"); assert!(alive(pid), "inconclusive reply {attempt} must hold work"); - assert_run_not_killing(); + assert_run_not_killing(&pid_path).await; } answer( &mut daemon.stream, @@ -893,7 +917,8 @@ fn close_case(name: &'static str, case: CloseCase) { .unwrap(), ) .await; - assert_hold_or_retry(&mut daemon.stream, pid, "describe error").await; + assert_hold_or_retry(&mut daemon.stream, pid, &pid_path, "describe error") + .await; } CloseCase::Timeout => { #[cfg(not(feature = "test-timing-hooks"))] @@ -904,7 +929,13 @@ fn close_case(name: &'static str, case: CloseCase) { ) .await .is_none()); - assert_hold_or_retry(&mut daemon.stream, pid, "describe timeout").await; + assert_hold_or_retry( + &mut daemon.stream, + pid, + &pid_path, + "describe timeout", + ) + .await; } CloseCase::NewEpoch => { answer( @@ -946,7 +977,7 @@ fn close_case(name: &'static str, case: CloseCase) { .await; } assert!(alive(pid), "a live epoch move must not kill"); - assert_run_not_killing(); + assert_run_not_killing(&pid_path).await; } CloseCase::NotLive => { answer( @@ -978,9 +1009,9 @@ fn close_case(name: &'static str, case: CloseCase) { if should_kill { wait_dead(pid).await; if run { - assert_run_kill_settled("run-session").await; + assert_run_kill_settled(&pid_path, "run-session").await; } else { - assert_scope_kill_settled().await; + assert_scope_kill_settled(&pid_path).await; } } if matches!(case, CloseCase::Duplicate) { @@ -1535,10 +1566,6 @@ fn codemode_background_task_killed_when_run_scope_ends_and_only_then() { Duration::from_secs(30), |input| async move { use subc_protocol::scope::ScopeStatus; - *OBSERVED_TASK - .get_or_init(|| Mutex::new(None)) - .lock() - .unwrap() = None; let mut daemon = open_scope_daemon_session(input).await; bind_head(&mut daemon.stream, &daemon.root1, 1, "head-session").await; // Two routes of one run: closing one is a transient close. @@ -1574,13 +1601,7 @@ fn codemode_background_task_killed_when_run_scope_ends_and_only_then() { ); let pid = child_pid(&pid_path).await; let _cleanup = ChildCleanup(pid); - let (registry, task) = OBSERVED_TASK - .get() - .unwrap() - .lock() - .unwrap() - .clone() - .unwrap(); + let (registry, task) = observed_task(&pid_path).await; // The task belongs to the head's session, so the agent reads it. assert!(registry .status(&task, "head-session", None, None, 0) @@ -1639,7 +1660,7 @@ fn codemode_background_task_killed_when_run_scope_ends_and_only_then() { ) .await; wait_dead(pid).await; - assert_run_kill_settled("head-session").await; + assert_run_kill_settled(&pid_path, "head-session").await; send_connection_goodbye(&mut daemon.stream).await; }, |_, _, _| {}, @@ -2404,20 +2425,19 @@ fn worker_codemode_run_cannot_see_or_kill_the_heads_bash_tasks() { "worker-run-head-tasks", Duration::from_secs(30), |input| async move { - *OBSERVED_TASK - .get_or_init(|| Mutex::new(None)) - .lock() - .unwrap() = None; let mut daemon = open_scope_daemon_session(input).await; bind_head(&mut daemon.stream, &daemon.root1, 1, "head-session").await; bind_worker(&mut daemon.stream, &daemon.root2, 2).await; bind_run(&mut daemon.stream, &daemon.root1, 4, worker_run_stamp()).await; - // The head starts a background task in its own session. + // The head starts a background task in its own session. The comment + // names a path in this test's own root, so the observed-task lookup + // finds this task and not another test's. + let marker = daemon.root1.join("head-background-task"); call( &mut daemon.stream, 1, 100, - json!({"command":"sleep 30", "background":true}), + json!({"command": format!("sleep 30 # {}", marker.display()), "background":true}), ) .await; let launch = read_frame_timeout(&mut daemon.stream, "head background launch").await; @@ -2427,13 +2447,7 @@ fn worker_codemode_run_cannot_see_or_kill_the_heads_bash_tasks() { "{}", String::from_utf8_lossy(&launch.body) ); - let (registry, task) = OBSERVED_TASK - .get() - .unwrap() - .lock() - .unwrap() - .clone() - .unwrap(); + let (registry, task) = observed_task(&marker).await; assert!(registry .status(&task, "head-session", None, None, 0) .is_some()); diff --git a/crates/aft/tests/integration/subc_bridge_test.rs b/crates/aft/tests/integration/subc_bridge_test.rs index d2f186776b..9a2a27f044 100644 --- a/crates/aft/tests/integration/subc_bridge_test.rs +++ b/crates/aft/tests/integration/subc_bridge_test.rs @@ -3332,7 +3332,9 @@ fn subc_bridge_management_surface_is_passive_closed_and_first_party() { fn subc_bridge_health_check_returns_root_status_report() { run_subc_bridge_test_inner( "subc_bridge_health_check_returns_root_status_report", - Duration::from_secs(30), + // Exceeds the driver's 30 s wait for an Ok health report, so a report + // that never turns Ok fails with its detail instead of the watchdog. + Duration::from_secs(60), Vec::new, drive_health_check_daemon, |_, executor, roots| { @@ -13424,20 +13426,38 @@ async fn drive_health_check_daemon(input: FakeDaemonInput) { // Bind acknowledgement follows configure commit, not health publication. // Wait for the actor to release its configure work, then acknowledge an // actual post-bind cache refresh instead of polling a throttled snapshot. + // + // Even after that barrier, a loaded runner can report Degraded for a + // moment: the rollup marks a root busy when one of its state locks is held + // while the census reads it, and the health reply itself marks the report + // Degraded when the executor scheduler lock is held at that instant. Both + // clear on their own, and the test worker only republishes the rollup on + // request, so each attempt requests a fresh rollup before asking again. + // Any other Degraded cause persists across attempts and still fails here, + // with the last report's detail in the message. let root_id = ProjectRootId::from_path(&root1).expect("bound root identity"); - tokio::time::timeout(Duration::from_secs(5), async { - while !executor.actor_is_idle(&root_id) { - tokio::time::sleep(Duration::from_millis(5)).await; + let deadline = Instant::now() + Duration::from_secs(30); + let mut corr = 20; + let report = loop { + tokio::time::timeout(Duration::from_secs(5), async { + while !executor.actor_is_idle(&root_id) { + tokio::time::sleep(Duration::from_millis(5)).await; + } + health_probe + .request_refresh() + .await + .expect("rollup published"); + }) + .await + .expect("post-bind health publication barrier"); + send_control_request(&mut stream, corr, ModuleControlRequest::HealthCheck {}).await; + let report = expect_health_check_report(&mut stream, corr).await; + if report.status == subc_protocol::session::HealthStatus::Ok || Instant::now() >= deadline { + break report; } - health_probe - .request_refresh() - .await - .expect("rollup published"); - }) - .await - .expect("post-bind health publication barrier"); - send_control_request(&mut stream, 20, ModuleControlRequest::HealthCheck {}).await; - let report = expect_health_check_report(&mut stream, 20).await; + corr += 1; + tokio::time::sleep(Duration::from_millis(50)).await; + }; let metrics = report.metrics.clone().expect("health check metrics"); let work = metrics.get("views_work").expect("production work counters"); assert_eq!(work["pid"], std::process::id()); @@ -13462,7 +13482,13 @@ async fn drive_health_check_daemon(input: FakeDaemonInput) { "missing production counter {field}: {work}" ); } - assert_eq!(report.status, subc_protocol::session::HealthStatus::Ok); + assert_eq!( + report.status, + subc_protocol::session::HealthStatus::Ok, + "health.check never reported Ok within 30 s; last detail: {:?}, stale_reason: {}", + report.detail, + metrics.get("stale_reason").unwrap_or(&Value::Null), + ); for (section, fields) in [ ( "lsp",