feat(core): run the scan walk at the configured io_priority (#173)

Adopts #170's priority mechanism in #169's parallel scanner - the
follow-up called out in #170's PR body ('the broad user-visible win
lands when the scanner's dedicated worker threads call
apply_to_current_thread').

- The walk's pooled spawn_blocking coordinator takes an RAII
begin_background_work guard, so the matcher build's ignore-file
collection I/O is shaped too and the pooled thread is restored on scan
end.
- Each transient walk worker thread (stat + include check + BLAKE3
hashing for deep-verify/coarse-FS) applies the priority once on its
first visit; workers are joined at walk end, so a one-shot apply with no
restore is correct.
- New scan_with_priority entry point;
scan/scan_with_latency/scan_with_progress keep their exact signatures
(Normal), so no test churn. The orchestrator passes its live
PriorityCell value - settings changes apply from the next scan.

A sibling PR (in flight) extends shaping to the large-file upload
pipeline.

Gates: cargo fmt, clippy -p driven-core -D warnings clean, cargo test -p
driven-core 393+ green.

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

https://claude.ai/code/session_01JLB3E2Jm7knNJd37fVpH8X

Author: pmaxhogan

crates/driven-core/src/orchestrator.rs

@@ -1565,12 +1565,17 @@ impl SyncOrchestrator {
                 scanned_total.store(scanned, std::sync::atomic::Ordering::Relaxed);
                 Box::pin(self.transition(OrchestratorState::Scanning { source_id, scanned }))
             };
-            crate::scanner::scan_with_progress(
+            crate::scanner::scan_with_priority(
                 source,
                 self.state.as_ref(),
                 mode,
                 self.latency.as_deref(),
                 Some(&on_scan_progress),
+                // Live SPEC s22 io_priority: the walk's coordinator + worker
+                // threads demote themselves so scan stat/hash I/O yields to
+                // foreground apps (the same cell the executor's bundle path
+                // reads).
+                self.priority.get(),
             )
             .await?
         };

crates/driven-core/src/scanner.rs

@@ -65,6 +65,7 @@ use anyhow::Context;
 use ignore::{DirEntry, ParallelVisitor, ParallelVisitorBuilder, WalkState};
 
 use crate::exclude::{build_source_matcher, build_walker_with_matcher, SourceMatcher};
+use crate::priority::WorkPriority;
 use crate::state::{FileStateRow, SourceRow, StateRepo};
 // `PlaceholderPolicy` is consumed only by `should_skip_placeholder`, which is
 // cfg-gated the same way; on a non-Windows, non-test lib build the fn (and thus
@@ -310,6 +311,12 @@ struct WalkCtx {
     /// Whether to time each file (the reservoir itself lives on the consumer
     /// side, which is what actually records the sample).
     capture_latency: bool,
+    /// The backup-work priority (SPEC s22 `io_priority`) each walk worker
+    /// thread applies to itself on its first visit, so the scan's stat + hash
+    /// I/O yields to foreground applications. Walker threads live only for the
+    /// walk (spawned and joined inside `build_parallel().visit()`), so a
+    /// one-shot apply with no restore is correct - nothing pooled is demoted.
+    priority: WorkPriority,
 }
 
 /// What a worker concluded about ONE file. Deliberately excludes every
@@ -372,6 +379,10 @@ struct WalkVisitor {
     /// Set once the consumer has gone away, so the walk stops instead of
     /// traversing the rest of the tree with nowhere to send the results.
     consumer_gone: bool,
+    /// Whether this worker thread has applied [`WalkCtx::priority`] to itself
+    /// yet. `build()` may run on the coordinating thread, so the apply happens
+    /// on the first `visit()` call, which is guaranteed to run on the worker.
+    priority_applied: bool,
 }
 
 impl WalkVisitor {
@@ -423,6 +434,10 @@ impl Drop for WalkVisitor {
 
 impl ParallelVisitor for WalkVisitor {
     fn visit(&mut self, result: Result<DirEntry, ignore::Error>) -> WalkState {
+        if !self.priority_applied {
+            self.priority_applied = true;
+            crate::priority::apply_to_current_thread(self.ctx.priority);
+        }
         let ctx = Arc::clone(&self.ctx);
         let entry = match result {
             Ok(e) => e,
@@ -517,6 +532,7 @@ impl<'s> ParallelVisitorBuilder<'s> for WalkVisitorBuilder {
             items: 0,
             last_flush: Instant::now(),
             consumer_gone: false,
+            priority_applied: false,
         })
     }
 }
@@ -667,6 +683,33 @@ pub async fn scan_with_progress(
     mode: ScanMode,
     latency: Option<&crate::telemetry::LatencyReservoir>,
     on_progress: Option<&ScanProgressSink<'_>>,
+) -> anyhow::Result<ScanResult> {
+    scan_with_priority(
+        source,
+        state,
+        mode,
+        latency,
+        on_progress,
+        WorkPriority::Normal,
+    )
+    .await
+}
+
+/// [`scan_with_progress`] plus the backup-work [`WorkPriority`] (SPEC s22
+/// `io_priority`) the walk should run at. The blocking coordinator task takes an
+/// RAII [`begin_background_work`](crate::priority::begin_background_work) guard
+/// (its thread is POOLED, so the demotion must be undone), and each transient
+/// walk worker thread applies the priority once on its first visit (no restore
+/// needed - workers die with the walk). `Normal` is exactly the old behaviour;
+/// the orchestrator passes its live [`PriorityCell`](crate::priority::PriorityCell)
+/// value so a settings change applies from the next scan.
+pub async fn scan_with_priority(
+    source: &SourceRow,
+    state: &dyn StateRepo,
+    mode: ScanMode,
+    latency: Option<&crate::telemetry::LatencyReservoir>,
+    on_progress: Option<&ScanProgressSink<'_>>,
+    priority: WorkPriority,
 ) -> anyhow::Result<ScanResult> {
     let known = state
         .load_source_file_state(source.id)
@@ -799,6 +842,10 @@ pub async fn scan_with_progress(
     let walk_known = Arc::clone(&known);
     let capture_latency = latency.is_some_and(|r| r.is_enabled());
     let walk_task = tokio::task::spawn_blocking(move || -> anyhow::Result<Arc<SourceMatcher>> {
+        // This spawn_blocking thread is POOLED, so the demotion is held in an
+        // RAII guard that restores normal priority when the walk ends; the
+        // matcher build's ignore-file collection I/O runs demoted too.
+        let _priority_guard = crate::priority::begin_background_work(priority);
         // ONE matcher for the whole scan: the walker's prune predicate, every
         // per-entry include check, and the orphan split share this `Arc` rather
         // than each rebuilding the cascade (the walker used to build a second,
@@ -814,6 +861,7 @@ pub async fn scan_with_progress(
             is_coarse,
             last_scan_end_ns,
             capture_latency,
+            priority,
         });
         let mut wb = build_walker_with_matcher(&walk_source, Arc::clone(&matcher));
         wb.threads(walk_threads());