omni-dev 0.46.0

//! `drive lease acquire` — the engine behind
//! [ADR-0080](../../../docs/adrs/adr-0080.md) §2: authenticate, then back
//! up, then record, then mint. Binary files back up as bytes to local disk;
//! a Google-native document (Sheet/Doc/Slide) backs up as a lossless
//! Drive-side `files.copy` into the account's configured backup folder
//! (§3's fidelity split) — refused outright, before authenticating at all,
//! when no backup folder is configured for this account, or when a live
//! lease already covers the target file (issue #1690). The latter refusal
//! is a fast path, not the authoritative gate — see [`acquire_inner`] — so
//! the narrow remaining race still spends a prompt and a backup; that
//! backup is then reclaimed (deleted/trashed) rather than left orphaned,
//! since `drive lease prune` can only ever see backups a ledger row
//! points at.

use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::Context as _;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use serde::Serialize;
use sha2::{Digest, Sha256};

use crate::cli::drive::format::JsonlSerialize;
use crate::cli::format::sanitize_for_terminal;
use crate::drive::client::DriveClient;
use crate::drive::files_api::FilesApi;
use crate::drive::lease::authenticate::{AuthOutcome, AuthPolicy, Authenticator};
use crate::drive::lease::ledger::{
    default_lock_wait_timeout, LeaseBackup, LeaseLedger, LeaseRecord, LedgerLock, ReleaseOutcome,
};
use crate::drive::types::DriveFile;

/// Per-call options for `drive lease acquire`.
#[derive(Debug, Clone)]
pub struct AcquireOptions {
    /// The file id to lease.
    pub file_id: String,
    /// Local directory byte backups are written under. Unused for a
    /// native-document target.
    pub backup_dir: PathBuf,
    /// Destination folder for a native document's Drive-side backup copy
    /// (ADR-0080 §3/§13) — the account's configured
    /// `lease_backup_folder_id`. `None` means a native-document target is
    /// refused outright; unused for a binary target.
    pub native_backup_folder_id: Option<String>,
    /// How long the lease stays live from the moment it is authorised.
    pub expiry: ChronoDuration,
    /// Which authentication policy to present (ADR-0080 §7).
    pub auth_policy: AuthPolicy,
    /// Path to the lease ledger. Production callers pass
    /// [`crate::drive::lease::ledger::ledger_path`]'s own result; tests
    /// pass a path under a `tempdir` so a test run never touches the real
    /// ledger.
    pub ledger_path: PathBuf,
    /// The global headless/off-macOS opt-out (ADR-0080 §8/§13, issue
    /// #1677): when `true`, an [`AuthOutcome::NoAuthenticator`] outcome —
    /// no prompt can reach a human in this context at all — is waived
    /// instead of refusing the lease, and the acquisition proceeds without a
    /// human ever having been prompted. An
    /// [`AuthOutcome::PolicyUnsatisfiable`] outcome is never waived (issue
    /// #1686). Resolved by
    /// `crate::drive::lease::settings::resolve_allow_headless`.
    pub allow_headless: bool,
    /// A lease this acquisition *replaces* — its row is released in the
    /// same locked ledger rewrite that inserts the new one, and it does not
    /// count as a blocker in either live-lease check (issue #1685).
    ///
    /// `drive lease restore` is the only caller that sets it, naming the
    /// backup token it is restoring from: restore mints its fresh lease on
    /// the same file that token already covers, so without this it refuses
    /// itself, reporting `AlreadyLeased` for the very token the user
    /// presented. Superseding rather than merely ignoring the row keeps
    /// [`LeaseLedger::live_lease_for_file`]'s one-live-lease-per-file
    /// invariant (#1664) exactly intact, and doing it atomically with the
    /// insert — rather than releasing before the prompt — means the file is
    /// never left uncovered, and a denied or failed acquisition leaves the
    /// superseded lease untouched.
    ///
    /// A token bound to a *different* file is ignored: the release only
    /// ever applies to a row already covering [`Self::file_id`].
    pub supersedes: Option<String>,
}

/// What happened.
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "status", rename_all = "kebab-case")]
pub enum AcquireResult {
    /// Authenticated, backed up, and recorded. Present the token to a
    /// gated write via `--lease`.
    Acquired {
        /// The opaque lease token.
        token: String,
        /// When this lease stops authorising writes.
        expires_at: DateTime<Utc>,
        /// Where the backup landed.
        backup: LeaseBackup,
        /// `true` when this acquisition proceeded under the headless
        /// opt-out (ADR-0080 §8/§13) instead of a real device-owner
        /// prompt — no human presence was verified. Carried into the audit
        /// record ([`record_attempt`]) so a waived acquisition is durably
        /// distinguishable from a normally-authorised one.
        headless_waiver: bool,
        /// The lease this acquisition released in the same locked rewrite
        /// that minted it ([`AcquireOptions::supersedes`], issue #1685) —
        /// `Some` only when [`insert_record`] actually released a live row,
        /// so a supersede that found no row, another file's row, or an
        /// already-dead row is reported as `None`. Carried out of the locked
        /// closure as a fact rather than re-derived afterwards from ledger
        /// timestamps, and into the audit record as `superseded_lease_id`.
        /// Omitted from the JSON shape when `None`, so a plain acquire's
        /// output is unchanged.
        #[serde(skip_serializing_if = "Option::is_none")]
        superseded_lease_id: Option<String>,
    },
    /// A live lease already covers this file — its token is returned for
    /// reuse rather than minting a second, independent one. Two leases
    /// concurrently acquired on the same file could each pass their own
    /// staleness check against the same now-stale snapshot, letting the
    /// second writer silently clobber the first's write (issue #1664
    /// review finding) — see
    /// [`LeaseLedger::live_lease_for_file`](super::ledger::LeaseLedger::live_lease_for_file)'s
    /// doc comment. A lock-free pre-check in [`acquire_inner`] refuses most
    /// of these *before* authenticating at all (issue #1690); only the
    /// narrow remaining race — another process inserting its own lease
    /// between that pre-check and the authoritative, lock-held check —
    /// still spends a prompt and a backup first, and that backup is
    /// reclaimed (deleted/trashed) automatically rather than left orphaned.
    AlreadyLeased {
        /// The existing lease's token — present this to `--lease` instead.
        token: String,
        /// When the existing lease expires.
        expires_at: DateTime<Utc>,
    },
    /// The target is a Google-native document and no backup folder is
    /// configured for this account (`lease_backup_folder_id`) — nowhere to
    /// put the required Drive-side copy.
    RefusedNativeDocument,
    /// The file changed while the backup was being taken, so the backup
    /// cannot be pinned to the version that would have been recorded
    /// (issue #1693). Nothing was minted and no ledger row was written;
    /// every backup this attempt took is reclaimed. See
    /// [`backup_is_pinned`] for how the two backup kinds detect this. A
    /// version-sandwich refusal has already been retried internally
    /// ([`SANDWICH_ATTEMPTS`], issue #1739), and its `detail` says so when
    /// more than one attempt was made; any other is a one-shot refusal
    /// that retrying the command is the intended response to.
    RefusedConcurrentChange {
        /// What changed, and which evidence detected it.
        detail: String,
    },
    /// A human answered the prompt and refused, or it timed out.
    Denied {
        /// The platform's own message.
        detail: String,
    },
    /// No prompt could be presented (ADR-0080 §8): either no authenticator
    /// exists in this context, or the requested policy cannot be satisfied
    /// in it (issue #1686). No backup was taken and no ledger row was
    /// written.
    Unavailable {
        /// Why no prompt could be presented.
        detail: String,
    },
    /// An API, filesystem, or ledger error.
    Failed {
        /// A human-readable summary of what failed.
        detail: String,
    },
}

impl JsonlSerialize for AcquireResult {
    fn write_jsonl(&self, out: &mut dyn std::io::Write) -> Result<(), anyhow::Error> {
        crate::cli::drive::format::write_scalar_jsonl(self, out)
    }
}

impl AcquireResult {
    /// The CLI's exit code for this outcome (issue #1775): `0` when the
    /// caller ends up holding a usable lease token — whether freshly minted
    /// or reused via [`Self::AlreadyLeased`] — `1` for every refusal or
    /// failure.
    pub fn exit_code(&self) -> i32 {
        match self {
            Self::Acquired { .. } | Self::AlreadyLeased { .. } => 0,
            Self::RefusedNativeDocument
            | Self::RefusedConcurrentChange { .. }
            | Self::Denied { .. }
            | Self::Unavailable { .. }
            | Self::Failed { .. } => 1,
        }
    }
}

/// Runs the four-step sequence ADR-0080 §2 specifies, in order, aborting at
/// the first failure with nothing further attempted, then records the
/// attempt to the audit sink regardless of outcome (ADR-0080 §11).
///
/// The audit record here is best-effort, not write-ahead/fail-closed the
/// way a leased *write*'s is: `drive lease acquire` mutates no Drive
/// content — it takes a backup and writes a ledger row, both already
/// durable by the time this function is about to return — so there is no
/// "mutating API call" for a write-ahead record to precede. A logging
/// failure here is warned, not surfaced as a failed acquisition: the
/// consent, the backup and the ledger row already happened, and turning a
/// genuine success into a reported failure because a follow-up log write
/// failed would make the tool's own output less trustworthy, not more.
pub async fn acquire(
    client: &DriveClient,
    opts: &AcquireOptions,
    authenticator: &dyn Authenticator,
) -> AcquireResult {
    acquire_with_timings(client, opts, authenticator, AcquireTimings::production()).await
}

/// How many backups [`take_pinned_backup`] takes before a version-sandwich
/// refusal stands (issue #1739). Drive defines `version` as reflecting
/// "every change made to the file on the server, even those not visible to
/// the user", so a one-off bump unrelated to any edit is plausible.
/// `files.copy` itself is not one: whether it bumps the *source* file's
/// `version` is undocumented, but a live acquire on 2026-09-19 left both a
/// Doc (11) and a Sheet (24) unmoved across the copy, and after it — so the
/// sandwich converges on a quiet native document in one attempt. Retrying
/// lets any other stray bump self-heal after the prompt has already been
/// answered, rather than handing the operator a refusal that reads exactly
/// like a collaborator's edit. Deliberately small: every attempt is a full
/// `files.copy` (or download) plus two metadata reads.
const SANDWICH_ATTEMPTS: u32 = 3;

/// The pause between two sandwich attempts — long enough for a transient
/// server-side bump to settle, short enough to be invisible next to the
/// Touch ID prompt that preceded it.
const SANDWICH_RETRY_DELAY: Duration = Duration::from_secs(1);

/// The waits [`acquire_with_timings`] observes. Production uses
/// [`AcquireTimings::production`]; tests shorten them so a busy-lock
/// timeout, or a sandwich retry, runs in milliseconds.
#[derive(Debug, Clone, Copy)]
pub(crate) struct AcquireTimings {
    /// Wait budget for the ledger insert's lock — the same seam as
    /// [`LedgerLock::acquire_waiting_blocking_with_timeout`].
    pub(crate) lock_wait: Duration,
    /// Pause between version-sandwich attempts ([`SANDWICH_RETRY_DELAY`]).
    pub(crate) sandwich_retry_delay: Duration,
}

impl AcquireTimings {
    /// The timings [`acquire`] uses.
    fn production() -> Self {
        Self {
            lock_wait: default_lock_wait_timeout(),
            sandwich_retry_delay: SANDWICH_RETRY_DELAY,
        }
    }
}

/// [`acquire`] with explicit [`AcquireTimings`], so the busy-lock timeout
/// arm, the backup reclaim that follows it, and the sandwich retry can be
/// driven in milliseconds.
pub(crate) async fn acquire_with_timings(
    client: &DriveClient,
    opts: &AcquireOptions,
    authenticator: &dyn Authenticator,
    timings: AcquireTimings,
) -> AcquireResult {
    let (result, disposition) = acquire_inner(client, opts, authenticator, timings).await;
    record_attempt(opts, &result, disposition.as_ref());
    result
}

/// The accepted lease-expiry range (ADR-0080 §5): at least a minute, at
/// most 24 hours. Enforced here, in the engine, not only by the CLI's own
/// `--expiry-minutes` parser (`cli::drive::lease::parse_expiry_minutes`,
/// which shares these constants) — a non-CLI caller of this engine (a
/// future MCP tool, a test helper reused outside tests) could otherwise
/// mint a lease whose expiry silently violates the ADR's intent, or one
/// that overflows `chrono::Duration` (issue #1664 review finding).
pub(crate) const MIN_EXPIRY_MINUTES: i64 = 1;
/// See [`MIN_EXPIRY_MINUTES`].
pub(crate) const MAX_EXPIRY_MINUTES: i64 = 24 * 60;

/// What became of a backup this attempt took, once [`acquire_inner`]
/// decided it is referenced by no ledger row — the belt-and-braces half of
/// issue #1690's fix, for the narrow race the lock-free pre-check in
/// [`acquire_inner`] cannot close. `None` everywhere in
/// [`acquire_inner`]'s return value means this attempt never took a
/// backup at all (refused/denied/unavailable before ever reaching step 2).
enum BackupDisposition {
    /// Deleted (bytes) or trashed (Drive copy) before returning.
    Reclaimed(LeaseBackup),
    /// Reclaiming it itself failed; the backup is still on disk/Drive,
    /// now truly orphaned — `drive lease prune` cannot see it either,
    /// since no ledger row points at it. Carried so the audit record can
    /// still name its location for a human to clean up.
    ReclaimFailed(LeaseBackup),
}

async fn acquire_inner(
    client: &DriveClient,
    opts: &AcquireOptions,
    authenticator: &dyn Authenticator,
    timings: AcquireTimings,
) -> (AcquireResult, Option<BackupDisposition>) {
    let expiry_minutes = opts.expiry.num_minutes();
    if !(MIN_EXPIRY_MINUTES..=MAX_EXPIRY_MINUTES).contains(&expiry_minutes) {
        return (
            AcquireResult::Failed {
                detail: format!(
                    "expiry must be between {MIN_EXPIRY_MINUTES} and {MAX_EXPIRY_MINUTES} \
                     minutes, got {expiry_minutes}"
                ),
            },
            None,
        );
    }

    let files_api = FilesApi::new(client);
    let target = match files_api.get_metadata(&opts.file_id).await {
        Ok(target) => target,
        Err(err) => {
            return (
                AcquireResult::Failed {
                    detail: err.to_string(),
                },
                None,
            )
        }
    };

    let is_native = target.is_google_native();
    if is_native && opts.native_backup_folder_id.is_none() {
        return (AcquireResult::RefusedNativeDocument, None);
    }

    if target.version.is_none() {
        return (
            AcquireResult::Failed {
                detail: "Drive did not return a `version` for this file; refusing to lease it \
                         without a staleness check"
                    .to_string(),
            },
            None,
        );
    }

    // Lock-free live-lease pre-check (issue #1690): cheap, and
    // deliberately *not* the authoritative gate — `insert_record`, taken
    // under the ledger lock inside `mint` below, remains
    // that (see its own doc comment). This one's only job is refusing
    // *before* spending the (up to 120s) Touch ID prompt and a real
    // backup on the common case this closes: a second `acquire` on a
    // file that already has a live lease — a "did I already lease this?"
    // retry, a lost token — not merely a rare race. A lease that goes
    // live or expires in the window between this check and the
    // authoritative one is still decided correctly there either way: this
    // fast path can only ever be *more* conservative than the real gate
    // (refusing a lease that in fact expires a moment later, which just
    // means the caller retries), never less, so the two can never
    // disagree in the direction that would matter.
    let pre_check = tokio::task::block_in_place(|| LeaseLedger::load(&opts.ledger_path));
    match pre_check {
        Ok(ledger) => {
            if let Some(existing) =
                ledger.live_lease_for_file(&opts.file_id, Utc::now(), opts.supersedes.as_deref())
            {
                return (
                    AcquireResult::AlreadyLeased {
                        token: existing.token.clone(),
                        expires_at: existing.expires_at,
                    },
                    None,
                );
            }
        }
        Err(err) => {
            return (
                AcquireResult::Failed {
                    detail: err.to_string(),
                },
                None,
            )
        }
    }

    // 1. Authenticate — consent gates the action, not merely possession of
    // the resulting token (ADR-0080 §2). Nothing below runs on refusal.
    // `target.name` is Drive-controlled (renamable by anyone with edit
    // access to the file) and is shown verbatim inside the OS consent
    // prompt, so it is sanitized the same way any other server-supplied
    // string reaching a terminal or prompt is elsewhere in this CLI —
    // stripping control characters and bidi-override code points closes
    // off a spoofed/deceptive file name misleading what the operator is
    // authorising.
    let reason = format!(
        "back up and lease-write '{}'",
        sanitize_for_terminal(&target.name)
    );
    // `authenticate` blocks synchronously on the human's answer, up to
    // `PROMPT_TIMEOUT` (120s) — `block_in_place` hands this worker
    // thread's other queued tasks off to the runtime's other workers for
    // the duration, so a single Touch ID prompt cannot stall unrelated
    // concurrent work on a shared multi-thread runtime (the daemon and
    // the MCP server both run one). `acquire` itself stays a plain `async
    // fn` — `block_in_place` runs the closure on the *current* thread, so
    // it needs no `'static`/`Send` bound on `authenticator`, unlike
    // `spawn_blocking`.
    let auth_outcome =
        tokio::task::block_in_place(|| authenticator.authenticate(&reason, opts.auth_policy));
    let headless_waiver = match auth_outcome {
        AuthOutcome::Authorized => false,
        AuthOutcome::Denied(detail) => return (AcquireResult::Denied { detail }, None),
        AuthOutcome::NoAuthenticator(detail) => {
            // ADR-0080 §8/§13: an explicit, per-installation opt-out lets
            // this proceed with no human ever having been prompted, rather
            // than refusing outright. `headless_waiver` on the eventual
            // `Acquired` result (and so the audit record) is what makes
            // this waiver durably visible.
            if !opts.allow_headless {
                return (AcquireResult::Unavailable { detail }, None);
            }
            true
        }
        // Never waived, whatever `allow_headless` says (issue #1686): a
        // human may be sitting right there with Touch ID locked out or the
        // lid shut, and that is not headless (ADR-0080 §8).
        AuthOutcome::PolicyUnsatisfiable(detail) => {
            let detail = if opts.allow_headless {
                format!(
                    "{detail} — the headless opt-out does not apply: it waives only a context \
                     with no authenticator at all, not a policy this session cannot satisfy"
                )
            } else {
                detail
            };
            return (AcquireResult::Unavailable { detail }, None);
        }
    };

    // The backup below must end up pinned to the version `mint` records,
    // or the lease silently vouches for content it never captured (issue
    // #1693). A byte backup proves that pinning by content — its SHA-256
    // against the post-backup `sha256Checksum` — and needs nothing extra
    // here. Everything else has no checksum to compare, so it falls back to
    // sandwiching the backup between two `version` reads: a native document
    // has no byte content at all, and Drive omits `sha256Checksum` only for
    // native and shortcut files, so a byte target without one is a
    // defensive case rather than an expected one (adding an md5/sha1 ladder
    // for it would mean a new hash dependency). The sandwich's near read is
    // taken inside `take_pinned_backup`, deliberately *after* the prompt, so
    // the up-to-120s window a human spends deciding — during which the file
    // may legitimately move — is outside the span being pinned.
    let needs_version_pin = is_native || target.sha256_checksum.is_none();
    let pinned = match take_pinned_backup(
        &files_api,
        opts,
        &target.name,
        is_native,
        needs_version_pin,
        timings.sandwich_retry_delay,
    )
    .await
    {
        Ok(pinned) => pinned,
        Err(refusal) => return *refusal,
    };

    // From here on a pinned backup exists, so every remaining exit must
    // account for it: `Acquired` because a ledger row now references it,
    // every other outcome by reclaiming it (issue #1690's belt-and-braces
    // half, for the pre-check's narrow remaining race).
    let backup = pinned.backup.clone();
    let result = mint(opts, pinned, headless_waiver, timings.lock_wait);
    if matches!(result, AcquireResult::Acquired { .. }) {
        return (result, None);
    }
    let disposition = reclaim_backup(&files_api, backup).await;
    (result, Some(disposition))
}

/// A backup [`take_pinned_backup`] has proven holds the content Drive
/// reports at `version` — what [`mint`] records against it.
struct PinnedBackup {
    backup: LeaseBackup,
    /// The file's `version`, read after the backup.
    version: String,
    /// The file's `modifiedTime`, from the same read as `version`.
    modified_time: Option<String>,
}

/// Steps 2–3's proof: takes the backup — bytes for a binary file, a
/// Drive-side copy for a native document (ADR-0080 §3) — and proves it is
/// pinned to the version about to be recorded (see [`backup_is_pinned`]).
///
/// A version-sandwich refusal ([`PinFailure::VersionMoved`]) is retried up
/// to [`SANDWICH_ATTEMPTS`] times, reclaiming each unpinned backup before
/// the next is taken (issue #1739). Nothing else is retried: a checksum
/// mismatch is positive proof the *content* moved, and each retry would
/// re-download up to 500 MB to find out whether it stopped; a missing
/// proof or an API error is not a race at all. Retrying also stops the
/// moment a reclaim fails, so a run of attempts leaves at most one orphan
/// — the one the returned disposition names for the audit record.
///
/// `Err` carries the finished outcome for `acquire_inner` to return: the
/// refusal or failure, and what became of the last backup, if one was
/// taken — boxed, since it otherwise dwarfs the `Ok` side
/// (`clippy::result_large_err`).
async fn take_pinned_backup(
    files_api: &FilesApi<'_>,
    opts: &AcquireOptions,
    name: &str,
    is_native: bool,
    needs_version_pin: bool,
    retry_delay: Duration,
) -> Result<PinnedBackup, Box<(AcquireResult, Option<BackupDisposition>)>> {
    let mut attempt = 1;
    loop {
        let pre_backup_version = if needs_version_pin {
            let version = read_pre_backup_version(files_api, &opts.file_id)
                .await
                .map_err(|detail| Box::new((AcquireResult::Failed { detail }, None)))?;
            Some(version)
        } else {
            None
        };

        let backup = take_backup(files_api, opts, name, is_native)
            .await
            .map_err(|err| {
                Box::new((
                    AcquireResult::Failed {
                        detail: err.to_string(),
                    },
                    None,
                ))
            })?;

        let failure = match prove_pinned(
            files_api,
            &opts.file_id,
            &backup,
            pre_backup_version.as_deref(),
        )
        .await
        {
            PinCheck::Pinned {
                version,
                modified_time,
            } => {
                return Ok(PinnedBackup {
                    backup,
                    version,
                    modified_time,
                })
            }
            PinCheck::Failed(detail) => {
                let disposition = reclaim_backup(files_api, backup).await;
                return Err(Box::new((
                    AcquireResult::Failed { detail },
                    Some(disposition),
                )));
            }
            PinCheck::Unpinned(failure) => failure,
        };

        let disposition = reclaim_backup(files_api, backup).await;
        let retry = failure.is_retryable()
            && attempt < SANDWICH_ATTEMPTS
            && matches!(disposition, BackupDisposition::Reclaimed(_));
        if !retry {
            return Err(Box::new((
                AcquireResult::RefusedConcurrentChange {
                    detail: failure.detail(
                        attempt,
                        matches!(disposition, BackupDisposition::Reclaimed(_)),
                    ),
                },
                Some(disposition),
            )));
        }
        tracing::info!(
            "drive lease acquire: the file's version moved across its backup (attempt \
             {attempt} of {SANDWICH_ATTEMPTS}); discarded that backup, retrying"
        );
        tokio::time::sleep(retry_delay).await;
        attempt += 1;
    }
}

/// The near half of the version sandwich: the file's `version` read
/// immediately before the backup. `Err` is the operator-facing detail for
/// [`AcquireResult::Failed`].
async fn read_pre_backup_version(
    files_api: &FilesApi<'_>,
    file_id: &str,
) -> Result<String, String> {
    match files_api.get_metadata(file_id).await {
        Ok(DriveFile {
            version: Some(version),
            ..
        }) => Ok(version),
        // Refused *here*, not left for `backup_is_pinned`'s no-proof arm:
        // that arm would refuse just the same, but only after the backup —
        // a download of up to 500 MB or a `files.copy` — had been taken and
        // reclaimed for nothing. Mirrors the pre-auth read's own
        // missing-`version` refusal.
        Ok(_) => Err(
            "Drive did not return a `version` for this file immediately before the \
                      backup; refusing to take a backup that could not be pinned to one"
                .to_string(),
        ),
        Err(err) => Err(err.to_string()),
    }
}

/// Step 2: a backup of the file as it stands now — bytes for a binary file,
/// a Drive-side copy for a native document (ADR-0080 §3).
/// `native_backup_folder_id` is guaranteed `Some` whenever `is_native`, by
/// `acquire_inner`'s up-front refusal. `name` only labels the backup.
async fn take_backup(
    files_api: &FilesApi<'_>,
    opts: &AcquireOptions,
    name: &str,
    is_native: bool,
) -> anyhow::Result<LeaseBackup> {
    if is_native {
        let folder_id = opts.native_backup_folder_id.as_deref().unwrap_or_default();
        native_backup(files_api, &opts.file_id, folder_id, name).await
    } else {
        byte_backup(files_api, &opts.file_id, &opts.backup_dir, name).await
    }
}

/// What [`prove_pinned`] concluded about one backup.
enum PinCheck {
    /// The backup provably holds the content Drive reports at `version`.
    Pinned {
        version: String,
        modified_time: Option<String>,
    },
    /// The backup cannot be pinned to the version that would be recorded.
    Unpinned(PinFailure),
    /// The post-backup read failed, or returned no `version` to record.
    Failed(String),
}

/// Step 3's proof: re-fetches metadata and checks the backup against it.
///
/// `version`/`modified_time` are re-fetched here rather than reused from
/// the `target` metadata read at the very top of `acquire_inner` — that
/// read happened before the (up to 120s) Touch ID prompt and before the
/// backup itself, so the file could have moved in the meantime. Recording
/// that earlier, now-possibly-stale version would break the token's own
/// invariant of being "bound to a specific backup and a specific Drive
/// version" (ADR-0080 §2/§4): the backup reflects whatever the file was at
/// backup time, so the recorded version must too, or the very next write
/// under this lease could spuriously refuse as stale (or, worse, pass a
/// staleness check against a version that doesn't match what was actually
/// backed up).
///
/// `pre_backup_version` is `Some` exactly when `acquire_inner` decided this
/// target needs the version sandwich rather than the checksum proof — see
/// [`backup_is_pinned`].
async fn prove_pinned(
    files_api: &FilesApi<'_>,
    file_id: &str,
    backup: &LeaseBackup,
    pre_backup_version: Option<&str>,
) -> PinCheck {
    let post_backup = match files_api.get_metadata(file_id).await {
        Ok(post_backup) => post_backup,
        Err(err) => return PinCheck::Failed(err.to_string()),
    };
    let Some(version) = post_backup.version else {
        return PinCheck::Failed(
            "Drive did not return a `version` for this file after the backup; refusing to \
             record a lease without a staleness check"
                .to_string(),
        );
    };
    // Re-fetching alone establishes only that `version` is current, never
    // that it is the version the backup above actually captured — the
    // download (`alt=media`, bytes only) and `files.copy` (whose reply
    // carries the *copy's* version) both leave that unstated, and for a
    // byte backup the gap spans the download, the hash and a disk write of
    // up to 500 MB. Proving it is what stops a lease vouching for content
    // it never backed up (issue #1693).
    match backup_is_pinned(
        backup,
        post_backup.sha256_checksum.as_deref(),
        &version,
        pre_backup_version,
    ) {
        Ok(()) => PinCheck::Pinned {
            version,
            modified_time: post_backup.modified_time,
        },
        Err(failure) => PinCheck::Unpinned(failure),
    }
}

/// Step 3's record and step 4's mint: checks the file's live lease under
/// the ledger lock and inserts the row for `pinned`. Only ever decides
/// mint-or-refuse — reclaiming the backup on a refusal is `acquire_inner`'s
/// job, so every exit past the backup step shares its one reclaim wrapper.
fn mint(
    opts: &AcquireOptions,
    pinned: PinnedBackup,
    headless_waiver: bool,
    lock_wait: Duration,
) -> AcquireResult {
    let PinnedBackup {
        backup,
        version,
        modified_time,
    } = pinned;
    let token = crate::request_log::new_id();
    let now = Utc::now();
    let expires_at = now + opts.expiry;
    let record = LeaseRecord {
        token: token.clone(),
        file_id: opts.file_id.clone(),
        version,
        modified_time,
        backup: backup.clone(),
        acquired_at: now,
        expires_at,
        released_at: None,
        superseded_by: None,
        restored_at: None,
        restored_sheet_id: None,
    };
    // Synchronous ledger I/O (lock, load, save) on the async runtime's
    // current thread — `block_in_place` hands its other queued tasks off
    // to the runtime's other workers for the duration, the same reasoning
    // the `authenticate` call in `acquire_inner` documents.
    let superseded = match tokio::task::block_in_place(|| {
        insert_record(
            record,
            &opts.ledger_path,
            opts.supersedes.as_deref(),
            lock_wait,
        )
    }) {
        Ok(InsertOutcome::Inserted { superseded }) => superseded,
        // Refuse rather than mint a second, independent lease — see
        // `AcquireResult::AlreadyLeased`'s and `insert_record`'s own doc
        // comments for why this stays the authoritative gate even with
        // the lock-free pre-check in `acquire_inner`. Reclaiming the
        // backup this attempt just took happens there, once it sees this
        // isn't `Acquired` — not here, since this function's only job is
        // deciding mint-or-refuse.
        Ok(InsertOutcome::AlreadyLeased(existing)) => {
            return AcquireResult::AlreadyLeased {
                token: existing.token,
                expires_at: existing.expires_at,
            };
        }
        Err(err) => {
            return AcquireResult::Failed {
                detail: err.to_string(),
            };
        }
    };

    // 4. Print the token (the caller's job) and exit 0.
    AcquireResult::Acquired {
        token,
        expires_at,
        backup,
        headless_waiver,
        superseded_lease_id: opts.supersedes.clone().filter(|_| superseded),
    }
}

/// Why [`backup_is_pinned`] could not pin a backup — one variant per arm,
/// so [`take_pinned_backup`] can tell the one worth retrying from the rest
/// without parsing the operator-facing text.
#[derive(Debug)]
enum PinFailure {
    /// The bytes backed up hash to `local`, but Drive reports `remote` for
    /// the file at `post_version`: the content moved under the download.
    ChecksumMismatch {
        remote: String,
        post_version: String,
        local: String,
    },
    /// The version sandwich's two reads disagree.
    VersionMoved { pre: String, post: String },
    /// Neither proof was available.
    NoProof { post_version: String },
}

impl PinFailure {
    /// Whether another attempt could plausibly succeed. Only a moved
    /// `version` is: it cannot tell a content edit from a bump unrelated
    /// to any edit, and the latter may well not recur (issue #1739).
    fn is_retryable(&self) -> bool {
        matches!(self, Self::VersionMoved { .. })
    }

    /// The operator-facing detail for
    /// [`AcquireResult::RefusedConcurrentChange`], after `attempts`
    /// backups, the last of which was `discarded` (reclaimed) or not.
    ///
    /// A version that moved on every one of the full [`SANDWICH_ATTEMPTS`]
    /// gets its own wording: "wait until the file is quiet" is the wrong
    /// advice if the version moves for a reason no quiet period will cure.
    /// A last backup that could not be discarded is said so, never claimed
    /// discarded — it is an orphan `drive lease prune` cannot see, and
    /// retrying stopped early because of it (see [`take_pinned_backup`]).
    fn detail(&self, attempts: u32, discarded: bool) -> String {
        let what = match self {
            Self::ChecksumMismatch {
                remote,
                post_version,
                local,
            } => format!(
                "the file changed while its backup was being taken: Drive reports checksum \
                 {remote} at version {post_version}, but the bytes backed up hash to {local}."
            ),
            Self::VersionMoved { pre, post } if attempts > 1 && discarded => format!(
                "the file's version changed across its backup on each of {attempts} attempts \
                 (last: version {pre} immediately before, {post} immediately after). A version \
                 that keeps moving may be changing for reasons unrelated to edits, so waiting \
                 for the file to be quiet may not help."
            ),
            Self::VersionMoved { pre, post } if attempts > 1 => format!(
                "the file's version changed across its backup on each of {attempts} attempts \
                 (last: version {pre} immediately before, {post} immediately after)."
            ),
            Self::VersionMoved { pre, post } => format!(
                "the file changed while its backup was being taken: version {pre} immediately \
                 before, {post} immediately after."
            ),
            Self::NoProof { post_version } => format!(
                "no proof available that the backup matches version {post_version}: Drive \
                 reported no checksum for the file after the backup and no pre-backup version \
                 was taken to fall back on."
            ),
        };
        let outcome = match (discarded, attempts > 1) {
            (true, false) => "No lease was minted and the backup was discarded — retry.",
            (true, true) => "No lease was minted and the backups were discarded.",
            (false, false) => {
                "No lease was minted, but the backup could not be discarded — its location is \
                 in the audit log (`omni-dev log --audit`); remove it by hand, then retry."
            }
            (false, true) => {
                "No lease was minted. Retrying stopped early because the last backup could not \
                 be discarded — its location is in the audit log (`omni-dev log --audit`); \
                 remove it by hand. Every earlier backup was discarded."
            }
        };
        format!("{what} {outcome}")
    }
}

/// Whether `backup` provably holds the content Drive reports at
/// `post_version`, so that recording `post_version` against it cannot
/// vouch for content the backup never captured (issue #1693). `Err` says
/// why not — see [`PinFailure`].
///
/// Two proofs, picked by what evidence Drive actually offers for this
/// target:
///
/// 1. **By content, for a byte backup with a `sha256Checksum`.** The digest
///    of the bytes on disk against the digest Drive reports for the file
///    *now*. Equal means the backup is that revision's content, whatever
///    happened meanwhile — which makes this strictly stronger than
///    comparing versions across the window: it also covers a download that
///    was not served as one coherent snapshot, something no pair of
///    metadata reads can see. It is also why a rename, move or ACL change
///    during the backup is *not* refused: those bump `version` without
///    touching content, so the pin still holds and refusing would be
///    spurious.
/// 2. **By sandwich, for everything else.** A native document has no byte
///    content to hash and `files.copy` reports no digest. It falls back to
///    requiring the `version` read just before the backup to equal the one
///    read just after. Sound but blunter: it cannot tell a content edit
///    from a metadata-only bump, so the latter refuses too. That
///    conservatism is the cost of having no checksum to compare, not a
///    judgement that the backup is bad — and why [`take_pinned_backup`]
///    retries this arm.
///
/// Which proof applies is decided by *this* function from the post-backup
/// read, while whether a pre-backup version exists to sandwich with was
/// decided by `acquire_inner` from the pre-auth read. For a native target
/// the two always agree. For a byte target they can differ: one that
/// reported no digest before the backup carries `pre_backup_version`, yet
/// is proven by content if a digest appears after it — the stronger proof
/// wins. (Drive omits `sha256Checksum` only for native and shortcut files,
/// so a byte target reaching the sandwich at all is a defensive path.) The
/// one way to arrive with *neither* proof is the reverse: a byte target
/// that reported a digest before the backup but none after it — a file
/// that stopped being binary content in between, itself a concurrent
/// change — and that refuses too: the two arms are exhaustive, never a
/// silent pass.
fn backup_is_pinned(
    backup: &LeaseBackup,
    remote_sha256: Option<&str>,
    post_version: &str,
    pre_backup_version: Option<&str>,
) -> Result<(), PinFailure> {
    if let (LeaseBackup::Bytes { sha256, .. }, Some(remote)) = (backup, remote_sha256) {
        if !sha256.eq_ignore_ascii_case(remote) {
            return Err(PinFailure::ChecksumMismatch {
                remote: remote.to_string(),
                post_version: post_version.to_string(),
                local: sha256.clone(),
            });
        }
        return Ok(());
    }

    match pre_backup_version {
        Some(pre) if pre != post_version => Err(PinFailure::VersionMoved {
            pre: pre.to_string(),
            post: post_version.to_string(),
        }),
        Some(_) => Ok(()),
        // Reached only when the target's digest vanished between the
        // pre-auth read (which decided no sandwich was needed) and the
        // post-backup one — or if `acquire_inner`'s decision and this
        // function's arms are ever changed out of step. (A sandwich read
        // that returned no `version` never gets this far:
        // `read_pre_backup_version` refuses that before taking the
        // backup.) Either way nothing is recorded unproven.
        None => Err(PinFailure::NoProof {
            post_version: post_version.to_string(),
        }),
    }
}

/// Deletes/trashes a backup this attempt itself just took, once
/// `acquire_inner` has decided it ends up referenced by no ledger row —
/// the belt-and-braces half of issue #1690's fix, for the narrow race the
/// lock-free pre-check above cannot close by itself. Reuses
/// [`super::prune::clear_backup`], which already handles both
/// [`LeaseBackup`] variants and tolerates an already-absent backup. Safe
/// specifically because `backup` is always one this very call just
/// created moments ago under a fresh identity — `write_backup`'s
/// `create_new`/`O_EXCL` open, or `native_backup`'s brand-new `files.copy`
/// id — never one an existing ledger row (live, expired, or already
/// pruned) could still point at.
///
/// A reclamation failure is warned, not surfaced: the primary outcome
/// (`AlreadyLeased`/`Failed`) is unaffected either way, and the orphan's
/// location still ends up in the best-effort audit record
/// ([`record_attempt`]) for `drive lease prune` — or a human — to find.
async fn reclaim_backup(files_api: &FilesApi<'_>, backup: LeaseBackup) -> BackupDisposition {
    match super::prune::clear_backup(files_api, &backup).await {
        Ok(()) => BackupDisposition::Reclaimed(backup),
        Err(err) => {
            let (location, ..) = backup.audit_fields();
            tracing::warn!(
                "drive lease acquire: failed to reclaim an orphaned backup at {}: {err} — \
                 `drive lease prune` cannot see it either, since no ledger row references it; \
                 remove it manually",
                location.unwrap_or_default()
            );
            BackupDisposition::ReclaimFailed(backup)
        }
    }
}

/// What [`insert_record`] did.
enum InsertOutcome {
    /// The record was inserted; the lease is minted. `superseded` is whether
    /// the `supersedes` token's row was live and got released in this same
    /// rewrite — `false` when it named no row, another file's row, or a row
    /// already expired or released, none of which this insert touched.
    Inserted { superseded: bool },
    /// A live lease already covered this record's `file_id` — nothing was
    /// inserted or overwritten. Carries that existing record so the caller
    /// can return its token instead — boxed, since it otherwise dwarfs the
    /// dataless `Inserted` variant (`clippy::large_enum_variant`), the same
    /// reasoning `restore.rs`'s `GateCheck::Ok` documents.
    AlreadyLeased(Box<LeaseRecord>),
}

/// Downloads a binary file's bytes and writes them to `backup_dir`
/// (ADR-0080 §3's binary-file case).
async fn byte_backup(
    files_api: &FilesApi<'_>,
    file_id: &str,
    backup_dir: &Path,
    name: &str,
) -> anyhow::Result<LeaseBackup> {
    let bytes = files_api.download(file_id).await?;
    let sha256 = crate::cli::drive::read::to_hex_string(&Sha256::digest(&bytes));
    let path = backup_file_path(backup_dir, file_id, name);
    // `write_backup` is synchronous filesystem I/O for a backup of arbitrary
    // size — `block_in_place` hands this worker thread's other queued tasks
    // off to the runtime's other workers for the duration, the same
    // reasoning the `authenticate` call above documents.
    tokio::task::block_in_place(|| write_backup(&path, &bytes))?;
    Ok(LeaseBackup::Bytes {
        path,
        sha256,
        size: bytes.len() as u64,
    })
}

/// Copies a native document into `backup_folder_id` via `files.copy`
/// (ADR-0080 §3's native-document case) — lossless, and restorable by a
/// human in the Drive UI even without this tool.
async fn native_backup(
    files_api: &FilesApi<'_>,
    file_id: &str,
    backup_folder_id: &str,
    name: &str,
) -> anyhow::Result<LeaseBackup> {
    let copy_name = backup_name(file_id, name);
    let copy = files_api
        .copy(file_id, backup_folder_id, &copy_name)
        .await?;
    Ok(LeaseBackup::DriveCopy { file_id: copy.id })
}

/// `<dir>` joined with [`backup_name`]'s result.
fn backup_file_path(dir: &Path, file_id: &str, name: &str) -> PathBuf {
    dir.join(backup_name(file_id, name))
}

/// `<YYYYMMDDTHHMMSSZ>-<fileId>-<name>` (ADR-0080 §3) — UTC, seconds
/// precision, the file id first to survive a name containing `/`. Shared by
/// the local byte-backup path ([`backup_file_path`] joins it under a
/// directory) and the native Drive-copy path (used directly as the copy's
/// own `name`).
fn backup_name(file_id: &str, name: &str) -> String {
    let timestamp = Utc::now().format("%Y%m%dT%H%M%SZ");
    let safe_name = name.replace('/', "_");
    format!("{timestamp}-{file_id}-{safe_name}")
}

/// Writes `bytes` to `path`, creating a missing `0700` parent directory
/// first — the same posture `crate::request_log`/the lease ledger use.
///
/// Opened `0600`-from-birth (never `std::fs::write`'s umask-derived mode)
/// — a backup is exactly the private, sensitive content this feature
/// exists to protect. Also `create_new` (`O_EXCL`) rather than a
/// truncating write: [`backup_file_path`]'s name has only whole-second
/// precision, so two acquisitions for the same file within one second
/// collide on the same path. Failing loudly here, instead of silently
/// overwriting, is what stops that rare collision from corrupting the
/// *earlier* lease's ledger row — its recorded `backup_sha256` would
/// otherwise no longer match the bytes actually on disk.
fn write_backup(path: &Path, bytes: &[u8]) -> anyhow::Result<()> {
    use std::io::Write as _;

    crate::daemon::paths::ensure_parent_dir_0700(path)?;
    let mut file = crate::daemon::paths::create_new_file_0600(path).map_err(|err| {
        // Distinguish a genuine same-second collision (the path already
        // existed) from the file being created fine but the follow-up
        // `fchmod` safety net failing — the latter is an unrelated
        // permissions/filesystem problem that "may already exist, retry"
        // would misdiagnose (issue #1664 review finding).
        if crate::daemon::paths::is_already_exists_error(&err) {
            err.context(format!(
                "Failed to create backup file at {} — it may already exist from a \
                 near-simultaneous `drive lease acquire` on the same file within the same \
                 second; retry",
                path.display()
            ))
        } else {
            err.context(format!(
                "Failed to create backup file at {} — not a same-second collision; check \
                 filesystem permissions",
                path.display()
            ))
        }
    })?;
    file.write_all(bytes)
        .with_context(|| format!("Failed to write backup to {}", path.display()))?;
    Ok(())
}

/// Inserts `record` into the ledger at `ledger_path` under the ledger lock,
/// loading and saving it in full — the atomic-rewrite contract
/// [`LeaseLedger`] documents. This lock-held check-then-insert is the
/// authoritative gate against two leases ever being live on the same file
/// at once (see [`LeaseLedger::live_lease_for_file`]'s doc comment) — unlike
/// a check made before taking the lock, nothing can race it.
///
/// Waits up to `lock_wait` for a busy lock rather than refusing (issue
/// #1738): by now this attempt has spent the Touch ID prompt and a full
/// backup, so hard-failing behind an unrelated lease operation would throw
/// both away. Blocking is safe here because the one production caller,
/// `mint`, already runs this under `block_in_place`.
///
/// `supersedes` names a lease this insert replaces
/// ([`AcquireOptions::supersedes`]), released here in the *same* rewrite that
/// inserts `record`, so the file is covered by exactly one live lease at
/// every instant.
fn insert_record(
    record: LeaseRecord,
    ledger_path: &Path,
    supersedes: Option<&str>,
    lock_wait: Duration,
) -> anyhow::Result<InsertOutcome> {
    let lock = LedgerLock::acquire_waiting_blocking_with_timeout(ledger_path, lock_wait)?;
    let new_token = record.token.clone();
    LeaseLedger::mutate(&lock, ledger_path, |ledger| {
        let now = Utc::now();
        // Order is load-bearing: check, then release, then insert.
        // `mutate` saves whatever the closure leaves behind even when
        // it returns a refusal, so releasing before this check would end the
        // superseded lease's window on a path that mints nothing to replace
        // it — losing the caller's lease to a *third* lease's race.
        if let Some(existing) = ledger.live_lease_for_file(&record.file_id, now, supersedes) {
            return InsertOutcome::AlreadyLeased(Box::new(existing.clone()));
        }
        // Scoped to a row already covering this same file, so naming another
        // file's token can never release it — the exclusion above is
        // likewise `file_id`-filtered, so the two agree on what "supersede"
        // can reach. `new_token` is stamped as `superseded_by` on the
        // released row (issue #1768) — the durable marker `prune`'s
        // `--max-size` filter uses to tell this deliberate supersede apart
        // from a plain `drive lease release`.
        let superseded = supersedes.is_some_and(|token| {
            ledger
                .get(token)
                .is_some_and(|r| r.file_id == record.file_id)
                && matches!(
                    ledger.release(token, now, Some(new_token.as_str())),
                    ReleaseOutcome::Released { .. }
                )
        });
        ledger.insert(record);
        InsertOutcome::Inserted { superseded }
    })
}

/// Whether `disposition` names a backup this attempt took and, if
/// reclaiming it failed, extends `base_verdict` to `<base>-backup-orphaned`
/// so an operator can grep for a leaked backup `drive lease prune` cannot
/// see (issue #1690) — see [`BackupDisposition`]. Returns `base_verdict`
/// unchanged, with no `backup_location`, when this attempt never took a
/// backup at all.
fn orphan_verdict(
    base_verdict: &str,
    disposition: Option<&BackupDisposition>,
) -> (String, Option<String>) {
    let backup = match disposition {
        None => return (base_verdict.to_string(), None),
        Some(BackupDisposition::Reclaimed(backup)) => backup,
        Some(BackupDisposition::ReclaimFailed(backup)) => {
            let (location, ..) = backup.audit_fields();
            return (format!("{base_verdict}-backup-orphaned"), location);
        }
    };
    let (location, ..) = backup.audit_fields();
    (base_verdict.to_string(), location)
}

/// Builds and writes the `kind: "audit"` record for one acquire attempt.
/// See [`acquire`]'s own doc comment for why this is best-effort rather than
/// write-ahead/fail-closed. `disposition` is `Some` only when this attempt
/// took a backup that turned out to be reclaimed or orphaned rather than
/// referenced by a ledger row (issue #1690) — see [`BackupDisposition`].
fn record_attempt(
    opts: &AcquireOptions,
    result: &AcquireResult,
    disposition: Option<&BackupDisposition>,
) {
    let auth_policy = Some(
        match opts.auth_policy {
            AuthPolicy::DeviceOwner => "device-owner",
            AuthPolicy::BiometricsOnly => "biometrics-only",
        }
        .to_string(),
    );
    let outcome = match result {
        AcquireResult::Acquired {
            token,
            backup,
            expires_at: _,
            headless_waiver,
            superseded_lease_id,
        } => {
            let (backup_location, backup_sha256, backup_size) = backup.audit_fields();
            // Re-read the just-written ledger row for the version/
            // modified_time actually recorded, rather than widening
            // `AcquireResult::Acquired` (a public, `--output json` wire
            // shape) with fields that exist only for this best-effort audit
            // record. A read failure here just omits them — the acquisition
            // itself already fully succeeded. (`superseded_lease_id` is the
            // deliberate exception: it is a fact of what the locked rewrite
            // did, not state a later read can recover, so it rides the
            // result — skip-serialized when `None`.)
            let (version_after, modified_time_after) = LeaseLedger::load(&opts.ledger_path)
                .ok()
                .and_then(|ledger| ledger.get(token).cloned())
                .map_or((None, None), |record| {
                    (Some(record.version), record.modified_time)
                });
            crate::request_log::AuditOutcome {
                command: vec!["drive".to_string(), "lease-acquire".to_string()],
                integration: "drive",
                file_id: opts.file_id.clone(),
                lease_id: Some(token.clone()),
                // ADR-0080 §8/§13, issue #1677: a distinct verdict, rather
                // than a separate field on this shared, free-form-vocabulary
                // struct (ADR-0080 §11), durably distinguishes an
                // acquisition that waived the human-presence guarantee from
                // a normally-authorised one.
                verdict: if *headless_waiver {
                    "acquired-headless-waiver".to_string()
                } else {
                    "acquired".to_string()
                },
                version_after,
                modified_time_after,
                backup_location,
                backup_sha256,
                backup_size,
                auth_policy,
                // `Some` only when `insert_record` really released a row
                // (issue #1685) — `AcquireResult::Acquired` carries that
                // fact out of the locked rewrite, so this never claims a
                // release the supersede did not make.
                superseded_lease_id: superseded_lease_id.clone(),
                ..Default::default()
            }
        }
        AcquireResult::AlreadyLeased { token, .. } => {
            let (verdict, backup_location) = orphan_verdict("already-leased", disposition);
            crate::request_log::AuditOutcome {
                command: vec!["drive".to_string(), "lease-acquire".to_string()],
                integration: "drive",
                file_id: opts.file_id.clone(),
                lease_id: Some(token.clone()),
                verdict,
                backup_location,
                auth_policy,
                ..Default::default()
            }
        }
        AcquireResult::RefusedNativeDocument => crate::request_log::AuditOutcome {
            command: vec!["drive".to_string(), "lease-acquire".to_string()],
            integration: "drive",
            file_id: opts.file_id.clone(),
            verdict: "refused-native-document".to_string(),
            auth_policy,
            ..Default::default()
        },
        // Routed through `orphan_verdict` like `already-leased`, not
        // emitted flat like the refusals above it: this one can only ever
        // fire *after* a real backup was taken, so it always names that
        // backup's location, and a failed reclaim of it must still carry
        // the `-backup-orphaned` suffix issue #1690 added.
        AcquireResult::RefusedConcurrentChange { detail } => {
            let (verdict, backup_location) =
                orphan_verdict("refused-concurrent-change", disposition);
            crate::request_log::AuditOutcome {
                command: vec!["drive".to_string(), "lease-acquire".to_string()],
                integration: "drive",
                file_id: opts.file_id.clone(),
                verdict,
                error: Some(detail.clone()),
                backup_location,
                auth_policy,
                ..Default::default()
            }
        }
        AcquireResult::Denied { detail } => crate::request_log::AuditOutcome {
            command: vec!["drive".to_string(), "lease-acquire".to_string()],
            integration: "drive",
            file_id: opts.file_id.clone(),
            verdict: "denied".to_string(),
            error: Some(detail.clone()),
            auth_policy,
            ..Default::default()
        },
        AcquireResult::Unavailable { detail } => crate::request_log::AuditOutcome {
            command: vec!["drive".to_string(), "lease-acquire".to_string()],
            integration: "drive",
            file_id: opts.file_id.clone(),
            verdict: "unavailable".to_string(),
            error: Some(detail.clone()),
            auth_policy,
            ..Default::default()
        },
        AcquireResult::Failed { detail } => {
            let (verdict, backup_location) = orphan_verdict("failed", disposition);
            crate::request_log::AuditOutcome {
                command: vec!["drive".to_string(), "lease-acquire".to_string()],
                integration: "drive",
                file_id: opts.file_id.clone(),
                verdict,
                error: Some(detail.clone()),
                backup_location,
                auth_policy,
                ..Default::default()
            }
        }
    };
    if let Err(err) = crate::request_log::record_audit_event(outcome) {
        tracing::warn!("drive lease acquire: failed to write audit record: {err}");
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::drive::auth::{DriveCredentials, DriveGrantedScopes};
    use crate::drive::lease::authenticate::Unsupported;
    use crate::drive::test_support::FakeAuthenticator;
    use crate::test_support::AuditLogGuard as AuditGuard;
    use crate::utils::secret::Secret;

    fn test_credentials() -> DriveCredentials {
        DriveCredentials {
            client_id: "client-1".to_string(),
            client_secret: Secret::new("secret-1"),
            refresh_token: Secret::new("refresh-1"),
            scope: DriveGrantedScopes::READONLY,
        }
    }

    async fn client_with_bootstrapped_token(server: &wiremock::MockServer) -> DriveClient {
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/token"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "access_token": "test-token",
                    "expires_in": 3600,
                })),
            )
            .mount(server)
            .await;

        let mut client = DriveClient::new(&server.uri(), &test_credentials()).unwrap();
        crate::drive::client::test_support::replace_session(
            &mut client,
            &test_credentials(),
            &format!("{}/token", server.uri()),
        );
        client
    }

    /// An [`Authenticator`] that panics if ever called, carrying its own
    /// panic message — for tests asserting that a refusal short-circuits
    /// before any authentication prompt is issued.
    struct PanicsIfCalled(&'static str);
    impl Authenticator for PanicsIfCalled {
        fn authenticate(&self, _reason: &str, _policy: AuthPolicy) -> AuthOutcome {
            panic!("{}", self.0);
        }
    }

    /// Builds options rooted at `dir` (a `tempdir`), so a test never
    /// touches the real backup directory or the real lease ledger.
    fn opts(dir: &Path) -> AcquireOptions {
        AcquireOptions {
            file_id: "f1".to_string(),
            backup_dir: dir.join("backups"),
            native_backup_folder_id: None,
            expiry: ChronoDuration::minutes(30),
            auth_policy: AuthPolicy::DeviceOwner,
            ledger_path: dir.join("lease-ledger.jsonl"),
            allow_headless: false,
            supersedes: None,
        }
    }

    /// A ledger-lock wait budget for tests: long enough that a holder
    /// released after ~100ms is always waited out, short enough that a
    /// regression into a hang fails in seconds rather than minutes.
    const TEST_LOCK_WAIT: Duration = Duration::from_secs(5);

    /// [`AcquireTimings`] for tests: [`TEST_LOCK_WAIT`], and no pause
    /// between sandwich attempts, so a retried acquisition runs in
    /// milliseconds rather than seconds.
    const FAST_TIMINGS: AcquireTimings = AcquireTimings {
        lock_wait: TEST_LOCK_WAIT,
        sandwich_retry_delay: Duration::ZERO,
    };

    #[tokio::test(flavor = "multi_thread")]
    async fn denied_authentication_takes_no_backup_and_writes_no_ledger_row() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .expect(1)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &FakeAuthenticator(AuthOutcome::Denied("no".to_string())),
        )
        .await;

        assert!(matches!(result, AcquireResult::Denied { .. }));
        assert!(!root.path().join("backups").exists());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn unavailable_authentication_takes_no_backup() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(&client, &opts(root.path()), &Unsupported).await;

        assert!(matches!(result, AcquireResult::Unavailable { .. }));
        assert!(!root.path().join("backups").exists());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn headless_opt_out_proceeds_without_an_authenticator() {
        // ADR-0080 §8/§13, issue #1677: the same `Unsupported` authenticator
        // as `unavailable_authentication_takes_no_backup` above, but with
        // `allow_headless: true` — this must now proceed to a real backup
        // and ledger row instead of refusing, with `headless_waiver` set so
        // the waiver is durably visible.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"bytes".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &AcquireOptions {
                allow_headless: true,
                ..opts(root.path())
            },
            &Unsupported,
        )
        .await;

        let AcquireResult::Acquired {
            headless_waiver, ..
        } = result
        else {
            panic!("expected Acquired, got {result:?}");
        };
        assert!(headless_waiver);
        assert!(root.path().join("backups").exists());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn headless_opt_out_never_waives_an_unsatisfiable_policy() {
        // Issue #1686: Touch ID locked out, or the lid shut, on an attended
        // Mac is not headless — `allow_headless: true` must still refuse,
        // take no backup, and say why the opt-out did not apply.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &AcquireOptions {
                allow_headless: true,
                ..opts(root.path())
            },
            &FakeAuthenticator(AuthOutcome::PolicyUnsatisfiable(
                "Biometry is locked out.".to_string(),
            )),
        )
        .await;

        let AcquireResult::Unavailable { detail } = result else {
            panic!("expected Unavailable, got {result:?}");
        };
        assert!(detail.starts_with("Biometry is locked out."), "{detail}");
        assert!(
            detail.contains("headless opt-out does not apply"),
            "{detail}"
        );
        assert!(!root.path().join("backups").exists());
        assert!(!root.path().join("lease-ledger.jsonl").exists());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn unsatisfiable_policy_without_the_opt_out_reports_the_platform_message() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &FakeAuthenticator(AuthOutcome::PolicyUnsatisfiable(
                "Authentication canceled.".to_string(),
            )),
        )
        .await;

        let AcquireResult::Unavailable { detail } = result else {
            panic!("expected Unavailable, got {result:?}");
        };
        assert_eq!(detail, "Authentication canceled.");
    }

    #[tokio::test]
    async fn get_metadata_failure_is_reported_as_failed_before_authenticating() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"code": 404, "message": "File not found"}
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &PanicsIfCalled("must not authenticate when metadata lookup already failed"),
        )
        .await;
        assert!(matches!(result, AcquireResult::Failed { .. }));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn download_failure_after_authorization_is_reported_as_failed() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        assert!(matches!(result, AcquireResult::Failed { .. }));
        assert!(!root.path().join("backups").exists());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_backup_write_collision_is_reported_as_failed() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());
        // Force `write_backup`'s `create_new` `open()` to fail deterministically
        // by putting a plain file where the backup directory should be, so the
        // `with_context` closure (otherwise dead in every other test here)
        // actually runs. Pre-creating the *exact* colliding path instead (the
        // same-second collision the doc comment describes) is racy under CI
        // load: the backup filename has only whole-second precision, and
        // enough time can pass between computing it here and `acquire()`
        // computing its own for the two to land in different seconds.
        std::fs::write(&test_opts.backup_dir, b"not a directory").unwrap();

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Failed { detail } = result else {
            panic!("expected Failed, got {result:?}");
        };
        assert!(detail.contains("Failed to create backup file"), "{detail}");
    }

    #[test]
    fn write_backup_reports_a_genuine_same_second_collision_distinctly() {
        // Exercises `write_backup` directly against a literal path, rather
        // than racing `acquire()`'s whole-second-precision timestamp — see
        // `a_backup_write_collision_is_reported_as_failed`'s doc comment for
        // why that race is avoided everywhere else.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("backup-file");
        std::fs::write(&path, b"already here").unwrap();

        let err = write_backup(&path, b"hello").unwrap_err();

        assert!(err.to_string().contains("may already exist"), "{err:?}");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_ledger_insert_failure_reclaims_the_backup_just_taken() {
        // A concurrent lease op holds the ledger lock across this attempt's
        // own `insert_record` for longer than its wait budget — the routine,
        // non-race way `Failed` still follows a full backup (issue #1690):
        // the lock-free pre-check can't see this, since it takes no lock
        // itself. `insert_record` waits out a busy lock (issue #1738), so
        // the budget is shortened through `acquire_with_timings` rather
        // than spending the default's minutes. A held `LedgerLock`, not a
        // bare `File::create` on the lock path (issue #1687): under `flock`,
        // the file's mere existence holds nothing — only an actual lock
        // does, and `flock` conflicts against a second `open()` even from
        // this same process.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());
        let _held = LedgerLock::acquire(&test_opts.ledger_path).unwrap();

        let result = acquire_with_timings(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
            AcquireTimings {
                lock_wait: Duration::from_millis(150),
                ..FAST_TIMINGS
            },
        )
        .await;

        assert!(
            matches!(&result, AcquireResult::Failed { detail } if detail.contains("timed out")),
            "{result:?}"
        );
        assert!(
            std::fs::read_dir(&test_opts.backup_dir)
                .unwrap()
                .next()
                .is_none(),
            "the backup this attempt took must be reclaimed, not left orphaned"
        );
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("failed"),
            "a successfully reclaimed backup takes no verdict suffix"
        );
        assert!(
            rec.context.contains_key("backup_location"),
            "a reclaimed backup's location must still be audited"
        );
    }

    #[test]
    fn acquire_result_serializes_to_jsonl() {
        use crate::cli::drive::format::JsonlSerialize;

        let mut buf = Vec::new();
        AcquireResult::Acquired {
            token: "tok-1".to_string(),
            expires_at: Utc::now(),
            backup: LeaseBackup::Bytes {
                path: PathBuf::from("/tmp/backup"),
                sha256: "deadbeef".to_string(),
                size: 0,
            },
            headless_waiver: false,
            superseded_lease_id: None,
        }
        .write_jsonl(&mut buf)
        .unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(text.contains("\"status\":\"acquired\""), "{text}");
        assert!(text.contains("tok-1"), "{text}");
    }

    #[tokio::test]
    async fn native_document_is_refused_before_authenticating() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/vnd.google-apps.document"
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        // An authenticator that panics if called at all — proves the
        // native-document refusal happens before authentication, per the
        // module doc ("mirroring drive edit's identical refusal").
        let result = acquire(
            &client,
            &opts(root.path()),
            &PanicsIfCalled("must not authenticate for a Google-native document"),
        )
        .await;
        assert!(matches!(result, AcquireResult::RefusedNativeDocument));
    }

    #[tokio::test]
    async fn missing_version_is_refused_before_authenticating() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "n", "mimeType": "application/pdf"
                })),
            )
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &PanicsIfCalled("must not authenticate without a version to lease against"),
        )
        .await;
        assert!(matches!(result, AcquireResult::Failed { .. }));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn authorized_acquisition_backs_up_bytes_and_records_a_ledger_row() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "42", "modifiedTime": "2026-09-11T00:00:00Z"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Acquired { token, backup, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        let LeaseBackup::Bytes { path, .. } = backup else {
            panic!("expected a Bytes backup for a binary file, got {backup:?}");
        };
        assert!(!token.is_empty());
        assert_eq!(std::fs::read(&path).unwrap(), b"hello");
        assert!(path
            .file_name()
            .unwrap()
            .to_string_lossy()
            .contains("f1-report.pdf"));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn recorded_version_reflects_a_post_backup_fetch_not_the_pre_auth_snapshot() {
        // Regression test for the lease-record TOCTOU (issue #1664): the
        // metadata read at the very top of `acquire` (before the
        // authenticator prompt) returns version "0". A foreign edit lands
        // during the prompt, so every later fetch returns "1" — and "1" is
        // what must end up in the ledger record, not the pre-auth "0"
        // (which would no longer match what was actually backed up).
        //
        // Since issue #1693 this payload (no `sha256Checksum`) also takes
        // the version-sandwich pin, so the fixture doubles as proof that an
        // edit landing during the *prompt* — outside the backup window the
        // sandwich guards — is absorbed and recorded, never refused: the
        // pre-backup and post-backup reads both see "1".
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "0", "modifiedTime": "2026-09-11T00:00:00Z"
                })),
            )
            .up_to_n_times(1)
            .with_priority(1)
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "1", "modifiedTime": "2026-09-11T00:05:00Z"
                })),
            )
            .with_priority(2)
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Acquired { token, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        let ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        let record = ledger.get(&token).expect("token must be in the ledger");
        assert_eq!(
            record.version, "1",
            "the recorded version must come from the post-backup fetch, not the pre-auth one"
        );
        assert_eq!(
            record.modified_time.as_deref(),
            Some("2026-09-11T00:05:00Z")
        );
    }

    // ── pinning the backup to the recorded version (issue #1693) ───────

    /// SHA-256 of the `b"hello"` body every download mock in this module
    /// serves — what a byte backup of it hashes to.
    const HELLO_SHA256: &str = "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824";

    /// Mounts a metadata mock that serves `first` for the first `n` reads
    /// and `rest` after — the staged-drift shape every pinning test needs.
    async fn mount_metadata_drift(
        server: &wiremock::MockServer,
        first: serde_json::Value,
        n: u64,
        rest: serde_json::Value,
    ) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(first))
            .up_to_n_times(n)
            .expect(n)
            .with_priority(1)
            .mount(server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(rest))
            .with_priority(2)
            .mount(server)
            .await;
    }

    async fn mount_hello_download(server: &wiremock::MockServer) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(server)
            .await;
    }

    fn assert_nothing_minted(test_opts: &AcquireOptions) {
        assert!(
            !test_opts.ledger_path.exists(),
            "no ledger row must be written"
        );
        assert!(
            std::fs::read_dir(&test_opts.backup_dir).is_ok_and(|mut d| d.next().is_none()),
            "the backup this attempt took must be reclaimed, not left orphaned"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_byte_backup_whose_checksum_disagrees_with_drive_is_refused() {
        // Drive's post-backup `sha256Checksum` is not what the downloaded
        // bytes hash to: the content moved under the download. The
        // metadata mock's `.expect(2)` is half the assertion — a byte
        // target *with* a checksum is pinned by content alone, so the
        // pre-backup read the version sandwich needs must not be made — and
        // a checksum mismatch is never retried (issue #1739), so no second
        // post-backup read follows either.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "3",
                    "sha256Checksum": "0000000000000000000000000000000000000000000000000000000000000000"
                })),
            )
            .expect(2)
            .mount(&server)
            .await;
        mount_hello_download(&server).await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::RefusedConcurrentChange { detail } = result else {
            panic!("expected RefusedConcurrentChange, got {result:?}");
        };
        assert!(detail.contains(HELLO_SHA256), "{detail}");
        assert!(detail.contains("version 3"), "{detail}");
        assert_nothing_minted(&test_opts);
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("refused-concurrent-change")
        );
        assert!(
            rec.context
                .get("backup_location")
                .is_some_and(|loc| loc.contains("f1-report.pdf")),
            "the reclaimed backup's location must still be audited: {:?}",
            rec.context.get("backup_location")
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_byte_backup_whose_checksum_matches_is_pinned_even_when_version_bumped() {
        // A rename/move/ACL change lands mid-backup: `version` goes "1" →
        // "2" but the content — and so the checksum — is untouched. The
        // backup provably *is* version 2's content, so the lease is minted
        // against "2", not refused. This is the property a bare version
        // comparison could never offer.
        let server = wiremock::MockServer::start().await;
        mount_metadata_drift(
            &server,
            serde_json::json!({
                "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                "version": "1", "sha256Checksum": HELLO_SHA256
            }),
            1,
            serde_json::json!({
                "id": "f1", "name": "renamed.pdf", "mimeType": "application/pdf",
                "version": "2", "sha256Checksum": HELLO_SHA256
            }),
        )
        .await;
        mount_hello_download(&server).await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Acquired { token, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        let ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        assert_eq!(ledger.get(&token).unwrap().version, "2");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_byte_target_without_a_checksum_falls_back_to_the_version_sandwich() {
        // No `sha256Checksum` at all (defensive: Drive omits it only for
        // native and shortcut files), so there is no content proof — the
        // backup is sandwiched between two `version` reads instead.
        // Pre-auth and pre-backup both see "1"; the post-backup read sees
        // "2", so that first backup is refused and discarded. The second
        // attempt reads "2" on both sides and is pinned (issue #1739). The
        // first mock's `.expect(2)` proves the pre-backup read *was* made
        // on this path; the download count proves the retry took a second
        // backup rather than reusing the first.
        let server = wiremock::MockServer::start().await;
        mount_metadata_drift(
            &server,
            serde_json::json!({
                "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "1"
            }),
            2,
            serde_json::json!({
                "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "2"
            }),
        )
        .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .expect(2)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire_with_timings(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
            FAST_TIMINGS,
        )
        .await;

        let AcquireResult::Acquired { token, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        let ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        assert_eq!(ledger.get(&token).unwrap().version, "2");
        assert_eq!(
            std::fs::read_dir(&test_opts.backup_dir).unwrap().count(),
            1,
            "the first attempt's backup must be reclaimed, leaving only the pinned one"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_sandwich_read_without_a_version_refuses_before_taking_the_backup() {
        // The pre-auth read carries a `version`, so the acquisition
        // proceeds past the prompt; the pre-backup sandwich read then comes
        // back without one. `backup_is_pinned`'s no-proof arm would refuse
        // this too — but only after the download — so `acquire_inner`
        // refuses first. The download mock's `.expect(0)` is the assertion.
        let server = wiremock::MockServer::start().await;
        mount_metadata_drift(
            &server,
            serde_json::json!({
                "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "1"
            }),
            1,
            serde_json::json!({
                "id": "f1", "name": "report.pdf", "mimeType": "application/pdf"
            }),
        )
        .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .expect(0)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Failed { detail } = result else {
            panic!("expected Failed, got {result:?}");
        };
        assert!(detail.contains("immediately before the backup"), "{detail}");
        assert!(!test_opts.ledger_path.exists(), "no ledger row");
        assert!(
            !test_opts.backup_dir.exists()
                || std::fs::read_dir(&test_opts.backup_dir).is_ok_and(|mut d| d.next().is_none()),
            "no backup may have been taken"
        );
    }

    /// The native-document shape of the sandwich tests: a spreadsheet
    /// target whose `files.copy` backup lands as `copy-1`, with the
    /// metadata drifting from `"7"` to `"8"` on the first post-backup read
    /// and staying there — a one-off bump. `copies` is how many
    /// `files.copy` calls the test expects.
    async fn mount_native_drift(server: &wiremock::MockServer, copies: u64) {
        mount_metadata_drift(
            server,
            serde_json::json!({
                "id": "f1", "name": "Budget",
                "mimeType": "application/vnd.google-apps.spreadsheet", "version": "7"
            }),
            2,
            serde_json::json!({
                "id": "f1", "name": "Budget",
                "mimeType": "application/vnd.google-apps.spreadsheet", "version": "8"
            }),
        )
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .expect(copies)
            .mount(server)
            .await;
    }

    /// A metadata responder whose `version` moves on *every* read —
    /// `base`'s other fields, with `version` counting up from `first`. The
    /// shape a file whose `version` moves for reasons unrelated to edits
    /// would present (issue #1739), which no number of retries can pin.
    struct EverMovingVersion {
        base: serde_json::Value,
        next: std::sync::atomic::AtomicU64,
    }

    impl wiremock::Respond for EverMovingVersion {
        fn respond(&self, _request: &wiremock::Request) -> wiremock::ResponseTemplate {
            let version = self.next.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            let mut body = self.base.clone();
            body["version"] = serde_json::Value::String(version.to_string());
            wiremock::ResponseTemplate::new(200).set_body_json(body)
        }
    }

    /// Mounts a mock that trashes `copy-1` successfully, expecting
    /// `times` calls.
    async fn mount_trash_copy(server: &wiremock::MockServer, times: u64) {
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .and(wiremock::matchers::body_json(
                serde_json::json!({"trashed": true}),
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "copy-1", "name": "backup", "trashed": true,
                })),
            )
            .expect(times)
            .mount(server)
            .await;
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_native_backup_whose_version_moved_once_is_retried_and_acquired() {
        // The one-off bump issue #1739 is about: the first copy is
        // sandwiched between "7" and "8" and discarded, the second between
        // "8" and "8" and pinned. The copy mock's `.expect(2)` and the
        // trash mock's `.expect(1)` are the retry and reclaim assertions.
        let server = wiremock::MockServer::start().await;
        mount_native_drift(&server, 2).await;
        mount_trash_copy(&server, 1).await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire_with_timings(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
            FAST_TIMINGS,
        )
        .await;

        let AcquireResult::Acquired { token, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        let ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        assert_eq!(ledger.get(&token).unwrap().version, "8");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_native_backup_whose_version_never_settles_is_refused_after_every_attempt() {
        // `version` moves on every read, so every attempt is refused —
        // each copy is trashed before the next is taken, and the final
        // refusal says the version kept moving rather than asking the
        // operator to wait for a quiet file. Reads: pre-auth "7", then
        // ("8", "9"), ("10", "11"), ("12", "13").
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(EverMovingVersion {
                base: serde_json::json!({
                    "id": "f1", "name": "Budget",
                    "mimeType": "application/vnd.google-apps.spreadsheet"
                }),
                next: std::sync::atomic::AtomicU64::new(7),
            })
            .expect(1 + 2 * u64::from(SANDWICH_ATTEMPTS))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .expect(u64::from(SANDWICH_ATTEMPTS))
            .mount(&server)
            .await;
        mount_trash_copy(&server, u64::from(SANDWICH_ATTEMPTS)).await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire_with_timings(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
            FAST_TIMINGS,
        )
        .await;

        let AcquireResult::RefusedConcurrentChange { detail } = result else {
            panic!("expected RefusedConcurrentChange, got {result:?}");
        };
        assert!(detail.contains("on each of 3 attempts"), "{detail}");
        assert!(detail.contains("version 12"), "{detail}");
        assert!(detail.contains("13 immediately after"), "{detail}");
        assert!(detail.contains("unrelated to edits"), "{detail}");
        assert!(!test_opts.ledger_path.exists(), "no ledger row");
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("refused-concurrent-change")
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_concurrent_change_whose_reclaim_fails_is_audited_as_backup_orphaned() {
        // A one-off drift, but trashing the copy fails — so retrying stops
        // there rather than risk a second orphan (the copy mock's
        // `.expect(1)` is that assertion, issue #1739). The refusal stands
        // with the one-shot wording, and the audit verdict carries issue
        // #1690's `-backup-orphaned` suffix so the leaked copy is greppable.
        let server = wiremock::MockServer::start().await;
        mount_native_drift(&server, 1).await;
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        assert!(matches!(
            result,
            AcquireResult::RefusedConcurrentChange { .. }
        ));
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("refused-concurrent-change-backup-orphaned")
        );
        assert_eq!(
            rec.context.get("backup_location").map(String::as_str),
            Some("copy-1")
        );
        assert!(
            rec.error
                .as_deref()
                .is_some_and(|e| e.contains("version 7")),
            "{:?}",
            rec.error
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_reclaim_failing_mid_retry_stops_early_and_never_claims_the_backup_discarded() {
        // `version` moves on every read; the first copy is trashed, but
        // trashing the second fails. Retrying stops there — the copy
        // mock's `.expect(2)` — and the refusal says the last backup could
        // not be discarded rather than claiming all were, and does not
        // claim the full retry budget was spent.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(EverMovingVersion {
                base: serde_json::json!({
                    "id": "f1", "name": "Budget",
                    "mimeType": "application/vnd.google-apps.spreadsheet"
                }),
                next: std::sync::atomic::AtomicU64::new(7),
            })
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .expect(2)
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "copy-1", "name": "backup", "trashed": true,
                })),
            )
            .up_to_n_times(1)
            .with_priority(1)
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .with_priority(2)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire_with_timings(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
            FAST_TIMINGS,
        )
        .await;

        let AcquireResult::RefusedConcurrentChange { detail } = result else {
            panic!("expected RefusedConcurrentChange, got {result:?}");
        };
        assert!(detail.contains("on each of 2 attempts"), "{detail}");
        assert!(detail.contains("stopped early"), "{detail}");
        assert!(
            !detail.contains("minted and the backups were discarded"),
            "{detail}"
        );
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("refused-concurrent-change-backup-orphaned")
        );
    }

    #[test]
    fn backup_is_pinned_refuses_when_neither_proof_is_available() {
        // A byte backup whose target reported a digest before the backup
        // (so no sandwich read was taken) but none after it has nothing
        // left to compare — and must refuse rather than silently pass.
        let backup = LeaseBackup::Bytes {
            path: PathBuf::from("/x"),
            sha256: HELLO_SHA256.to_string(),
            size: 5,
        };
        let err = backup_is_pinned(&backup, None, "7", None).unwrap_err();
        assert!(matches!(err, PinFailure::NoProof { .. }), "{err:?}");
        assert!(!err.is_retryable());
        let detail = err.detail(1, true);
        assert!(detail.contains("no proof available"), "{detail}");
    }

    #[test]
    fn only_a_moved_version_is_retried_and_its_final_wording_names_the_attempts() {
        let moved = PinFailure::VersionMoved {
            pre: "7".to_string(),
            post: "8".to_string(),
        };
        assert!(moved.is_retryable());
        let once = moved.detail(1, true);
        assert!(once.contains("version 7 immediately before"), "{once}");
        assert!(once.ends_with("backup was discarded — retry."), "{once}");
        let exhausted = moved.detail(SANDWICH_ATTEMPTS, true);
        assert!(exhausted.contains("on each of 3 attempts"), "{exhausted}");
        assert!(exhausted.contains("unrelated to edits"), "{exhausted}");
        assert!(!exhausted.contains("retry."), "{exhausted}");

        // A last backup that could not be reclaimed is never claimed
        // discarded, and an early stop does not claim the version kept
        // moving through the full retry budget.
        let orphaned_once = moved.detail(1, false);
        assert!(!orphaned_once.contains("was discarded"), "{orphaned_once}");
        assert!(
            orphaned_once.contains("could not be discarded"),
            "{orphaned_once}"
        );
        let stopped_early = moved.detail(2, false);
        assert!(
            stopped_early.contains("on each of 2 attempts"),
            "{stopped_early}"
        );
        assert!(stopped_early.contains("stopped early"), "{stopped_early}");
        assert!(
            !stopped_early.contains("minted and the backups were discarded"),
            "{stopped_early}"
        );
        assert!(
            !stopped_early.contains("unrelated to edits"),
            "{stopped_early}"
        );

        let mismatch = PinFailure::ChecksumMismatch {
            remote: "aa".to_string(),
            post_version: "3".to_string(),
            local: "bb".to_string(),
        };
        assert!(!mismatch.is_retryable());
    }

    #[test]
    fn backup_is_pinned_compares_checksums_case_insensitively() {
        let backup = LeaseBackup::Bytes {
            path: PathBuf::from("/x"),
            sha256: HELLO_SHA256.to_string(),
            size: 5,
        };
        let upper = HELLO_SHA256.to_ascii_uppercase();
        assert!(backup_is_pinned(&backup, Some(&upper), "1", None).is_ok());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn native_document_with_a_backup_folder_configured_copies_instead_of_refusing() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .expect(1)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut opts = opts(root.path());
        opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire(&client, &opts, &FakeAuthenticator(AuthOutcome::Authorized)).await;

        let AcquireResult::Acquired { backup, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };
        assert_eq!(
            backup,
            LeaseBackup::DriveCopy {
                file_id: "copy-1".to_string()
            }
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_failed_native_copy_after_authorization_is_reported_as_failed() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .expect(1)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut opts = opts(root.path());
        opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire(&client, &opts, &FakeAuthenticator(AuthOutcome::Authorized)).await;

        assert!(
            matches!(result, AcquireResult::Failed { .. }),
            "expected Failed, got {result:?}"
        );
        assert!(!opts.ledger_path.exists(), "no ledger row must be written");
    }

    /// Mounts the pre-auth `files.get` (version "1", consumed once) plus
    /// the byte download, leaving the post-backup `files.get` to the
    /// caller — for tests exercising a failure in that second fetch.
    async fn mount_pre_auth_metadata_and_download(server: &wiremock::MockServer) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "1"
                })),
            )
            // The pre-auth read *and* the pre-backup read (issue #1693) —
            // this payload carries no `sha256Checksum`, so acquire pins the
            // backup by sandwiching it between two `version` reads, and
            // both must see a healthy response for a test here to reach the
            // post-backup fetch it actually means to exercise.
            .up_to_n_times(2)
            .with_priority(1)
            .mount(server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(server)
            .await;
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_failed_post_backup_metadata_fetch_is_reported_as_failed() {
        let server = wiremock::MockServer::start().await;
        mount_pre_auth_metadata_and_download(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .with_priority(2)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        assert!(
            matches!(result, AcquireResult::Failed { .. }),
            "expected Failed, got {result:?}"
        );
        assert!(
            !test_opts.ledger_path.exists(),
            "no ledger row must be written"
        );
        assert!(
            std::fs::read_dir(&test_opts.backup_dir)
                .unwrap()
                .next()
                .is_none(),
            "the backup this attempt took must be reclaimed, not left orphaned (issue #1690)"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_post_backup_fetch_without_a_version_is_reported_as_failed() {
        // The pre-auth fetch carried a `version`, so the early refusal
        // does not fire; the post-backup fetch — the one the ledger row is
        // actually recorded from — omits it, and a lease with no staleness
        // check must not be minted.
        let server = wiremock::MockServer::start().await;
        mount_pre_auth_metadata_and_download(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf"
                })),
            )
            .with_priority(2)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Failed { detail } = result else {
            panic!("expected Failed, got {result:?}");
        };
        assert!(detail.contains("after the backup"), "{detail}");
        assert!(
            !test_opts.ledger_path.exists(),
            "no ledger row must be written"
        );
        assert!(
            std::fs::read_dir(&test_opts.backup_dir)
                .unwrap()
                .next()
                .is_none(),
            "the backup this attempt took must be reclaimed, not left orphaned (issue #1690)"
        );
    }

    // ── the audit sink (ADR-0080 §11) ──────────────────────────────────

    #[tokio::test(flavor = "multi_thread")]
    async fn an_acquired_lease_writes_an_audit_record_carrying_the_backup_and_version() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "42", "modifiedTime": "2026-09-11T00:00:00Z"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;

        let AcquireResult::Acquired { token, .. } = result else {
            panic!("expected Acquired, got {result:?}");
        };

        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(rec.kind, crate::request_log::RecordKind::Audit);
        assert_eq!(rec.context.get("lease_id"), Some(&token));
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("acquired")
        );
        assert_eq!(
            rec.context.get("version_after").map(String::as_str),
            Some("42")
        );
        assert_eq!(
            rec.context.get("auth_policy").map(String::as_str),
            Some("device-owner")
        );
        assert!(rec.context.contains_key("backup_sha256"));
    }

    #[tokio::test]
    async fn a_refused_acquisition_still_writes_an_audit_record() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &PanicsIfCalled("must not authenticate with no backup folder configured"),
        )
        .await;

        assert!(matches!(result, AcquireResult::RefusedNativeDocument));

        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("refused-native-document")
        );
        assert_eq!(rec.context.get("lease_id"), None);
    }

    #[tokio::test]
    async fn native_document_without_a_backup_folder_is_refused_before_authenticating() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());

        let result = acquire(
            &client,
            &opts(root.path()),
            &PanicsIfCalled("must not authenticate with no backup folder configured"),
        )
        .await;
        assert!(matches!(result, AcquireResult::RefusedNativeDocument));
    }

    // ── expiry bound enforced in the engine, not only the CLI (#1664) ───

    #[tokio::test]
    async fn expiry_below_the_minimum_is_refused_before_authenticating() {
        // No `/drive/v3/files/f1` mock is mounted — if the check below did
        // not gate before the metadata fetch, this would fail differently
        // (a connection/parse error, not this message).
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.expiry = ChronoDuration::minutes(0);

        let result = acquire(
            &client,
            &test_opts,
            &PanicsIfCalled("must not authenticate with an out-of-range expiry"),
        )
        .await;
        let AcquireResult::Failed { detail } = result else {
            panic!("expected Failed, got {result:?}");
        };
        assert!(detail.contains("expiry must be between"), "{detail}");
    }

    #[tokio::test]
    async fn expiry_above_the_maximum_is_refused_before_authenticating() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.expiry = ChronoDuration::minutes(MAX_EXPIRY_MINUTES + 1);

        let result = acquire(
            &client,
            &test_opts,
            &PanicsIfCalled("must not authenticate with an out-of-range expiry"),
        )
        .await;
        let AcquireResult::Failed { detail } = result else {
            panic!("expected Failed, got {result:?}");
        };
        assert!(detail.contains("expiry must be between"), "{detail}");
    }

    // ── refusing a second live lease on the same file (#1664) ───────────

    #[tokio::test(flavor = "multi_thread")]
    async fn a_second_acquisition_while_one_is_still_live_reuses_the_existing_token_instead_of_minting_a_second(
    ) {
        // Regression test for the TOCTOU this closes: two independently
        // acquired leases on the same file would otherwise each capture the
        // same version and each pass their own staleness check against it,
        // letting the second writer's write silently clobber the first's.
        // Refusing to mint the second lease at all closes that gap. The
        // second attempt's authenticator panics if called at all — proving
        // the lock-free pre-check (issue #1690) refuses it before ever
        // spending a Touch ID prompt, not merely before minting.
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let first = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;
        let AcquireResult::Acquired {
            token: first_token, ..
        } = first
        else {
            panic!("expected Acquired, got {first:?}");
        };

        let mut second_opts = test_opts.clone();
        second_opts.backup_dir = root.path().join("backups2");

        let second = acquire(
            &client,
            &second_opts,
            &PanicsIfCalled(
                "must not authenticate: the lock-free pre-check must refuse a second live \
                 lease before spending a prompt",
            ),
        )
        .await;
        let AcquireResult::AlreadyLeased { token, .. } = second else {
            panic!("expected AlreadyLeased, got {second:?}");
        };
        assert_eq!(token, first_token);
        assert!(
            !second_opts.backup_dir.exists(),
            "the pre-check must refuse before ever taking a backup"
        );

        // Only the first lease is live in the ledger.
        let ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        assert_eq!(
            ledger
                .live_lease_for_file("f1", Utc::now(), None)
                .map(|r| &r.token),
            Some(&first_token)
        );
    }

    // ── the belt-and-braces reclaim for the pre-check's narrow race (#1690) ──

    /// An authenticator that, as a side effect of authenticating, inserts a
    /// live lease for `file_id` into the ledger at `ledger_path` — used to
    /// simulate another process's `acquire` winning the race between this
    /// attempt's lock-free pre-check and its own `insert_record`, which the
    /// pre-check is not designed to close (only to make rare).
    struct InsertsALiveLeaseDuringAuth {
        ledger_path: PathBuf,
        file_id: String,
    }
    impl Authenticator for InsertsALiveLeaseDuringAuth {
        fn authenticate(&self, _reason: &str, _policy: AuthPolicy) -> AuthOutcome {
            LeaseLedger::mutate_locked(&self.ledger_path, |ledger| {
                ledger.insert(LeaseRecord {
                    token: "racer".to_string(),
                    file_id: self.file_id.clone(),
                    version: "1".to_string(),
                    modified_time: None,
                    backup: LeaseBackup::DriveCopy {
                        file_id: "racer-backup".to_string(),
                    },
                    acquired_at: Utc::now(),
                    expires_at: Utc::now() + ChronoDuration::minutes(30),
                    released_at: None,
                    superseded_by: None,
                    restored_at: None,
                    restored_sheet_id: None,
                });
            })
            .unwrap();
            AuthOutcome::Authorized
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_lease_inserted_during_the_prompt_reclaims_this_attempts_bytes_backup() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf", "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let result = acquire(
            &client,
            &test_opts,
            &InsertsALiveLeaseDuringAuth {
                ledger_path: test_opts.ledger_path.clone(),
                file_id: test_opts.file_id.clone(),
            },
        )
        .await;

        let AcquireResult::AlreadyLeased { token, .. } = result else {
            panic!("expected AlreadyLeased, got {result:?}");
        };
        assert_eq!(token, "racer");
        assert!(
            std::fs::read_dir(&test_opts.backup_dir)
                .unwrap()
                .next()
                .is_none(),
            "this attempt's own now-orphaned backup must be reclaimed"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_lease_inserted_during_the_prompt_reclaims_this_attempts_native_backup() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .expect(1)
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .and(wiremock::matchers::body_json(
                serde_json::json!({"trashed": true}),
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "copy-1", "name": "backup", "trashed": true,
                })),
            )
            .expect(1)
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire(
            &client,
            &test_opts,
            &InsertsALiveLeaseDuringAuth {
                ledger_path: test_opts.ledger_path.clone(),
                file_id: test_opts.file_id.clone(),
            },
        )
        .await;

        let AcquireResult::AlreadyLeased { token, .. } = result else {
            panic!("expected AlreadyLeased, got {result:?}");
        };
        assert_eq!(token, "racer");
        // The PATCH mock's `.expect(1)` above is the real assertion:
        // reclaiming this attempt's own orphaned Drive-copy backup means
        // trashing it.
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_reclamation_failure_is_audited_as_backup_orphaned_but_still_reports_already_leased()
    {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "f1", "name": "Budget", "mimeType": "application/vnd.google-apps.spreadsheet",
                "version": "7"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/drive/v3/files/f1/copy"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "copy-1", "name": "backup", "mimeType": "application/vnd.google-apps.spreadsheet"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PATCH"))
            .and(wiremock::matchers::path("/drive/v3/files/copy-1"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let mut test_opts = opts(root.path());
        test_opts.native_backup_folder_id = Some("backup-folder".to_string());

        let result = acquire(
            &client,
            &test_opts,
            &InsertsALiveLeaseDuringAuth {
                ledger_path: test_opts.ledger_path.clone(),
                file_id: test_opts.file_id.clone(),
            },
        )
        .await;

        assert!(matches!(result, AcquireResult::AlreadyLeased { .. }));
        let contents = std::fs::read_to_string(root.path().join("audit.jsonl")).unwrap();
        let rec: crate::request_log::LogRecord = serde_json::from_str(contents.trim_end()).unwrap();
        assert_eq!(
            rec.context.get("verdict").map(String::as_str),
            Some("already-leased-backup-orphaned")
        );
        assert_eq!(
            rec.context.get("backup_location").map(String::as_str),
            Some("copy-1")
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_second_acquisition_after_the_first_expires_mints_a_fresh_lease() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param_is_missing("alt"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "f1", "name": "report.pdf", "mimeType": "application/pdf",
                    "version": "1"
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/f1"))
            .and(wiremock::matchers::query_param("alt", "media"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(b"hello".to_vec()))
            .mount(&server)
            .await;
        let client = client_with_bootstrapped_token(&server).await;
        let root = tempfile::tempdir().unwrap();
        let _audit = AuditGuard::redirect(root.path());
        let test_opts = opts(root.path());

        let first = acquire(
            &client,
            &test_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;
        let AcquireResult::Acquired {
            token: first_token, ..
        } = first
        else {
            panic!("expected Acquired, got {first:?}");
        };
        // Backdate the first lease's expiry directly rather than waiting out
        // a real one — `MIN_EXPIRY_MINUTES` now rules out a sub-minute
        // `--expiry-minutes` that a real sleep-then-retry could use instead.
        let mut ledger = LeaseLedger::load(&test_opts.ledger_path).unwrap();
        let mut expired = ledger.get(&first_token).unwrap().clone();
        expired.expires_at = Utc::now() - ChronoDuration::minutes(1);
        ledger.insert(expired);
        ledger.save(&test_opts.ledger_path).unwrap();

        // A distinct `backup_dir`: `backup_name`'s timestamp has only
        // whole-second precision, so a second backup of the same file
        // within one test's runtime would otherwise collide on the same
        // path (see `write_backup`'s own doc comment).
        let mut second_opts = test_opts.clone();
        second_opts.backup_dir = root.path().join("backups2");

        let second = acquire(
            &client,
            &second_opts,
            &FakeAuthenticator(AuthOutcome::Authorized),
        )
        .await;
        let AcquireResult::Acquired {
            token: second_token,
            ..
        } = second
        else {
            panic!("expected Acquired, got {second:?}");
        };
        assert_ne!(second_token, first_token);
    }

    #[test]
    fn insert_record_waits_for_a_concurrent_holder_then_inserts() {
        // Issue #1738: an unrelated lease operation holding the ledger lock
        // must delay this insert, not fail it — by now the attempt has
        // already spent the prompt and the backup.
        let root = tempfile::tempdir().unwrap();
        let ledger_path = root.path().join("lease-ledger.jsonl");
        let held = LedgerLock::acquire(&ledger_path).unwrap();

        let releaser = std::thread::spawn(move || {
            std::thread::sleep(std::time::Duration::from_millis(100));
            drop(held);
        });

        let outcome =
            insert_record(live_row("new", "f1"), &ledger_path, None, TEST_LOCK_WAIT).unwrap();
        releaser.join().unwrap();

        assert!(matches!(
            outcome,
            InsertOutcome::Inserted { superseded: false }
        ));
        assert!(LeaseLedger::load(&ledger_path)
            .unwrap()
            .get("new")
            .is_some());
    }

    // ── supersede: `drive lease restore` replacing its own backup lease (#1685) ──

    /// A live row for `file_id`, so a supersede has something to replace.
    fn live_row(token: &str, file_id: &str) -> LeaseRecord {
        LeaseRecord {
            token: token.to_string(),
            file_id: file_id.to_string(),
            version: "1".to_string(),
            modified_time: None,
            backup: LeaseBackup::DriveCopy {
                file_id: format!("{token}-backup"),
            },
            acquired_at: Utc::now(),
            expires_at: Utc::now() + ChronoDuration::minutes(30),
            released_at: None,
            superseded_by: None,
            restored_at: None,
            restored_sheet_id: None,
        }
    }

    fn seed(ledger_path: &Path, records: Vec<LeaseRecord>) {
        LeaseLedger::mutate_locked(ledger_path, |ledger| {
            for record in records {
                ledger.insert(record);
            }
        })
        .unwrap();
    }

    #[test]
    fn superseding_releases_the_named_lease_in_the_same_rewrite_that_inserts() {
        let root = tempfile::tempdir().unwrap();
        let ledger_path = root.path().join("lease-ledger.jsonl");
        seed(&ledger_path, vec![live_row("old", "f1")]);

        let outcome = insert_record(
            live_row("new", "f1"),
            &ledger_path,
            Some("old"),
            TEST_LOCK_WAIT,
        )
        .unwrap();

        assert!(matches!(
            outcome,
            InsertOutcome::Inserted { superseded: true }
        ));
        let ledger = LeaseLedger::load(&ledger_path).unwrap();
        assert!(
            ledger.get("old").unwrap().released_at.is_some(),
            "the superseded row must be released"
        );
        assert_eq!(
            ledger.get("old").unwrap().superseded_by.as_deref(),
            Some("new"),
            "the superseded row must durably record which lease replaced it \
             (issue #1768), not just that it was released"
        );
        assert_eq!(
            ledger
                .live_lease_for_file("f1", Utc::now(), None)
                .map(|r| r.token.clone()),
            Some("new".to_string()),
            "exactly one live lease must cover the file afterwards"
        );
    }

    #[test]
    fn superseding_ignores_a_token_bound_to_another_file() {
        let root = tempfile::tempdir().unwrap();
        let ledger_path = root.path().join("lease-ledger.jsonl");
        seed(&ledger_path, vec![live_row("elsewhere", "f2")]);

        let outcome = insert_record(
            live_row("new", "f1"),
            &ledger_path,
            Some("elsewhere"),
            TEST_LOCK_WAIT,
        )
        .unwrap();

        assert!(
            matches!(outcome, InsertOutcome::Inserted { superseded: false }),
            "a row this insert did not release must not be reported as superseded"
        );
        let ledger = LeaseLedger::load(&ledger_path).unwrap();
        assert!(
            ledger.get("elsewhere").unwrap().released_at.is_none(),
            "naming another file's token must never release it"
        );
        assert_eq!(ledger.get("elsewhere").unwrap().superseded_by, None);
    }

    /// `mutate_locked` saves whatever the closure leaves behind, refusal or
    /// not, so releasing before the live-lease check would end the caller's
    /// own lease on a path that mints nothing to replace it.
    #[test]
    fn a_blocked_insert_does_not_release_the_superseded_lease() {
        let root = tempfile::tempdir().unwrap();
        let ledger_path = root.path().join("lease-ledger.jsonl");
        seed(
            &ledger_path,
            vec![live_row("old", "f1"), live_row("third-party", "f1")],
        );

        let outcome = insert_record(
            live_row("new", "f1"),
            &ledger_path,
            Some("old"),
            TEST_LOCK_WAIT,
        )
        .unwrap();

        assert!(
            matches!(&outcome, InsertOutcome::AlreadyLeased(rec) if rec.token == "third-party"),
            "a third live lease must still block the insert"
        );
        let ledger = LeaseLedger::load(&ledger_path).unwrap();
        assert!(
            ledger.get("old").unwrap().released_at.is_none(),
            "nothing was minted, so the superseded lease must survive intact"
        );
        assert!(ledger.get("new").is_none());
    }

    /// A supersede against a row that is already dead — expired or released
    /// earlier — is left alone, and must be reported that way, or the audit
    /// record would claim a release this insert never made.
    #[test]
    fn superseding_an_already_dead_row_is_not_reported_as_a_release() {
        let root = tempfile::tempdir().unwrap();
        let ledger_path = root.path().join("lease-ledger.jsonl");
        let mut old = live_row("old", "f1");
        let first_release = Utc::now() - ChronoDuration::hours(1);
        old.released_at = Some(first_release);
        seed(&ledger_path, vec![old]);

        let outcome = insert_record(
            live_row("new", "f1"),
            &ledger_path,
            Some("old"),
            TEST_LOCK_WAIT,
        )
        .unwrap();

        assert!(matches!(
            outcome,
            InsertOutcome::Inserted { superseded: false }
        ));
        assert_eq!(
            LeaseLedger::load(&ledger_path)
                .unwrap()
                .get("old")
                .unwrap()
                .released_at,
            Some(first_release),
            "the earlier release's timestamp must survive untouched"
        );
    }

    /// Drives `record_attempt` for an `Acquired` result carrying
    /// `superseded_lease_id`, returning what the audit record says.
    fn superseded_lease_id_on_record(root: &Path, superseded: Option<&str>) -> Option<String> {
        let audit = AuditGuard::redirect(root);
        let mut test_opts = opts(root);
        test_opts.supersedes = Some("old".to_string());
        seed(&test_opts.ledger_path, vec![live_row("new", "f1")]);
        record_attempt(
            &test_opts,
            &AcquireResult::Acquired {
                token: "new".to_string(),
                expires_at: Utc::now(),
                backup: LeaseBackup::DriveCopy {
                    file_id: "new-backup".to_string(),
                },
                headless_waiver: false,
                superseded_lease_id: superseded.map(str::to_string),
            },
            None,
        );
        audit
            .records()
            .pop()
            .and_then(|r| r.context.get("superseded_lease_id").cloned())
    }

    #[test]
    fn the_audit_record_names_the_lease_the_result_says_was_released() {
        let root = tempfile::tempdir().unwrap();
        assert_eq!(
            superseded_lease_id_on_record(root.path(), Some("old")),
            Some("old".to_string())
        );
    }

    /// `opts.supersedes` alone is never enough: only the result's own
    /// `superseded_lease_id` — the fact `insert_record` carried out of the
    /// locked rewrite — puts the field on the record.
    #[test]
    fn the_audit_record_omits_the_field_when_nothing_was_released() {
        let root = tempfile::tempdir().unwrap();
        assert_eq!(superseded_lease_id_on_record(root.path(), None), None);
    }

    /// Pins the CLI exit-code classification (issue #1775) for every
    /// variant — in particular that `AlreadyLeased` is `0`, since it hands
    /// back a usable token for reuse rather than refusing outright.
    #[test]
    fn exit_code_matches_the_documented_classification() {
        let cases: Vec<(AcquireResult, i32)> = vec![
            (
                AcquireResult::Acquired {
                    token: "tok".to_string(),
                    expires_at: Utc::now(),
                    backup: LeaseBackup::Bytes {
                        path: PathBuf::from("/tmp/backup"),
                        sha256: "deadbeef".to_string(),
                        size: 0,
                    },
                    headless_waiver: false,
                    superseded_lease_id: None,
                },
                0,
            ),
            (
                AcquireResult::AlreadyLeased {
                    token: "tok".to_string(),
                    expires_at: Utc::now(),
                },
                0,
            ),
            (AcquireResult::RefusedNativeDocument, 1),
            (
                AcquireResult::RefusedConcurrentChange {
                    detail: "moved".to_string(),
                },
                1,
            ),
            (
                AcquireResult::Denied {
                    detail: "no".to_string(),
                },
                1,
            ),
            (
                AcquireResult::Unavailable {
                    detail: "no authenticator".to_string(),
                },
                1,
            ),
            (
                AcquireResult::Failed {
                    detail: "boom".to_string(),
                },
                1,
            ),
        ];
        for (result, expected) in cases {
            assert_eq!(result.exit_code(), expected, "{result:?}");
        }
    }
}