omni_dev/drive/sheets/

write.rs

//! `drive sheets write`/`append`/`clear` engines — cell mutation gated by
//! the ADR-0071 folder write-permission rules (issue #1589,
//! [ADR-0073](../../../docs/adrs/adr-0073.md)).
//!
//! Single-target, so this follows `rename.rs`/`content_edit.rs`'s linear
//! shape — a public wrapper that logs, and an `_inner` that classifies then
//! mutates — rather than `file_move.rs`'s Plan/Execute split, which exists
//! to amortize a shared-destination fetch across a batch these verbs don't
//! have. `--dry-run` and a real run therefore share the same gate
//! classification *by construction* (same function, same early return), not
//! by convention.
//!
//! Two refusals happen **before** the gate, because they are not policy
//! decisions — the operation is simply nonsensical for that target:
//! anything that is not a spreadsheet, and a shortcut (even one pointing at
//! a spreadsheet, since we don't follow shortcuts). This is the mirror image
//! of `content_edit.rs`'s Google-native refusal.
//!
//! A third refusal is policy-adjacent but distinct: a target whose `parents`
//! are not visible. See [`WriteResult::RefusedNoVisibleParents`].

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

use serde::Serialize;

use crate::cli::drive::format::{write_scalar_jsonl, JsonlSerialize};
use crate::drive::client::DriveClient;
use crate::drive::files_api::FilesApi;
use crate::drive::lease::check::{
    conclude_native_leased_write, gate_optional_leased_write, FromLeaseRefusal, LeaseGateRefusal,
    LeasedWrite,
};
use crate::drive::sheets::a1;
use crate::drive::sheets::api::{SheetsApi, ValueInputOption};
use crate::drive::sheets::client::SheetsClient;
use crate::drive::sheets::grid_range;
use crate::drive::sheets::target_gate;
use crate::drive::sheets::types::UpdateValuesResponse;
use crate::drive::types::SheetTargetRefusal;
use crate::drive::write_gate::{self, DecidingRule, DriveOperation, FolderPermissionRule};
use crate::request_log::{self, DriveMutationOutcome};

/// Which cell mutation to perform.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WriteVerb {
    /// Overwrite the range's values.
    Write,
    /// Append rows after the last row of the range's table.
    Append,
    /// Clear the range's values, leaving formatting intact.
    Clear,
}

impl WriteVerb {
    /// The `operation` this verb records in the request log.
    ///
    /// `build_drive_mutation_record` shapes `command` as `["drive",
    /// <operation>]`, so these read as `drive sheets-write` in the log even
    /// though the CLI spells them `drive sheets write`.
    const fn log_operation(self) -> &'static str {
        match self {
            Self::Write => "sheets-write",
            Self::Append => "sheets-append",
            Self::Clear => "sheets-clear",
        }
    }

    /// Human-readable present-tense verb for CLI output.
    const fn label(self) -> &'static str {
        match self {
            Self::Write => "write",
            Self::Append => "append",
            Self::Clear => "clear",
        }
    }
}

/// Per-call options.
#[derive(Debug, Clone)]
pub struct WriteOptions {
    /// Spreadsheet id.
    pub spreadsheet_id: String,
    /// Which mutation to perform.
    pub verb: WriteVerb,
    /// An explicit A1 range, which may carry its own `Sheet!` prefix.
    pub range: Option<String>,
    /// A sheet title, supplying a prefix for a bare `range`.
    pub sheet: Option<String>,
    /// Row-major values to write. Empty for [`WriteVerb::Clear`].
    pub values: Vec<Vec<String>>,
    /// How the API should interpret the values.
    pub input: ValueInputOption,
    /// Classify only; never call a mutating endpoint.
    pub dry_run: bool,
    /// The lease token presented via `--lease`. Checked only when the
    /// deciding rule requires one
    /// ([`write_gate::decided_rule_requires_lease`], ADR-0080 §1/§9);
    /// `None` is only ever valid when it does not.
    pub lease_token: Option<String>,
    /// Path to the lease ledger the token is checked against. Production
    /// callers pass `crate::drive::lease::ledger::ledger_path`'s own
    /// result; tests pass a path under a `tempdir`.
    pub ledger_path: PathBuf,
}

/// What happened (or, under `--dry-run`, would happen).
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[serde(tag = "status", rename_all = "kebab-case")]
pub enum WriteResult {
    /// `--dry-run` of a `write`/`append`, and the gate would allow it.
    WouldWrite {
        /// Rows of input parsed, so a transposed or ragged input is visible
        /// before it lands.
        rows: usize,
        /// Widest row, likewise.
        columns: usize,
    },
    /// `--dry-run` of a `clear`, and the gate would allow it. Its own
    /// variant rather than a `WouldWrite` of `0 x 0`: a clear takes no
    /// values, so those dimensions would always be zero (#1941).
    WouldClear,
    /// The target is not a Google Sheet. Checked client-side before the
    /// gate: writing cells into a PDF isn't disallowed, it's meaningless.
    RefusedNotASpreadsheet {
        /// The target's actual MIME type.
        mime_type: String,
    },
    /// The target is a shortcut. Given its own variant rather than falling
    /// into `RefusedNotASpreadsheet`, because "this is not a spreadsheet" is
    /// a confusing thing to say about a shortcut *to* a spreadsheet — the
    /// same distinction `drive read --content` already draws.
    RefusedShortcut,
    /// The target has no parents this account can see, so the gate has no
    /// ancestor chain to evaluate — **and** no `file_id` rule named it
    /// either, so nothing granted it.
    ///
    /// Distinct from `Blocked { decided_by: None }`, which means "a chain
    /// was resolved and no rule matched". Conflating them would send an
    /// operator off to fix folder rules when no folder rule they could
    /// write would help. This is the common shape for a Sheet shared by
    /// link or email; since issue #1612 the fix is a `file_id` rule, and
    /// this variant is reached only once that lookup has come up empty.
    RefusedNoVisibleParents,
    /// The named sheet (from `--sheet`, or the range's own `Sheet!` prefix)
    /// does not exist in this workbook. Checked client-side, like
    /// find-replace/sort-range/auto-fill/paste and friends, rather than
    /// letting the API's opaque "Unable to parse range" stand in for it
    /// (#1941).
    RefusedSheetNotFound {
        /// The title that was not found.
        title: String,
        /// The titles that do exist.
        available: Vec<String>,
    },
    /// `write`/`append`'s `--values` resolved to zero rows, so there is
    /// nothing to write. `clear` takes no values and is unaffected.
    RefusedEmptyValues,
    /// The folder write-permission gate refused it.
    Blocked {
        /// The rule that decided the refusal, if any (`None` means the bare
        /// default policy — every write defaults deny).
        decided_by: Option<DecidingRule>,
    },
    /// No `--lease` was presented, and the deciding rule requires one
    /// (ADR-0080 §9).
    RefusedNoLease,
    /// The presented lease has expired, or was never a token this ledger
    /// knows about.
    RefusedLeaseExpired,
    /// The presented lease is bound to a different file id.
    RefusedLeaseWrongFile,
    /// The file has moved since the lease's recorded `version` — the
    /// staleness check (ADR-0080 §6).
    RefusedLeaseStale,
    /// A `write` succeeded.
    ///
    /// Every count is optional because the API may omit it.
    Written {
        /// The server-normalised range actually written.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_range: Option<String>,
        /// Rows the API reported changing.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_rows: Option<i64>,
        /// Columns the API reported changing.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_columns: Option<i64>,
        /// Cells the API reported changing.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_cells: Option<i64>,
    },
    /// An `append` succeeded. The same counts as [`Self::Written`], under
    /// its own status so a script can tell an append from an overwrite
    /// (#1941).
    Appended {
        /// The server-normalised range the rows landed in.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_range: Option<String>,
        /// Rows the API reported appending.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_rows: Option<i64>,
        /// Columns the API reported appending.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_columns: Option<i64>,
        /// Cells the API reported appending.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_cells: Option<i64>,
    },
    /// A `clear` succeeded. The API reports no counts for a clear, only
    /// the range.
    Cleared {
        /// The server-normalised range actually cleared.
        #[serde(skip_serializing_if = "Option::is_none")]
        updated_range: Option<String>,
    },
    /// An API or validation error.
    Failed {
        /// A human-readable summary of what failed.
        detail: String,
    },
}

impl FromLeaseRefusal for WriteResult {
    fn from_no_lease() -> Self {
        Self::RefusedNoLease
    }
    fn from_lease_expired() -> Self {
        Self::RefusedLeaseExpired
    }
    fn from_lease_wrong_file() -> Self {
        Self::RefusedLeaseWrongFile
    }
    fn from_lease_stale() -> Self {
        Self::RefusedLeaseStale
    }
    fn from_lease_failed(detail: String) -> Self {
        Self::Failed { detail }
    }
}

impl WriteResult {
    /// The `status` string the request log records.
    ///
    /// Hand-written rather than derived from the `#[serde(tag)]` shape,
    /// matching `MoveResult`/`EditResult`'s precedent of keeping the log's
    /// vocabulary decoupled from the wire format.
    fn log_status(&self) -> &'static str {
        match self {
            Self::WouldWrite { .. } => "would-write",
            Self::WouldClear => "would-clear",
            Self::RefusedNotASpreadsheet { .. } => "refused-not-a-spreadsheet",
            Self::RefusedShortcut => "refused-shortcut",
            Self::RefusedNoVisibleParents => "refused-no-visible-parents",
            Self::RefusedSheetNotFound { .. } => "refused-sheet-not-found",
            Self::RefusedEmptyValues => "refused-empty-values",
            Self::Blocked { .. } => "blocked",
            Self::RefusedNoLease => LeaseGateRefusal::NoLease.log_status(),
            Self::RefusedLeaseExpired => LeaseGateRefusal::Expired.log_status(),
            Self::RefusedLeaseWrongFile => LeaseGateRefusal::WrongFile.log_status(),
            Self::RefusedLeaseStale => LeaseGateRefusal::Stale.log_status(),
            Self::Written { .. } => "written",
            Self::Appended { .. } => "appended",
            Self::Cleared { .. } => "cleared",
            Self::Failed { .. } => "failed",
        }
    }
}

/// The full outcome of one attempt.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct WriteOutcome {
    /// The spreadsheet acted on.
    pub spreadsheet_id: String,
    /// Its name, when the metadata fetch got that far.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub file_name: Option<String>,
    /// The composed A1 range, when composition succeeded.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub range: Option<String>,
    /// The folder the gate evaluated against, when exactly one parent
    /// resolved it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub resolved_folder_id: Option<String>,
    /// Which mutation was attempted.
    ///
    /// Not serialised: the caller already knows which verb it invoked, and
    /// the JSON shape is a stable output contract. It is carried so
    /// [`describe`] can render an outcome from the outcome alone, rather
    /// than taking a verb a caller could mismatch against it.
    #[serde(skip)]
    pub verb: WriteVerb,
    /// What happened.
    pub result: WriteResult,
}

impl JsonlSerialize for WriteOutcome {
    fn write_jsonl(&self, out: &mut dyn std::io::Write) -> anyhow::Result<()> {
        write_scalar_jsonl(self, out)
    }
}

/// Runs one cell mutation, logging every attempt that isn't a dry run.
///
/// Never returns `Err`: every failure is a [`WriteResult`] variant, so the
/// caller renders one shape and the log records one shape. Exit code stays 0
/// regardless, matching ADR-0070 §10 / ADR-0071 §12.
pub async fn write(
    drive: &DriveClient,
    sheets: &SheetsClient,
    opts: &WriteOptions,
    rules: &[FolderPermissionRule],
) -> WriteOutcome {
    let started = Instant::now();
    let outcome = write_inner(drive, sheets, opts, rules).await;
    // The single logging site, and the only reason a dry run leaves no
    // record — `write_inner` never logs.
    if !opts.dry_run {
        record_attempt(&outcome, opts, started.elapsed());
    }
    outcome
}

async fn write_inner(
    drive: &DriveClient,
    sheets: &SheetsClient,
    opts: &WriteOptions,
    rules: &[FolderPermissionRule],
) -> WriteOutcome {
    let bare = |result| WriteOutcome {
        spreadsheet_id: opts.spreadsheet_id.clone(),
        file_name: None,
        range: None,
        resolved_folder_id: None,
        verb: opts.verb,
        result,
    };

    // `clear` takes no values; `write`/`append` with zero parsed rows have
    // nothing to send, and the API silently reports success on the request
    // it *is* sent (`Wrote to ...`) even though nothing changed — refuse
    // locally instead, matching `paste-data`'s empty-`--data` refusal
    // (#1941).
    if matches!(opts.verb, WriteVerb::Write | WriteVerb::Append) && opts.values.is_empty() {
        return bare(WriteResult::RefusedEmptyValues);
    }

    // Compose the range first: it is pure, and a conflicting
    // --sheet/--range pair should fail without spending a request.
    let range = match a1::compose(opts.sheet.as_deref(), opts.range.as_deref()) {
        Ok(range) => range,
        Err(err) => {
            return bare(WriteResult::Failed {
                detail: err.to_string(),
            })
        }
    };

    // ── Target resolution, pre-gate refusals, and the gate itself ──────
    // Shared with `structure.rs` via `target_gate::resolve`, so this whole
    // shape — the metadata fetch, the shortcut/non-spreadsheet checks, and
    // the file-id-then-ancestor-chain gate lookup — can't quietly drift
    // between the two engines.
    let (target, decision, resolved_folder_id, requires_lease) = match target_gate::resolve(
        drive,
        &opts.spreadsheet_id,
        DriveOperation::SheetsWrite,
        rules,
    )
    .await
    {
        target_gate::TargetGateOutcome::MetadataFetchFailed { detail } => {
            return bare(WriteResult::Failed { detail })
        }
        target_gate::TargetGateOutcome::Refused { target, refusal } => {
            let result = match refusal {
                SheetTargetRefusal::Shortcut => WriteResult::RefusedShortcut,
                SheetTargetRefusal::NotASpreadsheet { mime_type } => {
                    WriteResult::RefusedNotASpreadsheet { mime_type }
                }
                SheetTargetRefusal::NoVisibleParents => WriteResult::RefusedNoVisibleParents,
            };
            return WriteOutcome {
                spreadsheet_id: opts.spreadsheet_id.clone(),
                file_name: Some(target.name),
                range: Some(range),
                resolved_folder_id: None,
                verb: opts.verb,
                result,
            };
        }
        // A chain that could not be resolved is a refusal, never a
        // silent allow — ADR-0071 §3's highest-priority invariant.
        target_gate::TargetGateOutcome::GateFetchFailed { target, detail } => {
            return WriteOutcome {
                spreadsheet_id: opts.spreadsheet_id.clone(),
                file_name: Some(target.name),
                range: Some(range),
                resolved_folder_id: None,
                verb: opts.verb,
                result: WriteResult::Failed { detail },
            };
        }
        target_gate::TargetGateOutcome::Gated {
            target,
            decision,
            resolved_folder_id,
            requires_lease,
        } => (target, decision, resolved_folder_id, requires_lease),
    };

    let gated = |result| WriteOutcome {
        spreadsheet_id: opts.spreadsheet_id.clone(),
        file_name: Some(target.name.clone()),
        range: Some(range.clone()),
        resolved_folder_id: resolved_folder_id.clone(),
        verb: opts.verb,
        result,
    };

    if decision.verdict == write_gate::Verdict::Deny {
        return gated(WriteResult::Blocked {
            decided_by: decision.decided_by,
        });
    }

    // The named sheet may not exist — `--sheet`, when given, is
    // authoritative (`compose` already refused a `--range` naming a
    // *different* sheet); otherwise fall back to `range`'s own prefix, if
    // it has one. A bare, unprefixed range names no sheet to check here and
    // is left to the API, which defaults it to the first sheet. Checked
    // identically in `--dry-run` and a real run, like find-replace/sort's
    // own metadata fetch, rather than letting a doomed write/append/clear
    // pass its dry run and then fail live with an opaque "Unable to parse
    // range" (#1941).
    if let Some(title) = opts.sheet.clone().or_else(|| a1::sheet_title_of(&range)) {
        let api = SheetsApi::new(sheets);
        let workbook = match api.get_spreadsheet(&opts.spreadsheet_id).await {
            Ok(workbook) => workbook,
            Err(err) => {
                return gated(WriteResult::Failed {
                    detail: format!("{err:#}"),
                })
            }
        };
        if let Err(result) = grid_range::find_sheet_id(&workbook, &title, |title, available| {
            WriteResult::RefusedSheetNotFound { title, available }
        }) {
            return gated(result);
        }
    }

    if opts.dry_run {
        return gated(match opts.verb {
            WriteVerb::Clear => WriteResult::WouldClear,
            WriteVerb::Write | WriteVerb::Append => WriteResult::WouldWrite {
                rows: opts.values.len(),
                columns: opts.values.iter().map(Vec::len).max().unwrap_or(0),
            },
        });
    }

    // The lease check (ADR-0080 §9) sits here: after the permission gate
    // and the `--dry-run` branch, before the mutating call — see
    // `content_edit.rs::edit_inner`'s doc comment for the full reasoning,
    // shared verbatim by every leased engine. Sheets has no revision field
    // of any kind (§6), so the staleness check is a fresh `files.get`
    // immediately before the values call, not a reuse of the metadata
    // `target_gate::resolve` fetched before the (potentially slow)
    // ancestor-chain walk above.
    //
    // No request-building step sits between the `--dry-run` branch and the
    // gate below either — `opts.values`/`opts.input` are already-validated
    // caller-supplied data, passed straight to `values_update`/
    // `values_append`/`values_clear` once the gate succeeds — so the
    // build-before-gate invariant is satisfied with nothing to build. A
    // future fallible step added here must stay ahead of the gate, the same
    // way (#1688/#1742).
    let files_api = FilesApi::new(drive);
    let leased = LeasedWrite {
        log_prefix: "drive sheets write",
        operation: opts.verb.log_operation(),
        ledger_path: &opts.ledger_path,
        file_id: &opts.spreadsheet_id,
    };
    let lease_grant = match gate_optional_leased_write(
        leased,
        &files_api,
        requires_lease,
        opts.lease_token.as_deref(),
    )
    .await
    {
        Ok(grant) => grant,
        Err(err) => return gated(err.into_result()),
    };

    // ── The mutation ───────────────────────────────────────────────────
    let api = SheetsApi::new(sheets);
    let result = match opts.verb {
        WriteVerb::Write => api
            .values_update(&opts.spreadsheet_id, &range, &opts.values, opts.input)
            .await
            .map(into_written),
        WriteVerb::Append => api
            .values_append(&opts.spreadsheet_id, &range, &opts.values, opts.input)
            .await
            .map(|response| into_appended(response.updates.unwrap_or_default())),
        WriteVerb::Clear => api
            .values_clear(&opts.spreadsheet_id, &range)
            .await
            .map(|response| WriteResult::Cleared {
                updated_range: response.cleared_range,
            }),
    };

    let result =
        match conclude_native_leased_write(leased, &lease_grant, &files_api, result, |err| {
            format!("{err:#}")
        })
        .await
        {
            Ok(written) => written,
            Err(err) => WriteResult::Failed {
                detail: format!("{err:#}"),
            },
        };
    drop(lease_grant);

    gated(result)
}

/// Carries **every** count the API reported through to the outcome, not just
/// the cell count: `updated_rows`/`updated_columns` are what let a
/// transposed write be spotted in the request log after the fact, and
/// `docs/log.md` documents them as recorded.
fn into_written(response: UpdateValuesResponse) -> WriteResult {
    WriteResult::Written {
        updated_range: response.updated_range,
        updated_rows: response.updated_rows,
        updated_columns: response.updated_columns,
        updated_cells: response.updated_cells,
    }
}

/// [`into_written`] for an `append`, with the same counts.
fn into_appended(response: UpdateValuesResponse) -> WriteResult {
    WriteResult::Appended {
        updated_range: response.updated_range,
        updated_rows: response.updated_rows,
        updated_columns: response.updated_columns,
        updated_cells: response.updated_cells,
    }
}

/// Emits the `kind: "drivemutation"` record.
///
/// Inside the engine, never the CLI layer, so a future MCP caller cannot
/// bypass it — and so a `Blocked` outcome, which makes zero API calls, still
/// leaves a trace. Same reasoning as `content_edit.rs::record_attempt`.
fn record_attempt(outcome: &WriteOutcome, opts: &WriteOptions, duration: Duration) {
    let error = match &outcome.result {
        WriteResult::Failed { detail } => Some(detail.clone()),
        _ => None,
    };
    let decided_by = match &outcome.result {
        WriteResult::Blocked { decided_by } => decided_by.as_ref(),
        _ => None,
    };
    let decided_by = write_gate::decided_by_log_fields(decided_by);
    let (updated_range, updated_rows, updated_columns, updated_cells) = match &outcome.result {
        WriteResult::Written {
            updated_range,
            updated_rows,
            updated_columns,
            updated_cells,
        }
        | WriteResult::Appended {
            updated_range,
            updated_rows,
            updated_columns,
            updated_cells,
        } => (
            updated_range.clone(),
            *updated_rows,
            *updated_columns,
            *updated_cells,
        ),
        WriteResult::Cleared { updated_range } => (updated_range.clone(), None, None, None),
        _ => (None, None, None, None),
    };

    request_log::record_drive_mutation(DriveMutationOutcome {
        operation: opts.verb.log_operation(),
        file_id: outcome.spreadsheet_id.clone(),
        file_name: outcome.file_name.clone().unwrap_or_default(),
        status: outcome.result.log_status().to_string(),
        resolved_folder_id: outcome.resolved_folder_id.clone(),
        decided_by_folder_id: decided_by.folder_id,
        decided_by_depth: decided_by.depth,
        decided_by_file_id: decided_by.file_id,
        range: outcome.range.clone(),
        updated_range,
        updated_rows,
        updated_columns,
        updated_cells,
        error,
        duration,
        ..Default::default()
    });
}

/// Renders an outcome as a single human-readable line.
///
/// Lives here rather than in the CLI layer so the CLI and a future MCP
/// caller describe an outcome identically.
#[must_use]
pub fn describe(outcome: &WriteOutcome) -> String {
    let verb = outcome.verb;
    let name = outcome
        .file_name
        .as_deref()
        .unwrap_or(&outcome.spreadsheet_id);
    let range = outcome.range.as_deref().unwrap_or("(unresolved range)");
    match &outcome.result {
        WriteResult::WouldClear => format!("Would clear: {range} of '{name}'"),
        WriteResult::WouldWrite { rows, columns } => format!(
            "Would {}: {rows} row(s) x {columns} column(s) into {range} of '{name}'",
            verb.label()
        ),
        WriteResult::RefusedNotASpreadsheet { mime_type } => format!(
            "Refused: '{name}' is not a Google Sheet (mimeType: {mime_type}); \
             `drive sheets {}` only works on spreadsheets",
            verb.label()
        ),
        WriteResult::RefusedShortcut => format!(
            "Refused: '{name}' is a shortcut; `drive sheets {}` doesn't follow shortcuts — \
             resolve the target spreadsheet's id and use that instead",
            verb.label()
        ),
        WriteResult::RefusedNoVisibleParents => format!(
            "Refused: '{name}' has no parent folder visible to this account, so no folder \
             rule can apply to it. This is normal for a Sheet shared by link or email. \
             Grant it by id instead: add {{\"file_id\": \"<spreadsheet id>\", \"allow\": \
             [\"sheets-write\"]}} to write_permissions.rules. (Adding it to a folder in your \
             own Drive and granting that folder `sheets-write` also works.)"
        ),
        WriteResult::RefusedSheetNotFound { title, available } => {
            grid_range::no_sheet_refusal(&format!("'{name}'"), title, available)
        }
        WriteResult::RefusedEmptyValues => format!(
            "Refused: `drive sheets {}` was given no values to write (--values is empty)",
            verb.label()
        ),
        WriteResult::Blocked { decided_by } => match decided_by {
            Some(rule) => format!(
                "Blocked: {range} of '{name}' — refused by rule on {} {}{}",
                rule.kind_label(),
                rule.id(),
                rule.depth_suffix()
            ),
            None => format!(
                "Blocked: {range} of '{name}' — refused by default policy (no matching rule)"
            ),
        },
        WriteResult::RefusedNoLease => LeaseGateRefusal::NoLease
            .describe_line(&outcome.spreadsheet_id, &format!("'{name}'"))
            .unwrap_or_default(),
        WriteResult::RefusedLeaseExpired => LeaseGateRefusal::Expired
            .describe_line(&outcome.spreadsheet_id, &format!("'{name}'"))
            .unwrap_or_default(),
        WriteResult::RefusedLeaseWrongFile => LeaseGateRefusal::WrongFile
            .describe_line(&outcome.spreadsheet_id, &format!("'{name}'"))
            .unwrap_or_default(),
        WriteResult::RefusedLeaseStale => LeaseGateRefusal::Stale
            .describe_line(&outcome.spreadsheet_id, &format!("'{name}'"))
            .unwrap_or_default(),
        // Each success is its own variant, never inferred from an absent
        // cell count: the API is allowed to omit the counts (see
        // `UpdateValuesResponse`), and reading "a clear" into that would
        // report a destructive outcome for a write that was nothing of the
        // sort.
        WriteResult::Cleared { updated_range } => format!(
            "Cleared {} of '{name}'",
            updated_range.as_deref().unwrap_or(range)
        ),
        WriteResult::Appended {
            updated_range,
            updated_cells,
            ..
        } => {
            let where_ = updated_range.as_deref().unwrap_or(range);
            match updated_cells {
                Some(cells) => format!("Appended {cells} cell(s) to {where_} of '{name}'"),
                None => format!("Appended to {where_} of '{name}'"),
            }
        }
        WriteResult::Written {
            updated_range,
            updated_cells,
            ..
        } => {
            let where_ = updated_range.as_deref().unwrap_or(range);
            match updated_cells {
                Some(cells) => format!("Wrote {cells} cell(s) to {where_} of '{name}'"),
                None => format!("Wrote to {where_} of '{name}'"),
            }
        }
        WriteResult::Failed { detail } => {
            format!("Failed: {range} of '{name}': {detail}")
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::drive::test_support::seed_lease;

    /// Every `describe` arm renders to exactly one line, for every verb.
    ///
    /// Load-bearing rather than cosmetic. `describe` interpolates a
    /// Drive-supplied file name, a server-supplied range and the deciding
    /// rule's id, and its CLI caller sanitizes the **whole rendered line**
    /// rather than each interpolation — which it can only do because no arm
    /// emits a newline of its own. `sanitize_for_terminal` strips control
    /// characters, so a multi-line arm added later would be silently
    /// flattened into one run-on line rather than failing anything.
    ///
    /// The exhaustive `match` in `every_write_result` is the other half: a
    /// new `WriteResult` variant will not compile until it is listed there,
    /// so it cannot reach the terminal without passing through this check.
    #[test]
    fn every_describe_arm_renders_a_single_line() {
        for verb in [WriteVerb::Write, WriteVerb::Append, WriteVerb::Clear] {
            for result in every_write_result() {
                let outcome = WriteOutcome {
                    spreadsheet_id: "sheet-1".to_string(),
                    file_name: Some("Quarterly Plan".to_string()),
                    range: Some("Sheet1!A1:B2".to_string()),
                    resolved_folder_id: None,
                    verb,
                    result,
                };
                let rendered = describe(&outcome);
                assert_eq!(
                    rendered.lines().count(),
                    1,
                    "describe emitted {} lines for {:?}/{verb:?}: {rendered:?}",
                    rendered.lines().count(),
                    outcome.result
                );
                assert!(
                    !rendered.chars().any(char::is_control),
                    "describe emitted a control character for {:?}/{verb:?}: {rendered:?}",
                    outcome.result
                );
            }
        }
    }

    /// One of every `WriteResult` variant.
    ///
    /// The `match` is exhaustive and wildcard-free on purpose: adding a
    /// variant breaks this build, which is what forces the new arm through
    /// `every_describe_arm_renders_a_single_line`.
    fn every_write_result() -> Vec<WriteResult> {
        let all = vec![
            WriteResult::WouldWrite {
                rows: 2,
                columns: 3,
            },
            WriteResult::WouldClear,
            WriteResult::RefusedNotASpreadsheet {
                mime_type: "application/pdf".to_string(),
            },
            WriteResult::RefusedShortcut,
            WriteResult::RefusedNoVisibleParents,
            WriteResult::RefusedSheetNotFound {
                title: "Nope".to_string(),
                available: vec!["Sheet1".to_string()],
            },
            WriteResult::RefusedEmptyValues,
            WriteResult::Blocked { decided_by: None },
            WriteResult::Blocked {
                decided_by: Some(DecidingRule::Folder {
                    folder_id: "folder-1".to_string(),
                    depth: 2,
                }),
            },
            WriteResult::Blocked {
                decided_by: Some(DecidingRule::File {
                    file_id: "sheet-1".to_string(),
                }),
            },
            WriteResult::RefusedNoLease,
            WriteResult::RefusedLeaseExpired,
            WriteResult::RefusedLeaseWrongFile,
            WriteResult::RefusedLeaseStale,
            WriteResult::Written {
                updated_range: Some("Sheet1!A1:B2".to_string()),
                updated_rows: Some(2),
                updated_columns: Some(2),
                updated_cells: Some(4),
            },
            WriteResult::Written {
                updated_range: None,
                updated_rows: None,
                updated_columns: None,
                updated_cells: None,
            },
            WriteResult::Appended {
                updated_range: Some("Sheet1!A3:B3".to_string()),
                updated_rows: Some(1),
                updated_columns: Some(2),
                updated_cells: Some(2),
            },
            WriteResult::Cleared {
                updated_range: Some("Sheet1!A1:B2".to_string()),
            },
            WriteResult::Failed {
                detail: "the API said no".to_string(),
            },
        ];
        for result in &all {
            match result {
                WriteResult::WouldWrite { .. }
                | WriteResult::WouldClear
                | WriteResult::RefusedNotASpreadsheet { .. }
                | WriteResult::RefusedShortcut
                | WriteResult::RefusedNoVisibleParents
                | WriteResult::RefusedSheetNotFound { .. }
                | WriteResult::RefusedEmptyValues
                | WriteResult::Blocked { .. }
                | WriteResult::RefusedNoLease
                | WriteResult::RefusedLeaseExpired
                | WriteResult::RefusedLeaseWrongFile
                | WriteResult::RefusedLeaseStale
                | WriteResult::Written { .. }
                | WriteResult::Appended { .. }
                | WriteResult::Cleared { .. }
                | WriteResult::Failed { .. } => (),
            }
        }
        all
    }

    use crate::drive::auth::{DriveCredentials, DriveGrantedScopes};
    use crate::drive::sheets::client::SHEETS_API_URL;
    use crate::drive::types::GOOGLE_SHEET_MIME_TYPE;
    use crate::drive::write_gate::Verdict;
    use crate::test_support::env::MapEnv;
    use crate::utils::secret::Secret;
    use std::collections::HashSet;

    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,
        }
    }

    /// Both clients against one wiremock server, sharing an OAuth session.
    ///
    /// `replace_session` must run before the derive: it swaps the Drive
    /// client's whole transport, so deriving first would leave the Sheets
    /// client pointed at the real `oauth2.googleapis.com`.
    async fn clients(server: &wiremock::MockServer) -> (DriveClient, SheetsClient) {
        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 drive = DriveClient::new(&server.uri(), &test_credentials()).unwrap();
        crate::drive::client::test_support::replace_session(
            &mut drive,
            &test_credentials(),
            &format!("{}/token", server.uri()),
        );
        let env = MapEnv::new().with(SHEETS_API_URL, &server.uri());
        let sheets = SheetsClient::from_drive_client_with(&env, &drive).unwrap();
        (drive, sheets)
    }

    /// `version: "1"` throughout — matches [`opts`]'s default seeded
    /// lease, so any test reaching the mutating call has a live, non-stale
    /// lease by construction (ADR-0080 §9).
    fn mount_file(id: &str, mime_type: &str, parents: &[&str]) -> wiremock::Mock {
        let parents: Vec<&str> = parents.to_vec();
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path(format!("/drive/v3/files/{id}")))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": id, "name": id, "mimeType": mime_type, "parents": parents,
                    "version": "1",
                })),
            )
    }

    fn mount_folder(id: &str) -> wiremock::Mock {
        mount_file(id, "application/vnd.google-apps.folder", &[])
    }

    fn allow_rule(folder: &str) -> FolderPermissionRule {
        FolderPermissionRule {
            folder_id: Some(folder.to_string()),
            file_id: None,
            recursive: true,
            allow: std::iter::once(DriveOperation::SheetsWrite).collect(),
            deny: HashSet::default(),
            require_lease: true,
        }
    }

    /// Seeds a fresh, isolated ledger with a live lease for `"sheet-1"` at
    /// version `"1"` (matching [`mount_file`]'s default) and returns
    /// options carrying it. Every existing test built before the lease
    /// (ADR-0080 §9) reaches its mutating call this way by construction;
    /// see `structure.rs::opts`'s doc comment (same rationale, same
    /// leaked-tempdir mechanism) for why this doesn't need touching each
    /// test individually.
    fn opts(verb: WriteVerb, dry_run: bool) -> WriteOptions {
        let ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let token = seed_lease(&ledger_path, "sheet-1", "1");
        WriteOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb,
            range: Some("A1:B2".to_string()),
            sheet: None,
            values: vec![vec!["a".to_string(), "b".to_string()]],
            input: ValueInputOption::UserEntered,
            dry_run,
            lease_token: Some(token),
            ledger_path,
        }
    }

    // ── refusals that must precede the gate and the network ────────────

    #[tokio::test]
    async fn non_spreadsheet_is_refused_before_any_gate_or_sheets_call() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", "application/pdf", &["parent-1"])
            .mount(&server)
            .await;
        // Deliberately no mock for parent-1 (the gate never runs) and none
        // for any Sheets endpoint — proves the refusal short-circuits both,
        // even though the rule set below would otherwise permit the write.
        // Run as an `append` so the "names the verb the user typed" assertion
        // below is about a verb this outcome was actually produced by.
        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Append, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert!(matches!(
            outcome.result,
            WriteResult::RefusedNotASpreadsheet { .. }
        ));
        let text = describe(&outcome);
        assert!(text.contains("is not a Google Sheet"), "{text}");
        assert!(text.contains("drive sheets append"), "{text}");
    }

    #[tokio::test]
    async fn shortcut_is_refused_with_its_own_message_not_the_generic_one() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            "application/vnd.google-apps.shortcut",
            &["parent-1"],
        )
        .mount(&server)
        .await;
        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert!(matches!(outcome.result, WriteResult::RefusedShortcut));
        let text = describe(&outcome);
        assert!(text.contains("is a shortcut"), "{text}");
        assert!(!text.contains("is not a Google Sheet"), "{text}");
    }

    #[tokio::test]
    async fn a_sheet_with_no_visible_parents_is_refused_distinctly_from_a_blocked_one() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // The shape a Sheet shared by link comes back as.
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &[])
            .mount(&server)
            .await;
        let outcome = write(&drive, &sheets, &opts(WriteVerb::Write, false), &[]).await;
        assert!(matches!(
            outcome.result,
            WriteResult::RefusedNoVisibleParents
        ));
        // The message must not send the operator off to fix rules that
        // could never apply — it names the one rule shape that works.
        let text = describe(&outcome);
        assert!(text.contains("no parent folder visible"), "{text}");
        assert!(text.contains("file_id"), "{text}");
        assert!(text.contains("Adding it to a folder"), "{text}");
    }

    // ── the gate ───────────────────────────────────────────────────────

    #[tokio::test]
    async fn denied_target_makes_zero_sheets_calls() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        // No Sheets mock: any call would 404 and surface as Failed.
        let outcome = write(&drive, &sheets, &opts(WriteVerb::Write, false), &[]).await;
        assert!(
            matches!(outcome.result, WriteResult::Blocked { decided_by: None }),
            "{:?}",
            outcome.result
        );
        let text = describe(&outcome);
        assert!(
            text.contains("refused by default policy (no matching rule)"),
            "{text}"
        );
    }

    #[tokio::test]
    async fn a_blocked_by_rule_names_the_deciding_folder_in_the_message() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        let deny_rule = FolderPermissionRule {
            folder_id: Some("parent-1".to_string()),
            file_id: None,
            recursive: true,
            allow: HashSet::default(),
            deny: std::iter::once(DriveOperation::SheetsWrite).collect(),
            require_lease: true,
        };
        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[deny_rule],
        )
        .await;
        let text = describe(&outcome);
        assert!(
            text.contains("refused by rule on folder parent-1"),
            "{text}"
        );
    }

    #[tokio::test]
    async fn an_edit_rule_alone_does_not_permit_a_cell_write() {
        // The consequence of ADR-0073 §3, asserted end-to-end: an existing
        // `allow: ["edit"]` rule must not silently grant cell writes.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        let edit_only = FolderPermissionRule {
            folder_id: Some("parent-1".to_string()),
            file_id: None,
            recursive: true,
            allow: std::iter::once(DriveOperation::Edit).collect(),
            deny: HashSet::default(),
            require_lease: true,
        };
        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[edit_only],
        )
        .await;
        assert!(matches!(outcome.result, WriteResult::Blocked { .. }));
    }

    #[tokio::test]
    async fn ancestor_chain_fetch_failure_produces_failed_not_allow() {
        // ADR-0071 §3's highest-priority invariant, inherited here: an
        // unresolvable chain must never read as "no rule applies".
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/parent-1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;
        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert!(matches!(outcome.result, WriteResult::Failed { .. }));
    }

    // ── dry run ────────────────────────────────────────────────────────

    #[tokio::test]
    async fn dry_run_reports_dimensions_and_calls_no_values_endpoint() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        // No Sheets mock mounted.
        let mut o = opts(WriteVerb::Write, true);
        o.values = vec![
            vec!["a".to_string(), "b".to_string(), "c".to_string()],
            vec!["d".to_string()],
        ];
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(
            outcome.result,
            WriteResult::WouldWrite {
                rows: 2,
                columns: 3
            }
        );
        let text = describe(&outcome);
        assert!(text.contains("2 row(s) x 3 column(s)"), "{text}");
    }

    #[tokio::test]
    async fn write_refuses_an_unknown_sheet_identically_in_dry_run_and_a_real_run() {
        // `write --sheet Nope --dry-run` used to say `Would write ...` and
        // only fail live with `HTTP 400: Unable to parse range` (#1941).
        // find-replace/sort-range/auto-fill/paste all refuse this locally
        // in both modes; write/append/clear now do too.
        for dry_run in [true, false] {
            let server = wiremock::MockServer::start().await;
            let (drive, sheets) = clients(&server).await;
            mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
                .mount(&server)
                .await;
            mount_folder("parent-1").mount(&server).await;
            wiremock::Mock::given(wiremock::matchers::method("GET"))
                .and(wiremock::matchers::path("/v4/spreadsheets/sheet-1"))
                .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({
                        "spreadsheetId": "sheet-1",
                        "sheets": [{"properties": {"sheetId": 0, "title": "Sheet1"}}]
                    }),
                ))
                .mount(&server)
                .await;
            // No mock for the values endpoint: a call to it would fail the
            // test with a connection/404 error, distinct from the expected
            // refusal.
            let mut o = opts(WriteVerb::Write, dry_run);
            o.sheet = Some("Nope".to_string());
            o.range = None;
            let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
            assert_eq!(
                outcome.result,
                WriteResult::RefusedSheetNotFound {
                    title: "Nope".to_string(),
                    available: vec!["Sheet1".to_string()],
                },
                "dry_run={dry_run}"
            );
            let text = describe(&outcome);
            assert!(text.contains("no sheet titled 'Nope'"), "{text}");
            assert!(text.contains("Available: 'Sheet1'"), "{text}");
        }
    }

    #[tokio::test]
    async fn sheet_title_check_metadata_fetch_failure_produces_failed() {
        // The `--sheet`-existence check (#1941) does its own
        // `get_spreadsheet` call; a failure there must surface as `Failed`,
        // not panic or read as success.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/v4/spreadsheets/sheet-1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;
        let mut o = opts(WriteVerb::Write, false);
        o.sheet = Some("Sheet1".to_string());
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert!(matches!(outcome.result, WriteResult::Failed { .. }));
    }

    #[tokio::test]
    async fn write_and_append_refuse_empty_values_before_any_request() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // No mocks at all: the refusal must fire before even the metadata
        // fetch.
        for verb in [WriteVerb::Write, WriteVerb::Append] {
            let mut o = opts(verb, true);
            o.values = vec![];
            let outcome = write(&drive, &sheets, &o, &[]).await;
            assert_eq!(outcome.result, WriteResult::RefusedEmptyValues, "{verb:?}");
            let text = describe(&outcome);
            assert!(text.contains("no values to write"), "{text}");
        }
    }

    #[tokio::test]
    async fn clear_is_unaffected_by_empty_values() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        let mut o = opts(WriteVerb::Clear, true);
        o.values = vec![];
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::WouldClear);
    }

    #[tokio::test]
    async fn dry_run_surfaces_the_same_blocked_reasoning_as_a_real_denied_run() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .expect(2)
            .mount(&server)
            .await;
        mount_folder("parent-1").expect(2).mount(&server).await;

        let dry = write(&drive, &sheets, &opts(WriteVerb::Write, true), &[]).await;
        let real = write(&drive, &sheets, &opts(WriteVerb::Write, false), &[]).await;
        assert_eq!(dry.result, real.result);
    }

    // ── successful mutations ───────────────────────────────────────────

    #[tokio::test]
    async fn allowed_write_calls_values_update_once_with_the_input_option() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2",
            ))
            .and(wiremock::matchers::query_param(
                "valueInputOption",
                "USER_ENTERED",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "updatedRange": "'Q1'!A1:B2", "updatedRows": 1, "updatedCells": 2,
                })),
            )
            .expect(1)
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert_eq!(
            outcome.result,
            WriteResult::Written {
                updated_range: Some("'Q1'!A1:B2".to_string()),
                updated_rows: Some(1),
                updated_columns: None,
                updated_cells: Some(2),
            }
        );
        assert!(describe(&outcome).starts_with("Wrote 2 cell(s) "));
    }

    #[tokio::test]
    async fn raw_input_option_is_sent_when_requested() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::query_param("valueInputOption", "RAW"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
            .expect(1)
            .mount(&server)
            .await;

        let mut o = opts(WriteVerb::Write, false);
        o.input = ValueInputOption::Raw;
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert!(matches!(outcome.result, WriteResult::Written { .. }));
    }

    // ── the Drive write lease (ADR-0080 §9) ─────────────────────────────

    fn allow_rule_no_lease(folder: &str) -> FolderPermissionRule {
        FolderPermissionRule {
            require_lease: false,
            ..allow_rule(folder)
        }
    }

    #[tokio::test]
    async fn refuses_without_a_lease_when_the_rule_requires_one() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        // No PUT mock mounted — a refusal must make zero mutating calls.

        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = None;
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::RefusedNoLease);
    }

    #[tokio::test]
    async fn refuses_an_unknown_lease_token() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;

        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = Some("bogus-token".to_string());
        // Never seeded — no ledger exists at this fresh path.
        o.ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::RefusedLeaseExpired);
    }

    #[tokio::test]
    async fn refuses_a_lease_bound_to_a_different_file() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;

        let ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let token = seed_lease(&ledger_path, "other-sheet", "1");
        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = Some(token);
        o.ledger_path = ledger_path;
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::RefusedLeaseWrongFile);
    }

    #[tokio::test]
    async fn refuses_a_stale_lease() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // Live version "1" (`mount_file`'s default) but the lease was
        // acquired at "0" — the file has moved since.
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;

        let ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let token = seed_lease(&ledger_path, "sheet-1", "0");
        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = Some(token);
        o.ledger_path = ledger_path;
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::RefusedLeaseStale);
    }

    #[tokio::test]
    async fn reports_a_lock_acquisition_failure_as_failed() {
        // A pre-existing lock file simulates another `drive lease`
        // operation genuinely in progress — reported as an operational
        // failure, not folded into `RefusedLeaseExpired`.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;

        let o = opts(WriteVerb::Write, false);
        // Under `flock` (issue #1687), a busy lock now waits rather than
        // hard-failing (`check_and_lock_lease` -> `acquire_waiting`), so a
        // held `LedgerLock` no longer reproduces an immediate failure here.
        // A directory at the lock path does: opening it for write fails
        // outright with an I/O error, which is never retried.
        let mut lock_path = o.ledger_path.clone().into_os_string();
        lock_path.push(".lock");
        std::fs::create_dir(std::path::PathBuf::from(lock_path)).unwrap();

        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert!(matches!(outcome.result, WriteResult::Failed { .. }));
    }

    #[tokio::test]
    async fn a_failed_pre_lease_refetch_is_reported_as_failed_with_no_values_call() {
        // The gate's own resolve step succeeds off the first `files.get`,
        // but the fresh re-fetch feeding the staleness check (ADR-0080 §6)
        // fails — the write must report `Failed` and never reach the
        // values endpoint.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .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/sheet-1"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .with_priority(2)
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .respond_with(wiremock::ResponseTemplate::new(200))
            .expect(0)
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert!(matches!(outcome.result, WriteResult::Failed { .. }));
    }

    #[tokio::test]
    async fn a_rule_that_does_not_require_a_lease_skips_the_check_entirely() {
        // No lease token and no ledger seeded, with a rule that sets
        // `require_lease: false` — the write must still succeed, taking
        // the `else { None }` branch and never consulting the ledger.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "updatedRange": "'Q1'!A1:B2", "updatedRows": 1, "updatedCells": 2,
                })),
            )
            .expect(1)
            .mount(&server)
            .await;

        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = None;
        let outcome = write(&drive, &sheets, &o, &[allow_rule_no_lease("parent-1")]).await;
        assert!(matches!(outcome.result, WriteResult::Written { .. }));
    }

    #[tokio::test]
    async fn a_rule_that_does_not_require_a_lease_still_refuses_a_stale_one_if_presented() {
        // ADR-0080 §13: `require_lease: false` relaxes the *requirement*,
        // not the *meaning* — a token volunteered anyway is checked exactly
        // like a required one, including staleness.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // Live version "1" (`mount_file`'s default) but the lease was
        // acquired at "0" — the file has moved since.
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        // No PUT mock mounted — a refusal must make zero mutating calls.

        let ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let token = seed_lease(&ledger_path, "sheet-1", "0");
        let mut o = opts(WriteVerb::Write, false);
        o.lease_token = Some(token);
        o.ledger_path = ledger_path;
        let outcome = write(&drive, &sheets, &o, &[allow_rule_no_lease("parent-1")]).await;
        assert_eq!(outcome.result, WriteResult::RefusedLeaseStale);
    }

    // ── describing an outcome ──────────────────────────────────────────

    #[tokio::test]
    async fn a_write_whose_response_omits_the_counts_is_not_described_as_a_clear() {
        // `UpdateValuesResponse` deliberately tolerates missing counts, so
        // "no cell count" must never stand in for "this was a clear" — that
        // would report a destructive outcome for a plain write.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        let text = describe(&outcome);
        assert!(!text.contains("Cleared"), "{text}");
        assert!(text.starts_with("Wrote to "), "{text}");
    }

    #[tokio::test]
    async fn a_clear_dry_run_omits_the_always_zero_dimensions() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Clear, true),
            &[allow_rule("parent-1")],
        )
        .await;
        let text = describe(&outcome);
        assert!(!text.contains("row(s)"), "{text}");
        assert_eq!(text, "Would clear: A1:B2 of 'sheet-1'");
    }

    #[tokio::test]
    async fn append_reads_its_counts_from_the_nested_updates_object() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2:append",
            ))
            .and(wiremock::matchers::query_param(
                "insertDataOption",
                "INSERT_ROWS",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "tableRange": "'Q1'!A1:B3",
                    "updates": {"updatedRange": "'Q1'!A4:B4", "updatedCells": 2},
                })),
            )
            .expect(1)
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Append, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert_eq!(
            outcome.result,
            WriteResult::Appended {
                updated_range: Some("'Q1'!A4:B4".to_string()),
                updated_rows: None,
                updated_columns: None,
                updated_cells: Some(2),
            }
        );
        // A successful append must not read identically to an overwrite.
        assert!(describe(&outcome).starts_with("Appended 2 cell(s) "));
    }

    #[tokio::test]
    async fn clear_calls_the_clear_endpoint_and_reports_the_cleared_range() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2:clear",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"clearedRange": "'Q1'!A1:B2"})),
            )
            .expect(1)
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Clear, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert_eq!(
            outcome.result,
            WriteResult::Cleared {
                updated_range: Some("'Q1'!A1:B2".to_string()),
            }
        );
        assert!(describe(&outcome).starts_with("Cleared "));
    }

    #[tokio::test]
    async fn a_403_surfaces_the_write_scope_hint() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        // The google.rpc envelope Sheets actually returns — the hint only
        // fires because `error_reason` understands `status` too.
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .respond_with(
                wiremock::ResponseTemplate::new(403).set_body_json(serde_json::json!({
                    "error": {"code": 403, "message": "The caller does not have permission",
                              "status": "PERMISSION_DENIED"},
                })),
            )
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        let WriteResult::Failed { detail } = &outcome.result else {
            panic!("expected Failed, got {:?}", outcome.result);
        };
        assert!(detail.contains("--write-file"), "{detail}");
        assert!(detail.contains("--write-full"), "{detail}");
    }

    // ── range composition ──────────────────────────────────────────────

    #[tokio::test]
    async fn a_conflicting_sheet_and_range_fails_before_any_request() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // No mocks at all — not even files.get.
        let mut o = opts(WriteVerb::Write, false);
        o.range = Some("Other!A1".to_string());
        o.sheet = Some("Mine".to_string());
        let outcome = write(&drive, &sheets, &o, &[]).await;
        let WriteResult::Failed { detail } = &outcome.result else {
            panic!("expected Failed, got {:?}", outcome.result);
        };
        assert!(detail.contains("already names a sheet"), "{detail}");
    }

    #[tokio::test]
    async fn a_sheet_title_with_a_space_is_encoded_on_the_wire() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/v4/spreadsheets/sheet-1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "spreadsheetId": "sheet-1",
                    "sheets": [{"properties": {"sheetId": 0, "title": "My Sheet"}}]
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/'My%20Sheet'!A1:B2",
            ))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
            .expect(1)
            .mount(&server)
            .await;

        let mut o = opts(WriteVerb::Write, false);
        o.sheet = Some("My Sheet".to_string());
        let outcome = write(&drive, &sheets, &o, &[allow_rule("parent-1")]).await;
        assert!(matches!(outcome.result, WriteResult::Written { .. }));
    }

    // ── verb metadata ──────────────────────────────────────────────────

    #[test]
    fn log_operations_are_distinct_and_kebab_cased() {
        assert_eq!(WriteVerb::Write.log_operation(), "sheets-write");
        assert_eq!(WriteVerb::Append.log_operation(), "sheets-append");
        assert_eq!(WriteVerb::Clear.log_operation(), "sheets-clear");
    }

    #[test]
    fn labels_are_distinct_present_tense_verbs() {
        assert_eq!(WriteVerb::Write.label(), "write");
        assert_eq!(WriteVerb::Append.label(), "append");
        assert_eq!(WriteVerb::Clear.label(), "clear");
    }

    /// The JSON `status` names what happened, per verb (#1941): a clear's
    /// dry run carries no always-zero `rows`/`columns`, and a real append or
    /// clear is not reported as `written`.
    #[test]
    fn json_statuses_name_the_verb() {
        let json = |result: WriteResult| serde_json::to_value(&result).unwrap();
        assert_eq!(
            json(WriteResult::WouldClear),
            serde_json::json!({"status": "would-clear"})
        );
        assert_eq!(
            json(WriteResult::Cleared {
                updated_range: Some("Q1!A1:B2".to_string())
            }),
            serde_json::json!({"status": "cleared", "updated_range": "Q1!A1:B2"})
        );
        assert_eq!(
            json(WriteResult::Appended {
                updated_range: None,
                updated_rows: Some(1),
                updated_columns: Some(2),
                updated_cells: Some(2),
            })["status"],
            "appended"
        );
    }

    #[test]
    fn log_status_covers_every_variant() {
        assert_eq!(
            WriteResult::WouldWrite {
                rows: 0,
                columns: 0
            }
            .log_status(),
            "would-write"
        );
        assert_eq!(
            WriteResult::RefusedNotASpreadsheet {
                mime_type: String::new()
            }
            .log_status(),
            "refused-not-a-spreadsheet"
        );
        assert_eq!(
            WriteResult::RefusedShortcut.log_status(),
            "refused-shortcut"
        );
        assert_eq!(
            WriteResult::RefusedNoVisibleParents.log_status(),
            "refused-no-visible-parents"
        );
        assert_eq!(
            WriteResult::Blocked { decided_by: None }.log_status(),
            "blocked"
        );
        assert_eq!(
            WriteResult::Written {
                updated_range: None,
                updated_rows: None,
                updated_columns: None,
                updated_cells: None,
            }
            .log_status(),
            "written"
        );
        assert_eq!(WriteResult::WouldClear.log_status(), "would-clear");
        assert_eq!(
            WriteResult::Appended {
                updated_range: None,
                updated_rows: None,
                updated_columns: None,
                updated_cells: None,
            }
            .log_status(),
            "appended"
        );
        assert_eq!(
            WriteResult::Cleared {
                updated_range: None
            }
            .log_status(),
            "cleared"
        );
        assert_eq!(
            WriteResult::Failed {
                detail: String::new()
            }
            .log_status(),
            "failed"
        );
        assert_eq!(
            WriteResult::RefusedEmptyValues.log_status(),
            "refused-empty-values"
        );
        assert_eq!(WriteResult::RefusedNoLease.log_status(), "refused-no-lease");
        assert_eq!(
            WriteResult::RefusedLeaseExpired.log_status(),
            "refused-lease-expired"
        );
        assert_eq!(
            WriteResult::RefusedLeaseWrongFile.log_status(),
            "refused-lease-wrong-file"
        );
        assert_eq!(
            WriteResult::RefusedLeaseStale.log_status(),
            "refused-lease-stale"
        );
    }

    #[test]
    fn write_jsonl_emits_one_line_of_json() {
        let outcome = WriteOutcome {
            spreadsheet_id: "sheet-1".to_string(),
            file_name: Some("Budget".to_string()),
            range: Some("A1:B2".to_string()),
            resolved_folder_id: Some("parent-1".to_string()),
            verb: WriteVerb::Write,
            result: WriteResult::Written {
                updated_range: Some("A1:B2".to_string()),
                updated_rows: Some(1),
                updated_columns: Some(2),
                updated_cells: Some(2),
            },
        };
        let mut buf = Vec::new();
        outcome.write_jsonl(&mut buf).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert_eq!(text.matches('\n').count(), 1);
        let parsed: serde_json::Value = serde_json::from_str(text.trim()).unwrap();
        assert_eq!(parsed["result"]["status"], "written");
    }

    #[test]
    fn gate_denies_sheets_write_by_default() {
        let decision =
            write_gate::resolve(&["folder".to_string()], DriveOperation::SheetsWrite, &[]);
        assert_eq!(decision.verdict, Verdict::Deny);
    }

    // ── file-id rules (issue #1612) ────────────────────────────────────

    #[tokio::test]
    async fn a_file_rule_grants_a_sheet_with_no_visible_parents() {
        // The case issue #1612 exists for: before file rules there was no
        // rule an operator could write that would permit this at all.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &[])
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "updatedRange": "'Q1'!A1:B2", "updatedRows": 1, "updatedCells": 2,
                })),
            )
            .expect(1)
            .mount(&server)
            .await;

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[FolderPermissionRule::file("sheet-1").allowing([DriveOperation::SheetsWrite])],
        )
        .await;

        assert!(
            matches!(outcome.result, WriteResult::Written { .. }),
            "{:?}",
            outcome.result
        );
    }

    #[tokio::test]
    async fn a_file_rule_denies_even_when_a_parent_folder_would_allow() {
        // Depth −1 beats depth 0 in the restrictive direction too.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        // No mock for parent-1 and none for any Sheets endpoint: the file
        // rule must decide before either is reached.

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[
                allow_rule("parent-1"),
                FolderPermissionRule::file("sheet-1").denying([DriveOperation::SheetsWrite]),
            ],
        )
        .await;

        match &outcome.result {
            WriteResult::Blocked { decided_by } => {
                let rule = decided_by.as_ref().expect("a file rule decided this");
                assert_eq!(rule.kind_label(), "file");
                assert_eq!(rule.id(), "sheet-1");
            }
            other => panic!("expected Blocked, got {other:?}"),
        }
        let text = describe(&outcome);
        assert!(text.contains("refused by rule on file sheet-1"), "{text}");
        assert!(!text.contains("depth"), "a file rule has no depth: {text}");
    }

    #[tokio::test]
    async fn the_no_visible_parents_message_names_a_file_rule_as_the_fix() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &[])
            .mount(&server)
            .await;
        let outcome = write(&drive, &sheets, &opts(WriteVerb::Write, false), &[]).await;
        let text = describe(&outcome);
        assert!(text.contains("file_id"), "{text}");
        assert!(text.contains("sheets-write"), "{text}");
    }

    // ── the write's own audit trail (ADR-0080 §11) ─────────────────────

    #[tokio::test]
    async fn a_leased_write_concludes_its_audit_pair_with_allowed() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "updatedRange": "'Q1'!A1:B2", "updatedRows": 1, "updatedCells": 2,
                })),
            )
            .mount(&server)
            .await;
        let dir = tempfile::tempdir().unwrap();
        let audit = crate::test_support::AuditLogGuard::redirect(dir.path());

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        assert!(matches!(outcome.result, WriteResult::Written { .. }));

        let records = audit.records();
        assert_eq!(audit.verdicts(), ["pending", "allowed"], "{records:?}");
        // The verb, not the engine — the same `["drive", <log_operation>]`
        // this write's `drivemutation` record carries, so an auditor can
        // see which verb ran without joining back to `log.jsonl`.
        assert_eq!(records[0].command, ["drive", "sheets-write"]);
        // The post-write `files.get` the ledger refresh pays for also
        // feeds the outcome record (`mount_file` answers version "1").
        assert_eq!(
            records[1].context.get("version_after").map(String::as_str),
            Some("1")
        );
    }

    #[tokio::test]
    async fn a_leased_write_that_fails_concludes_its_audit_pair_with_the_error() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", GOOGLE_SHEET_MIME_TYPE, &["parent-1"])
            .mount(&server)
            .await;
        mount_folder("parent-1").mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("PUT"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1/values/A1:B2",
            ))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .mount(&server)
            .await;
        let dir = tempfile::tempdir().unwrap();
        let audit = crate::test_support::AuditLogGuard::redirect(dir.path());

        let outcome = write(
            &drive,
            &sheets,
            &opts(WriteVerb::Write, false),
            &[allow_rule("parent-1")],
        )
        .await;
        let WriteResult::Failed { detail } = &outcome.result else {
            panic!("expected Failed, got {:?}", outcome.result);
        };

        let records = audit.records();
        assert_eq!(audit.verdicts(), ["pending", "failed"], "{records:?}");
        assert_eq!(records[1].error.as_deref(), Some(detail.as_str()));
    }
}