test: close the coverage gap on the bottleneck classifier's Rust surface

PR #311's coverage gate flagged the new backend surface as under-covered
(84.34% vs main's 84.56%). Adds targeted unit tests for every area the gate
report highlighted:

- bottleneck_hub.rs: extracted the AppHandle-independent decision logic out
  of tick_once into pure, directly-testable helpers (account_signal,
  aggregate_accounts, build_snapshot, should_emit), each covering a branch
  the monolithic async function previously hid from a unit test. Also covers
  BottleneckHub::push/latest directly.
- SyncOrchestrator::pacer_backoff_remaining_ms / backend_label: default
  (unwired), wired-but-clear (FakePacer), and wired-and-throttled (a real
  AimdPacer that has actually rate-limited) cases; plus a driven-app test
  proving FakeOrchestrator inherits the Orchestrator trait's default bodies.
- AimdPacer::backoff_remaining_ms: already covered by the previous commit's
  test, still verified against a FakeClock in and after the backoff window.
- IoCounters::add_hashed/snapshot: already covered by the previous commit's
  test.
- Executor credit paths: new inline_upload (buffered, <4MiB) and
  stream_upload (streaming cpu_stage, >=4MiB) tests asserting the hash
  counter is credited the full plaintext size in both pipelines.
- Scanner credit path: a direct scan_with_priority call (bypassing the
  scan()/scan_with_progress() wrappers, which always pass None) proving a
  deep-verify re-hash credits the hash counter.
- assembly.rs: bottleneck_backend_label covers all four BackendKind variants
  with distinct labels.
- app_state.rs: bottleneck_hub()/set_bottleneck_task()/shutdown_bottleneck_task()
  round-trip (register, signal, join, idempotent no-op).

Residual uncovered lines are structural: tick_once's AppHandle/AppState
plumbing, the bottleneck_status Tauri command (needs a real tauri::State,
which this codebase has no harness to construct - matches the pre-existing
untested pattern for io_throughput_series/get_sync_status), the
emit_sync_bottleneck event (needs a real AppHandle, matches
emit_sync_io_throughput's own untested baseline), and the lib.rs sampler
spawn/drain wiring (needs full app boot). None of these are new patterns -
each mirrors an already-uncovered sibling that predates this PR.

Local cargo llvm-cov (same command as .github/workflows/coverage.yml) moved
83.73% -> 83.96% on this diff; cargo test / clippy -D warnings / fmt all
clean (1032 driven-core + driven-app tests passing).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019xKUm9vH4ifb5LHR5szy1v

Author: pmaxhogan

crates/driven-core/src/executor.rs

@@ -12295,6 +12295,65 @@ mod tests {
         );
     }
 
+    /// issue #308 bottleneck classifier: a fresh small-file upload (the
+    /// BUFFERED `inline_upload` path, below [`PIPELINE_THRESHOLD`]) credits
+    /// the hash counter with the whole plaintext, alongside the existing
+    /// disk-read credit - both happen in the same `read_hash_encrypt` pass.
+    #[tokio::test]
+    async fn inline_upload_credits_hashed_bytes() {
+        let h = harness().await;
+        let body = vec![7u8; 4096];
+        let (rel, size) = h.write_file("small.bin", &body);
+
+        let io = Arc::new(crate::iostat::IoCounters::default());
+        let exec = h.executor().with_io_counters(io.clone());
+        let out = exec
+            .execute(
+                &h.source,
+                &h.upload_plan(&rel, size),
+                &noop_progress,
+                &noop_outcome,
+            )
+            .await
+            .unwrap();
+        assert!(matches!(out[0], OpOutcome::Done { .. }), "got {:?}", out[0]);
+
+        let snap = io.snapshot();
+        assert_eq!(snap.hashed_bytes, size, "the whole plaintext was hashed");
+        assert_eq!(snap.disk_read_bytes, size, "and read from disk");
+    }
+
+    /// issue #308 bottleneck classifier: a fresh large-file upload (the
+    /// STREAMING `cpu_stage` path, at/above [`PIPELINE_THRESHOLD`]) credits
+    /// the hash counter chunk-by-chunk as it streams, summing to the whole
+    /// plaintext by the time the upload completes.
+    #[tokio::test]
+    async fn stream_upload_credits_hashed_bytes() {
+        let h = harness().await;
+        let size_bytes = (PIPELINE_THRESHOLD + 64 * 1024) as usize;
+        let body: Vec<u8> = (0..size_bytes).map(|i| (i % 251) as u8).collect();
+        let (rel, size) = h.write_file("streamed.bin", &body);
+
+        let io = Arc::new(crate::iostat::IoCounters::default());
+        let exec = h.executor().with_io_counters(io.clone());
+        let out = exec
+            .execute(
+                &h.source,
+                &h.upload_plan(&rel, size),
+                &noop_progress,
+                &noop_outcome,
+            )
+            .await
+            .unwrap();
+        assert!(matches!(out[0], OpOutcome::Done { .. }), "got {:?}", out[0]);
+
+        let snap = io.snapshot();
+        assert_eq!(
+            snap.hashed_bytes, size,
+            "every streamed chunk's bytes were credited to the hash counter"
+        );
+    }
+
     /// [`ResumeAcc`]'s gauge accounting is symmetric across push / partial
     /// drain / clear, the drain clamps at the buffered length, and DROP
     /// refunds whatever is left - the guarantee the error-unwind test relies

crates/driven-core/src/orchestrator.rs

@@ -4458,6 +4458,86 @@ mod tests {
         }
     }
 
+    #[test]
+    fn pacer_backoff_and_backend_label_default_when_unwired() {
+        // issue #308: a trait object built without `with_pacer` /
+        // `with_backend_label` (the historical construction, and every test
+        // fake elsewhere in this crate) must read as "clear" with a generic
+        // label, never panic or silently misreport a backoff.
+        let account = AccountId::new_v4();
+        let dir = tempfile::tempdir().unwrap();
+        let src = source_in(account, dir.path());
+        let exec = Arc::new(RecordingExecutor::default());
+        let (orch, _clock) = build(
+            account,
+            vec![src],
+            exec,
+            power_on_ac(),
+            Arc::new(FakeNet::online()),
+            OrchestratorConfig::default(),
+        );
+        assert_eq!(orch.pacer_backoff_remaining_ms(), None);
+        assert_eq!(orch.backend_label(), "your destination");
+    }
+
+    #[test]
+    fn pacer_backoff_delegates_to_the_wired_pacer_when_clear() {
+        // issue #308: a wired-but-not-throttling pacer (`FakePacer`'s default
+        // `backoff_remaining_ms`, which it does not override) still reads as
+        // clear through the orchestrator - `with_pacer` alone must not
+        // fabricate a backoff.
+        let account = AccountId::new_v4();
+        let dir = tempfile::tempdir().unwrap();
+        let src = source_in(account, dir.path());
+        let exec = Arc::new(RecordingExecutor::default());
+        let (orch, _clock) = build(
+            account,
+            vec![src],
+            exec,
+            power_on_ac(),
+            Arc::new(FakeNet::online()),
+            OrchestratorConfig::default(),
+        );
+        let orch = orch
+            .with_pacer(Arc::new(FakePacer::default()))
+            .with_backend_label("Drive");
+        assert_eq!(orch.pacer_backoff_remaining_ms(), None);
+        assert_eq!(orch.backend_label(), "Drive");
+    }
+
+    #[test]
+    fn pacer_backoff_reports_the_live_deadline_from_a_throttled_pacer() {
+        // issue #308: a REAL AimdPacer that has actually throttled reports its
+        // live remaining-ms through the orchestrator, non-blocking (no
+        // `.await`, no clock advance needed to observe it).
+        let account = AccountId::new_v4();
+        let dir = tempfile::tempdir().unwrap();
+        let src = source_in(account, dir.path());
+        let exec = Arc::new(RecordingExecutor::default());
+        let (orch, _clock) = build(
+            account,
+            vec![src],
+            exec,
+            power_on_ac(),
+            Arc::new(FakeNet::online()),
+            OrchestratorConfig::default(),
+        );
+
+        let pacer_clock: Arc<dyn Clock> = Arc::new(FakeClock::new());
+        let pacer = crate::pacer::AimdPacer::new(pacer_clock, None);
+        pacer.note_response(crate::pacer::ResponseClass::RateLimited {
+            retry_after: std::time::Duration::from_secs(5),
+        });
+        let pacer: Arc<dyn Pacer> = Arc::new(pacer);
+        let orch = orch.with_pacer(pacer).with_backend_label("S3");
+
+        let remaining = orch
+            .pacer_backoff_remaining_ms()
+            .expect("the pacer is mid-backoff");
+        assert!(remaining >= 5_000, "remaining_ms = {remaining}");
+        assert_eq!(orch.backend_label(), "S3");
+    }
+
     #[test]
     fn effective_cap_throttles_only_when_metered_and_throttle_mode() {
         let base = OrchestratorConfig {

crates/driven-core/src/scanner.rs

@@ -2080,6 +2080,51 @@ mod tests {
         assert!(res.new_or_changed.is_empty(), "{:?}", res.new_or_changed);
     }
 
+    /// issue #308 bottleneck classifier: a deep-verify re-hash credits the
+    /// app-global hash counter with the WHOLE file's bytes, whether the
+    /// content turns out unchanged (this test) or changed (the mismatch
+    /// path re-hashes exactly the same way before it can tell the
+    /// difference). `scan_with_priority` is called directly (rather than the
+    /// `scan`/`scan_with_progress` wrappers, which always pass `None`) since
+    /// only it threads the counters through.
+    #[tokio::test]
+    async fn deep_verify_credits_the_hash_counter() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path();
+        let p = root.join("a.txt");
+        write(&p, b"hello");
+
+        let src = source_at(root);
+        let state = FakeState::default();
+        let (size, mtime) = stat_of(&p);
+        state.put(row(
+            src.id,
+            "a.txt",
+            size,
+            mtime,
+            *blake3::hash(b"hello").as_bytes(),
+        ));
+
+        let io = Arc::new(crate::iostat::IoCounters::default());
+        let res = scan_with_priority(
+            &src,
+            &state,
+            ScanMode::DeepVerify,
+            None,
+            None,
+            WorkPriority::Normal,
+            Some(io.clone()),
+        )
+        .await
+        .unwrap();
+        assert!(res.new_or_changed.is_empty(), "{:?}", res.new_or_changed);
+        assert_eq!(
+            io.snapshot().hashed_bytes,
+            size,
+            "the whole file was hashed"
+        );
+    }
+
     /// Issue #35 item e: a BUNDLED member - a `file_state` row with
     /// `drive_file_id = NULL` (its bytes live inside a `.tar.gz` bundle), status
     /// Synced, and a matching stored hash - must NOT be re-emitted as changed by a

src-tauri/src/app_state.rs

@@ -2062,6 +2062,63 @@ pub(crate) mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    #[test]
+    fn fake_orchestrator_inherits_the_orchestrator_traits_default_bottleneck_methods() {
+        // issue #308: `FakeOrchestrator` deliberately does not override
+        // `pacer_backoff_remaining_ms` / `backend_label` - it exists to
+        // exercise the `Orchestrator` trait's DEFAULT bodies (a pacer-less /
+        // label-less trait object must read as clear with a generic label,
+        // never panic), the same contract the bottleneck sampler leans on
+        // for any account whose orchestrator was built without those seams.
+        let orch = FakeOrchestrator::new();
+        assert_eq!(orch.pacer_backoff_remaining_ms(), None);
+        assert_eq!(orch.backend_label(), "your destination");
+    }
+
+    #[tokio::test]
+    async fn bottleneck_runtime_hub_task_and_shutdown_round_trip() {
+        // issue #308: the bottleneck sampler's runtime bookkeeping. Unlike
+        // `IostatRuntime` there is no "install" step - the hub is always
+        // available, defaulting to `NotBackingUp` - so this only needs to
+        // cover the getter plus the set/shutdown task pair (mirrors
+        // `set_iostat_task`/`shutdown_iostat_task`'s no-orphan drain).
+        let (state, dir) = temp_state().await;
+        let app_state = AppState::new(
+            state,
+            HashMap::new(),
+            RemoteMode::Fake,
+            default_fake_registry(),
+        );
+
+        let hub = app_state.bottleneck_hub();
+        assert_eq!(
+            hub.latest().state,
+            crate::bottleneck_hub::BottleneckState::NotBackingUp
+        );
+
+        // No task registered yet: shutdown is a safe no-op.
+        assert!(app_state.shutdown_bottleneck_task().is_none());
+
+        // Register a task that exits promptly on the shutdown signal (the
+        // real sampler's own shape), then confirm shutdown signals + hands
+        // back the handle so the quit drain can join it.
+        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+        let task = tokio::spawn(async move {
+            let _ = shutdown_rx.changed().await;
+        });
+        app_state.set_bottleneck_task(task, shutdown_tx);
+
+        let handle = app_state
+            .shutdown_bottleneck_task()
+            .expect("the just-registered task round-trips");
+        handle.await.unwrap();
+
+        // Taken: a second shutdown call is again a safe no-op.
+        assert!(app_state.shutdown_bottleneck_task().is_none());
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[tokio::test]
     async fn vss_helper_manager_installs_and_shutdown_is_noop() {
         // Issue #25: AppState owns the least-privilege VSS helper broker manager -

src-tauri/src/assembly.rs

@@ -1397,13 +1397,42 @@ struct SourceProgressEvent {
 
 #[cfg(test)]
 mod tests {
-    use super::{classify_bridge_event, BridgeAction};
+    use super::{bottleneck_backend_label, classify_bridge_event, BridgeAction};
     use driven_core::orchestrator::OrchestratorConfig;
     use driven_core::state::sqlite::SqliteStateRepo;
     use driven_core::state::StateRepo;
     use driven_core::types::{AccountId, ActivityEntry, ExecProgress, OrchestratorEvent, SourceId};
     use tokio::sync::broadcast::error::RecvError;
 
+    /// issue #308: every `BackendKind` variant maps to a distinct, non-empty
+    /// display label for the bottleneck classifier's "X rate-limited"
+    /// sub-line, matching the mockup's short-form wording ("Drive", not
+    /// "Google Drive").
+    #[test]
+    fn bottleneck_backend_label_covers_every_backend_kind_distinctly() {
+        use driven_remote::BackendKind;
+        let labels: Vec<&'static str> = BackendKind::ALL
+            .iter()
+            .map(|&kind| bottleneck_backend_label(kind))
+            .collect();
+        assert_eq!(bottleneck_backend_label(BackendKind::GoogleDrive), "Drive");
+        assert_eq!(bottleneck_backend_label(BackendKind::S3), "S3");
+        assert_eq!(
+            bottleneck_backend_label(BackendKind::LocalFolder),
+            "your local folder"
+        );
+        assert_eq!(bottleneck_backend_label(BackendKind::Sftp), "SFTP");
+        for label in &labels {
+            assert!(!label.is_empty());
+        }
+        let unique: std::collections::HashSet<_> = labels.iter().collect();
+        assert_eq!(
+            unique.len(),
+            labels.len(),
+            "every backend gets its own label"
+        );
+    }
+
     /// M7-P1-1: a broadcast `Lagged` MUST classify as an `ActivityReconcile`
     /// (carrying the dropped count) so the bridge emits `activity:lagged` and the
     /// webview reconciles the dropped rows from the durable `activity_log` -