Skip to content

Flake batch 6: health-check retry, watch and scope waits, scope-revocation task lookup #1053

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ualtinok merged 7 commits into from
Oct 27, 2026
Merged

Flake batch 6: health-check retry, watch and scope waits, scope-revocation task lookup #1053

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f90ca92
mason: retry health.check until Ok in bridge health test and print de…
ualtinok Oct 27, 2026
117ac30
Name flake batch 6's descriptor after its branch
ualtinok Oct 27, 2026
9417845
Flake batch 6 descriptor matches the batch format
ualtinok Oct 27, 2026
51c0d38
Watch and scope-workdir tests wait past their own budgets on slow run…
ualtinok Oct 27, 2026
8c60beb
mason: scope revocation tests wait for their own observed bash task
ualtinok Oct 27, 2026
3214bd2
Scope revocation tests allow 30 s for the spawn dispatch to report it…
ualtinok Oct 27, 2026
b1ff5b4
Worker-run head-task test looks up its task by a marker in its own root
ualtinok Oct 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions benchmarks/aft-search/slice-descriptors/land-flake-batch-6.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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."
}
5 changes: 4 additions & 1 deletion crates/aft/src/commands/bash_watch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
2 changes: 1 addition & 1 deletion crates/aft/src/subc/scope_workdir_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
194 changes: 104 additions & 90 deletions crates/aft/tests/integration/scope_revocation_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,58 +10,103 @@ async fn open_scope_daemon_session(input: FakeDaemonInput) -> FakeDaemonSession
.0
}

static OBSERVED_TASK: OnceLock<Mutex<Option<(aft::bash_background::BgTaskRegistry, String)>>> =
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<Vec<ObservedTask>> = 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,
"unconfirmed scope close must not start a kill"
);
}

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();
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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"))]
Expand All @@ -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,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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 => {
Expand All @@ -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,
Expand Down Expand Up @@ -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"))]
Expand All @@ -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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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;
},
|_, _, _| {},
Expand Down Expand Up @@ -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;
Expand All @@ -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());
Expand Down
Loading
Loading