land: main-flake-fixes

benchmarks/aft-search/slice-descriptors/land-main-flake-fixes.json

@@ -0,0 +1,13 @@
+{
+  "slice_class": "engine_unwired",
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "fixtures": [
+    "crates/aft/tests/integration/tool_call_parity_test.rs"
+  ],
+  "fenced_files": [
+    "crates/aft/src/commands/semantic_search/mod.rs",
+    "crates/aft/src/search_index.rs"
+  ],
+  "note": "Test-only changes inside the fenced files: the lane-paused search test installs a ready trigram index instead of waiting on a background build, and the long-line grep test runs its scan on a frozen test clock. No production search code changes; benchmark rows must remain byte-identical."
+}

crates/aft/src/commands/semantic_search/mod.rs

@@ -8824,13 +8824,22 @@ mod tests {
         use crate::synapse_embed::pause_test_support::*;
         crate::semantic_index::with_lane_pause_clock(START, |_clock| {
             let project = tempfile::tempdir().unwrap();
-            std::fs::write(
-                project.path().join("remix-panel.ts"),
-                "// where remix panel mounted option switching handled\nexport const panel = true;\n",
-            )
-            .unwrap();
+            let source_file = project.path().join("remix-panel.ts");
+            let source =
+                "// where remix panel mounted option switching handled\nexport const panel = true;\n";
+            std::fs::write(&source_file, source).unwrap();
             let ctx = test_context(project.path());
             ctx.set_canonical_cache_root(project.path().to_path_buf());
+            // The keyword lane must already be ready. Without a resident
+            // trigram index the search starts a background build and serves
+            // only if that build finishes inside its short wait budget; on a
+            // loaded machine it does not, and with the paused semantic lane
+            // also unavailable the search refuses with
+            // `search_lanes_unavailable` before any pause notice is written.
+            let mut index = SearchIndex::new();
+            index.index_file(&source_file, source.as_bytes());
+            index.ready = true;
+            *ctx.search_index().write().unwrap() = Some(index);
             // The cold build is waiting out the pause: no index is loaded, so
             // the search never embeds its query and takes the lane state from
             // the backend health snapshot.

crates/aft/src/search_index.rs

@@ -3087,7 +3087,7 @@ impl SearchIndexSnapshot {
         max_files: Option<usize>,
         scan_budget: Duration,
     ) -> (GrepResult, GrepQueryPhaseTimings) {
-        self.search_grep_profiled_with_filters_and_query_and_limits_with_clock(
+        self.search_grep_profiled_with_filters_and_query_and_limits_with_deadline(
             pattern,
             query,
             filters,
@@ -3096,11 +3096,11 @@ impl SearchIndexSnapshot {
             path_exclusion,
             max_files,
             scan_budget,
-            Instant::now,
+            || GrepScanDeadline::after_with_clock(scan_budget, Instant::now),
         )
     }
 
-    fn search_grep_profiled_with_filters_and_query_and_limits_with_clock(
+    fn search_grep_profiled_with_filters_and_query_and_limits_with_deadline(
         &self,
         pattern: &CompiledPattern,
         query: &RegexQuery,
@@ -3110,7 +3110,7 @@ impl SearchIndexSnapshot {
         path_exclusion: Option<GrepPathExclusion>,
         max_files: Option<usize>,
         scan_budget: Duration,
-        now: fn() -> Instant,
+        make_deadline: impl FnOnce() -> GrepScanDeadline,
     ) -> (GrepResult, GrepQueryPhaseTimings) {
         let _io_scope = crate::bounded_io::enter_if_absent(scan_budget);
         let matcher = match pattern {
@@ -3149,7 +3149,7 @@ impl SearchIndexSnapshot {
         let engine_capped = AtomicBool::new(false);
         let stop_after = max_results.saturating_mul(2);
         let stop_scan = Arc::new(AtomicBool::new(false));
-        let deadline = GrepScanDeadline::after_with_clock(scan_budget, now);
+        let deadline = make_deadline();
         let io_budget = crate::bounded_io::current();
         let verification_claims = AtomicUsize::new(0);
         let claim_verification = || {
@@ -3722,7 +3722,7 @@ fn search_candidate_file(
         return Vec::new();
     }
 
-    let content = match read_indexed_file_bytes(&file.path, deadline.at) {
+    let content = match read_indexed_file_bytes(&file.path, deadline.read_at) {
         Ok(content) => content,
         Err(error) => {
             if error.kind() == std::io::ErrorKind::TimedOut {
@@ -3939,16 +3939,20 @@ pub(crate) struct GrepScanDeadline {
     /// `None` when the budget is too large to represent as an instant, which
     /// leaves the scan unbounded in time.
     at: Option<Instant>,
+    /// File reads use real time, even when a test injects a scan clock.
+    read_at: Option<Instant>,
     now: fn() -> Instant,
     reached: AtomicBool,
 }
 
 impl GrepScanDeadline {
     fn after_with_clock(budget: Duration, now: fn() -> Instant) -> Self {
+        let at = now()
+            .checked_add(budget)
+            .map(|at| crate::bounded_io::current().map_or(at, |budget| at.min(budget.deadline)));
         Self {
-            at: now().checked_add(budget).map(|at| {
-                crate::bounded_io::current().map_or(at, |budget| at.min(budget.deadline))
-            }),
+            at,
+            read_at: at,
             now,
             reached: AtomicBool::new(false),
         }
@@ -10673,6 +10677,17 @@ mod tests {
         assert_ready_indexed_literal_match(r"foo\.bar");
     }
 
+    thread_local! {
+        static FROZEN_GREP_SCAN_START: Instant = Instant::now();
+    }
+
+    /// A scan clock that ignores real time: a fixed per-thread instant plus
+    /// whatever the test has added to the injected advance.
+    fn frozen_grep_scan_now() -> Instant {
+        FROZEN_GREP_SCAN_START.with(|start| *start)
+            + GREP_SCAN_CLOCK_ADVANCE.with(std::cell::Cell::get)
+    }
+
     /// Resets the injected clock after a deadline test, including on panic.
     struct GrepScanClockGuard;
 
@@ -10798,21 +10813,23 @@ mod tests {
         max_results: usize,
         budget: Duration,
     ) -> GrepResult {
-        indexed_grep_with_clock(index, root, pattern, max_results, budget, Instant::now)
+        indexed_grep_with_deadline(index, root, pattern, max_results, budget, || {
+            GrepScanDeadline::after_with_clock(budget, Instant::now)
+        })
     }
 
-    fn indexed_grep_with_clock(
+    fn indexed_grep_with_deadline(
         index: &SearchIndex,
         root: &Path,
         pattern: &CompiledPattern,
         max_results: usize,
         budget: Duration,
-        now: fn() -> Instant,
+        make_deadline: impl FnOnce() -> GrepScanDeadline,
     ) -> GrepResult {
         let query = decompose_grep_pattern(pattern);
         index
             .snapshot()
-            .search_grep_profiled_with_filters_and_query_and_limits_with_clock(
+            .search_grep_profiled_with_filters_and_query_and_limits_with_deadline(
                 pattern,
                 &query,
                 &PathFilters::default(),
@@ -10821,7 +10838,7 @@ mod tests {
                 None,
                 None,
                 budget,
-                now,
+                make_deadline,
             )
             .0
     }
@@ -10864,14 +10881,34 @@ mod tests {
             other => panic!("patterns must cover all three matcher arms: {other:?}"),
         }
 
-        for (arm, pattern) in &arms {
+        for (arm_index, (arm, pattern)) in arms.iter().enumerate() {
+            let budget = Duration::from_secs(2);
+            // Only the scan clock is frozen. Bounded file reads retain a real
+            // deadline with enough slack for a runner pause between arms.
+            let read_budget = Duration::from_secs(60);
+            let _io_scope = crate::bounded_io::enter_if_absent(read_budget);
+            let _clock = GrepScanClockGuard::install();
             let calls = std::rc::Rc::new(std::cell::Cell::new(0usize));
             let probe_calls = calls.clone();
-            let _probe = install_line_details_probe(move || probe_calls.set(probe_calls.get() + 1));
-            // The budget only lets a regressed, per-occurrence scan end; the
-            // assertion is on the work done, not on the time taken.
-            let result =
-                indexed_grep_with_budget(&index, &project, pattern, 100, Duration::from_secs(2));
+            // The scan clock is frozen and moves only when line details are
+            // built: half the budget per build. One build, the correct work,
+            // never reaches the deadline however slow the runner is; a
+            // regressed per-occurrence scan reaches it on its second build and
+            // ends quickly, failing the count below.
+            let _probe = install_line_details_probe(move || {
+                probe_calls.set(probe_calls.get() + 1);
+                GREP_SCAN_CLOCK_ADVANCE.with(|advance| advance.set(advance.get() + budget / 2));
+            });
+            let result = indexed_grep_with_deadline(&index, &project, pattern, 100, budget, || {
+                let mut deadline = GrepScanDeadline::after_with_clock(budget, frozen_grep_scan_now);
+                deadline.read_at = Instant::now().checked_add(read_budget);
+                if arm_index == 1 {
+                    // The second read must survive a pause longer than the
+                    // scan budget while the scan clock remains frozen.
+                    std::thread::sleep(Duration::from_secs(3));
+                }
+                deadline
+            });
 
             assert_eq!(
                 calls.get(),
@@ -10987,14 +11024,9 @@ mod tests {
             }
         });
 
-        let result = indexed_grep_with_clock(
-            &index,
-            &project,
-            &pattern,
-            10_000,
-            budget,
-            grep_scan_test_now,
-        );
+        let result = indexed_grep_with_deadline(&index, &project, &pattern, 10_000, budget, || {
+            GrepScanDeadline::after_with_clock(budget, grep_scan_test_now)
+        });
 
         assert!(result.scan_deadline_reached);
         assert!(result.engine_capped);

crates/aft/src/views/semantic_runtime_tests.rs

@@ -1377,6 +1377,7 @@ fn small_chunk_reuse_fill_completes_with_cold_cap_held() {
     runtime.driver().take_reconcile_work();
     runtime.plane().take_source_reads();
     let schedule = fast_schedule(root.path());
+    let cold = Arc::clone(&schedule.limiter);
     let held = schedule.limiter.try_acquire().unwrap();
     let (tx, rx) = crossbeam_channel::bounded(1);
     let worker = {
@@ -1401,10 +1402,21 @@ fn small_chunk_reuse_fill_completes_with_cold_cap_held() {
                 .unwrap();
         })
     };
-    let completed = rx.recv_timeout(Duration::from_secs(2));
-    if completed.is_err() {
-        slot.clear();
-    }
+    // Decide by what the fill does, not by how long it takes: it either
+    // finishes, or it queues on the held cold-build cap, which is the defect.
+    // A slow machine only makes the loop run longer; the test runner's
+    // overall timeout bounds a genuine hang elsewhere.
+    let completed = loop {
+        match rx.recv_timeout(Duration::from_millis(5)) {
+            Ok(result) => break Some(result),
+            Err(crossbeam_channel::RecvTimeoutError::Timeout) => {}
+            Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break None,
+        }
+        if !cold.census().queued.is_empty() {
+            slot.clear();
+            break None;
+        }
+    };
     drop(held);
     worker.join().unwrap();
     let (settled, lines) = completed.expect("reuse-only fill waited for the cold-build cap");

crates/aft/tests/integration/subc_storm_test.rs

@@ -1926,7 +1926,7 @@ async fn drive_epoch_rebind_rejection_daemon(input: FakeDaemonInput) {
         "storm-epoch-current",
         storm_project_config(false, false, false, 0),
     );
-    expect_ack_within(&mut rx, corr, BIND_ACK_BOUND).await;
+    expect_setup_bind_ack(&mut rx, corr).await;
     for epoch in [INSTALLED_EPOCH, INSTALLED_EPOCH - 1] {
         corr += 1;
         send_bind_epoch(
@@ -3187,6 +3187,32 @@ async fn expect_ack_within(rx: &mut mpsc::UnboundedReceiver<Frame>, corr: u64, b
     }
 }
 
+/// Waits for the reply to a bind that only sets up a scenario and whose
+/// latency is not what the test checks. A cold first bind does real work
+/// (topology probe, retention record, owner claim, index setup) that a loaded
+/// runner can stretch past any fixed window, so this waits for the reply
+/// itself: the module always answers a bind, with an ack or, at its own
+/// pending-bind deadline, a named refusal. A refusal fails here with its body
+/// instead of being skipped and reported as a timeout. The scenario's
+/// watchdog bounds a reply that never comes.
+async fn expect_setup_bind_ack(rx: &mut mpsc::UnboundedReceiver<Frame>, corr: u64) {
+    loop {
+        let frame = rx
+            .recv()
+            .await
+            .unwrap_or_else(|| panic!("EOF waiting for setup RouteBindAck {corr}"));
+        if is_ack(&frame, corr) {
+            return;
+        }
+        if frame.header.ty == FrameType::Error && frame.header.corr == corr {
+            panic!(
+                "setup bind {corr} was refused: {}",
+                String::from_utf8_lossy(&frame.body)
+            );
+        }
+    }
+}
+
 async fn expect_tool_response(
     rx: &mut mpsc::UnboundedReceiver<Frame>,
     corr: u64,