app_state.rs

//! [`AppState`]: the Tauri-managed shared application state (SPEC s11).
//!
//! Held behind Tauri's `State<AppState>` and reached by every IPC command
//! (SPEC s11.3). It owns the [`StateRepo`] handle plus one orchestrator
//! handle per account - the `Arc<dyn Orchestrator>` control surface and the
//! `JoinHandle` of its spawned run loop (SPEC s5: one orchestrator per
//! account). The remote-construction mode records whether assembly built
//! real `GoogleDriveStore`s or the in-memory fake (`DRIVEN_USE_FAKE_REMOTE`).

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use driven_core::orchestrator::Orchestrator;
use driven_core::state::StateRepo;
use driven_core::types::AccountId;
use driven_drive::fake::InMemoryRemoteStore;
use driven_power::SleepWakeMonitor;
use tokio::sync::{watch, Mutex};
use tokio::task::JoinHandle;

/// How the per-account remote store was constructed at assembly time.
///
/// Recorded so IPC / diagnostics can tell a real run from a fake-backed one
/// (`DRIVEN_USE_FAKE_REMOTE=1` selects [`RemoteMode::Fake`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RemoteMode {
    /// Real `GoogleDriveStore` built from the keyring refresh token.
    RealGoogleDrive,
    /// `InMemoryRemoteStore` (test / dev; `DRIVEN_USE_FAKE_REMOTE=1`).
    Fake,
}

/// One account's live orchestrator: the control-surface handle plus EVERY
/// per-account tokio task spawned by `assembly::build_account` (SPEC s5,
/// ROADMAP M5 "no orphaned tokio tasks"; DESIGN s5.10.2 in-flight drain).
///
/// R-P1-1: a clean Quit must leave NO orphaned tasks. Four tasks are spawned
/// per account and ALL are tracked here so [`Self::shutdown`] can drain them:
/// - `run_loop`: [`Orchestrator::run`]. Stopped via `Orchestrator::shutdown()`.
/// - `watcher_bridge`: forwards `NotifyWatcher` scan-ticks into the
///   orchestrator. The watcher owns the `mpsc::Sender`, so its `recv().await`
///   never closes on its own; it must be signalled via [`Self::bridge_shutdown`]
///   (it `select!`s on that watch) or aborted.
/// - `event_bridge`: forwards the orchestrator's `OrchestratorEvent` broadcast
///   to the tray + webview. It ends naturally when the broadcast closes (the
///   orchestrator dropped) but is ALSO signalled so quit does not have to wait
///   on a `Lagged`/slow consumer; aborted on timeout.
/// - `power_poller`: the `RealPowerSource` 30s poll loop. It loops forever (no
///   natural end), so its handle is KEPT and ABORTED on shutdown - dropping it
///   (the old bug) orphaned the task.
///
/// Held so IPC can drive the orchestrator (`trigger` / `set_paused` / `state`)
/// and so [`Self::shutdown`] can stop + join every task on quit.
pub struct AccountHandle {
    /// The per-account orchestrator control surface.
    pub orchestrator: Arc<dyn Orchestrator>,
    /// B2: the per-account LIVE crypto provider. Held so the source-command
    /// layer can REFRESH its source metadata (`crypto.refresh(..)`) after a
    /// source add / toggle / remove, so a mid-session encrypted source's key is
    /// resolved on the next tick (not stranded `Unavailable` until restart).
    pub crypto: Arc<crate::crypto_provider_impl::KeystoreCryptoProvider>,
    /// The spawned run-loop task. Behind a `Mutex<Option<..>>` so the shutdown
    /// path can TAKE + await it by value; `None` once drained.
    run_loop: Mutex<Option<JoinHandle<()>>>,
    /// The watcher-bridge task (forwards scan-ticks), or `None` when no enabled
    /// source produced a watcher. Drained on shutdown.
    watcher_bridge: Mutex<Option<JoinHandle<()>>>,
    /// The orchestrator-event -> tray/IPC bridge task. Drained on shutdown.
    event_bridge: Mutex<Option<JoinHandle<()>>>,
    /// The power-source poller task. Looped forever; ABORTED on shutdown.
    power_poller: Mutex<Option<JoinHandle<()>>>,
    /// The OS sleep/wake EDGE monitor (DESIGN s5.10.1, issue #33), or `None`
    /// when its per-OS registration failed at build time (the app then degrades
    /// to the 30 s poll). TORN DOWN on shutdown - its `stop()` unregisters the
    /// Win32 suspend/resume callback / stops the macOS `CFRunLoop` thread /
    /// aborts the Linux logind DBus task - so no OS handle, thread, or task
    /// outlives the account.
    sleep_wake_monitor: Mutex<Option<SleepWakeMonitor>>,
    /// Shutdown signal the watcher + event bridges `select!` on (R-P1-1). Set to
    /// `true` by [`Self::shutdown`] so a bridge whose source never closes
    /// (the watcher owns its `Sender`) still exits promptly.
    bridge_shutdown: watch::Sender<bool>,
}

/// The SHORT per-task graceful-drain budget on quit (DESIGN s5.10.2) for the
/// AUXILIARY tasks (watcher bridge, event bridge, power poller): await each this
/// long before aborting it. These tasks carry no in-flight upload work - they
/// only forward signals or poll - so they should stop near-instantly once the
/// run loop has exited; the short budget keeps a single wedged bridge/poller
/// from holding the join indefinitely.
const TASK_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);

/// R2-P2-1: the run loop's OWN graceful-drain budget on quit. The run loop is
/// the ONLY per-account task that may be mid-`execute` (a large in-flight
/// upload), so DESIGN s5.10.2's "let the current cycle finish" guarantee must
/// give it the FULL drain window - not the short [`TASK_DRAIN_TIMEOUT`] the
/// signal-only bridges use. Before this fix the run loop shared the 5s bridge
/// budget, so a >5s in-flight upload was aborted on explicit Quit even though
/// the intended drain window was ~20s.
///
/// R3-P1-1: `lib.rs`'s quit sweep no longer wraps the per-account drains in an
/// outer timeout (that could drop a cancellation-unsafe drain mid-abort and
/// orphan a task); instead it runs every account's `shutdown()` concurrently and
/// lets each self-bound by this budget (plus the short auxiliary one).
pub const RUN_LOOP_DRAIN_TIMEOUT: Duration = Duration::from_secs(20);

/// The collected per-account task handles + the bridge shutdown sender, returned
/// by `assembly::build_account` and stored on [`AccountHandle`]. Groups the four
/// tracked tasks so the constructor signature stays readable (R-P1-1).
pub struct AccountTasks {
    /// B2: the per-account live crypto provider (so the handle can expose it for
    /// refresh on a source change).
    pub crypto: Arc<crate::crypto_provider_impl::KeystoreCryptoProvider>,
    /// [`Orchestrator::run`] loop.
    pub run_loop: JoinHandle<()>,
    /// Watcher -> orchestrator scan-tick bridge, or `None` if none was spawned.
    pub watcher_bridge: Option<JoinHandle<()>>,
    /// Orchestrator-event -> tray/IPC bridge.
    pub event_bridge: JoinHandle<()>,
    /// Power-source poll loop.
    pub power_poller: JoinHandle<()>,
    /// The OS sleep/wake EDGE monitor (DESIGN s5.10.1, issue #33), or `None`
    /// when its per-OS registration failed (degrade to the 30 s poll).
    pub sleep_wake_monitor: Option<SleepWakeMonitor>,
    /// The sender the watcher + event bridges `select!` on for shutdown.
    pub bridge_shutdown: watch::Sender<bool>,
}

impl AccountHandle {
    /// Build a handle from the orchestrator control surface + the collected
    /// per-account task set (R-P1-1).
    #[must_use]
    pub fn new(orchestrator: Arc<dyn Orchestrator>, tasks: AccountTasks) -> Self {
        Self {
            orchestrator,
            crypto: tasks.crypto,
            run_loop: Mutex::new(Some(tasks.run_loop)),
            watcher_bridge: Mutex::new(tasks.watcher_bridge),
            event_bridge: Mutex::new(Some(tasks.event_bridge)),
            power_poller: Mutex::new(Some(tasks.power_poller)),
            sleep_wake_monitor: Mutex::new(tasks.sleep_wake_monitor),
            bridge_shutdown: tasks.bridge_shutdown,
        }
    }

    /// Stop + JOIN every per-account task so quit leaves NO orphaned tokio task
    /// (R-P1-1, ROADMAP M5 acceptance; DESIGN s5.10.2 graceful drain).
    ///
    /// Order (R2-P2-1):
    /// 1. signal the orchestrator to stop after its in-flight cycle
    ///    (`Orchestrator::shutdown()`), and signal the bridges via
    ///    [`Self::bridge_shutdown`] (the watcher bridge's source never closes on
    ///    its own);
    /// 2. drain the RUN LOOP FIRST with its OWN full [`RUN_LOOP_DRAIN_TIMEOUT`]
    ///    budget - it is the only task that may be mid-upload, so DESIGN s5.10.2's
    ///    "let the current cycle finish" applies to it and it must NOT be cut off
    ///    by the short bridge budget;
    /// 3. ONLY AFTER the run loop has exited, drain the auxiliary tasks (watcher
    ///    bridge, event bridge, power poller) with the short [`TASK_DRAIN_TIMEOUT`]
    ///    - they carry no upload work and stop promptly once the loop is gone.
    ///
    /// For EVERY tracked handle: await-with-timeout, and on timeout abort the
    /// task AND AWAIT the aborted handle - so the task is truly GONE, not merely
    /// abort-requested. The power poller loops forever, so it always takes the
    /// abort path; the others normally drain cleanly.
    ///
    /// Idempotent: a second call finds every handle already taken and returns
    /// immediately. Errors awaiting a task (cancelled / panicked) are swallowed -
    /// the post-condition is "the task is no longer running".
    pub async fn shutdown(&self) {
        // 1) Signal stop. The orchestrator finishes its current cycle then
        //    returns; the bridges observe the watch flip and select! out.
        self.orchestrator.shutdown();
        // A send error only means there are no live bridge receivers (already
        // gone) - benign.
        let _ = self.bridge_shutdown.send(true);

        // 2) Drain the run loop FIRST with the FULL in-flight budget (R2-P2-1):
        //    a large upload mid-`execute` gets the intended ~20s to finish
        //    rather than being aborted at the 5s bridge budget.
        drain_or_abort(&self.run_loop, RUN_LOOP_DRAIN_TIMEOUT).await;

        // 3) THEN drain the signal-only auxiliary tasks with the SHORT budget.
        //    With the run loop already gone they should stop near-instantly.
        drain_or_abort(&self.watcher_bridge, TASK_DRAIN_TIMEOUT).await;
        drain_or_abort(&self.event_bridge, TASK_DRAIN_TIMEOUT).await;
        drain_or_abort(&self.power_poller, TASK_DRAIN_TIMEOUT).await;

        // 4) Tear down the OS sleep/wake monitor (issue #33): its `stop()`
        //    unregisters the Win32 suspend/resume callback / stops the macOS
        //    CFRunLoop thread / aborts the Linux logind DBus task, so no OS
        //    handle, thread, or task outlives the account. A `None` (registration
        //    failed at build time) is a no-op.
        if let Some(monitor) = self.sleep_wake_monitor.lock().await.take() {
            monitor.stop();
        }
    }
}

/// Take the handle out of `slot` and drive it to a true stop: await it up to
/// `budget`; on timeout `abort()` it and AWAIT the aborted handle so the task is
/// genuinely finished before this returns (R-P1-1). A `None` slot (already
/// drained / never spawned) is a no-op. R2-P2-1: the budget is a parameter so
/// the run loop gets the full [`RUN_LOOP_DRAIN_TIMEOUT`] while the auxiliary
/// tasks use the short [`TASK_DRAIN_TIMEOUT`].
///
/// `tokio::time::timeout` MOVES the `JoinHandle` into itself and, on elapse,
/// DROPS it - and a dropped `JoinHandle` does NOT cancel its task (it merely
/// detaches it). So we capture an [`tokio::task::AbortHandle`] BEFORE the
/// timeout, and on elapse abort via it, then RE-AWAIT the same task via a second
/// `JoinHandle` we also kept... which `timeout` consumed. To avoid that, we do
/// not hand the original handle to `timeout`; we `select!` between the handle and
/// a sleep so the handle stays in scope and can be re-awaited after an abort.
async fn drain_or_abort(slot: &Mutex<Option<JoinHandle<()>>>, budget: Duration) {
    let Some(mut handle) = slot.lock().await.take() else {
        return;
    };
    let abort = handle.abort_handle();
    tokio::select! {
        // Bias toward the task finishing: if it completes within the budget we
        // take this arm and never abort.
        biased;
        _join_result = &mut handle => {
            // Joined cleanly (or the task panicked - either way it is gone).
        }
        () = tokio::time::sleep(budget) => {
            // Budget elapsed: request cancellation, then AWAIT the same handle
            // so the task is genuinely finished (a JoinError::cancelled is the
            // expected, swallowed result) before we return.
            abort.abort();
            let _ = handle.await;
        }
    }
}

/// The Tauri-managed application state (SPEC s11).
pub struct AppState {
    /// The SQLite state layer (SPEC s2), shared by every account + IPC path.
    state: Arc<dyn StateRepo>,
    /// Per-account orchestrator handles, keyed by [`AccountId`].
    ///
    /// A2: behind a sync [`std::sync::Mutex`] (never held across an await -
    /// only ever for a quick insert / clone-out) so the wizard can HOT-SPAWN a
    /// brand-new account's orchestrator mid-session (`finish_add_account` ->
    /// `spawn_account`) and `sync_now` finds it without a restart. Each handle
    /// is an [`Arc`] so a caller can clone it out and drive / await it after
    /// releasing the map lock.
    accounts: std::sync::Mutex<HashMap<AccountId, Arc<AccountHandle>>>,
    /// How remotes were constructed this run (real Drive vs in-memory fake).
    remote_mode: RemoteMode,
    /// C5-P2-1: per-account pause "generation" token. Every pause/resume bumps
    /// the account's generation; a TIMED pause captures the new generation and
    /// its detached auto-resume timer only fires if the generation still
    /// matches when it wakes. A newer pause/resume (e.g. a `pause(None)`
    /// indefinite pause issued before the old timer fires) bumps the generation
    /// and thereby CANCELS the stale timer's auto-resume. Behind a sync `Mutex`
    /// (only ever held for a counter bump/read, never across an await).
    pause_generations: std::sync::Mutex<HashMap<AccountId, u64>>,
    /// C1 (SPEC s11.6.1): one-shot dialog-token -> path bindings. The backend
    /// OWNS the native folder / save-file dialogs; each returns an opaque token
    /// bound to the path the USER actually chose. A path-bearing write command
    /// (`add_source`, `export_diagnostic_bundle`) must present a token that maps
    /// to exactly that path - so the (untrusted) webview can never inject an
    /// arbitrary path. Single-use (spent when the write commits) with a bounded
    /// TTL so a leaked token cannot be replayed later. Behind a sync `Mutex`
    /// (only ever held for a quick insert / take, never across an await).
    dialog_tokens: std::sync::Mutex<HashMap<String, DialogTokenBinding>>,
    /// R2-P1-1: per-account ASYNC lock serialising the FIRST-encrypted-source
    /// critical section (ensure-master-key -> stamp -> insert source). Without
    /// it two concurrent `add_source` calls on an account whose
    /// `encryption_master_key_id` is still NULL could BOTH generate DIFFERENT
    /// master keys into the same keychain slot and wrap different source keys -
    /// leaving one source permanently unrestorable. The lock (a `tokio::Mutex`
    /// so it can be held across the awaited DB write) makes the second add see
    /// the master key the first installed and wrap under the SAME key. Keyed by
    /// account; the inner map is behind a sync `Mutex` only to hand out the
    /// per-account `Arc<tokio::Mutex<()>>` (never held across an await).
    ensure_master_key_locks: std::sync::Mutex<HashMap<AccountId, Arc<Mutex<()>>>>,
    /// R2-P1-2: per-account in-memory fake remote store, shared between the
    /// Drive-folder picker (`pick_drive_folder`) and the orchestrator's
    /// uploader (assembly `build_remote`) so a folder id the picker mints in
    /// fake mode is visible to the uploader. Created on demand, one instance per
    /// account ([`InMemoryRemoteStore`] is `Clone` over a shared `Arc<Mutex>`, so
    /// every clone sees the same backing objects). Only ever populated in
    /// [`RemoteMode::Fake`]; in real mode it stays empty.
    fake_remote_stores: FakeRemoteStores,
    /// M8: live restore-job records, keyed by job id. Each entry carries the
    /// latest [`RestoreJobStatus`](crate::commands::dtos::RestoreJobStatus)
    /// snapshot (the background task writes it on every progress tick, so
    /// `get_restore_job` can serve a webview that subscribed late / missed an
    /// event), plus the per-job CANCEL control + spawned [`JoinHandle`] so
    /// `cancel_restore_job` and the app-shutdown drain can stop an in-flight job
    /// (M8-P1-1). Behind a sync `Mutex` (only ever held for a quick insert /
    /// clone-out / take, never across an await). Terminal entries are retained so
    /// a late poll still sees the result, but they are TTL-pruned + count-capped
    /// (M8-P2-3) so a long-running tray app does not leak snapshots forever.
    restore_jobs: std::sync::Mutex<HashMap<String, RestoreJobEntry>>,
    /// M9a (SPEC s15.2): the in-app updater runtime - the pending checked update
    /// (held so `install_update` stages + applies the SAME object the check
    /// found) plus the periodic-check task handle + shutdown signal, so the
    /// app-quit drain joins it with NO orphan (mirrors the M5 no-orphan
    /// bookkeeping).
    updater: UpdaterRuntime,
    /// 2026-08-14 follow-up: live disk/network throughput sampling runtime.
    iostat: IostatRuntime,
    /// issue #308 (2026-08-17 follow-up): live bottleneck-classification
    /// sampling runtime (the Activity dashboard's Bottleneck stat tile).
    bottleneck: BottleneckRuntime,
    /// issue #309: the debug-logging-mode 24h auto-off watchdog's task handle
    /// and shutdown signal, so the app-quit drain joins it with no orphan
    /// (mirrors [`UpdaterRuntime`]/[`IostatRuntime`]). No shared "hub" field
    /// like those two - the watchdog only reads/writes the persisted settings
    /// KV directly, nothing else on `AppState` needs to observe it.
    debug_mode: DebugModeRuntime,
    /// The ONE in-flight streaming exclusion preview
    /// ([`crate::commands::exclusion_stream`]). The exclusion editor re-previews
    /// on every rule edit, so without a single-slot registry a user tweaking
    /// globs over a large folder would stack N concurrent full-tree walks;
    /// starting a preview cancels the one it supersedes, and the editor cancels
    /// the last one when it closes. Behind an `Arc` because the blocking walk
    /// task deregisters itself when it ends, outliving the command that spawned
    /// it.
    exclusion_previews: Arc<crate::commands::exclusion_stream::PreviewRegistry>,
    /// The exclusion editor's in-memory folder-tree cache
    /// ([`crate::commands::preview_cache`]). Walking the source folder and
    /// classifying it under the candidate rules are separate jobs, and only the
    /// second one changes when a glob is edited - so the first pass over a root
    /// records what it found here and every later pass re-classifies from
    /// memory, turning a minutes-long re-walk per keystroke into milliseconds.
    /// Editor-scoped: dropped when the previewed root changes and freed when the
    /// editor closes. Behind an `Arc` for the same reason as the registry above
    /// (the blocking pass outlives the command that spawned it).
    preview_tree_cache: Arc<crate::commands::preview_cache::PreviewTreeCache>,
    /// M9b (SPEC s16): the anonymous-telemetry runtime - the periodic ping task
    /// handle + shutdown signal, so the app-quit drain joins it with NO orphan
    /// (mirrors the M9a updater bookkeeping). The ping itself is best-effort and
    /// holds no state here beyond the task control surface.
    telemetry: TelemetryRuntime,
    /// M9c D4 (M6 R4-P1-1, DATA-SAFETY); R4-P1-1 made DURABLE: per-source
    /// recovery-phrase ACK gate, a reconstructed-on-startup MIRROR of the durable
    /// `recovery_phrase_acks` table. The FIRST encrypted source for an account is
    /// persisted DISABLED (excluded from the scheduler + manual sync, which filter
    /// on `enabled`) and a pending-ack record is written DURABLY in the same
    /// transaction as the source insert + master-key stamp. The source is only
    /// ENABLED once `ack_recovery_phrase_saved` lands - and that ack is REJECTED
    /// unless a real backend `reveal_recovery_phrase` was recorded first
    /// (`revealed == true`). So a user can never tick "I saved it" without the
    /// backend having actually revealed the phrase, and no encrypted backups run
    /// before the recovery phrase is durably saveable - closing the
    /// unrestorable-backup window EVEN ACROSS A CRASH/RESTART.
    ///
    /// R4-P1-1: the DURABLE table is the source of truth for every gate decision
    /// (the command layer reads/writes it via [`StateRepo`]); this in-memory map is
    /// a reconstructed mirror so a fresh process resumes the exact pending-ack gate,
    /// kept in sync as the commands mutate the durable state. Behind a sync `Mutex`
    /// (only ever held for a quick insert / read / take, never across an await).
    recovery_acks: std::sync::Mutex<HashMap<driven_core::types::SourceId, RecoveryAckState>>,
    /// Issue #25 (DESIGN s5.3.1): the app-side least-privilege VSS helper broker
    /// lifecycle owner, or `None` when the helper is not in play (off Windows, the
    /// app is already elevated, or the `windows.vss_helper` setting is off). Built
    /// ONCE by `assembly::build_and_spawn` and installed here so (a) the quit sweep
    /// can shut the broker down (no elevated process outlives the app) and (b)
    /// `get_vss_helper_status` can report truthful liveness. Behind a sync `Mutex`
    /// (set once at boot, read for status/shutdown - never held across an await).
    vss_helper: std::sync::Mutex<Option<Arc<crate::vss_helper::VssHelperManager>>>,
    /// DESIGN s5.3.2: the macOS sibling of [`Self::vss_helper`] - the app-side
    /// APFS snapshot broker lifecycle owner, or `None` when it is not in play
    /// (off macOS, or the bundled sidecar could not be located). Built ONCE by
    /// `assembly::build_and_spawn` and installed here so (a) the quit sweep can
    /// shut the broker down (no root process outlives the app) and (b)
    /// `get_apfs_helper_status` can report truthful liveness.
    apfs_helper: std::sync::Mutex<Option<Arc<crate::apfs_helper::ApfsHelperManager>>>,
}

/// M9c D4: one pending recovery-phrase ack - the owning account (so the ack can
/// reconfigure it once the source is enabled) and whether the backend has actually
/// REVEALED the phrase (`reveal_recovery_phrase`). `ack_recovery_phrase_saved` is
/// rejected unless `revealed` is true. R4-P1-1: a mirror of the durable
/// `recovery_phrase_acks` row.
struct RecoveryAckState {
    /// The account owning the pending-ack source (used to reconfigure on enable).
    account_id: AccountId,
    /// True once `reveal_recovery_phrase` has actually returned the phrase from the
    /// backend for this source. The ack is ineffective until this is set.
    revealed: bool,
}

/// M9a (SPEC s15.2): the in-app updater runtime state held on [`AppState`].
///
/// `pending` holds the [`tauri_plugin_updater::Update`] the most recent check
/// found (manual or periodic) PLUS the channel it came from, so `install_update`
/// can `download_and_install` the SAME object without re-resolving the manifest
/// AND emit `updater:downloaded` with the REAL channel (R1-P2-3). It is TAKEN on
/// install but RESTORED on a download/install failure (R1-P2-2) so the banner's
/// next Install retries instead of failing "no pending update"; a fresh check
/// also re-populates it. `task` + `shutdown` track the single app-wide
/// periodic-check task so the quit drain stops + joins it with no orphan.
#[derive(Default)]
pub struct UpdaterRuntime {
    /// The update the latest check found (with its channel string), awaiting
    /// install; `None` when up to date / not yet checked / already installed.
    pending: std::sync::Mutex<Option<(tauri_plugin_updater::Update, String)>>,
    /// The spawned periodic-check task, behind `Option` so the shutdown drain
    /// can TAKE + await it by value; `None` once drained / never spawned.
    task: std::sync::Mutex<Option<JoinHandle<()>>>,
    /// The shutdown signal the periodic-check task `select!`s on, so it exits
    /// promptly on quit rather than waiting out its 6h interval.
    shutdown: std::sync::Mutex<Option<watch::Sender<bool>>>,
}

/// 2026-08-14 follow-up: the live disk/network throughput runtime held on
/// [`AppState`] - the app-global counters+ring hub every executor credits,
/// plus the sampler task's lifecycle slots (mirrors [`UpdaterRuntime`] so the
/// quit drain stops + joins it with no orphan).
#[derive(Default)]
pub struct IostatRuntime {
    /// The counters + trailing-sample ring. Installed by assembly (the SAME
    /// hub whose counters the executors were built with); the quiesced boot
    /// path keeps the default hub (all-zero graphs).
    hub: std::sync::Mutex<Arc<crate::iostat_hub::IoStatHub>>,
    /// The spawned sampler task, behind `Option` so the shutdown drain can
    /// TAKE + await it by value; `None` once drained / never spawned.
    task: std::sync::Mutex<Option<JoinHandle<()>>>,
    /// The shutdown signal the sampler `select!`s on.
    shutdown: std::sync::Mutex<Option<watch::Sender<bool>>>,
}

/// issue #308 (2026-08-17 follow-up): the live bottleneck-classification
/// runtime held on [`AppState`] - the latest-snapshot hub plus the sampler
/// task's lifecycle slots (mirrors [`IostatRuntime`], which this sampler
/// reads from). Unlike `IostatRuntime` there is nothing to "install" from
/// assembly: the hub reads the app-global IO counters and the accounts map
/// straight off `AppState` each tick, so the default hub is already correct
/// even in the quiesced boot path (it just classifies `NotBackingUp`).
#[derive(Default)]
pub struct BottleneckRuntime {
    /// The latest-snapshot hub the sampler pushes into and the
    /// `bottleneck_status` command reads.
    hub: Arc<crate::bottleneck_hub::BottleneckHub>,
    /// The spawned sampler task, behind `Option` so the shutdown drain can
    /// TAKE + await it by value; `None` once drained / never spawned.
    task: std::sync::Mutex<Option<JoinHandle<()>>>,
    /// The shutdown signal the sampler `select!`s on.
    shutdown: std::sync::Mutex<Option<watch::Sender<bool>>>,
}

/// issue #309: the debug-logging-mode 24h auto-off watchdog's runtime state
/// held on [`AppState`] - just the task's lifecycle slots (mirrors
/// [`UpdaterRuntime`]/[`TelemetryRuntime`]'s task+shutdown pair). No "hub"
/// field like [`BottleneckRuntime`]/[`IostatRuntime`]: the watchdog reads and
/// writes the persisted `global.debug_logging_*` settings directly via its
/// `StateRepo` handle, so there is nothing else on `AppState` for another
/// caller to read.
#[derive(Default)]
pub struct DebugModeRuntime {
    /// The spawned watchdog task, behind `Option` so the shutdown drain can
    /// TAKE + await it by value; `None` once drained / never spawned.
    task: std::sync::Mutex<Option<JoinHandle<()>>>,
    /// The shutdown signal the watchdog `select!`s on.
    shutdown: std::sync::Mutex<Option<watch::Sender<bool>>>,
}

/// M9b (SPEC s16): the anonymous-telemetry runtime state held on [`AppState`].
///
/// `task` + `shutdown` track the single app-wide periodic-ping task so the quit
/// drain stops + joins it with no orphan (mirrors [`UpdaterRuntime`]). The ping
/// reads settings + aggregates on each tick, so no payload state lives here.
#[derive(Default)]
pub struct TelemetryRuntime {
    /// The spawned periodic-ping task, behind `Option` so the shutdown drain can
    /// TAKE + await it by value; `None` once drained / never spawned.
    task: std::sync::Mutex<Option<JoinHandle<()>>>,
    /// The shutdown signal the periodic-ping task `select!`s on, so it exits
    /// promptly on quit rather than waiting out its 24h interval.
    shutdown: std::sync::Mutex<Option<watch::Sender<bool>>>,
    /// M9b (P1-2): a cancellation flag flipped to `true` the instant
    /// `set_telemetry_enabled(false)` commits, so an IN-FLIGHT ping that is
    /// mid-build aborts BEFORE its network send (the disable is honored
    /// immediately, not merely on the next 24h tick). Re-armed to `false` on
    /// re-enable. Shared with the ping task via [`AppState::telemetry_cancel`].
    cancel: Arc<std::sync::atomic::AtomicBool>,
    /// M9b R3-P1-2: the SEND-ADMISSION gate shared between the ping path and the
    /// disable path. The ping ACQUIRES this lock, re-checks the cancel flag + pref
    /// UNDER it, and starts the network send while holding it; the disable path
    /// sets [`Self::cancel`] FIRST and then ACQUIRES the SAME gate to coordinate,
    /// so a disable can never be admitted concurrently with a send's final
    /// re-check, and a send that begins after the disable observed the gate sees
    /// the cancel flag already set and aborts. A `tokio::Mutex` so it can be held
    /// across the awaited send. Shared via [`AppState::telemetry_send_gate`].
    send_gate: Arc<Mutex<()>>,
    /// DESIGN s13: the app-global latency sampler shared into every account's
    /// executor + orchestrator (the SAME `Arc`), read at ping-build time for the
    /// scan / upload-per-MB percentiles. Default-ON; boot replaces it via
    /// [`AppState::install_telemetry_latency`] with one initialized from the
    /// persisted `telemetry.enabled` pref BEFORE any capture, and the enable/
    /// disable toggle flips it in lockstep with the pref
    /// (`telemetry::apply_enabled_change`). Shared via [`AppState::telemetry_latency`].
    latency: Arc<driven_core::telemetry::LatencyReservoir>,
}

/// M8 (P2-3): max number of TERMINAL restore-job records retained for late
/// polling. Active (non-terminal) jobs are never evicted by the cap; only
/// finished ones are pruned once this many accumulate (oldest-terminal first).
const MAX_RETAINED_TERMINAL_JOBS: usize = 32;

/// M8 (P2-3): how long a TERMINAL restore-job record is retained for a late
/// `get_restore_job` poll before it is eligible for pruning. Generous (the
/// webview reconciles right after a job ends) but bounded so the map cannot grow
/// without limit across many restores in one long-lived session.
const TERMINAL_JOB_TTL: Duration = Duration::from_secs(3600);

/// M8 (P1-1): the per-job cancellation control shared between the spawned restore
/// task and the IPC / shutdown paths. A plain [`AtomicBool`] checked between
/// frames in the stream loop (no extra dependency): set once, observed
/// monotonically. Cloned (`Arc`) into the spawned task.
pub type RestoreCancel = Arc<AtomicBool>;

/// M8: one tracked restore job - its latest status snapshot, the instant it
/// reached a terminal state (for TTL pruning, `None` while running), the shared
/// cancel flag, and the spawned task handle (taken on cancel / shutdown so the
/// drain can await it).
struct RestoreJobEntry {
    /// The latest status snapshot served by `get_restore_job`.
    status: crate::commands::dtos::RestoreJobStatus,
    /// When the job reached a terminal state, for TTL pruning; `None` while it
    /// is still running.
    terminal_at: Option<Instant>,
    /// The shared cancel flag the spawned task observes between frames.
    cancel: RestoreCancel,
    /// The spawned job task, behind `Option` so the shutdown drain can TAKE +
    /// await it by value; `None` once the job finished or was drained.
    handle: std::sync::Mutex<Option<JoinHandle<()>>>,
}

/// R2-P1-2: the shared per-account fake-remote-store registry. An `Arc` so
/// assembly (which builds the orchestrator's store BEFORE [`AppState`] exists)
/// and [`AppState`] hold the SAME map - the orchestrator's fake store and the
/// picker's fake store are then guaranteed to be the same instance per account.
pub type FakeRemoteStores = Arc<std::sync::Mutex<HashMap<AccountId, InMemoryRemoteStore>>>;

/// R2-P1-2: get-or-create the fake remote store for `account` in `registry`.
/// A free function (not a method) so assembly's pre-[`AppState`] boot phase -
/// which builds the orchestrator's store before `AppState` exists - shares the
/// SAME registry the picker later reads via [`AppState::fake_remote_store`].
/// Returns a clone (the store wraps a shared `Arc<Mutex>`).
#[must_use]
pub fn fake_remote_store_in(
    registry: &FakeRemoteStores,
    account: AccountId,
) -> InMemoryRemoteStore {
    registry
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .entry(account)
        .or_insert_with(|| new_fake_store(account))
        .clone()
}

/// Env var naming a JSON [`FaultPlan`](driven_drive::fake::fault_plan::FaultPlan)
/// file applied to every fake store this process creates (agent QA harness).
/// DOUBLY gated: it is only honoured when the fake remote is also selected
/// (`DRIVEN_USE_FAKE_REMOTE=1`), so the plan can never affect a real backend.
pub const ENV_TEST_FAULT_PLAN: &str = "DRIVEN_TEST_FAULT_PLAN";

/// Build a fresh fake store for the registry, arming the env-configured fault
/// plan when the harness gate is open (see [`resolve_fault_plan`]).
fn new_fake_store(account: AccountId) -> InMemoryRemoteStore {
    let store = InMemoryRemoteStore::new();
    let use_fake = std::env::var(crate::assembly::ENV_USE_FAKE_REMOTE)
        .map(|v| v == "1")
        .unwrap_or(false);
    let plan_path = std::env::var(ENV_TEST_FAULT_PLAN).ok();
    match resolve_fault_plan(use_fake, plan_path.as_deref()) {
        FaultPlanResolution::None => store,
        FaultPlanResolution::Armed(plan) => {
            tracing::info!(
                target: "driven::app::state",
                account_id = %account,
                ?plan,
                "agent QA harness: fault plan armed on the fake remote store"
            );
            plan.apply_to(store)
        }
        FaultPlanResolution::Error(msg) => {
            // Fail LOUDLY but not fatally: store creation is infallible by
            // contract (the picker + assembly paths cannot surface an error
            // here). The harness asserts plan-armed via this log line + the
            // fault actually firing, so a broken plan cannot silently pass.
            tracing::error!(
                target: "driven::app::state",
                account_id = %account,
                %msg,
                "agent QA harness: DRIVEN_TEST_FAULT_PLAN was set but NOT applied"
            );
            store
        }
    }
}

/// Outcome of resolving the env-configured fault plan (pure decision, so the
/// gating is unit-testable without touching process-global env vars).
#[derive(Debug, PartialEq)]
enum FaultPlanResolution {
    /// No plan configured (or the fake-remote gate is closed with no plan set).
    None,
    /// A plan parsed and ready to arm.
    Armed(driven_drive::fake::fault_plan::FaultPlan),
    /// The plan was requested but could not be honoured - surfaced as an
    /// ERROR log so a harness scenario can never silently run fault-free.
    Error(String),
}

/// Pure gating + parse for the [`ENV_TEST_FAULT_PLAN`] seam:
/// - no `plan_path`: nothing to do;
/// - `plan_path` set while the fake remote is NOT selected: an error (the
///   harness misconfigured the environment - the plan would never fire);
/// - otherwise read + parse the JSON plan, propagating read/parse failures.
fn resolve_fault_plan(use_fake_remote: bool, plan_path: Option<&str>) -> FaultPlanResolution {
    use driven_drive::fake::fault_plan::FaultPlan;
    let Some(path) = plan_path.filter(|p| !p.is_empty()) else {
        return FaultPlanResolution::None;
    };
    if !use_fake_remote {
        return FaultPlanResolution::Error(format!(
            "{ENV_TEST_FAULT_PLAN} is set ({path}) but DRIVEN_USE_FAKE_REMOTE=1 is not - \
             the fault plan only exists on the fake remote, so this configuration is a \
             harness bug"
        ));
    }
    let json = match std::fs::read_to_string(path) {
        Ok(j) => j,
        Err(e) => return FaultPlanResolution::Error(format!("failed to read {path}: {e}")),
    };
    match FaultPlan::from_json(&json) {
        Ok(plan) => FaultPlanResolution::Armed(plan),
        Err(e) => FaultPlanResolution::Error(format!("failed to parse {path}: {e}")),
    }
}

/// M8 (P2-3): prune retained TERMINAL restore-job records so the map cannot grow
/// unbounded across many restores in one long-lived tray session. Two bounds:
/// 1. TTL: a terminal job older than [`TERMINAL_JOB_TTL`] is dropped.
/// 2. count cap: if more than [`MAX_RETAINED_TERMINAL_JOBS`] terminal jobs
///    remain, the OLDEST-terminal ones are dropped down to the cap.
///
/// Active (non-terminal) jobs are NEVER pruned - only finished ones - so an
/// in-flight job's status + cancel handle always survive.
fn prune_terminal_jobs(map: &mut HashMap<String, RestoreJobEntry>) {
    let now = Instant::now();
    // 1) TTL: drop terminal jobs older than the retention window.
    map.retain(|_, e| match e.terminal_at {
        Some(t) => now.duration_since(t) < TERMINAL_JOB_TTL,
        None => true,
    });
    // 2) count cap: if too many terminal jobs remain, drop the oldest.
    let mut terminal: Vec<(String, Instant)> = map
        .iter()
        .filter_map(|(id, e)| e.terminal_at.map(|t| (id.clone(), t)))
        .collect();
    if terminal.len() > MAX_RETAINED_TERMINAL_JOBS {
        terminal.sort_by_key(|(_, t)| *t);
        let drop_n = terminal.len() - MAX_RETAINED_TERMINAL_JOBS;
        for (id, _) in terminal.into_iter().take(drop_n) {
            map.remove(&id);
        }
    }
}

/// C1: one backend-minted dialog-token binding - the path the user chose via a
/// native dialog plus the instant the binding expires (single-use TTL).
struct DialogTokenBinding {
    /// The path the native dialog returned (a folder for the folder dialog, a
    /// concrete file path for the save dialog).
    path: std::path::PathBuf,
    /// When the token stops being valid (mint time + [`DIALOG_TOKEN_TTL`]).
    /// Stored as the deadline rather than the mint instant so tests can force
    /// expiry without `Instant` subtraction (which can underflow soon after
    /// boot in CI).
    expires_at: std::time::Instant,
}

/// C1: how long a backend-minted dialog token stays valid.
///
/// This must cover the LONGEST honest gap between the native dialog and the
/// write command that spends the token - which is NOT "immediately": the
/// add-source wizard picks the local folder first and then walks the whole
/// tree for the exclusion preview, browses the destination, and waits on the
/// user to think, which on a big source is well over the 5 minutes this
/// originally allowed. An expired token used to surface as `local.io_error`
/// ("disk error"), sending users to check a healthy drive. An hour keeps the
/// replay window bounded (the token is also single-use and process-local)
/// without a live wizard session ever outrunning it.
const DIALOG_TOKEN_TTL: Duration = Duration::from_secs(3600);

impl AppState {
    /// Build the managed state from the state repo, the per-account handles,
    /// the remote-construction mode, and the shared fake-remote-store registry
    /// (called by `assembly::build_and_spawn`). The `fake_remote_stores` map is
    /// the SAME one assembly threaded into `build_remote`, so the orchestrator's
    /// fake store and the picker's fake store are one instance per account
    /// (R2-P1-2).
    #[must_use]
    pub fn new(
        state: Arc<dyn StateRepo>,
        accounts: HashMap<AccountId, AccountHandle>,
        remote_mode: RemoteMode,
        fake_remote_stores: FakeRemoteStores,
    ) -> Self {
        let accounts = accounts
            .into_iter()
            .map(|(id, handle)| (id, Arc::new(handle)))
            .collect();
        Self {
            state,
            accounts: std::sync::Mutex::new(accounts),
            remote_mode,
            pause_generations: std::sync::Mutex::new(HashMap::new()),
            dialog_tokens: std::sync::Mutex::new(HashMap::new()),
            ensure_master_key_locks: std::sync::Mutex::new(HashMap::new()),
            fake_remote_stores,
            restore_jobs: std::sync::Mutex::new(HashMap::new()),
            updater: UpdaterRuntime::default(),
            iostat: IostatRuntime::default(),
            bottleneck: BottleneckRuntime::default(),
            debug_mode: DebugModeRuntime::default(),
            exclusion_previews: Arc::default(),
            preview_tree_cache: Arc::default(),
            telemetry: TelemetryRuntime::default(),
            recovery_acks: std::sync::Mutex::new(HashMap::new()),
            vss_helper: std::sync::Mutex::new(None),
            apfs_helper: std::sync::Mutex::new(None),
        }
    }

    // --- issue #25: least-privilege VSS helper broker (DESIGN s5.3.1) --------

    /// Install the VSS helper broker lifecycle manager built by
    /// `assembly::build_and_spawn` (Windows + un-elevated + `windows.vss_helper`
    /// on). Called once at boot before `.manage(..)`; a subsequent call replaces
    /// it (defensive - boot installs exactly once).
    pub fn set_vss_helper_manager(&self, manager: Arc<crate::vss_helper::VssHelperManager>) {
        *self.vss_helper.lock().unwrap_or_else(|e| e.into_inner()) = Some(manager);
    }

    /// The installed VSS helper broker manager, if one is in play. `None` off
    /// Windows / when elevated / when the setting is off. Returns a cloned `Arc`
    /// so `get_vss_helper_status` can read liveness without holding the lock.
    #[must_use]
    pub fn vss_helper_manager(&self) -> Option<Arc<crate::vss_helper::VssHelperManager>> {
        self.vss_helper
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// Shut the VSS helper broker down at app quit (best-effort; a no-op if the
    /// broker was never launched or no manager is in play), so no elevated
    /// process outlives the app session. Mirrors the M9a/M9b task drains.
    pub fn shutdown_vss_helper(&self) {
        if let Some(manager) = self.vss_helper_manager() {
            manager.shutdown();
        }
    }

    // --- DESIGN s5.3.2: macOS APFS snapshot broker --------------------------

    /// Install the APFS broker lifecycle manager built by
    /// `assembly::build_and_spawn` (macOS only). Called once at boot before
    /// `.manage(..)`; a subsequent call replaces it.
    pub fn set_apfs_helper_manager(&self, manager: Arc<crate::apfs_helper::ApfsHelperManager>) {
        *self.apfs_helper.lock().unwrap_or_else(|e| e.into_inner()) = Some(manager);
    }

    /// The installed APFS broker manager, if one is in play. `None` off macOS.
    /// Returns a cloned `Arc` so `get_apfs_helper_status` can read liveness
    /// without holding the lock.
    #[must_use]
    pub fn apfs_helper_manager(&self) -> Option<Arc<crate::apfs_helper::ApfsHelperManager>> {
        self.apfs_helper
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// Shut the APFS broker down at app quit (best-effort; a no-op when no
    /// manager is in play), so no root process - and no mounted snapshot -
    /// outlives the app session. The broker also self-terminates when the app
    /// pid it was launched with exits, so this is belt-and-braces.
    pub fn shutdown_apfs_helper(&self) {
        if let Some(manager) = self.apfs_helper_manager() {
            manager.shutdown();
        }
    }

    /// Shut the VSS helper broker down BEFORE an in-app update installer runs
    /// (issue #125), returning whether a manager was disabled (so the caller can
    /// RE-ARM it if the install then fails - see
    /// [`Self::rearm_vss_helper_after_failed_update`]).
    ///
    /// The Windows NSIS updater overwrites the bundled `driven-vss-helper.exe`
    /// sidecar, but its stock process-kill only targets the MAIN binary
    /// (`driven-app.exe`) - never the sidecar. A running elevated broker holds an
    /// open handle to its own exe, so the install fails with "Error opening file
    /// for writing: ...driven-vss-helper.exe". `download_and_install` runs the
    /// NSIS installer synchronously (`/P /R`), so the broker must be gone BEFORE
    /// that call.
    ///
    /// Unlike the app-quit [`Self::shutdown_vss_helper`] (a bare `shutdown()`),
    /// this uses `set_enabled(false)`, which is a SUPERSET: it performs the same
    /// Shutdown+reap (including abandoning + reaping a `Pending` launch per the
    /// #113 generation semantics) AND disables the manager so a still-running
    /// sync that hits a locked file cannot RE-LAUNCH the elevated broker (and
    /// re-lock the exe) during the potentially-long `download_and_install`
    /// window. A memoised session decline is reset to `NotAttempted` by the
    /// underlying shutdown; that is inherent to any shutdown-based sweep and
    /// harmless here (an update is user-consented and a successful install
    /// restarts the app anyway).
    ///
    /// Best-effort + idempotent: a no-op (returns `false`) when no manager is in
    /// play (off Windows / elevated / setting off).
    pub fn shutdown_vss_helper_for_update(&self) -> bool {
        if let Some(manager) = self.vss_helper_manager() {
            manager.set_enabled(false);
            true
        } else {
            false
        }
    }

    /// Re-arm the VSS helper broker after a FAILED update install (issue #125):
    /// the app keeps running, so undo the
    /// [`Self::shutdown_vss_helper_for_update`] disable so locked-file backup is
    /// available again on demand (a LAZY re-launch on the next locked file - no
    /// forced UAC prompt), rather than staying silently degraded until the next
    /// app restart. Best-effort + idempotent; a no-op when no manager is in play.
    pub fn rearm_vss_helper_after_failed_update(&self) {
        if let Some(manager) = self.vss_helper_manager() {
            manager.set_enabled(true);
        }
    }

    // --- M9c D4: recovery-phrase ACK gate (M6 R4-P1-1, DATA-SAFETY) ---------

    /// Lock the recovery-ack map, recovering a poisoned lock (house rule: never
    /// panic on a poisoned lock).
    fn lock_recovery_acks(
        &self,
    ) -> std::sync::MutexGuard<'_, HashMap<driven_core::types::SourceId, RecoveryAckState>> {
        self.recovery_acks.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// R4-P1-1: reconstruct the in-memory recovery-ack mirror from the DURABLE
    /// `recovery_phrase_acks` table on startup, so a process that restarts mid
    /// onboarding (after the first encrypted source + master key were persisted but
    /// before reveal+ack) resumes the EXACT pending-ack gate - the disabled source
    /// is still reveal/ackable and no second encrypted source can enable without the
    /// durable ack. Called once after assembly builds [`AppState`]. Errors reading
    /// the table are logged and treated as "no pending acks" (the durable table is
    /// still the gate source of truth for the command layer; the mirror is a
    /// convenience), but a healthy boot always succeeds.
    pub async fn reconstruct_recovery_acks_from_db(&self) {
        match self.state.list_pending_recovery_acks().await {
            Ok(rows) => {
                let mut map = self.lock_recovery_acks();
                map.clear();
                for r in rows {
                    map.insert(
                        r.source_id,
                        RecoveryAckState {
                            account_id: r.account_id,
                            revealed: r.revealed,
                        },
                    );
                }
                tracing::info!(
                    pending_recovery_acks = map.len(),
                    "reconstructed durable recovery-phrase ack gate from SQLite (R4-P1-1)"
                );
            }
            Err(err) => {
                tracing::error!(%err, "failed to reconstruct recovery-phrase ack gate from SQLite; the durable table still gates the commands");
            }
        }
    }

    /// M9c D4 / R4-P1-1: register in the in-memory MIRROR that `source` (on
    /// `account`) was persisted DISABLED and is awaiting a recovery-phrase ack. The
    /// DURABLE record is written atomically with the source insert by
    /// `add_source` (via [`StateRepo::insert_first_encrypted_source_pending_ack`]);
    /// this only mirrors that durable state into the in-memory map. Until the ack
    /// lands the source stays disabled, so the scheduler + manual sync (which filter
    /// on `enabled`) never back it up.
    pub fn register_pending_recovery_ack(
        &self,
        source: driven_core::types::SourceId,
        account: AccountId,
    ) {
        self.lock_recovery_acks().insert(
            source,
            RecoveryAckState {
                account_id: account,
                revealed: false,
            },
        );
    }

    /// M9c D4: record that the backend actually REVEALED the phrase for `source`
    /// (`reveal_recovery_phrase`). Returns `true` if `source` had a pending ack
    /// (so the reveal is meaningful), `false` for an unknown / already-acked source.
    /// The ack is only accepted after this has been recorded.
    pub fn record_recovery_reveal(&self, source: driven_core::types::SourceId) -> bool {
        match self.lock_recovery_acks().get_mut(&source) {
            Some(entry) => {
                entry.revealed = true;
                true
            }
            None => false,
        }
    }

    /// M9c D4: whether `source` has a pending recovery ack AND its phrase has been
    /// revealed by the backend - the precondition `ack_recovery_phrase_saved`
    /// enforces. `None` means no pending ack at all; `Some(false)` means pending but
    /// the phrase was never revealed (the ack must be rejected); `Some(true)` means
    /// the ack may proceed.
    #[must_use]
    pub fn recovery_reveal_recorded(&self, source: driven_core::types::SourceId) -> Option<bool> {
        self.lock_recovery_acks().get(&source).map(|e| e.revealed)
    }

    /// M9c D4: the account owning `source`'s pending recovery ack, if one exists.
    #[must_use]
    pub fn pending_recovery_ack_account(
        &self,
        source: driven_core::types::SourceId,
    ) -> Option<AccountId> {
        self.lock_recovery_acks().get(&source).map(|e| e.account_id)
    }

    /// M9c D4: clear the pending recovery ack for `source` once it has been enabled
    /// (the ack succeeded). Idempotent - clearing an unknown source is a no-op.
    pub fn clear_pending_recovery_ack(&self, source: driven_core::types::SourceId) {
        self.lock_recovery_acks().remove(&source);
    }

    // --- M9a updater runtime (SPEC s15.2) ----------------------------------

    /// M9a: record the [`tauri_plugin_updater::Update`] a check found PLUS the
    /// channel string it came from, so a subsequent `install_update` stages +
    /// applies the SAME object without re-resolving the manifest AND emits the
    /// REAL channel on `updater:downloaded` (R1-P2-3). Overwrites any prior
    /// pending update (a newer check supersedes an older one). `None` clears it.
    pub fn set_pending_update(&self, update: Option<(tauri_plugin_updater::Update, String)>) {
        *self
            .updater
            .pending
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = update;
    }

    /// M9a: TAKE (single-use) the pending update + its channel for installation.
    /// `None` when no check has found an update (so `install_update` returns a
    /// clear "nothing to install" error rather than guessing). On a failed
    /// install the caller RESTORES it via [`Self::set_pending_update`] so the
    /// banner's next Install retries (R1-P2-2).
    #[must_use]
    pub fn take_pending_update(&self) -> Option<(tauri_plugin_updater::Update, String)> {
        self.updater
            .pending
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    /// M9a (R2-P1-3): PEEK (non-consuming) the pending update as an owned
    /// snapshot, so the `get_pending_update_info` IPC can hydrate the webview's
    /// updater store on startup WITHOUT taking the pending update (install still
    /// needs it). Returns `(version, notes, published_at_rfc3339, channel)` while
    /// holding the lock briefly; `None` when no check has recorded an update.
    /// Kept as owned primitives so `AppState` stays decoupled from the updater
    /// DTO mapping (updater.rs builds the `UpdateInfo`).
    #[must_use]
    pub fn peek_pending_update(&self) -> Option<(String, Option<String>, Option<String>, String)> {
        self.updater
            .pending
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .as_ref()
            .map(|(update, channel)| {
                (
                    update.version.clone(),
                    update.body.clone().filter(|b| !b.is_empty()),
                    update.date.map(|d| d.to_string()),
                    channel.clone(),
                )
            })
    }

    /// M9a: register the spawned periodic-check task + its shutdown sender so the
    /// app-quit drain can stop + join it with no orphan. Called once after the
    /// updater task is spawned in `lib.rs` setup.
    pub fn set_updater_task(&self, task: JoinHandle<()>, shutdown: watch::Sender<bool>) {
        *self.updater.task.lock().unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self
            .updater
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(shutdown);
    }

    /// M9a: signal the periodic-check task to stop and TAKE its handle so the
    /// caller can await it (the app-quit drain). Returns `None` if no task is
    /// tracked (never spawned / already drained). Mirrors the M8 restore-job
    /// no-orphan take pattern.
    #[must_use]
    pub fn shutdown_updater_task(&self) -> Option<JoinHandle<()>> {
        if let Some(tx) = self
            .updater
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            // A send error only means the receiver is already gone - benign.
            let _ = tx.send(true);
        }
        self.updater
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    // --- 2026-08-14 follow-up: io throughput runtime ------------------------

    /// The live-throughput hub (counters + trailing ring) the sampler and the
    /// `io_throughput_series` command read.
    pub fn iostat_hub(&self) -> Arc<crate::iostat_hub::IoStatHub> {
        self.iostat
            .hub
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// Install the assembly-built hub (the one whose counters the executors
    /// credit). Called once at boot before `.manage(..)`; the quiesced boot
    /// path skips it and keeps the default (all-zero) hub.
    pub fn install_iostat_hub(&self, hub: Arc<crate::iostat_hub::IoStatHub>) {
        *self.iostat.hub.lock().unwrap_or_else(|e| e.into_inner()) = hub;
    }

    /// Register the spawned sampler task + its shutdown sender so the app-quit
    /// drain can stop + join it with no orphan (mirrors the updater task).
    pub fn set_iostat_task(&self, task: JoinHandle<()>, shutdown: watch::Sender<bool>) {
        *self.iostat.task.lock().unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self
            .iostat
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(shutdown);
    }

    /// Signal the sampler to stop and TAKE its handle so the quit drain can
    /// await it. Mirrors [`Self::shutdown_updater_task`].
    #[must_use]
    pub fn shutdown_iostat_task(&self) -> Option<JoinHandle<()>> {
        if let Some(tx) = self
            .iostat
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            let _ = tx.send(true);
        }
        self.iostat
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    // --- issue #308: bottleneck-classification runtime -----------------------

    /// The live bottleneck-classification hub the sampler pushes into and the
    /// `bottleneck_status` command reads. Always available (no "install"
    /// step needed - it reads `AppState` directly each tick).
    pub fn bottleneck_hub(&self) -> Arc<crate::bottleneck_hub::BottleneckHub> {
        self.bottleneck.hub.clone()
    }

    /// Register the spawned sampler task + its shutdown sender so the app-quit
    /// drain can stop + join it with no orphan (mirrors [`Self::set_iostat_task`]).
    pub fn set_bottleneck_task(&self, task: JoinHandle<()>, shutdown: watch::Sender<bool>) {
        *self
            .bottleneck
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self
            .bottleneck
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(shutdown);
    }

    /// Signal the sampler to stop and TAKE its handle so the quit drain can
    /// await it. Mirrors [`Self::shutdown_iostat_task`].
    #[must_use]
    pub fn shutdown_bottleneck_task(&self) -> Option<JoinHandle<()>> {
        if let Some(tx) = self
            .bottleneck
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            let _ = tx.send(true);
        }
        self.bottleneck
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    // --- issue #309: debug-logging-mode watchdog runtime ----------------------

    /// Register the spawned watchdog task + its shutdown sender so the
    /// app-quit drain can stop + join it with no orphan (mirrors
    /// [`Self::set_bottleneck_task`]).
    pub fn set_debug_mode_task(&self, task: JoinHandle<()>, shutdown: watch::Sender<bool>) {
        *self
            .debug_mode
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self
            .debug_mode
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(shutdown);
    }

    /// Signal the watchdog to stop and TAKE its handle so the quit drain can
    /// await it. Mirrors [`Self::shutdown_bottleneck_task`].
    #[must_use]
    pub fn shutdown_debug_mode_task(&self) -> Option<JoinHandle<()>> {
        if let Some(tx) = self
            .debug_mode
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            let _ = tx.send(true);
        }
        self.debug_mode
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    // --- M9b telemetry runtime (SPEC s16) ----------------------------------

    /// M9b: register the spawned periodic-ping task + its shutdown sender so the
    /// app-quit drain can stop + join it with no orphan. Called once after the
    /// telemetry task is spawned in `lib.rs` setup (mirrors the M9a updater task).
    pub fn set_telemetry_task(&self, task: JoinHandle<()>, shutdown: watch::Sender<bool>) {
        *self
            .telemetry
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self
            .telemetry
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(shutdown);
    }

    /// M9b (P1-2): the shared telemetry cancellation flag. The ping task clones
    /// this `Arc` and checks it right before its network send; the IPC toggle flips
    /// it via [`set_telemetry_cancelled`] so a disable aborts an in-flight ping
    /// immediately.
    #[must_use]
    pub fn telemetry_cancel(&self) -> Arc<std::sync::atomic::AtomicBool> {
        Arc::clone(&self.telemetry.cancel)
    }

    /// M9b (P1-2): set the telemetry cancellation flag. `set_telemetry_enabled`
    /// flips it to `true` when disabling (abort any in-flight send immediately) and
    /// back to `false` when re-enabling (re-arm the ping path).
    pub fn set_telemetry_cancelled(&self, cancelled: bool) {
        self.telemetry
            .cancel
            .store(cancelled, std::sync::atomic::Ordering::SeqCst);
    }

    /// M9b R3-P1-2: the shared SEND-ADMISSION gate. The ping path holds it across
    /// its final cancel/pref re-check + network send; the disable path acquires it
    /// (after setting cancel) so a disable is never admitted concurrently with a
    /// send and any send starting after the disable sees the cancel flag set.
    #[must_use]
    pub fn telemetry_send_gate(&self) -> Arc<Mutex<()>> {
        Arc::clone(&self.telemetry.send_gate)
    }

    /// DESIGN s13: the app-global latency reservoir. The ping task snapshots it at
    /// build time (and resets it after a successful send); the enable/disable
    /// toggle flips its capture gate via `telemetry::apply_enabled_change`. Returns
    /// a cloned `Arc` - the SAME instance every account's executor + orchestrator
    /// records into.
    #[must_use]
    pub fn telemetry_latency(&self) -> Arc<driven_core::telemetry::LatencyReservoir> {
        Arc::clone(&self.telemetry.latency)
    }

    /// DESIGN s13: install the boot-built latency reservoir (initialized from the
    /// persisted `telemetry.enabled` pref) so [`AppState`] shares the SAME `Arc`
    /// the executors + orchestrators were wired with in assembly. Called once
    /// before `.manage(..)`, replacing the default-ON placeholder.
    pub fn install_telemetry_latency(
        &mut self,
        latency: Arc<driven_core::telemetry::LatencyReservoir>,
    ) {
        self.telemetry.latency = latency;
    }

    /// M9b: signal the periodic-ping task to stop and TAKE its handle so the
    /// caller can await it (the app-quit drain). Returns `None` if no task is
    /// tracked (never spawned / already drained). Mirrors `shutdown_updater_task`.
    #[must_use]
    pub fn shutdown_telemetry_task(&self) -> Option<JoinHandle<()>> {
        if let Some(tx) = self
            .telemetry
            .shutdown
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            // A send error only means the receiver is already gone - benign.
            let _ = tx.send(true);
        }
        self.telemetry
            .task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
    }

    /// Lock the restore-jobs map, recovering a poisoned lock (house rule: never
    /// panic on a poisoned lock).
    fn lock_restore_jobs(&self) -> std::sync::MutexGuard<'_, HashMap<String, RestoreJobEntry>> {
        self.restore_jobs.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// M8 (P1-1; R3-P1-2): SEED a restore job - its initial status snapshot, shared
    /// cancel flag, AND its already-spawned task [`JoinHandle`] - in ONE locked
    /// insert. Registering the handle ATOMICALLY with the seed (rather than seeding
    /// first and attaching the handle in a separate call) closes the R3-P1-2 race:
    /// `cancel_all_restore_jobs` (app quit) can NEVER observe a seeded restore job
    /// that lacks an awaitable handle. The command spawns the task behind a START
    /// BARRIER (the task does no filesystem work until released), seeds it here with
    /// its handle, THEN releases the barrier - so a quit anywhere in the spawn
    /// window finds a drainable handle and the task creates no partial temp.
    /// `get_restore_job` can serve the job immediately and `cancel_restore_job`
    /// works the moment the job is seeded. Also opportunistically prunes terminal
    /// jobs (P2-3).
    pub fn seed_restore_job(
        &self,
        status: crate::commands::dtos::RestoreJobStatus,
        cancel: RestoreCancel,
        handle: JoinHandle<()>,
    ) {
        let mut map = self.lock_restore_jobs();
        map.insert(
            status.job_id.clone(),
            RestoreJobEntry {
                status,
                terminal_at: None,
                cancel,
                handle: std::sync::Mutex::new(Some(handle)),
            },
        );
        prune_terminal_jobs(&mut map);
    }

    /// M8: record the latest status snapshot for a restore job (the background
    /// task calls this on every progress tick). `get_restore_job` reads it back.
    /// A terminal (`done`) snapshot stamps the terminal time so the entry becomes
    /// eligible for TTL pruning (P2-3). Updating an unknown id (e.g. a job whose
    /// terminal record was already pruned) is a benign no-op.
    pub fn put_restore_job(&self, status: crate::commands::dtos::RestoreJobStatus) {
        let mut map = self.lock_restore_jobs();
        let done = status.done;
        if let Some(entry) = map.get_mut(&status.job_id) {
            entry.status = status;
            if done && entry.terminal_at.is_none() {
                entry.terminal_at = Some(Instant::now());
            }
        }
        prune_terminal_jobs(&mut map);
    }

    /// M8: the current status snapshot for `job_id`, if the job exists (live or
    /// terminal). `None` for an unknown / forged id so `get_restore_job` surfaces
    /// an error rather than fabricating a status.
    #[must_use]
    pub fn restore_job(&self, job_id: &str) -> Option<crate::commands::dtos::RestoreJobStatus> {
        self.lock_restore_jobs()
            .get(job_id)
            .map(|e| e.status.clone())
    }

    /// M8 (R1-P1-2): request cancellation of a running restore job FROM THE UI
    /// WITHOUT detaching it. Sets the shared cancel flag (the spawned task observes
    /// it between frames, deletes any in-flight temp, and emits a terminal
    /// CANCELLED status) but LEAVES the task handle tracked on the job entry, so
    /// the M5-style app-shutdown drain ([`Self::cancel_all_restore_jobs`]) still
    /// awaits/aborts it - a UI cancel never orphans the task. The task clears its
    /// own handle on exit via [`Self::finish_restore_job_handle`]. Returns `true`
    /// if a tracked job's flag was set, `false` for an unknown / already-finished
    /// id (cancellation is idempotent).
    pub fn signal_cancel_restore_job(&self, job_id: &str) -> bool {
        let map = self.lock_restore_jobs();
        match map.get(job_id) {
            Some(entry) => {
                entry.cancel.store(true, Ordering::SeqCst);
                true
            }
            None => false,
        }
    }

    /// M8 (P1-1): request cancellation of a running restore job AND TAKE its task
    /// handle. Sets the shared cancel flag (the spawned task observes it between
    /// frames, deletes any in-flight temp, and emits a terminal CANCELLED status)
    /// and returns the spawned [`JoinHandle`] if the job is still tracked +
    /// running, so the CALLER can await it. `None` for an unknown / already-
    /// finished id - cancellation is idempotent (a second call, or a call after
    /// the job ended, is a no-op).
    ///
    /// R1-P1-2: this is reserved for callers that take ownership of draining the
    /// handle (e.g. a dedicated shutdown sweep). The UI cancel path must use
    /// [`Self::signal_cancel_restore_job`] instead so the handle stays tracked for
    /// the shutdown drain - taking + dropping the handle would DETACH the task.
    #[must_use]
    pub fn cancel_restore_job(&self, job_id: &str) -> Option<JoinHandle<()>> {
        let map = self.lock_restore_jobs();
        let entry = map.get(job_id)?;
        entry.cancel.store(true, Ordering::SeqCst);
        let handle = entry
            .handle
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take();
        handle
    }

    /// M8 (P1-1): mark a job's handle as drained (the spawned task finished and
    /// took its own handle). Called by the job task on its way out so the shutdown
    /// drain does not try to await an already-finished handle.
    pub fn finish_restore_job_handle(&self, job_id: &str) {
        if let Some(entry) = self.lock_restore_jobs().get(job_id) {
            let _ = entry
                .handle
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .take();
        }
    }

    /// Test-only: number of restore-job records currently tracked (live +
    /// terminal). R2-P2-1 uses this to assert that a `restore_files` whose fallible
    /// setup fails leaves NO lingering job entry.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn restore_jobs_len(&self) -> usize {
        self.lock_restore_jobs().len()
    }

    /// M8 (P1-1): signal-cancel EVERY in-flight restore job and TAKE their task
    /// handles, so the app-shutdown drain can await them (mirrors the M5 no-orphan
    /// AccountHandle drain). Sets each job's cancel flag, so each task deletes its
    /// in-flight temp + emits a terminal CANCELLED status before returning, then
    /// returns the handles for the caller to `join` so quit leaves no orphaned
    /// restore task and no partial files.
    #[must_use]
    pub fn cancel_all_restore_jobs(&self) -> Vec<JoinHandle<()>> {
        let map = self.lock_restore_jobs();
        let mut handles = Vec::new();
        for entry in map.values() {
            entry.cancel.store(true, Ordering::SeqCst);
            if let Some(h) = entry
                .handle
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .take()
            {
                handles.push(h);
            }
        }
        handles
    }

    /// C1: mint a one-shot dialog token bound to `path` (the path a native
    /// backend dialog returned), returning the opaque token string the webview
    /// threads back into the write command. Also opportunistically evicts
    /// expired bindings so the map cannot grow unbounded.
    pub fn mint_dialog_token(&self, path: std::path::PathBuf) -> String {
        let token = uuid::Uuid::new_v4().to_string();
        let mut map = self.lock_dialog_tokens();
        let now = std::time::Instant::now();
        map.retain(|_, b| now < b.expires_at);
        map.insert(
            token.clone(),
            DialogTokenBinding {
                path,
                expires_at: now + DIALOG_TOKEN_TTL,
            },
        );
        token
    }

    /// C1: TAKE (single-use) the path bound to `token`, if it exists and has not
    /// expired. Returns `None` for an unknown / already-consumed / expired token
    /// so a path-bearing command can REJECT a path the user did not pick via a
    /// backend dialog. Consuming the token here makes it single-use.
    pub fn take_dialog_token(&self, token: &str) -> Option<std::path::PathBuf> {
        let mut map = self.lock_dialog_tokens();
        let binding = map.remove(token)?;
        if std::time::Instant::now() >= binding.expires_at {
            return None;
        }
        Some(binding.path)
    }

    /// C1 / R1-P1-2: PEEK (non-consuming) the path bound to `token`, if it
    /// exists and has not expired. Unlike [`Self::take_dialog_token`] this does
    /// NOT consume the token, so a read-only, idempotent, repeatable command
    /// (`preview_exclusions`, which the user re-runs as they tweak globs) can
    /// resolve the dialog-derived path without spending the single use the
    /// subsequent `add_source` write needs. The TTL still bounds replay; only a
    /// path-bearing WRITE consumes the token.
    pub fn peek_dialog_token(&self, token: &str) -> Option<std::path::PathBuf> {
        let map = self.lock_dialog_tokens();
        let binding = map.get(token)?;
        if std::time::Instant::now() >= binding.expires_at {
            return None;
        }
        Some(binding.path.clone())
    }

    /// Test-only: force `token`'s binding past its TTL so expiry behaviour can
    /// be exercised without sleeping or `Instant` arithmetic that underflows
    /// soon after boot.
    #[cfg(test)]
    pub fn expire_dialog_token_for_test(&self, token: &str) {
        let mut map = self.lock_dialog_tokens();
        if let Some(b) = map.get_mut(token) {
            b.expires_at = std::time::Instant::now();
        }
    }

    /// The streaming-exclusion-preview registry (see the
    /// [`Self::exclusion_previews`] field docs). Returns an owned `Arc` so the
    /// spawned blocking walk can deregister itself after the command that
    /// started it has returned.
    pub fn exclusion_previews(&self) -> Arc<crate::commands::exclusion_stream::PreviewRegistry> {
        Arc::clone(&self.exclusion_previews)
    }

    /// The exclusion editor's folder-tree cache (see the
    /// [`Self::preview_tree_cache`] field docs). Returns an owned `Arc` so the
    /// spawned blocking pass can keep filling it after the command that started
    /// it has returned.
    pub fn preview_tree_cache(&self) -> Arc<crate::commands::preview_cache::PreviewTreeCache> {
        Arc::clone(&self.preview_tree_cache)
    }

    /// Lock the dialog-token map, recovering a poisoned lock.
    fn lock_dialog_tokens(&self) -> std::sync::MutexGuard<'_, HashMap<String, DialogTokenBinding>> {
        self.dialog_tokens.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// Lock the accounts map, recovering a poisoned lock (house rule: never
    /// panic on a poisoned lock).
    fn lock_accounts(&self) -> std::sync::MutexGuard<'_, HashMap<AccountId, Arc<AccountHandle>>> {
        self.accounts.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// A2: insert (or replace) the live handle for `account`, returning the
    /// PRIOR handle if one was present (so the caller can shut it down to avoid
    /// orphaning its tasks - mirrors the M5 no-orphan bookkeeping). Called by
    /// the wizard hot-spawn path after `finish_add_account`.
    pub fn insert_account(
        &self,
        account: AccountId,
        handle: AccountHandle,
    ) -> Option<Arc<AccountHandle>> {
        self.lock_accounts().insert(account, Arc::new(handle))
    }

    /// A2: remove + return the live handle for `account` (so the caller can
    /// shut it down before deleting the account's rows). `None` if the account
    /// has no running orchestrator (never spawned / needs_reauth).
    pub fn remove_account_handle(&self, account: AccountId) -> Option<Arc<AccountHandle>> {
        self.lock_accounts().remove(&account)
    }

    /// Bump and return the new pause generation for `account` (C5-P2-1). Called
    /// on every pause/resume so any in-flight timed-resume timer for that
    /// account is superseded.
    #[must_use]
    pub fn bump_pause_generation(&self, account: AccountId) -> u64 {
        let mut map = self
            .pause_generations
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let gen = map.entry(account).or_insert(0);
        *gen = gen.wrapping_add(1);
        *gen
    }

    /// `true` if `account`'s current pause generation still equals `token`
    /// (C5-P2-1): the timed-resume timer auto-resumes ONLY when this holds, so a
    /// newer pause/resume that bumped the generation cancels the stale timer.
    #[must_use]
    pub fn pause_generation_matches(&self, account: AccountId, token: u64) -> bool {
        let map = self
            .pause_generations
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        map.get(&account).copied() == Some(token)
    }

    /// Borrow the shared [`StateRepo`] (SPEC s2).
    #[must_use]
    pub fn state(&self) -> &Arc<dyn StateRepo> {
        &self.state
    }

    /// The orchestrator handle for `account`, if present. Returns a cloned
    /// [`Arc`] so the caller can drive / await it after the map lock is
    /// released (A2: the map is now behind a sync mutex).
    #[must_use]
    pub fn account(&self, account: AccountId) -> Option<Arc<AccountHandle>> {
        self.lock_accounts().get(&account).cloned()
    }

    /// Snapshot every account's orchestrator handle (for "all accounts"
    /// commands like `sync_now(None)` / `pause_sync`). Returns owned
    /// `(AccountId, Arc<AccountHandle>)` pairs so callers can iterate + await
    /// without holding the map lock across an await (A2).
    #[must_use]
    pub fn accounts(&self) -> Vec<(AccountId, Arc<AccountHandle>)> {
        self.lock_accounts()
            .iter()
            .map(|(id, handle)| (*id, handle.clone()))
            .collect()
    }

    /// How remotes were constructed this run.
    #[must_use]
    pub fn remote_mode(&self) -> RemoteMode {
        self.remote_mode
    }

    /// R2-P1-1: the per-account ensure-master-key async lock (get-or-create).
    /// `add_source` holds this across the ENTIRE first-encrypted-source critical
    /// section (prepare master key -> stamp -> insert) so two concurrent adds
    /// serialise and the second wraps under the master key the first installed.
    /// Returns a cloned `Arc` so the caller can `.lock().await` after releasing
    /// the (sync) map lock.
    #[must_use]
    pub fn ensure_master_key_lock(&self, account: AccountId) -> Arc<Mutex<()>> {
        self.ensure_master_key_locks
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .entry(account)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    }

    /// R2-P1-2: the per-account fake remote store (get-or-create). Shared
    /// between the Drive-folder picker and the orchestrator's uploader so a
    /// folder id minted by the picker in fake mode is visible to the uploader.
    /// Returns a clone (the store wraps a shared `Arc<Mutex>`, so the clone sees
    /// the same backing objects). Only meaningful in [`RemoteMode::Fake`].
    #[must_use]
    pub fn fake_remote_store(&self, account: AccountId) -> InMemoryRemoteStore {
        fake_remote_store_in(&self.fake_remote_stores, account)
    }

    /// R2-P1-2: a clone of the shared fake-remote-store registry handle, so the
    /// wizard hot-spawn path (`assembly::spawn_account`) builds the new account's
    /// orchestrator store FROM the same registry the picker reads.
    #[must_use]
    pub fn fake_remote_stores(&self) -> FakeRemoteStores {
        self.fake_remote_stores.clone()
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use driven_core::orchestrator::{Orchestrator, OrchestratorConfig, TickSource};
    use driven_core::types::OrchestratorState;
    use std::sync::atomic::{AtomicBool, Ordering};

    /// A no-source crypto provider for the shutdown tests (B2: `AccountTasks`
    /// now carries the live crypto provider; these tests only exercise the
    /// task-drain bookkeeping, so an empty provider suffices).
    fn test_crypto() -> Arc<crate::crypto_provider_impl::KeystoreCryptoProvider> {
        Arc::new(crate::crypto_provider_impl::KeystoreCryptoProvider::new(
            AccountId::new_v4(),
            Vec::new(),
        ))
    }

    /// A no-op [`Orchestrator`] whose `run()` returns as soon as `shutdown()` is
    /// signalled - mirroring the real run loop's graceful-drain contract, so the
    /// R-P1-1 shutdown test exercises the clean-join path for the run loop while
    /// the poller (which loops forever) exercises the abort-and-await path.
    ///
    /// `pub(crate)` (banner Task 3, spec 2026-08-01): the `sync_now` bypass-
    /// ordering test in `commands::sync` reuses this fake rather than
    /// duplicating a second one, recording `trigger`/`bypass_gates_once` calls
    /// via [`Self::calls`] so that test can assert the bypass is armed BEFORE
    /// the manual trigger.
    pub(crate) struct FakeOrchestrator {
        shutdown: watch::Sender<bool>,
        shutdown_rx: watch::Receiver<bool>,
        calls: std::sync::Mutex<Vec<&'static str>>,
    }

    impl FakeOrchestrator {
        pub(crate) fn new() -> Self {
            let (shutdown, shutdown_rx) = watch::channel(false);
            Self {
                shutdown,
                shutdown_rx,
                calls: std::sync::Mutex::new(Vec::new()),
            }
        }

        /// The `trigger`/`bypass_gates_once` calls this fake observed, in
        /// invocation order (banner Task 3).
        pub(crate) fn calls(&self) -> Vec<&'static str> {
            self.calls.lock().unwrap_or_else(|e| e.into_inner()).clone()
        }
    }

    #[async_trait::async_trait]
    impl Orchestrator for FakeOrchestrator {
        async fn run(&self) -> anyhow::Result<()> {
            let mut rx = self.shutdown_rx.clone();
            loop {
                if *rx.borrow() {
                    return Ok(());
                }
                if rx.changed().await.is_err() {
                    return Ok(());
                }
            }
        }
        async fn trigger(&self, _reason: TickSource) {
            self.calls
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .push("trigger");
        }
        async fn set_paused(&self, _paused: bool) {}
        fn bypass_gates_once(&self) {
            self.calls
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .push("bypass_gates_once");
        }
        async fn state(&self) -> OrchestratorState {
            OrchestratorState::Idle { last_run_at: None }
        }
        async fn reconfigure(&self, _config: OrchestratorConfig) {}
        fn shutdown(&self) {
            let _ = self.shutdown.send(true);
        }
    }

    /// Build an [`AccountHandle`] around `orchestrator` with trivial (already-
    /// finished) task handles, for command-layer tests that only need a real
    /// `AccountHandle` present in `AppState::accounts()` (banner Task 3: the
    /// `sync_now` bypass-ordering test). These tests never exercise
    /// `AccountHandle::shutdown`'s task-draining, so the task handles need not
    /// do anything.
    pub(crate) fn build_fake_account_handle(orchestrator: Arc<dyn Orchestrator>) -> AccountHandle {
        let (bridge_shutdown, _rx) = watch::channel(false);
        AccountHandle::new(
            orchestrator,
            AccountTasks {
                crypto: test_crypto(),
                run_loop: tokio::spawn(async {}),
                watcher_bridge: None,
                event_bridge: tokio::spawn(async {}),
                power_poller: tokio::spawn(async {}),
                sleep_wake_monitor: None,
                bridge_shutdown,
            },
        )
    }

    #[tokio::test]
    async fn shutdown_joins_every_per_account_task_no_orphans() {
        // R-P1-1: `AccountHandle::shutdown` must leave ZERO orphaned tasks - the
        // run loop, watcher bridge, event bridge, AND the forever-looping power
        // poller must all be finished (joined or aborted-and-awaited) when it
        // returns. This is the M5 "no orphaned tokio tasks" acceptance, modelled
        // on the EXACT task shapes assembly spawns.
        let orchestrator: Arc<dyn Orchestrator> = Arc::new(FakeOrchestrator::new());

        // The bridge shutdown signal both bridges select! on (the watcher bridge
        // never closes on its own; the poller loops forever).
        let (bridge_shutdown, _rx0) = watch::channel(false);

        // Watcher bridge: an mpsc whose Sender we KEEP (modelling NotifyWatcher
        // owning the sender), so recv() never returns None on its own - the
        // bridge can only end via the shutdown signal.
        let (_watch_tx, mut watch_rx) = tokio::sync::mpsc::channel::<u32>(4);
        let watcher_bridge = {
            let mut shutdown = bridge_shutdown.subscribe();
            tokio::spawn(async move {
                loop {
                    tokio::select! {
                        _ = watch_rx.recv() => {}
                        res = shutdown.changed() => {
                            match res {
                                Ok(()) if *shutdown.borrow() => break,
                                Ok(()) => {}
                                Err(_) => break,
                            }
                        }
                    }
                }
            })
        };

        // Event bridge: a broadcast whose Sender we KEEP, so recv() blocks
        // indefinitely - it can only end via the shutdown signal.
        let (_evt_tx, mut evt_rx) = tokio::sync::broadcast::channel::<u32>(4);
        let event_bridge = {
            let mut shutdown = bridge_shutdown.subscribe();
            tokio::spawn(async move {
                loop {
                    tokio::select! {
                        res = shutdown.changed() => {
                            match res {
                                Ok(()) if *shutdown.borrow() => break,
                                Ok(()) => {}
                                Err(_) => break,
                            }
                        }
                        _ = evt_rx.recv() => {}
                    }
                }
            })
        };

        // Power poller: loops FOREVER with no shutdown path (exactly the real
        // `RealPowerSource::spawn_poller` shape) - only abort can stop it. A flag
        // proves it actually ran (and was not a no-op) before being aborted.
        let poller_ran = Arc::new(AtomicBool::new(false));
        let power_poller = {
            let poller_ran = poller_ran.clone();
            tokio::spawn(async move {
                poller_ran.store(true, Ordering::SeqCst);
                let mut ticker = tokio::time::interval(Duration::from_millis(10));
                loop {
                    ticker.tick().await;
                }
            })
        };

        // Capture abort handles BEFORE moving the JoinHandles into the handle, so
        // the test can independently assert each task is finished afterwards.
        let run_loop = {
            let orch = orchestrator.clone();
            tokio::spawn(async move {
                let _ = orch.run().await;
            })
        };
        let run_loop_abort = run_loop.abort_handle();
        let watcher_abort = watcher_bridge.abort_handle();
        let event_abort = event_bridge.abort_handle();
        let poller_abort = power_poller.abort_handle();

        let handle = AccountHandle::new(
            orchestrator,
            AccountTasks {
                crypto: test_crypto(),
                run_loop,
                watcher_bridge: Some(watcher_bridge),
                event_bridge,
                power_poller,
                sleep_wake_monitor: None,
                bridge_shutdown,
            },
        );

        // Let the poller actually start before we shut down.
        tokio::time::sleep(Duration::from_millis(30)).await;

        // The whole shutdown must complete well within the per-task budgets (the
        // run loop + bridges drain cleanly; the poller is aborted). Bound it so a
        // regression (a task that never stops) fails instead of hanging.
        tokio::time::timeout(Duration::from_secs(10), handle.shutdown())
            .await
            .expect("shutdown must complete (no task left orphaned)");

        assert!(
            poller_ran.load(Ordering::SeqCst),
            "the power poller task must have actually started"
        );

        // EVERY task is now finished (cleanly joined or aborted-and-awaited).
        assert!(run_loop_abort.is_finished(), "run loop must be finished");
        assert!(
            watcher_abort.is_finished(),
            "watcher bridge must be finished"
        );
        assert!(event_abort.is_finished(), "event bridge must be finished");
        assert!(
            poller_abort.is_finished(),
            "power poller must be finished (aborted, not orphaned)"
        );

        // Idempotent: a second shutdown is a no-op and does not panic / hang.
        tokio::time::timeout(Duration::from_secs(2), handle.shutdown())
            .await
            .expect("second shutdown is an immediate no-op");
    }

    /// An [`Orchestrator`] whose `run()` keeps working for `drain` AFTER the
    /// shutdown signal (modelling an in-flight upload that needs time to finish
    /// the current cycle), setting `finished_cleanly` only when it returns of
    /// its OWN accord (never if it was aborted mid-sleep).
    struct SlowDrainOrchestrator {
        shutdown: watch::Sender<bool>,
        shutdown_rx: watch::Receiver<bool>,
        drain: Duration,
        finished_cleanly: Arc<AtomicBool>,
    }

    impl SlowDrainOrchestrator {
        fn new(drain: Duration, finished_cleanly: Arc<AtomicBool>) -> Self {
            let (shutdown, shutdown_rx) = watch::channel(false);
            Self {
                shutdown,
                shutdown_rx,
                drain,
                finished_cleanly,
            }
        }
    }

    #[async_trait::async_trait]
    impl Orchestrator for SlowDrainOrchestrator {
        async fn run(&self) -> anyhow::Result<()> {
            let mut rx = self.shutdown_rx.clone();
            // Wait for the shutdown signal, then "finish the in-flight cycle"
            // (sleep `drain`) before returning. If aborted during the sleep,
            // `finished_cleanly` stays false.
            while !*rx.borrow() {
                if rx.changed().await.is_err() {
                    break;
                }
            }
            tokio::time::sleep(self.drain).await;
            self.finished_cleanly.store(true, Ordering::SeqCst);
            Ok(())
        }
        async fn trigger(&self, _reason: TickSource) {}
        async fn set_paused(&self, _paused: bool) {}
        async fn state(&self) -> OrchestratorState {
            OrchestratorState::Idle { last_run_at: None }
        }
        async fn reconfigure(&self, _config: OrchestratorConfig) {}
        fn shutdown(&self) {
            let _ = self.shutdown.send(true);
        }
    }

    /// R2-P2-1: a run loop that needs LONGER than the short 5s
    /// [`TASK_DRAIN_TIMEOUT`] to finish its in-flight cycle - but less than the
    /// full [`RUN_LOOP_DRAIN_TIMEOUT`] - must drain CLEANLY (DESIGN s5.10.2),
    /// NOT be aborted at the bridge budget. Uses paused time so the long drain
    /// is instant.
    #[tokio::test]
    async fn run_loop_gets_full_drain_budget_not_the_short_bridge_timeout() {
        // Paused virtual time so the long in-flight drain is instant (no real
        // multi-second sleep). `tokio::time::pause` auto-advances when every
        // task is waiting on a timer.
        tokio::time::pause();
        // The in-flight cycle takes ~12s to wind down: well past the 5s short
        // budget, comfortably inside the 20s run-loop budget.
        let drain = Duration::from_secs(12);
        assert!(
            drain > TASK_DRAIN_TIMEOUT && drain < RUN_LOOP_DRAIN_TIMEOUT,
            "test fixture must straddle the two budgets"
        );

        let finished_cleanly = Arc::new(AtomicBool::new(false));
        let orchestrator: Arc<dyn Orchestrator> =
            Arc::new(SlowDrainOrchestrator::new(drain, finished_cleanly.clone()));

        let (bridge_shutdown, _rx0) = watch::channel(false);
        let run_loop = {
            let orch = orchestrator.clone();
            tokio::spawn(async move {
                let _ = orch.run().await;
            })
        };
        let run_loop_abort = run_loop.abort_handle();

        // No watcher/event/poller for this focused test: the run loop is what
        // matters. `None` slots are drained as no-ops.
        let (_evt_tx, evt_rx) = tokio::sync::broadcast::channel::<u32>(4);
        drop(evt_rx);
        let event_bridge = tokio::spawn(async {});
        let power_poller = tokio::spawn(async {});

        let handle = AccountHandle::new(
            orchestrator,
            AccountTasks {
                crypto: test_crypto(),
                run_loop,
                watcher_bridge: None,
                event_bridge,
                power_poller,
                sleep_wake_monitor: None,
                bridge_shutdown,
            },
        );

        // Bound by the full run-loop budget + a margin: the run loop must finish
        // by draining cleanly (not by being aborted at 5s).
        tokio::time::timeout(
            RUN_LOOP_DRAIN_TIMEOUT + Duration::from_secs(5),
            handle.shutdown(),
        )
        .await
        .expect("shutdown must complete within the full run-loop budget");

        assert!(
            finished_cleanly.load(Ordering::SeqCst),
            "the run loop must drain CLEANLY (finish its in-flight cycle), not be aborted at the 5s bridge budget (R2-P2-1)"
        );
        assert!(run_loop_abort.is_finished(), "run loop must be finished");
    }

    /// One account's tracked task set plus the abort handles needed to assert
    /// every task finished after shutdown - built to the EXACT shapes assembly
    /// spawns (slow-draining run loop + forever-looping power poller).
    struct BuiltAccount {
        handle: AccountHandle,
        run_loop_abort: tokio::task::AbortHandle,
        poller_abort: tokio::task::AbortHandle,
    }

    /// Build a slow account: a run loop that needs `drain` to wind down after the
    /// stop signal (modelling a large in-flight upload) and a power poller that
    /// loops forever (only abort can stop it). Mirrors the per-account task set
    /// in `assembly::build_account` so the R3-P1-1 concurrent-drain test is real.
    fn build_slow_account(drain: Duration) -> BuiltAccount {
        let finished_cleanly = Arc::new(AtomicBool::new(false));
        let orchestrator: Arc<dyn Orchestrator> =
            Arc::new(SlowDrainOrchestrator::new(drain, finished_cleanly));
        let (bridge_shutdown, _rx0) = watch::channel(false);

        let run_loop = {
            let orch = orchestrator.clone();
            tokio::spawn(async move {
                let _ = orch.run().await;
            })
        };
        let run_loop_abort = run_loop.abort_handle();

        // Power poller: loops FOREVER with no shutdown path (only abort stops it).
        let power_poller = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(Duration::from_millis(10));
            loop {
                ticker.tick().await;
            }
        });
        let poller_abort = power_poller.abort_handle();

        let event_bridge = tokio::spawn(async {});

        let handle = AccountHandle::new(
            orchestrator,
            AccountTasks {
                crypto: test_crypto(),
                run_loop,
                watcher_bridge: None,
                event_bridge,
                power_poller,
                sleep_wake_monitor: None,
                bridge_shutdown,
            },
        );
        BuiltAccount {
            handle,
            run_loop_abort,
            poller_abort,
        }
    }

    /// R3-P1-1: shutting down MULTIPLE accounts must run their `shutdown()`
    /// futures CONCURRENTLY (as `lib.rs::shutdown_orchestrators` now does via
    /// `futures::future::join_all`) and leave ZERO orphaned tasks - even when
    /// every run loop is slow AND every poller would run forever. The bug this
    /// guards: a serial drain under one outer timeout could fire MID-drain and
    /// drop a cancellation-unsafe `drain_or_abort` (whose `JoinHandle` is already
    /// taken), detaching the aborted task. Uses paused virtual time so the long
    /// per-account drains are instant, and asserts the WHOLE concurrent sweep
    /// finishes well under the SUM of the two accounts' budgets.
    #[tokio::test]
    async fn concurrent_shutdown_of_multiple_slow_accounts_leaves_no_orphans() {
        tokio::time::pause();
        // Each run loop needs ~15s to wind down (inside the 20s run-loop budget),
        // and each poller loops forever (always aborted). Two accounts.
        let drain = Duration::from_secs(15);
        assert!(
            drain < RUN_LOOP_DRAIN_TIMEOUT,
            "each account's drain must fit inside its own run-loop budget"
        );

        let a = build_slow_account(drain);
        let b = build_slow_account(drain);

        // Let both pollers actually start before shutting down.
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Drive BOTH accounts' shutdown CONCURRENTLY, exactly like the quit sweep.
        // If they ran serially the elapsed virtual time would be ~2x the run-loop
        // drain; concurrently it is ~1x. Bound the whole sweep by ONE run-loop
        // budget + margin (which is LESS than two serial drains would need), so a
        // regression to serial draining fails this timeout.
        tokio::time::timeout(
            RUN_LOOP_DRAIN_TIMEOUT + Duration::from_secs(5),
            futures::future::join_all([a.handle.shutdown(), b.handle.shutdown()]),
        )
        .await
        .expect(
            "concurrent multi-account shutdown must complete (no task orphaned, drains overlap)",
        );

        // EVERY task across BOTH accounts is finished (joined or aborted-and-
        // awaited) - no orphans.
        assert!(
            a.run_loop_abort.is_finished(),
            "account A run loop must be finished"
        );
        assert!(
            a.poller_abort.is_finished(),
            "account A poller must be finished (aborted, not orphaned)"
        );
        assert!(
            b.run_loop_abort.is_finished(),
            "account B run loop must be finished"
        );
        assert!(
            b.poller_abort.is_finished(),
            "account B poller must be finished (aborted, not orphaned)"
        );
    }

    /// Open a temp-backed state repo for the AppState dialog-token tests.
    async fn temp_state() -> (Arc<dyn StateRepo>, std::path::PathBuf) {
        use driven_core::state::sqlite::SqliteStateRepo;
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let dir = std::env::temp_dir().join(format!("driven-appstate-test-{nonce}-{:p}", &nonce));
        std::fs::create_dir_all(&dir).expect("create temp dir");
        let repo = SqliteStateRepo::open(&dir.join("state.db"))
            .await
            .expect("open state repo");
        (Arc::new(repo), dir)
    }

    /// An empty fake-remote-store registry for the AppState tests.
    fn default_fake_registry() -> super::FakeRemoteStores {
        Arc::new(std::sync::Mutex::new(HashMap::new()))
    }

    #[tokio::test]
    async fn dialog_token_round_trips_and_is_single_use() {
        // C1 (SPEC s11.6.1): a minted token maps to its path exactly once; a
        // second take (or an unknown token) returns None so a leaked / replayed
        // token cannot authorise a second write.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        let path = std::path::PathBuf::from("/home/u/backups");
        let token = app_state.mint_dialog_token(path.clone());

        // First take resolves the bound path.
        assert_eq!(app_state.take_dialog_token(&token), Some(path));
        // Single-use: a second take is rejected.
        assert_eq!(app_state.take_dialog_token(&token), None);
        // An unknown token is rejected.
        assert_eq!(app_state.take_dialog_token("not-a-real-token"), None);

        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 debug_mode_runtime_task_and_shutdown_round_trip() {
        // Issue #309: the debug-logging-mode watchdog's runtime bookkeeping.
        // No hub getter to cover (unlike bottleneck/iostat) - just the
        // set/shutdown task pair, mirrors
        // `bottleneck_runtime_hub_task_and_shutdown_round_trip`.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        // No task registered yet: shutdown is a safe no-op.
        assert!(app_state.shutdown_debug_mode_task().is_none());

        // Register a task that exits promptly on the shutdown signal (the
        // real watchdog'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_debug_mode_task(task, shutdown_tx);

        let handle = app_state
            .shutdown_debug_mode_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_debug_mode_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 -
        // `None` until installed at boot, returned once set, and
        // `shutdown_vss_helper` is a best-effort no-op both when no manager is in
        // play and when the (never-launched) manager is installed.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        // No manager yet; shutdown is a safe no-op.
        assert!(app_state.vss_helper_manager().is_none());
        app_state.shutdown_vss_helper();

        // Install one with an injected launch fn (no real UAC / process),
        // disabled so it never spawns a launch thread.
        let manager = Arc::new(crate::vss_helper::VssHelperManager::with_launch_fn(
            std::env::temp_dir().join("driven-vss-helper.exe"),
            std::env::temp_dir(),
            false,
            Box::new(|| Ok(())),
        ));
        app_state.set_vss_helper_manager(manager);
        assert!(app_state.vss_helper_manager().is_some());
        // The broker was never launched, so shutdown stays a no-op (no panic /
        // no connect).
        app_state.shutdown_vss_helper();

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn shutdown_vss_helper_for_update_disables_then_rearm_restores() {
        // Issue #125: the updater path disables the broker BEFORE
        // `download_and_install` so a live/relaunching elevated helper cannot hold
        // its own exe open while the NSIS installer overwrites it. On a FAILED
        // install (app keeps running) the caller re-arms it so locked-file backup
        // is not left silently degraded.
        use driven_vss_helper::HelperLauncher; // brings `is_available` into scope
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        // No manager: both calls are safe no-ops; the pre-install sweep reports it
        // disabled nothing.
        assert!(!app_state.shutdown_vss_helper_for_update());
        app_state.rearm_vss_helper_after_failed_update();

        // Install an ENABLED manager with an injected launch (no real UAC /
        // process). Not launched yet -> NotAttempted + enabled == launchable.
        let manager = Arc::new(crate::vss_helper::VssHelperManager::with_launch_fn(
            std::env::temp_dir().join("driven-vss-helper.exe"),
            std::env::temp_dir(),
            true,
            Box::new(|| Ok(())),
        ));
        app_state.set_vss_helper_manager(manager.clone());
        assert!(
            manager.is_available(),
            "an enabled, not-yet-tried broker is available on demand"
        );

        // The pre-install sweep disables it (so a mid-download locked file cannot
        // re-launch the broker) and reports it acted.
        assert!(app_state.shutdown_vss_helper_for_update());
        assert!(
            !manager.is_available(),
            "disabled broker is not available -> cannot re-launch during the install"
        );

        // A failed install re-arms it so backup is available again on demand.
        app_state.rearm_vss_helper_after_failed_update();
        assert!(
            manager.is_available(),
            "re-arm after a failed install restores on-demand launchability"
        );

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn recovery_ack_gate_requires_a_recorded_backend_reveal() {
        // M9c D4 (M6 R4-P1-1, DATA-SAFETY): the ack gate `ack_recovery_phrase_saved`
        // enforces is the AppState predicate `recovery_reveal_recorded`. A source
        // registered as pending-ack is NOT acknowledgeable until the backend records
        // a real reveal; an unknown source has no pending ack at all.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        let source = driven_core::types::SourceId::new_v4();
        let account = AccountId::new_v4();

        // Unknown source: no pending ack -> the ack would be rejected (None).
        assert_eq!(app_state.recovery_reveal_recorded(source), None);
        assert_eq!(app_state.pending_recovery_ack_account(source), None);

        // Register a pending ack (add_source's first-encrypted-source path).
        app_state.register_pending_recovery_ack(source, account);
        // Pending but NOT revealed: the ack must be rejected (Some(false)).
        assert_eq!(app_state.recovery_reveal_recorded(source), Some(false));
        assert_eq!(
            app_state.pending_recovery_ack_account(source),
            Some(account)
        );

        // Record a backend reveal (reveal_recovery_phrase). Now the ack may proceed.
        assert!(
            app_state.record_recovery_reveal(source),
            "recording a reveal for a pending source returns true"
        );
        assert_eq!(app_state.recovery_reveal_recorded(source), Some(true));

        // Recording a reveal for an UNKNOWN source is a no-op (false).
        assert!(!app_state.record_recovery_reveal(driven_core::types::SourceId::new_v4()));

        // After the ack succeeds, clearing the pending entry makes a replay a no-op.
        app_state.clear_pending_recovery_ack(source);
        assert_eq!(app_state.recovery_reveal_recorded(source), None);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn peek_dialog_token_is_non_consuming() {
        // R1-P1-2: `preview_exclusions` PEEKS the dialog token (non-consuming) so
        // the user can re-run the preview as they tweak globs AND the subsequent
        // `add_source` still has the single TAKE it needs. Peeking N times then
        // taking once must all resolve the same path; a take after that is
        // rejected.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        let path = std::path::PathBuf::from("/home/u/preview-root");
        let token = app_state.mint_dialog_token(path.clone());

        // Multiple peeks all resolve the path WITHOUT consuming the token.
        assert_eq!(app_state.peek_dialog_token(&token), Some(path.clone()));
        assert_eq!(app_state.peek_dialog_token(&token), Some(path.clone()));
        // The single TAKE (what add_source uses) still works after the peeks.
        assert_eq!(app_state.take_dialog_token(&token), Some(path));
        // Now consumed: a further peek AND take both return None.
        assert_eq!(app_state.peek_dialog_token(&token), None);
        assert_eq!(app_state.take_dialog_token(&token), None);
        // An unknown token never resolves.
        assert_eq!(app_state.peek_dialog_token("nope"), None);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn expired_dialog_token_is_rejected_by_peek_and_take() {
        // C1: a token past its TTL must resolve for NEITHER the non-consuming
        // peek nor the consuming take - the wizard gets the stable
        // `internal.stale_dialog_token` rejection (never a phantom disk error)
        // and re-picking mints a fresh, working token.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        let path = std::path::PathBuf::from("/home/u/slow-wizard-root");
        let token = app_state.mint_dialog_token(path.clone());
        assert_eq!(app_state.peek_dialog_token(&token), Some(path.clone()));

        app_state.expire_dialog_token_for_test(&token);
        assert_eq!(app_state.peek_dialog_token(&token), None);
        assert_eq!(app_state.take_dialog_token(&token), None);

        // Re-picking (a fresh mint of the same path) works immediately.
        let fresh = app_state.mint_dialog_token(path.clone());
        assert_eq!(app_state.take_dialog_token(&fresh), Some(path));

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn ensure_master_key_lock_is_shared_per_account_and_serialises() {
        // R2-P1-1: two `add_source` calls for the SAME account must get the SAME
        // lock (so they serialise); different accounts get DISTINCT locks (so they
        // do not block each other). Also prove the lock actually serialises a
        // critical section (no overlap).
        use std::sync::atomic::{AtomicUsize, Ordering};
        let (state, dir) = temp_state().await;
        let app_state = Arc::new(AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        ));
        let acct_a = AccountId::new_v4();
        let acct_b = AccountId::new_v4();

        // Same account -> same Arc; different account -> different Arc.
        let l1 = app_state.ensure_master_key_lock(acct_a);
        let l2 = app_state.ensure_master_key_lock(acct_a);
        let l3 = app_state.ensure_master_key_lock(acct_b);
        assert!(Arc::ptr_eq(&l1, &l2), "same account shares one lock");
        assert!(
            !Arc::ptr_eq(&l1, &l3),
            "different accounts get distinct locks"
        );

        // Serialisation: two tasks contend on acct_a's lock; the in-section
        // counter must never exceed 1 (no overlap).
        let max_in_section = Arc::new(AtomicUsize::new(0));
        let in_section = Arc::new(AtomicUsize::new(0));
        let mut handles = Vec::new();
        for _ in 0..8 {
            let app_state = app_state.clone();
            let in_section = in_section.clone();
            let max_in_section = max_in_section.clone();
            handles.push(tokio::spawn(async move {
                let lock = app_state.ensure_master_key_lock(acct_a);
                let _guard = lock.lock().await;
                let now = in_section.fetch_add(1, Ordering::SeqCst) + 1;
                max_in_section.fetch_max(now, Ordering::SeqCst);
                // Yield so an overlapping task would be observed if the lock
                // failed to serialise.
                tokio::task::yield_now().await;
                in_section.fetch_sub(1, Ordering::SeqCst);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
        assert_eq!(
            max_in_section.load(Ordering::SeqCst),
            1,
            "the per-account lock must serialise the critical section (no overlap)"
        );
        let _ = std::fs::remove_dir_all(dir);
    }

    /// Build a minimal terminal/active restore status for the job tests.
    fn restore_status(job_id: &str, done: bool) -> crate::commands::dtos::RestoreJobStatus {
        crate::commands::dtos::RestoreJobStatus {
            job_id: job_id.to_string(),
            total_files: 1,
            completed_files: 0,
            failed_files: 0,
            total_bytes: 0,
            bytes_done: 0,
            current_file: None,
            done,
            cancelled: false,
            files: Vec::new(),
        }
    }

    #[tokio::test]
    async fn cancel_restore_job_sets_flag_and_returns_handle_and_drains() {
        // M8-P1-1: cancelling a registered job sets its cancel flag (the task
        // observes it + exits), returns the task handle so the caller can await
        // it, and is idempotent for an unknown / already-cancelled id.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        let cancel: RestoreCancel = Arc::new(AtomicBool::new(false));
        let observed_cancel = cancel.clone();
        // A task that loops until the cancel flag is set (models the stream loop).
        let handle = tokio::spawn(async move {
            loop {
                if observed_cancel.load(Ordering::SeqCst) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        });
        app_state.seed_restore_job(restore_status("job-1", false), cancel.clone(), handle);

        // Cancel: flag set + handle returned.
        let returned = app_state.cancel_restore_job("job-1");
        assert!(cancel.load(Ordering::SeqCst), "cancel flag must be set");
        let returned = returned.expect("a running job returns its handle");
        // The task observes the flag and exits; awaiting it completes.
        tokio::time::timeout(Duration::from_secs(2), returned)
            .await
            .expect("cancelled task must finish")
            .expect("task joined cleanly");

        // Idempotent: a second cancel (handle already taken) returns None.
        assert!(app_state.cancel_restore_job("job-1").is_none());
        // Unknown id: None.
        assert!(app_state.cancel_restore_job("nope").is_none());

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn ui_cancel_sets_flag_but_keeps_handle_for_shutdown_drain() {
        // R1-P1-2: the UI cancel path (`signal_cancel_restore_job`) must ONLY set
        // the cancel flag and LEAVE the handle tracked, so the M5-style shutdown
        // drain still awaits/aborts the task - a UI cancel never orphans it. This
        // models the exact sequence: UI cancel signals the flag, the task is still
        // tracked, then the shutdown drain (`cancel_all_restore_jobs`) takes the
        // handle and joins it (no orphan).
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        let cancel: RestoreCancel = Arc::new(AtomicBool::new(false));
        let observed_cancel = cancel.clone();
        // A task that loops until the cancel flag is set (models the stream loop).
        let handle = tokio::spawn(async move {
            loop {
                if observed_cancel.load(Ordering::SeqCst) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        });
        let task_finished = handle.abort_handle();
        app_state.seed_restore_job(restore_status("ui-job", false), cancel.clone(), handle);

        // UI cancel: flag set, but the handle is LEFT tracked (not taken/dropped).
        assert!(
            app_state.signal_cancel_restore_job("ui-job"),
            "signalling a tracked job returns true"
        );
        assert!(cancel.load(Ordering::SeqCst), "cancel flag must be set");

        // The shutdown drain still finds the handle (it was NOT detached by the UI
        // cancel) and can await it - i.e. the task is accounted-for, not orphaned.
        let handles = app_state.cancel_all_restore_jobs();
        assert_eq!(
            handles.len(),
            1,
            "the UI-cancelled job's handle must STILL be tracked for the shutdown drain"
        );
        for h in handles {
            tokio::time::timeout(Duration::from_secs(2), h)
                .await
                .expect("the UI-cancelled task must be joinable on shutdown (not orphaned)")
                .expect("joined cleanly");
        }
        assert!(
            task_finished.is_finished(),
            "the task must have actually finished (cancel flag observed)"
        );

        // Idempotent: signalling an unknown id is a benign false.
        assert!(!app_state.signal_cancel_restore_job("nope"));

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn cancel_all_restore_jobs_drains_every_in_flight_job() {
        // M8-P1-1: the shutdown path cancels EVERY in-flight restore job and
        // returns their handles so quit can await them (no orphaned restore task).
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        for i in 0..3 {
            let cancel: RestoreCancel = Arc::new(AtomicBool::new(false));
            let observed = cancel.clone();
            let handle = tokio::spawn(async move {
                loop {
                    if observed.load(Ordering::SeqCst) {
                        break;
                    }
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
            });
            let id = format!("job-{i}");
            app_state.seed_restore_job(restore_status(&id, false), cancel, handle);
        }

        let handles = app_state.cancel_all_restore_jobs();
        assert_eq!(handles.len(), 3, "every in-flight job handle is returned");
        for h in handles {
            tokio::time::timeout(Duration::from_secs(2), h)
                .await
                .expect("each cancelled restore task must finish")
                .expect("joined cleanly");
        }
        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn terminal_restore_jobs_are_count_capped() {
        // M8-P2-3: terminal restore-job records are count-capped so the map does
        // not leak across many restores. Register MORE than the cap terminal jobs
        // and assert only the cap remains (plus any active jobs, which we add one
        // of to prove active jobs are NEVER pruned).
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );

        // A trivial already-finished handle (the pruning test does not exercise the
        // task itself; seed_restore_job now requires a handle, R3-P1-2).
        let noop_handle = || tokio::spawn(async {});

        // One ACTIVE job that must survive pruning.
        let active_cancel: RestoreCancel = Arc::new(AtomicBool::new(false));
        app_state.seed_restore_job(
            restore_status("active", false),
            active_cancel,
            noop_handle(),
        );

        // Register many TERMINAL jobs (seed active, then put a done snapshot to
        // stamp terminal_at + trigger pruning).
        let total = MAX_RETAINED_TERMINAL_JOBS + 10;
        for i in 0..total {
            let id = format!("done-{i}");
            let cancel: RestoreCancel = Arc::new(AtomicBool::new(false));
            app_state.seed_restore_job(restore_status(&id, false), cancel, noop_handle());
            app_state.put_restore_job(restore_status(&id, true));
        }

        let map = app_state.lock_restore_jobs();
        let terminal = map.values().filter(|e| e.terminal_at.is_some()).count();
        assert!(
            terminal <= MAX_RETAINED_TERMINAL_JOBS,
            "terminal jobs must be capped at {MAX_RETAINED_TERMINAL_JOBS}, got {terminal}"
        );
        assert!(
            map.contains_key("active"),
            "an active job must never be pruned"
        );
        drop(map);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn dialog_tokens_are_distinct_per_mint() {
        // Two mints yield distinct tokens each bound to its own path.
        let (state, dir) = temp_state().await;
        let app_state = AppState::new(
            state,
            HashMap::new(),
            RemoteMode::Fake,
            default_fake_registry(),
        );
        let a = app_state.mint_dialog_token(std::path::PathBuf::from("/a"));
        let b = app_state.mint_dialog_token(std::path::PathBuf::from("/b"));
        assert_ne!(a, b);
        assert_eq!(
            app_state.take_dialog_token(&b),
            Some(std::path::PathBuf::from("/b"))
        );
        assert_eq!(
            app_state.take_dialog_token(&a),
            Some(std::path::PathBuf::from("/a"))
        );
        let _ = std::fs::remove_dir_all(dir);
    }

    // --- agent QA harness: DRIVEN_TEST_FAULT_PLAN gating ---------------------

    #[test]
    fn fault_plan_unset_resolves_none() {
        assert_eq!(
            super::resolve_fault_plan(true, None),
            FaultPlanResolution::None
        );
        assert_eq!(
            super::resolve_fault_plan(false, None),
            FaultPlanResolution::None
        );
        // Empty value = unset (mirrors the non_empty_env convention).
        assert_eq!(
            super::resolve_fault_plan(true, Some("")),
            FaultPlanResolution::None
        );
    }

    #[test]
    fn fault_plan_without_fake_remote_is_a_loud_error() {
        // The plan can only ever fire on the fake store; setting it in real
        // mode is a harness bug that must NOT pass silently.
        match super::resolve_fault_plan(false, Some("/tmp/plan.json")) {
            FaultPlanResolution::Error(msg) => {
                assert!(
                    msg.contains("DRIVEN_USE_FAKE_REMOTE"),
                    "actionable message: {msg}"
                );
            }
            other => panic!("expected Error, got {other:?}"),
        }
    }

    #[test]
    fn fault_plan_missing_file_is_an_error() {
        match super::resolve_fault_plan(true, Some("/definitely/not/here.json")) {
            FaultPlanResolution::Error(msg) => {
                assert!(msg.contains("failed to read"), "actionable message: {msg}");
            }
            other => panic!("expected Error, got {other:?}"),
        }
    }

    #[test]
    fn fault_plan_parses_and_arms() {
        let dir = std::env::temp_dir().join(format!("driven-fault-plan-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("plan.json");
        std::fs::write(
            &path,
            r#"{"network_drop_after": 3, "dest_folder_readonly": true}"#,
        )
        .unwrap();
        match super::resolve_fault_plan(true, Some(path.to_str().unwrap())) {
            FaultPlanResolution::Armed(plan) => {
                assert_eq!(plan.network_drop_after, Some(3));
                assert!(plan.dest_folder_readonly);
            }
            other => panic!("expected Armed, got {other:?}"),
        }
        // A typo'd fault name must surface, not arm-nothing (deny_unknown_fields).
        std::fs::write(&path, r#"{"network_drop_afer": 3}"#).unwrap();
        match super::resolve_fault_plan(true, Some(path.to_str().unwrap())) {
            FaultPlanResolution::Error(msg) => {
                assert!(msg.contains("failed to parse"), "actionable message: {msg}");
            }
            other => panic!("expected Error, got {other:?}"),
        }
        let _ = std::fs::remove_dir_all(dir);
    }
}