omni_dev/drive/sheets/

filter.rs

//! The basic filter and filter views via `spreadsheets.batchUpdate` (issue
//! #1794, [ADR-0081](../../../docs/adrs/adr-0081.md)).
//!
//! Gated by [`DriveOperation::SheetsStructure`] — every mutating verb here
//! reaches that operation, unlike `structure.rs`'s split between
//! `SheetsStructure` and `SheetsDelete`. ADR-0081 §1 settles this: a filter
//! hides rows, which is view state, not data.
//!
//! **One exception: `set-basic-filter --sort-by`** (issue #1940). A basic
//! filter's `sortSpecs` are not view state — the server applies them by
//! physically reordering the rows of the filtered range, `sortRange`-style,
//! and the reorder survives `clear-basic-filter`. That is the same
//! permutation `randomize-range` performs, which live verification found
//! carries each row's formatting, notes and data-validation rules with it
//! (ADR-0083 §5), so it takes the same union of [`DriveOperation::SheetsWrite`]
//! and [`DriveOperation::SheetsStructure`] — see
//! `FilterVerb::gate_operations`. Its preview and report carry
//! `sort-range`'s record-integrity caveats. A filter view's sort
//! (`add-filter-view`/`update-filter-view --sort-by`) is per-view and never
//! touches the grid, so it stays on `SheetsStructure` alone.
//!
//! Two shapes, each with its own upsert/CRUD story:
//!
//! - **The basic filter** has at most one per sheet, so `set-basic-filter`
//!   is a plain upsert — no existing state to merge with — and
//!   `clear-basic-filter` needs no identifier beyond the sheet
//!   (`ClearBasicFilterRequest` carries only a `sheetId`).
//! - **Filter views** are many, named and directly id-addressed. Unlike
//!   `protection.rs`'s protected ranges (which have no user-facing handle
//!   and so are resolved by exact range match), a filter view's
//!   `filterViewId` is the stable handle `list-filter-views` discovers, so
//!   `update-filter-view`/`delete-filter-view` take it directly via
//!   `--filter-view-id` — there is no ambiguous-match case to handle.
//!
//! `list-filter-views` is read-only and ungated, like `list-protections`
//! (ADR-0081).
//!
//! **Two documented cuts, not silent gaps** — matching `validation.rs`'s own
//! framing of its curated condition surface:
//!
//! - **No `duplicate-filter-view`.** The issue's own background section
//!   names `duplicateFilterView` as part of the API surface, but its
//!   "Proposed verbs" list omits it; this module ships exactly that list.
//! - **`FilterCriteria` supports only `hiddenValues`** — the literal
//!   "uncheck a value in the dropdown" filter, the most common real use.
//!   Condition-based filter criteria (the same `BooleanCondition`
//!   vocabulary `validation.rs` curates for data validation) is not
//!   exposed.
//!
//! Shape mirrors `protection.rs`: compose a target, resolve it against a
//! freshly-fetched workbook, gate, dry-run, mutate, log.

use std::collections::BTreeMap;
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::{
    applied_reply_unreadable_line, BatchUpdateOutcome, RetryHint, SheetsApi,
};
use crate::drive::sheets::client::SheetsClient;
use crate::drive::sheets::grid_range;
use crate::drive::sheets::target_gate;
use crate::drive::sheets::types::{
    AddFilterViewRequest, BasicFilter, BatchUpdateRequestItem, ClearBasicFilterRequest,
    DeleteFilterViewRequest, FilterCriteria, FilterView, FilterViewUpdate, GridRange,
    SetBasicFilterRequest, SortOrder, SortSpec, Spreadsheet, UpdateFilterViewRequest,
};
use crate::drive::types::SheetTargetRefusal;
use crate::drive::write_gate::{self, DecidingRule, DriveOperation, FolderPermissionRule};
use crate::request_log::{self, DriveMutationOutcome};

/// Appended to `update-filter-view`'s summary when the change is sent as a
/// re-creation ([`FilterViewWrite::Replace`]).
const REPLACE_NOTE: &str = " — re-created under the same id, since Sheets cannot remove a \
     sort column in place";

/// The gate for every verb that only changes view state.
const GATE_STRUCTURE_ONLY: &[DriveOperation] = &[DriveOperation::SheetsStructure];
/// The gate for `set-basic-filter --sort-by`, which physically reorders
/// rows (module docs). `SheetsWrite` first:
/// `target_gate::TargetGateUnionOutcome::Gated::denied` names the first of
/// these that denied, and under the `sheets-structure`-only grant every
/// other verb here needs, `sheets-write` is the operation actually missing.
const GATE_WRITE_AND_STRUCTURE: &[DriveOperation] =
    &[DriveOperation::SheetsWrite, DriveOperation::SheetsStructure];

/// Which mutation to perform.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FilterVerb {
    /// Upsert the basic filter on a sheet.
    SetBasicFilter {
        /// The sheet to filter.
        sheet: String,
        /// The filtered A1 range, optionally carrying its own `Sheet!`
        /// prefix.
        range: String,
        /// `<COL>:<asc|desc>` sort specs, in priority order.
        sort_by: Vec<String>,
        /// `<COL>:<v1,v2,...>` hidden-value criteria, one per column.
        hide_values: Vec<String>,
    },
    /// Remove a sheet's basic filter.
    ClearBasicFilter {
        /// The sheet to clear the basic filter from.
        sheet: String,
    },
    /// Add a named filter view.
    AddFilterView {
        /// The sheet to filter.
        sheet: String,
        /// The filtered A1 range, optionally carrying its own `Sheet!`
        /// prefix.
        range: String,
        /// A human-readable name for the view.
        title: Option<String>,
        /// `<COL>:<asc|desc>` sort specs, in priority order.
        sort_by: Vec<String>,
        /// `<COL>:<v1,v2,...>` hidden-value criteria, one per column.
        hide_values: Vec<String>,
    },
    /// Change an existing filter view's title, range, sort order, or hidden
    /// values.
    UpdateFilterView {
        /// Which filter view to change, discovered via `list-filter-views`.
        filter_view_id: i64,
        /// A sheet title, supplying a prefix for a bare `range`, when
        /// changing the filtered range.
        sheet: Option<String>,
        /// The new filtered A1 range, when changing it.
        range: Option<String>,
        /// The new title, when changing it.
        title: Option<String>,
        /// Sort specs to merge in, replacing any existing entry sharing a
        /// column, else appending.
        sort_by: Vec<String>,
        /// Hidden-value criteria to merge in, replacing any existing entry
        /// for the named column.
        hide_values: Vec<String>,
        /// Reset the sort order to empty before applying `sort_by`.
        clear_sort: bool,
        /// Reset the criteria to empty before applying `hide_values`.
        clear_criteria: bool,
    },
    /// Remove a filter view.
    DeleteFilterView {
        /// Which filter view to remove, discovered via `list-filter-views`.
        filter_view_id: i64,
    },
}

impl FilterVerb {
    const fn log_operation(&self) -> &'static str {
        match self {
            Self::SetBasicFilter { .. } => "sheets-set-basic-filter",
            Self::ClearBasicFilter { .. } => "sheets-clear-basic-filter",
            Self::AddFilterView { .. } => "sheets-add-filter-view",
            Self::UpdateFilterView { .. } => "sheets-update-filter-view",
            Self::DeleteFilterView { .. } => "sheets-delete-filter-view",
        }
    }

    const fn label(&self) -> &'static str {
        match self {
            Self::SetBasicFilter { .. } => "set-basic-filter",
            Self::ClearBasicFilter { .. } => "clear-basic-filter",
            Self::AddFilterView { .. } => "add-filter-view",
            Self::UpdateFilterView { .. } => "update-filter-view",
            Self::DeleteFilterView { .. } => "delete-filter-view",
        }
    }

    /// Whether this verb physically reorders the grid's rows: only
    /// `set-basic-filter` with at least one `--sort-by` (issue #1940). A
    /// filter view's sort is per-view and reorders nothing.
    fn reorders_rows(&self) -> bool {
        matches!(self, Self::SetBasicFilter { sort_by, .. } if !sort_by.is_empty())
    }

    /// The operations this verb's gate is the union of, in the order a
    /// refusal reports them — `SheetsStructure` alone, except for a row
    /// reorder (see [`Self::reorders_rows`] and the module docs).
    fn gate_operations(&self) -> &'static [DriveOperation] {
        if self.reorders_rows() {
            GATE_WRITE_AND_STRUCTURE
        } else {
            GATE_STRUCTURE_ONLY
        }
    }
}

/// Per-call options.
#[derive(Debug, Clone)]
pub struct FilterOptions {
    /// Spreadsheet id.
    pub spreadsheet_id: String,
    /// Which mutation to perform.
    pub verb: FilterVerb,
    /// Classify and describe only; never call `batchUpdate`.
    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 FilterResult {
    /// `--dry-run`, and the gate would allow it.
    WouldChange {
        /// A human-readable summary of the effect.
        summary: String,
        /// For a row reorder (`set-basic-filter --sort-by`) whose range
        /// is narrower than its sheet, the record-integrity caveat
        /// `sort-range` leads with (ADR-0083 §6). Omitted otherwise.
        #[serde(skip_serializing_if = "Option::is_none")]
        width_warning: Option<String>,
    },
    /// The target is not a Google Sheet.
    RefusedNotASpreadsheet {
        /// The target's actual MIME type.
        mime_type: String,
    },
    /// The target is a shortcut, which we do not follow.
    RefusedShortcut,
    /// The target has no parents this account can see.
    RefusedNoVisibleParents,
    /// The named sheet does not exist in this workbook.
    RefusedSheetNotFound {
        /// The title that was not found.
        title: String,
        /// The titles that do exist.
        available: Vec<String>,
    },
    /// The verb's own arguments were invalid — a bad `--sort-by`/
    /// `--hide-values` format, a malformed `--sheet`/`--range` pair, or
    /// (for `update-filter-view`) nothing to change.
    RefusedInvalidRange {
        /// What was wrong and why.
        detail: String,
    },
    /// `update-filter-view`/`delete-filter-view` named a `--filter-view-id`
    /// that does not exist in this workbook.
    RefusedFilterViewNotFound {
        /// The id that was not found.
        filter_view_id: i64,
    },
    /// The folder write-permission gate refused it.
    Blocked {
        /// Which of `FilterVerb::gate_operations` denied first — only
        /// ever not `sheets-structure` for `set-basic-filter --sort-by`'s
        /// union gate, where which of the two denied is the actionable
        /// part of the message.
        operation: DriveOperation,
        /// The rule that decided the refusal, if any.
        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,
    /// The workbook is already in the requested state
    /// (`clear-basic-filter` on a sheet with no basic filter), so no
    /// request is sent. Reported identically by a dry run and a real run.
    Unchanged {
        /// Why nothing needed to change.
        detail: String,
    },
    /// The mutation succeeded.
    Changed {
        /// Same summary as [`Self::WouldChange`].
        summary: String,
        /// The filter view's stable id — server-assigned for
        /// `add-filter-view`, otherwise the one resolved against. `None`
        /// for `set-basic-filter`/`clear-basic-filter`.
        filter_view_id: Option<i64>,
        /// Same as [`Self::WouldChange`]'s.
        #[serde(skip_serializing_if = "Option::is_none")]
        width_warning: Option<String>,
    },
    /// The API answered 2xx — the mutation **was applied** — but its reply
    /// could not be read (issue #2021). Distinct from [`Self::Failed`]: a
    /// retry here could duplicate a filter view. The server-assigned id of an
    /// added filter view was in that reply, so it is unknown.
    AppliedReplyUnreadable {
        /// Same summary as [`Self::WouldChange`].
        summary: String,
        /// The filter view id resolved before the call — `Some` for
        /// `update-filter-view`/`delete-filter-view`, `None` otherwise. Never
        /// the server-assigned id of an added view (that was in the reply).
        filter_view_id: Option<i64>,
        /// Same as [`Self::Changed`]'s.
        #[serde(skip_serializing_if = "Option::is_none")]
        width_warning: Option<String>,
        /// Why the reply could not be read.
        detail: String,
    },
    /// An API or validation error.
    Failed {
        /// A human-readable summary of what failed.
        detail: String,
    },
}

impl FromLeaseRefusal for FilterResult {
    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 FilterResult {
    fn log_status(&self) -> &'static str {
        match self {
            Self::WouldChange { .. } => "would-change",
            Self::RefusedNotASpreadsheet { .. } => "refused-not-a-spreadsheet",
            Self::RefusedShortcut => "refused-shortcut",
            Self::RefusedNoVisibleParents => "refused-no-visible-parents",
            Self::RefusedSheetNotFound { .. } => "refused-sheet-not-found",
            Self::RefusedInvalidRange { .. } => "refused-invalid-range",
            Self::RefusedFilterViewNotFound { .. } => "refused-filter-view-not-found",
            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::Unchanged { .. } => "unchanged",
            Self::Changed { .. } => "changed",
            Self::AppliedReplyUnreadable { .. } => "applied-reply-unreadable",
            Self::Failed { .. } => "failed",
        }
    }
}

/// The full outcome of one attempt.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct FilterOutcome {
    /// The spreadsheet acted on.
    pub spreadsheet_id: String,
    /// Its Drive file name, when the metadata fetch got that far.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub file_name: Option<String>,
    /// The folder the gate evaluated against.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub resolved_folder_id: Option<String>,
    /// The sheet a `set-basic-filter`/`clear-basic-filter` resolved
    /// against, once known — the only identifier that can distinguish
    /// which sheet in a multi-sheet workbook was affected, since a basic
    /// filter (unlike a filter view) is scoped to a sheet rather than
    /// id-addressed (issue #1794, `docs/log.md`). `None` before the
    /// workbook resolves it, and for every other verb.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sheet_id: Option<i64>,
    /// Which mutation was attempted. Not serialised.
    #[serde(skip)]
    pub verb: FilterVerb,
    /// What happened.
    pub result: FilterResult,
}

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

/// Runs one filter mutation, logging every attempt that isn't a dry run.
pub async fn filter(
    drive: &DriveClient,
    sheets: &SheetsClient,
    opts: &FilterOptions,
    rules: &[FolderPermissionRule],
) -> FilterOutcome {
    let started = Instant::now();
    let outcome = filter_inner(drive, sheets, opts, rules).await;
    if !opts.dry_run {
        record_attempt(&outcome, opts, started.elapsed());
    }
    outcome
}

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

    let sort_specs = match parse_sort_specs(sort_by_flags(&opts.verb)) {
        Ok(specs) => specs,
        Err(detail) => return bare(FilterResult::RefusedInvalidRange { detail }),
    };
    let criteria = match parse_hidden_values(hide_values_flags(&opts.verb)) {
        Ok(criteria) => criteria,
        Err(detail) => return bare(FilterResult::RefusedInvalidRange { detail }),
    };
    let composed_range = match compose_target(&opts.verb) {
        Ok(composed) => composed,
        Err(detail) => return bare(FilterResult::RefusedInvalidRange { detail }),
    };
    if let Err(detail) = validate_verb(&opts.verb) {
        return bare(FilterResult::RefusedInvalidRange { detail });
    }

    let operations = opts.verb.gate_operations();

    let (target, verdict, denied, resolved_folder_id, requires_lease) =
        match target_gate::resolve_all(drive, &opts.spreadsheet_id, operations, rules).await {
            target_gate::TargetGateUnionOutcome::MetadataFetchFailed { detail } => {
                return bare(FilterResult::Failed { detail })
            }
            target_gate::TargetGateUnionOutcome::Refused { target, refusal } => {
                let result = match refusal {
                    SheetTargetRefusal::Shortcut => FilterResult::RefusedShortcut,
                    SheetTargetRefusal::NotASpreadsheet { mime_type } => {
                        FilterResult::RefusedNotASpreadsheet { mime_type }
                    }
                    SheetTargetRefusal::NoVisibleParents => FilterResult::RefusedNoVisibleParents,
                };
                return FilterOutcome {
                    spreadsheet_id: opts.spreadsheet_id.clone(),
                    file_name: Some(target.name),
                    resolved_folder_id: None,
                    sheet_id: None,
                    verb: opts.verb.clone(),
                    result,
                };
            }
            target_gate::TargetGateUnionOutcome::GateFetchFailed { target, detail } => {
                return FilterOutcome {
                    spreadsheet_id: opts.spreadsheet_id.clone(),
                    file_name: Some(target.name),
                    resolved_folder_id: None,
                    sheet_id: None,
                    verb: opts.verb.clone(),
                    result: FilterResult::Failed { detail },
                };
            }
            target_gate::TargetGateUnionOutcome::Gated {
                target,
                verdict,
                denied,
                resolved_folder_id,
                requires_lease,
            } => (target, verdict, denied, resolved_folder_id, requires_lease),
        };

    // Returns before the sheet/range resolution just below always have
    // `sheet_id: None` — there is nothing to resolve it against yet.
    // Distinct from `gated` below (never a shadow of it) so a future early
    // return accidentally added between the two is a compile error
    // ("cannot find value `gated`"), not a silent bind to the wrong
    // closure.
    let pre_gated = |result| FilterOutcome {
        spreadsheet_id: opts.spreadsheet_id.clone(),
        file_name: Some(target.name.clone()),
        resolved_folder_id: resolved_folder_id.clone(),
        sheet_id: None,
        verb: opts.verb.clone(),
        result,
    };

    if verdict == write_gate::Verdict::Deny {
        // `denied` is `Some` whenever the verdict is `Deny`; the fallback
        // only keeps this total.
        let (operation, decided_by) = denied.unwrap_or((operations[0], None));
        return pre_gated(FilterResult::Blocked {
            operation,
            decided_by,
        });
    }

    let api = SheetsApi::new(sheets);
    let workbook = match api
        .get_spreadsheet_with_filter_views(&opts.spreadsheet_id)
        .await
    {
        Ok(workbook) => workbook,
        Err(err) => {
            return pre_gated(FilterResult::Failed {
                detail: format!("{err:#}"),
            })
        }
    };

    // `update-filter-view --range` with no `--sheet` means the view's own
    // sheet.
    let composed_range = match (&opts.verb, composed_range) {
        (
            FilterVerb::UpdateFilterView {
                filter_view_id,
                sheet: None,
                ..
            },
            Some(composed),
        ) => {
            // An unknown id is reported as such here, before the bare range
            // could be refused for naming no sheet.
            let current_sheet = match find_existing_filter_view(&workbook, *filter_view_id) {
                Ok(view) => view.range.as_ref().map(|range| range.sheet_id),
                Err(result) => return pre_gated(result),
            };
            Some(grid_range::default_sheet_prefix(
                &workbook,
                composed,
                current_sheet,
            ))
        }
        (_, composed) => composed,
    };
    let resolved_target =
        match resolve_sheet_target(&workbook, &opts.verb, composed_range.as_deref()) {
            Ok(resolved) => resolved,
            Err(result) => return pre_gated(result),
        };
    let sheet_id = resolved_target.map(|grid| grid.sheet_id);

    // `set-basic-filter`/`clear-basic-filter` have no id-addressed handle
    // the way a filter view does, so the sheet id resolved above is the
    // only thing that can tell an audit record which sheet in a
    // multi-sheet workbook was affected (docs/log.md).
    let gated = |result| FilterOutcome {
        spreadsheet_id: opts.spreadsheet_id.clone(),
        file_name: Some(target.name.clone()),
        resolved_folder_id: resolved_folder_id.clone(),
        sheet_id: match &opts.verb {
            FilterVerb::SetBasicFilter { .. } | FilterVerb::ClearBasicFilter { .. } => sheet_id,
            _ => None,
        },
        verb: opts.verb.clone(),
        result,
    };

    let existing = match &opts.verb {
        FilterVerb::UpdateFilterView { filter_view_id, .. }
        | FilterVerb::DeleteFilterView { filter_view_id } => {
            match find_existing_filter_view(&workbook, *filter_view_id) {
                Ok(existing) => Some(existing),
                Err(result) => return gated(result),
            }
        }
        FilterVerb::SetBasicFilter { .. }
        | FilterVerb::ClearBasicFilter { .. }
        | FilterVerb::AddFilterView { .. } => None,
    };

    // Before the dry-run branch, so a dry run reports the no-op the real
    // run would. The workbook fetch above already carries each sheet's
    // basic filter.
    if let FilterVerb::ClearBasicFilter { sheet } = &opts.verb {
        let has_filter = sheet_id
            .and_then(|id| grid_range::find_sheet_by_id(&workbook, id))
            .is_some_and(|found| found.basic_filter.is_some());
        if !has_filter {
            return gated(FilterResult::Unchanged {
                detail: format!("sheet '{sheet}' has no basic filter to clear"),
            });
        }
    }

    // A sort column past the target range's width is silently accepted
    // client-side, but the API 500s live rather than rejecting cleanly
    // (`set-basic-filter --sort-by 9:asc` on a 5-column range) — validate
    // it locally, matching `add-pivot-table`'s column-offset check.
    // `update-filter-view` with no new range falls back to the existing
    // view's range, since that is what the sort would apply to. A range
    // whose column bounds are unknown (an open-ended row span) is left
    // unchecked, as elsewhere.
    let effective_grid = resolved_target.or_else(|| existing.and_then(|view| view.range));
    if let Some(grid) = effective_grid {
        if let (Some(start), Some(end)) = (grid.start_column_index, grid.end_column_index) {
            if let Some(spec) = sort_specs
                .iter()
                .find(|spec| spec.dimension_index < start || spec.dimension_index >= end)
            {
                return gated(FilterResult::RefusedInvalidRange {
                    detail: format!(
                        "sort column {} is outside the selected range's columns {start}..{end} \
                         (zero-based, end exclusive)",
                        spec.dimension_index
                    ),
                });
            }
        }
    }

    // `update-filter-view`'s write is decided here, before the dry-run
    // return, because it can refuse (a re-creation with no range to
    // re-create over) and because a re-creation is worth saying in the
    // preview. It only merges onto the already-fetched view, so it costs
    // no call.
    let update_write = match &opts.verb {
        FilterVerb::UpdateFilterView {
            filter_view_id,
            title,
            clear_sort,
            clear_criteria,
            ..
        } => {
            let Some(existing) = existing else {
                // omni-dev: coverage ignore-line reason="find_existing_filter_view returns Some for UpdateFilterView or has already returned RefusedFilterViewNotFound; this else-arm exists only to unwrap the shared Option"
                unreachable!("existing is resolved for UpdateFilterView above")
            };
            match build_update(
                existing,
                *filter_view_id,
                title,
                resolved_target,
                sort_specs.clone(),
                criteria.clone(),
                *clear_sort,
                *clear_criteria,
            ) {
                Ok(write) => Some(write),
                Err(detail) => return gated(FilterResult::RefusedInvalidRange { detail }),
            }
        }
        _ => None,
    };

    let mut summary = describe_effect(&opts.verb);
    if matches!(update_write, Some(FilterViewWrite::Replace(_))) {
        summary.push_str(REPLACE_NOTE);
    }
    // A delete is addressed by a bare id, so name what the id resolved to
    // — the view was already fetched to confirm it exists.
    if let (FilterVerb::DeleteFilterView { .. }, Some(view)) = (&opts.verb, existing) {
        summary.push_str(&describe_existing_view(&workbook, view));
    }
    let width_warning = if opts.verb.reorders_rows() {
        resolved_target
            .as_ref()
            .and_then(|grid| reorder_width_warning(&workbook, grid))
    } else {
        None
    };

    if opts.dry_run {
        return gated(FilterResult::WouldChange {
            summary,
            width_warning,
        });
    }

    // Every branch here reuses `resolved_target` rather than re-parsing
    // `composed_range` a second time; `resolve_sheet_target` already did
    // that work once, above.
    let (requests, existing_id) = match &opts.verb {
        FilterVerb::SetBasicFilter { .. } => {
            let Some(grid) = resolved_target else {
                // omni-dev: coverage ignore-line reason="resolve_sheet_target returns Some for SetBasicFilter or has already returned its refusal; this else-arm exists only to unwrap the shared Option"
                unreachable!("resolved_target is resolved for SetBasicFilter above")
            };
            (
                vec![BatchUpdateRequestItem::SetBasicFilter(
                    SetBasicFilterRequest {
                        filter: BasicFilter {
                            range: Some(grid),
                            sort_specs,
                            criteria,
                        },
                    },
                )],
                None,
            )
        }
        FilterVerb::ClearBasicFilter { .. } => {
            let Some(sheet_id) = sheet_id else {
                // omni-dev: coverage ignore-line reason="resolve_sheet_target returns Some for ClearBasicFilter or has already returned its refusal; this else-arm exists only to unwrap the shared Option"
                unreachable!("sheet_id is resolved for ClearBasicFilter above")
            };
            (
                vec![BatchUpdateRequestItem::ClearBasicFilter(
                    ClearBasicFilterRequest { sheet_id },
                )],
                None,
            )
        }
        FilterVerb::AddFilterView { title, .. } => {
            let Some(grid) = resolved_target else {
                // omni-dev: coverage ignore-line reason="resolve_sheet_target returns Some for AddFilterView or has already returned its refusal; this else-arm exists only to unwrap the shared Option"
                unreachable!("resolved_target is resolved for AddFilterView above")
            };
            (
                vec![BatchUpdateRequestItem::AddFilterView(
                    AddFilterViewRequest {
                        filter: FilterView {
                            filter_view_id: None,
                            title: title.clone(),
                            range: Some(grid),
                            named_range_id: None,
                            table_id: None,
                            sort_specs,
                            criteria,
                        },
                    },
                )],
                None,
            )
        }
        FilterVerb::UpdateFilterView { filter_view_id, .. } => {
            let Some(write) = update_write else {
                // omni-dev: coverage ignore-line reason="update_write is built for UpdateFilterView above or has already returned its refusal; this else-arm exists only to unwrap the shared Option"
                unreachable!("update_write is built for UpdateFilterView above")
            };
            (write.into_requests(*filter_view_id), Some(*filter_view_id))
        }
        FilterVerb::DeleteFilterView { filter_view_id } => (
            vec![BatchUpdateRequestItem::DeleteFilterView(
                DeleteFilterViewRequest {
                    filter_id: *filter_view_id,
                },
            )],
            Some(*filter_view_id),
        ),
    };

    // The lease check (ADR-0080 §9) sits here: after the permission gate
    // and the `--dry-run` branch, before the mutating call — see
    // `protection.rs::protection_inner`'s doc comment for the full
    // reasoning, shared verbatim by every leased engine.
    let files_api = FilesApi::new(drive);
    let leased = LeasedWrite {
        log_prefix: "drive sheets filter",
        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()),
    };

    let result = match conclude_native_leased_write(
        leased,
        &lease_grant,
        &files_api,
        api.batch_update(&opts.spreadsheet_id, requests).await,
        |err| format!("{err:#}"),
    )
    .await
    {
        Ok(BatchUpdateOutcome::Applied(response)) => {
            let filter_view_id = added_filter_view_id(&response).or(existing_id);
            FilterResult::Changed {
                summary,
                filter_view_id,
                width_warning,
            }
        }
        // NotIdempotent: a second `add-filter-view` would add a second view, and a
        // second `sort-range` would re-sort already-moved rows.
        Ok(BatchUpdateOutcome::AppliedReplyUnreadable { detail }) => {
            FilterResult::AppliedReplyUnreadable {
                summary,
                filter_view_id: existing_id,
                width_warning,
                detail,
            }
        }
        Err(err) => FilterResult::Failed {
            detail: format!("{err:#}"),
        },
    };
    drop(lease_grant);
    gated(result)
}

/// `sort-range`'s range-width-vs-sheet-width caveat for a row reorder over
/// `grid` (ADR-0083 §6): a reorder moves only the cells inside the range,
/// so any data in the same rows but outside the selected columns is left
/// behind. An open-ended column span reaches the sheet's last column.
fn reorder_width_warning(workbook: &Spreadsheet, grid: &GridRange) -> Option<String> {
    let allocated_columns = grid_range::find_sheet_by_id(workbook, grid.sheet_id)
        .and_then(|sheet| sheet.properties.as_ref())
        .and_then(|props| props.grid_properties.as_ref())
        .and_then(|props| props.column_count);
    let start_column = grid.start_column_index.unwrap_or(0);
    match grid.end_column_index.or(allocated_columns) {
        Some(end_column) => {
            grid_range::width_warning("sorting", start_column, end_column, allocated_columns)
        }
        // Open-ended to the right on a sheet of unknown width: only a
        // span that also starts past column A can leave columns behind.
        None if start_column == 0 => None,
        None => grid_range::width_warning("sorting", start_column, start_column, None),
    }
}

/// `--sort-by` flags for whichever verb carries them.
fn sort_by_flags(verb: &FilterVerb) -> &[String] {
    match verb {
        FilterVerb::SetBasicFilter { sort_by, .. }
        | FilterVerb::AddFilterView { sort_by, .. }
        | FilterVerb::UpdateFilterView { sort_by, .. } => sort_by,
        FilterVerb::ClearBasicFilter { .. } | FilterVerb::DeleteFilterView { .. } => &[],
    }
}

/// `--hide-values` flags for whichever verb carries them.
fn hide_values_flags(verb: &FilterVerb) -> &[String] {
    match verb {
        FilterVerb::SetBasicFilter { hide_values, .. }
        | FilterVerb::AddFilterView { hide_values, .. }
        | FilterVerb::UpdateFilterView { hide_values, .. } => hide_values,
        FilterVerb::ClearBasicFilter { .. } | FilterVerb::DeleteFilterView { .. } => &[],
    }
}

/// Parses `<COL>:<asc|desc>` flags into [`SortSpec`]s, in the order given.
fn parse_sort_specs(flags: &[String]) -> Result<Vec<SortSpec>, String> {
    flags
        .iter()
        .map(|flag| {
            let (col, order) = flag.split_once(':').ok_or_else(|| {
                format!("'{flag}' is not COLUMN:asc|desc — expected e.g. '0:asc'")
            })?;
            let dimension_index: i64 = col
                .trim()
                .parse()
                .map_err(|_| format!("'{col}' in '{flag}' is not a column index"))?;
            let sort_order = match order.trim().to_ascii_lowercase().as_str() {
                "asc" | "ascending" => SortOrder::Ascending,
                "desc" | "descending" => SortOrder::Descending,
                other => return Err(format!("'{other}' in '{flag}' is not 'asc' or 'desc'")),
            };
            Ok(SortSpec {
                dimension_index,
                sort_order,
            })
        })
        .collect()
}

/// Parses `<COL>:<v1,v2,...>` flags into a `criteria` map.
fn parse_hidden_values(flags: &[String]) -> Result<BTreeMap<String, FilterCriteria>, String> {
    let mut criteria = BTreeMap::new();
    for flag in flags {
        let (col, values) = flag.split_once(':').ok_or_else(|| {
            format!("'{flag}' is not COLUMN:VALUE[,VALUE...] — expected e.g. '1:Foo,Bar'")
        })?;
        let dimension_index: i64 = col
            .trim()
            .parse()
            .map_err(|_| format!("'{col}' in '{flag}' is not a column index"))?;
        if values.is_empty() {
            return Err(format!("'{flag}' names no values to hide"));
        }
        let hidden_values: Vec<String> = values.split(',').map(str::to_string).collect();
        let key = dimension_index.to_string();
        // A `BTreeMap::insert` on a repeated column would silently drop the
        // earlier `--hide-values` and keep only the last one, with the
        // dry-run preview (which walks the raw flags, not this map) echoing
        // both as if they combined (#1941).
        if criteria.contains_key(&key) {
            return Err(format!(
                "--hide-values for column {dimension_index} is given more than once"
            ));
        }
        criteria.insert(key, FilterCriteria::hiding(hidden_values));
    }
    Ok(criteria)
}

/// Composes the `--sheet`/`--range` pair a verb carries into one string,
/// exactly like `protection.rs`/`format.rs`. Verbs with no range flags at
/// all (`ClearBasicFilter`, `DeleteFilterView`) never call this.
fn compose_target(verb: &FilterVerb) -> Result<Option<String>, String> {
    match verb {
        FilterVerb::SetBasicFilter { sheet, range, .. }
        | FilterVerb::AddFilterView { sheet, range, .. } => a1::compose(Some(sheet), Some(range))
            .map(Some)
            .map_err(|err| err.to_string()),
        FilterVerb::UpdateFilterView { sheet, range, .. } => {
            match (sheet.as_deref(), range.as_deref()) {
                (None, None) => Ok(None),
                (sheet, range) => a1::compose(sheet, range)
                    .map(Some)
                    .map_err(|err| err.to_string()),
            }
        }
        FilterVerb::ClearBasicFilter { .. } | FilterVerb::DeleteFilterView { .. } => Ok(None),
    }
}

/// Rejects a verb whose own arguments are internally inconsistent, cheaply,
/// before ever fetching the workbook.
fn validate_verb(verb: &FilterVerb) -> Result<(), String> {
    if let FilterVerb::UpdateFilterView {
        sheet,
        range,
        title,
        sort_by,
        hide_values,
        clear_sort,
        clear_criteria,
        ..
    } = verb
    {
        let nothing_to_change = sheet.is_none()
            && range.is_none()
            && title.is_none()
            && sort_by.is_empty()
            && hide_values.is_empty()
            && !clear_sort
            && !clear_criteria;
        if nothing_to_change {
            return Err(
                "nothing to change: pass --title, --sheet/--range, --sort-by, --hide-values, \
                 --clear-sort, or --clear-criteria"
                    .to_string(),
            );
        }
    }
    Ok(())
}

/// Resolves the sheet (and, for every verb but `ClearBasicFilter`, the full
/// [`GridRange`]) a verb targets — once, so callers never need to re-parse
/// the same `--sheet`/`--range` composition a second time to get the range
/// they already resolved a sheet id from. `ClearBasicFilter`'s grid carries
/// only `sheet_id`, with every other bound `Default`, since it names a
/// sheet directly rather than a range within it — the same "whole sheet"
/// shape `protection.rs::resolve_grid`'s `--whole-sheet` case uses.
/// `UpdateFilterView` may target no sheet at all (leaving the existing
/// view's range untouched), hence the `Option`; `DeleteFilterView` never
/// does.
fn resolve_sheet_target(
    workbook: &Spreadsheet,
    verb: &FilterVerb,
    composed: Option<&str>,
) -> Result<Option<GridRange>, FilterResult> {
    match verb {
        FilterVerb::SetBasicFilter { .. } | FilterVerb::AddFilterView { .. } => {
            let composed = composed.unwrap_or_default();
            let (_, grid) = grid_range::resolve_grid_range(
                workbook,
                composed,
                |detail| FilterResult::RefusedInvalidRange { detail },
                |title, available| FilterResult::RefusedSheetNotFound { title, available },
            )?;
            Ok(Some(grid))
        }
        FilterVerb::ClearBasicFilter { sheet } => {
            let sheet_id = find_sheet_id(workbook, sheet)?;
            Ok(Some(GridRange {
                sheet_id,
                ..Default::default()
            }))
        }
        FilterVerb::UpdateFilterView { .. } => match composed {
            Some(composed) => {
                let (_, grid) = grid_range::resolve_grid_range(
                    workbook,
                    composed,
                    |detail| FilterResult::RefusedInvalidRange { detail },
                    |title, available| FilterResult::RefusedSheetNotFound { title, available },
                )?;
                Ok(Some(grid))
            }
            None => Ok(None),
        },
        FilterVerb::DeleteFilterView { .. } => Ok(None),
    }
}

fn find_sheet_id(workbook: &Spreadsheet, title: &str) -> Result<i64, FilterResult> {
    grid_range::find_sheet_id(workbook, title, |title, available| {
        FilterResult::RefusedSheetNotFound { title, available }
    })
}

/// Finds the one filter view whose id exactly matches `filter_view_id`.
/// Filter views are directly id-addressed (the id comes from
/// `list-filter-views`), so unlike `protection.rs::find_existing_protection`
/// there is no ambiguous-match case — ids are unique by construction.
fn find_existing_filter_view(
    workbook: &Spreadsheet,
    filter_view_id: i64,
) -> Result<&FilterView, FilterResult> {
    workbook
        .sheets
        .iter()
        .flat_map(|sheet| sheet.filter_views.iter())
        .find(|view| view.filter_view_id == Some(filter_view_id))
        .ok_or(FilterResult::RefusedFilterViewNotFound { filter_view_id })
}

/// How `update-filter-view` reaches the server (issue #1931).
///
/// `updateFilterView` can only *merge* (see [`FilterViewUpdate`]): it has
/// no way to remove a sort column. So an update that drops one is sent
/// instead as a delete of the view plus an `addFilterView` carrying the
/// view's own id and the complete resulting state, in one atomic
/// `batchUpdate`. Sheets honours the requested id, so the view keeps its
/// `filterViewId` (live-verified).
#[derive(Debug, Clone, PartialEq, Eq)]
enum FilterViewWrite {
    /// A plain `updateFilterView`.
    Update(UpdateFilterViewRequest),
    /// Delete and re-add the view under its own id, with this full state.
    Replace(FilterView),
}

impl FilterViewWrite {
    fn into_requests(self, filter_view_id: i64) -> Vec<BatchUpdateRequestItem> {
        match self {
            Self::Update(update) => vec![BatchUpdateRequestItem::UpdateFilterView(update)],
            Self::Replace(filter) => vec![
                BatchUpdateRequestItem::DeleteFilterView(DeleteFilterViewRequest {
                    filter_id: filter_view_id,
                }),
                BatchUpdateRequestItem::AddFilterView(AddFilterViewRequest { filter }),
            ],
        }
    }
}

/// Builds the write for `update-filter-view` from the existing view's
/// current state.
///
/// The resulting sort order is computed client-side: `--clear-sort`
/// starts from empty, else from the existing order; each parsed `SortSpec`
/// then replaces any entry sharing its `dimension_index` (else appends).
/// When that result still names every column the view already sorts by,
/// it is sent as-is in an `updateFilterView` — Sheets puts the sent list
/// first and appends the existing columns it doesn't name, so there are
/// none left over to append. When it drops one (only `--clear-sort` can),
/// the view is re-created instead ([`FilterViewWrite::Replace`]), since no
/// `updateFilterView` body removes a sort column.
///
/// Criteria merge per column on the server, so an update sends only the
/// columns that change: each `--hide-values` column, plus — under
/// `--clear-criteria` — a `{}` reset for every other column the view
/// currently filters on. A column neither names is left alone rather than
/// re-sent, so criteria this crate doesn't model survive.
///
/// Errors when a re-creation is needed but there is no range to re-create
/// the view over: the existing view read back with none (it is not bound
/// to a grid range this crate models) and no `--range` was given.
#[allow(clippy::too_many_arguments)]
fn build_update(
    existing: &FilterView,
    filter_view_id: i64,
    title: &Option<String>,
    range: Option<GridRange>,
    sort_by: Vec<SortSpec>,
    hide_values: BTreeMap<String, FilterCriteria>,
    clear_sort: bool,
    clear_criteria: bool,
) -> Result<FilterViewWrite, String> {
    let sort_specs = (clear_sort || !sort_by.is_empty()).then(|| {
        let mut resulting = if clear_sort {
            Vec::new()
        } else {
            existing.sort_specs.clone()
        };
        for spec in sort_by {
            if let Some(slot) = resulting
                .iter_mut()
                .find(|s| s.dimension_index == spec.dimension_index)
            {
                *slot = spec;
            } else {
                resulting.push(spec);
            }
        }
        resulting
    });
    let removes_sort_column = sort_specs.as_ref().is_some_and(|resulting| {
        existing.sort_specs.iter().any(|old| {
            !resulting
                .iter()
                .any(|new| new.dimension_index == old.dimension_index)
        })
    });

    if removes_sort_column {
        // A view bound to a table or named range is re-bound, not pinned
        // to its current range: Sheets rejects `addFilterView` with both a
        // binding and a `range` (live-verified), and a binding follows its
        // table/named range as it grows.
        let binding = existing
            .table_id
            .as_ref()
            .map(|id| format!("table {id}"))
            .or_else(|| {
                existing
                    .named_range_id
                    .as_ref()
                    .map(|id| format!("named range {id}"))
            });
        let range = match (binding, range) {
            (Some(binding), Some(_)) => {
                return Err(format!(
                    "removing a sort column re-creates filter view {filter_view_id}, which is \
                     bound to {binding}; re-creating it with --sheet/--range would unbind it, so \
                     drop --sheet/--range, or change the range in a separate update first"
                ))
            }
            (Some(_), None) => None,
            (None, range) => Some(range.or(existing.range).ok_or_else(|| {
                format!(
                    "removing a sort column re-creates filter view {filter_view_id}, but it has \
                     no grid range to re-create it over; pass --sheet/--range"
                )
            })?),
        };
        let mut criteria: BTreeMap<String, FilterCriteria> = if clear_criteria {
            BTreeMap::new()
        } else {
            existing
                .criteria
                .iter()
                .filter(|(_, criteria)| !criteria.is_empty())
                .map(|(col, criteria)| (col.clone(), criteria.clone()))
                .collect()
        };
        criteria.extend(hide_values);
        return Ok(FilterViewWrite::Replace(FilterView {
            filter_view_id: Some(filter_view_id),
            title: title.clone().or_else(|| existing.title.clone()),
            range,
            named_range_id: existing.named_range_id.clone(),
            table_id: existing.table_id.clone(),
            sort_specs: sort_specs.unwrap_or_default(),
            criteria,
        }));
    }

    let criteria = (clear_criteria || !hide_values.is_empty()).then(|| {
        let mut changes: BTreeMap<String, FilterCriteria> = if clear_criteria {
            existing
                .criteria
                .iter()
                .filter(|(_, criteria)| !criteria.is_empty())
                .map(|(col, _)| (col.clone(), FilterCriteria::default()))
                .collect()
        } else {
            BTreeMap::new()
        };
        changes.extend(hide_values);
        changes
    });

    let update = FilterViewUpdate {
        filter_view_id,
        title: title.clone(),
        range,
        sort_specs,
        criteria,
    };
    let fields = [
        ("title", update.title.is_some()),
        ("range", update.range.is_some()),
        ("sortSpecs", update.sort_specs.is_some()),
        ("criteria", update.criteria.is_some()),
    ]
    .into_iter()
    .filter_map(|(name, present)| present.then_some(name))
    .collect::<Vec<_>>()
    .join(",");
    Ok(FilterViewWrite::Update(UpdateFilterViewRequest {
        filter: update,
        fields,
    }))
}

fn added_filter_view_id(
    response: &crate::drive::sheets::types::BatchUpdateResponse,
) -> Option<i64> {
    response
        .replies
        .iter()
        .find_map(|reply| reply.add_filter_view.as_ref())
        .and_then(|added| added.filter.as_ref())
        .and_then(|filter| filter.filter_view_id)
}

fn describe_effect(verb: &FilterVerb) -> String {
    match verb {
        FilterVerb::SetBasicFilter {
            sort_by,
            hide_values,
            ..
        } => describe_filter_effect("set basic filter", sort_by, hide_values),
        FilterVerb::ClearBasicFilter { .. } => "clear basic filter".to_string(),
        FilterVerb::AddFilterView {
            title,
            sort_by,
            hide_values,
            ..
        } => {
            let named = title
                .as_deref()
                .map_or_else(String::new, |t| format!(" '{t}'"));
            describe_filter_effect(&format!("add filter view{named}"), sort_by, hide_values)
        }
        FilterVerb::UpdateFilterView {
            title,
            sort_by,
            hide_values,
            clear_sort,
            clear_criteria,
            ..
        } => {
            let mut parts = Vec::new();
            if let Some(title) = title {
                parts.push(format!("title='{title}'"));
            }
            if *clear_sort {
                parts.push("clear sort".to_string());
            }
            if !sort_by.is_empty() {
                parts.push(format!("sort {}", sort_by.join(",")));
            }
            if *clear_criteria {
                parts.push("clear criteria".to_string());
            }
            if !hide_values.is_empty() {
                parts.push(format!("hide {}", hide_values.join(",")));
            }
            if parts.is_empty() {
                "update filter view".to_string()
            } else {
                format!("update filter view ({})", parts.join(" "))
            }
        }
        FilterVerb::DeleteFilterView { .. } => "delete filter view".to_string(),
    }
}

/// Identifies an existing filter view for a summary: its title and range,
/// e.g. ` 'Open items' over 'Q1'!A1:D10`. A range that is not fully bounded
/// falls back to [`grid_range::render_grid_range`]; a view with neither
/// renders nothing.
fn describe_existing_view(workbook: &Spreadsheet, view: &FilterView) -> String {
    let mut text = String::new();
    if let Some(title) = view.title.as_deref().filter(|t| !t.is_empty()) {
        text.push_str(&format!(" '{title}'"));
    }
    if let Some(grid) = &view.range {
        let a1 = grid_range::sheet_title_by_id(workbook, grid.sheet_id)
            .and_then(|sheet| grid_range::bounded_range_to_a1(&sheet, grid))
            .unwrap_or_else(|| grid_range::render_grid_range(grid));
        text.push_str(&format!(" over {a1}"));
    }
    text
}

fn describe_filter_effect(prefix: &str, sort_by: &[String], hide_values: &[String]) -> String {
    let mut parts = Vec::new();
    if !sort_by.is_empty() {
        parts.push(format!("sort {}", sort_by.join(",")));
    }
    if !hide_values.is_empty() {
        parts.push(format!("hide {}", hide_values.join(",")));
    }
    if parts.is_empty() {
        prefix.to_string()
    } else {
        format!("{prefix} ({})", parts.join(" "))
    }
}

fn record_attempt(outcome: &FilterOutcome, opts: &FilterOptions, duration: Duration) {
    request_log::record_drive_mutation(mutation_record(outcome, opts, duration));
}

/// The request-log record for one attempt. Split from [`record_attempt`] so
/// the logged fields are unit-testable.
fn mutation_record(
    outcome: &FilterOutcome,
    opts: &FilterOptions,
    duration: Duration,
) -> DriveMutationOutcome {
    let error = match &outcome.result {
        FilterResult::Failed { detail } | FilterResult::AppliedReplyUnreadable { detail, .. } => {
            Some(detail.clone())
        }
        _ => None,
    };
    let decided_by = match &outcome.result {
        FilterResult::Blocked { decided_by, .. } => decided_by.as_ref(),
        _ => None,
    };
    let decided_by = write_gate::decided_by_log_fields(decided_by);
    let filter_view_id = match &outcome.result {
        FilterResult::Changed { filter_view_id, .. }
        | FilterResult::AppliedReplyUnreadable { filter_view_id, .. } => *filter_view_id,
        _ => None,
    };
    let fields_changed = match &outcome.result {
        // An applied change always records what it was, whatever the verb.
        FilterResult::AppliedReplyUnreadable { summary, .. } => Some(summary.clone()),
        FilterResult::Changed { summary, .. }
            if matches!(opts.verb, FilterVerb::UpdateFilterView { .. }) =>
        {
            Some(summary.clone())
        }
        _ => None,
    };

    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,
        sheet_id: outcome.sheet_id,
        filter_view_id,
        fields_changed,
        error,
        duration,
        ..Default::default()
    }
}

/// Renders an outcome as human-readable text.
#[must_use]
pub fn describe(outcome: &FilterOutcome) -> String {
    describe_lines(outcome).join("\n")
}

/// Renders an outcome as its individual lines, none of which contains a
/// newline.
#[must_use]
pub fn describe_lines(outcome: &FilterOutcome) -> Vec<String> {
    let verb = &outcome.verb;
    let book = outcome.file_name.as_deref().map_or_else(
        || format!("'{}'", outcome.spreadsheet_id),
        |n| format!("'{n}'"),
    );
    match &outcome.result {
        FilterResult::WouldChange {
            summary,
            width_warning,
        } => change_lines(
            verb,
            format!("Would {summary} in {book}"),
            width_warning.as_deref(),
            false,
        ),
        FilterResult::RefusedNotASpreadsheet { mime_type } => vec![format!(
            "Refused: {book} is not a Google Sheet (mimeType: {mime_type}); \
             `drive sheets {}` only works on spreadsheets",
            verb.label()
        )],
        FilterResult::RefusedShortcut => vec![format!(
            "Refused: {book} is a shortcut; `drive sheets {}` doesn't follow shortcuts",
            verb.label()
        )],
        FilterResult::RefusedNoVisibleParents => {
            let ops = verb
                .gate_operations()
                .iter()
                .map(|op| format!("\"{op}\""))
                .collect::<Vec<_>>()
                .join(", ");
            vec![format!(
                "Refused: {book} has no parent folder visible to this account, so no folder \
                 rule can apply to it. Grant it by id instead: add {{\"file_id\": \
                 \"<spreadsheet id>\", \"allow\": [{ops}]}} to write_permissions.rules."
            )]
        }
        FilterResult::RefusedSheetNotFound { title, available } => {
            vec![grid_range::no_sheet_refusal(&book, title, available)]
        }
        FilterResult::RefusedInvalidRange { detail } => vec![format!("Refused: {detail}")],
        FilterResult::RefusedFilterViewNotFound { filter_view_id } => vec![format!(
            "Refused: {book} has no filter view with id {filter_view_id}; run \
             `drive sheets list-filter-views` to see what exists"
        )],
        FilterResult::Blocked {
            operation,
            decided_by,
        } => vec![match decided_by {
            Some(rule) => format!(
                "Blocked: {} on {book} refused by rule on {} {}{}",
                verb.label(),
                rule.kind_label(),
                rule.id(),
                rule.depth_suffix()
            ),
            None => format!(
                "Blocked: {} on {book} refused by default policy (no matching rule for \
                 {operation})",
                verb.label()
            ),
        }],
        FilterResult::RefusedNoLease => LeaseGateRefusal::NoLease
            .describe_line(&outcome.spreadsheet_id, &book)
            .into_iter()
            .collect(),
        FilterResult::RefusedLeaseExpired => LeaseGateRefusal::Expired
            .describe_line(&outcome.spreadsheet_id, &book)
            .into_iter()
            .collect(),
        FilterResult::RefusedLeaseWrongFile => LeaseGateRefusal::WrongFile
            .describe_line(&outcome.spreadsheet_id, &book)
            .into_iter()
            .collect(),
        FilterResult::RefusedLeaseStale => LeaseGateRefusal::Stale
            .describe_line(&outcome.spreadsheet_id, &book)
            .into_iter()
            .collect(),
        FilterResult::Changed {
            summary,
            filter_view_id,
            width_warning,
        } => {
            let id = filter_view_id.map_or_else(String::new, |id| format!(" (id {id})"));
            change_lines(
                verb,
                format!("Applied: {summary}{id} in {book}"),
                width_warning.as_deref(),
                true,
            )
        }
        FilterResult::Unchanged { detail } => vec![format!("Unchanged: {detail} in {book}")],
        FilterResult::AppliedReplyUnreadable {
            summary,
            filter_view_id,
            width_warning,
            detail,
        } => {
            let id = filter_view_id.map_or_else(String::new, |id| format!(" (id {id})"));
            let mut lines = vec![applied_reply_unreadable_line(
                &format!("{summary}{id}"),
                &book,
                RetryHint::NotIdempotent,
                detail,
            )];
            if matches!(verb, FilterVerb::AddFilterView { .. }) {
                lines.push(
                    "  the new filter view's id was in the unreadable reply; run \
                     `list-filter-views` to find it"
                        .to_string(),
                );
            }
            // The plan-time caveats `Changed` prints around its summary: the
            // summary is already in the line above, so keep only the rest.
            lines.extend(
                change_lines(verb, String::new(), width_warning.as_deref(), true)
                    .into_iter()
                    .filter(|line| !line.is_empty()),
            );
            lines
        }
        FilterResult::Failed { detail } => vec![format!("Failed: {detail}")],
    }
}

/// A would-change/changed report's lines. A row reorder
/// (`set-basic-filter --sort-by`) gets `sort-range`'s caveats around the
/// summary — the width warning first, when there is one, then the
/// out-of-range-references line (ADR-0083 §6); every other verb is the
/// summary alone.
fn change_lines(
    verb: &FilterVerb,
    summary: String,
    width_warning: Option<&str>,
    changed: bool,
) -> Vec<String> {
    if !verb.reorders_rows() {
        return vec![summary];
    }
    grid_range::reorder_caveat_lines("sorting", summary, width_warning, changed, None)
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::drive::auth::{DriveCredentials, DriveGrantedScopes};
    use crate::drive::sheets::client::SHEETS_API_URL;
    use crate::drive::sheets::types::Sheet;
    use crate::drive::test_support::seed_lease;
    use crate::test_support::env::MapEnv;
    use crate::utils::secret::Secret;
    use std::collections::HashSet;

    // ── parse_sort_specs / parse_hidden_values ──────────────────────────

    #[test]
    fn parse_sort_specs_accepts_asc_and_desc() {
        let specs = parse_sort_specs(&["0:asc".to_string(), "2:desc".to_string()]).unwrap();
        assert_eq!(
            specs,
            vec![
                SortSpec {
                    dimension_index: 0,
                    sort_order: SortOrder::Ascending
                },
                SortSpec {
                    dimension_index: 2,
                    sort_order: SortOrder::Descending
                },
            ]
        );
    }

    #[test]
    fn parse_sort_specs_rejects_missing_colon() {
        let err = parse_sort_specs(&["0asc".to_string()]).unwrap_err();
        assert!(err.contains("COLUMN:asc|desc"), "{err}");
    }

    #[test]
    fn parse_sort_specs_rejects_non_numeric_column() {
        let err = parse_sort_specs(&["x:asc".to_string()]).unwrap_err();
        assert!(err.contains("not a column index"), "{err}");
    }

    #[test]
    fn parse_sort_specs_rejects_unknown_direction() {
        let err = parse_sort_specs(&["0:sideways".to_string()]).unwrap_err();
        assert!(err.contains("not 'asc' or 'desc'"), "{err}");
    }

    #[test]
    fn parse_hidden_values_builds_a_criteria_map() {
        let criteria = parse_hidden_values(&["1:Foo,Bar".to_string()]).unwrap();
        assert_eq!(
            criteria.get("1").unwrap().hidden_values,
            vec!["Foo".to_string(), "Bar".to_string()]
        );
    }

    #[test]
    fn parse_hidden_values_rejects_missing_colon() {
        let err = parse_hidden_values(&["1Foo".to_string()]).unwrap_err();
        assert!(err.contains("COLUMN:VALUE"), "{err}");
    }

    #[test]
    fn parse_hidden_values_rejects_empty_value_list() {
        let err = parse_hidden_values(&["1:".to_string()]).unwrap_err();
        assert!(err.contains("names no values"), "{err}");
    }

    #[test]
    fn parse_hidden_values_rejects_a_repeated_column() {
        // `--hide-values 1:Cherry --hide-values 1:Apple` silently kept only
        // the last one via `BTreeMap::insert`, while the dry-run preview
        // (which walks the raw flags) echoed both (#1941).
        let err =
            parse_hidden_values(&["1:Cherry".to_string(), "1:Apple".to_string()]).unwrap_err();
        assert!(err.contains("column 1 is given more than once"), "{err}");
    }

    // ── log_operation / label ────────────────────────────────────────────

    #[test]
    fn every_verb_has_a_distinct_log_operation_and_label() {
        let verbs = [
            FilterVerb::SetBasicFilter {
                sheet: String::new(),
                range: String::new(),
                sort_by: Vec::new(),
                hide_values: Vec::new(),
            },
            FilterVerb::ClearBasicFilter {
                sheet: String::new(),
            },
            FilterVerb::AddFilterView {
                sheet: String::new(),
                range: String::new(),
                title: None,
                sort_by: Vec::new(),
                hide_values: Vec::new(),
            },
            FilterVerb::UpdateFilterView {
                filter_view_id: 1,
                sheet: None,
                range: None,
                title: None,
                sort_by: Vec::new(),
                hide_values: Vec::new(),
                clear_sort: false,
                clear_criteria: false,
            },
            FilterVerb::DeleteFilterView { filter_view_id: 1 },
        ];
        let ops: HashSet<&str> = verbs.iter().map(FilterVerb::log_operation).collect();
        let labels: HashSet<&str> = verbs.iter().map(FilterVerb::label).collect();
        assert_eq!(ops.len(), verbs.len());
        assert_eq!(labels.len(), verbs.len());
    }

    // ── validate_verb ────────────────────────────────────────────────────

    #[test]
    fn validate_verb_rejects_update_filter_view_with_nothing_to_change() {
        let verb = FilterVerb::UpdateFilterView {
            filter_view_id: 1,
            sheet: None,
            range: None,
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        };
        let err = validate_verb(&verb).unwrap_err();
        assert!(err.contains("nothing to change"), "{err}");
    }

    #[test]
    fn validate_verb_accepts_update_filter_view_with_only_clear_sort() {
        let verb = FilterVerb::UpdateFilterView {
            filter_view_id: 1,
            sheet: None,
            range: None,
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: true,
            clear_criteria: false,
        };
        assert!(validate_verb(&verb).is_ok());
    }

    // ── build_update ─────────────────────────────────────────────────────

    fn filter_view(id: i64) -> FilterView {
        FilterView {
            filter_view_id: Some(id),
            ..Default::default()
        }
    }

    fn spec(dimension_index: i64, sort_order: SortOrder) -> SortSpec {
        SortSpec {
            dimension_index,
            sort_order,
        }
    }

    fn hiding(col: &str, values: &[&str]) -> BTreeMap<String, FilterCriteria> {
        let mut criteria = BTreeMap::new();
        criteria.insert(
            col.to_string(),
            FilterCriteria::hiding(values.iter().map(|v| (*v).to_string()).collect()),
        );
        criteria
    }

    /// A view shaped like the one the #1931 live probe started from:
    /// sorted by columns 2 then 3, hiding `b` in column 1.
    fn sorted_filtered_view(id: i64) -> FilterView {
        FilterView {
            filter_view_id: Some(id),
            title: Some("Old".to_string()),
            range: Some(GridRange {
                sheet_id: 0,
                start_row_index: Some(0),
                end_row_index: Some(6),
                start_column_index: Some(0),
                end_column_index: Some(4),
            }),
            named_range_id: None,
            table_id: None,
            sort_specs: vec![
                spec(2, SortOrder::Descending),
                spec(3, SortOrder::Descending),
            ],
            criteria: hiding("1", &["b"]),
        }
    }

    /// The `batchUpdate` `requests` array the write turns into, as JSON —
    /// what actually reaches the wire, which is where #1931 hid.
    fn wire(write: FilterViewWrite, filter_view_id: i64) -> serde_json::Value {
        serde_json::to_value(write.into_requests(filter_view_id)).unwrap()
    }

    #[test]
    fn build_update_sends_only_the_changed_criteria_columns() {
        let mut existing = filter_view(3);
        existing.criteria = hiding("0", &["Keep"]);
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            Vec::new(),
            hiding("1", &["New"]),
            false,
            false,
        )
        .unwrap();
        // Sheets merges criteria per column, so column 0 is left out of
        // the body rather than re-sent — which is also what keeps any
        // criteria this crate doesn't model on it intact.
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "criteria",
                "filter": {"filterViewId": 3, "criteria": {"1": {"hiddenValues": ["New"]}}},
            }}])
        );
    }

    #[test]
    fn build_update_clear_criteria_resets_each_filtered_column_with_an_empty_object() {
        let mut existing = sorted_filtered_view(3);
        existing.criteria.extend(hiding("3", &["x"]));
        // A column an earlier reset left behind as `{}` needs no second
        // reset.
        existing
            .criteria
            .insert("0".to_string(), FilterCriteria::default());
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            Vec::new(),
            hiding("3", &["y"]),
            false,
            true,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "criteria",
                "filter": {"filterViewId": 3, "criteria": {
                    "1": {},
                    "3": {"hiddenValues": ["y"]},
                }},
            }}])
        );
    }

    #[test]
    fn build_update_clear_criteria_on_an_unfiltered_view_still_names_the_field() {
        let write = build_update(
            &filter_view(3),
            3,
            &None,
            None,
            Vec::new(),
            BTreeMap::new(),
            false,
            true,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "criteria",
                "filter": {"filterViewId": 3, "criteria": {}},
            }}])
        );
    }

    #[test]
    fn build_update_sort_spec_replaces_matching_column_and_appends_others() {
        let mut existing = filter_view(3);
        existing.sort_specs = vec![spec(0, SortOrder::Ascending), spec(1, SortOrder::Ascending)];
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            vec![
                spec(0, SortOrder::Descending),
                spec(2, SortOrder::Ascending),
            ],
            BTreeMap::new(),
            false,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "sortSpecs",
                "filter": {"filterViewId": 3, "sortSpecs": [
                    {"dimensionIndex": 0, "sortOrder": "DESCENDING"},
                    {"dimensionIndex": 1, "sortOrder": "ASCENDING"},
                    {"dimensionIndex": 2, "sortOrder": "ASCENDING"},
                ]},
            }}])
        );
    }

    #[test]
    fn build_update_sort_by_a_new_column_sends_the_whole_list() {
        // The issue's non-clear case: `--sort-by 1:asc` on `[2 desc]`.
        let mut existing = filter_view(3);
        existing.sort_specs = vec![spec(2, SortOrder::Descending)];
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            vec![spec(1, SortOrder::Ascending)],
            BTreeMap::new(),
            false,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "sortSpecs",
                "filter": {"filterViewId": 3, "sortSpecs": [
                    {"dimensionIndex": 2, "sortOrder": "DESCENDING"},
                    {"dimensionIndex": 1, "sortOrder": "ASCENDING"},
                ]},
            }}])
        );
    }

    #[test]
    fn build_update_title_only_sends_no_sort_or_criteria() {
        let write = build_update(
            &sorted_filtered_view(3),
            3,
            &Some("New title".to_string()),
            None,
            Vec::new(),
            BTreeMap::new(),
            false,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "title",
                "filter": {"filterViewId": 3, "title": "New title"},
            }}])
        );
    }

    #[test]
    fn build_update_clear_sort_then_sort_by_recreates_the_view_with_exactly_that_order() {
        // The issue's second symptom: `--clear-sort --sort-by 0:asc` on
        // `[2 desc, 3 desc]` came back `[0 asc, 2 desc, 3 desc]`, because
        // `updateFilterView` merges. Re-creating is the only way to drop 2
        // and 3.
        let write = build_update(
            &sorted_filtered_view(3),
            3,
            &None,
            None,
            vec![spec(0, SortOrder::Ascending)],
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([
                {"deleteFilterView": {"filterId": 3}},
                {"addFilterView": {"filter": {
                    "filterViewId": 3,
                    "title": "Old",
                    "range": {"sheetId": 0, "startRowIndex": 0, "endRowIndex": 6,
                              "startColumnIndex": 0, "endColumnIndex": 4},
                    "sortSpecs": [{"dimensionIndex": 0, "sortOrder": "ASCENDING"}],
                    "criteria": {"1": {"hiddenValues": ["b"]}},
                }}},
            ])
        );
    }

    #[test]
    fn build_update_clear_sort_alone_recreates_the_view_with_no_sort() {
        let write = build_update(
            &sorted_filtered_view(3),
            3,
            &None,
            None,
            Vec::new(),
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        let FilterViewWrite::Replace(view) = write else {
            panic!("expected a re-creation, got {write:?}");
        };
        assert!(view.sort_specs.is_empty());
        assert_eq!(view.criteria, hiding("1", &["b"]));
        assert_eq!(view.filter_view_id, Some(3));
    }

    #[test]
    fn build_update_clear_sort_on_an_unsorted_view_is_a_plain_update() {
        let mut existing = sorted_filtered_view(3);
        existing.sort_specs.clear();
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            Vec::new(),
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "sortSpecs",
                "filter": {"filterViewId": 3, "sortSpecs": []},
            }}])
        );
    }

    #[test]
    fn build_update_clear_sort_keeping_every_column_reorders_in_place() {
        // Every existing column is still named, so the server's merge
        // (sent entries first) yields exactly `[3 desc, 2 asc]` — live-
        // verified — with no re-creation.
        let write = build_update(
            &sorted_filtered_view(3),
            3,
            &None,
            None,
            vec![
                spec(3, SortOrder::Descending),
                spec(2, SortOrder::Ascending),
            ],
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([{"updateFilterView": {
                "fields": "sortSpecs",
                "filter": {"filterViewId": 3, "sortSpecs": [
                    {"dimensionIndex": 3, "sortOrder": "DESCENDING"},
                    {"dimensionIndex": 2, "sortOrder": "ASCENDING"},
                ]},
            }}])
        );
    }

    #[test]
    fn build_update_recreation_applies_every_other_change_too() {
        let mut existing = sorted_filtered_view(3);
        // A reset column and an unmodelled condition, both as a read-back
        // would carry them.
        existing
            .criteria
            .insert("0".to_string(), FilterCriteria::default());
        let condition: FilterCriteria = serde_json::from_value(serde_json::json!({
            "condition": {"type": "NUMBER_GREATER", "values": [{"userEnteredValue": "2"}]}
        }))
        .unwrap();
        existing.criteria.insert("2".to_string(), condition.clone());
        let range = GridRange {
            sheet_id: 0,
            start_row_index: Some(0),
            end_row_index: Some(20),
            start_column_index: Some(0),
            end_column_index: Some(4),
        };
        let write = build_update(
            &existing,
            3,
            &Some("Renamed".to_string()),
            Some(range),
            Vec::new(),
            hiding("3", &["x"]),
            true,
            false,
        )
        .unwrap();
        let mut criteria = hiding("1", &["b"]);
        criteria.insert("2".to_string(), condition);
        criteria.extend(hiding("3", &["x"]));
        assert_eq!(
            write,
            FilterViewWrite::Replace(FilterView {
                filter_view_id: Some(3),
                title: Some("Renamed".to_string()),
                range: Some(range),
                named_range_id: None,
                table_id: None,
                sort_specs: Vec::new(),
                criteria,
            })
        );
    }

    #[test]
    fn build_update_recreation_with_clear_criteria_keeps_only_the_new_columns() {
        let write = build_update(
            &sorted_filtered_view(3),
            3,
            &None,
            None,
            Vec::new(),
            hiding("3", &["x"]),
            true,
            true,
        )
        .unwrap();
        let FilterViewWrite::Replace(view) = write else {
            panic!("expected a re-creation, got {write:?}");
        };
        assert_eq!(view.criteria, hiding("3", &["x"]));
    }

    #[test]
    fn build_update_recreation_without_a_range_is_refused() {
        let mut existing = sorted_filtered_view(3);
        existing.range = None;
        let detail = build_update(
            &existing,
            3,
            &None,
            None,
            Vec::new(),
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap_err();
        assert!(detail.contains("--sheet/--range"), "{detail}");
    }

    #[test]
    fn build_update_recreation_rebinds_a_table_bound_view_without_its_range() {
        // Sheets reports a table-bound view's resolved range too, but
        // rejects an `addFilterView` carrying both ("Only one of tableId and
        // range may be set") — live-verified — so only the binding goes.
        let mut existing = sorted_filtered_view(3);
        existing.table_id = Some("454645209".to_string());
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            vec![spec(0, SortOrder::Ascending)],
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        assert_eq!(
            wire(write, 3),
            serde_json::json!([
                {"deleteFilterView": {"filterId": 3}},
                {"addFilterView": {"filter": {
                    "filterViewId": 3,
                    "title": "Old",
                    "tableId": "454645209",
                    "sortSpecs": [{"dimensionIndex": 0, "sortOrder": "ASCENDING"}],
                    "criteria": {"1": {"hiddenValues": ["b"]}},
                }}},
            ])
        );
    }

    #[test]
    fn build_update_recreation_rebinds_a_named_range_bound_view_without_its_range() {
        let mut existing = sorted_filtered_view(3);
        existing.named_range_id = Some("2014020481".to_string());
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            Vec::new(),
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap();
        let FilterViewWrite::Replace(view) = write else {
            panic!("expected a re-creation, got {write:?}");
        };
        assert_eq!(view.named_range_id.as_deref(), Some("2014020481"));
        assert_eq!(view.table_id, None);
        assert_eq!(view.range, None);
    }

    #[test]
    fn build_update_recreation_of_a_bound_view_with_a_new_range_is_refused() {
        let mut existing = sorted_filtered_view(3);
        existing.table_id = Some("454645209".to_string());
        let detail = build_update(
            &existing,
            3,
            &None,
            sorted_filtered_view(3).range,
            Vec::new(),
            BTreeMap::new(),
            true,
            false,
        )
        .unwrap_err();
        assert!(detail.contains("bound to table 454645209"), "{detail}");
        assert!(detail.contains("would unbind it"), "{detail}");
    }

    #[test]
    fn build_update_plain_update_of_a_bound_view_sends_no_binding() {
        // Only the re-creation re-sends the binding; an in-place update
        // leaves it alone.
        let mut existing = sorted_filtered_view(3);
        existing.table_id = Some("454645209".to_string());
        let write = build_update(
            &existing,
            3,
            &None,
            None,
            vec![spec(1, SortOrder::Ascending)],
            BTreeMap::new(),
            false,
            false,
        )
        .unwrap();
        let FilterViewWrite::Update(update) = write else {
            panic!("expected an in-place update, got {write:?}");
        };
        assert_eq!(update.fields, "sortSpecs");
    }

    #[test]
    fn filter_criteria_keeps_unmodelled_fields_across_a_round_trip() {
        let json = serde_json::json!({
            "hiddenValues": ["a"],
            "condition": {"type": "TEXT_CONTAINS", "values": [{"userEnteredValue": "x"}]},
        });
        let criteria: FilterCriteria = serde_json::from_value(json.clone()).unwrap();
        assert!(!criteria.is_empty());
        assert_eq!(serde_json::to_value(&criteria).unwrap(), json);
        assert!(FilterCriteria::default().is_empty());
        assert_eq!(
            serde_json::to_value(FilterCriteria::default()).unwrap(),
            serde_json::json!({})
        );
    }

    #[test]
    fn add_filter_view_wire_shape_is_unchanged() {
        let request = AddFilterViewRequest {
            filter: FilterView {
                filter_view_id: None,
                title: Some("T".to_string()),
                range: sorted_filtered_view(3).range,
                named_range_id: None,
                table_id: None,
                sort_specs: vec![spec(2, SortOrder::Descending)],
                criteria: hiding("1", &["b"]),
            },
        };
        assert_eq!(
            serde_json::to_string(&request).unwrap(),
            r#"{"filter":{"title":"T","range":{"sheetId":0,"startRowIndex":0,"endRowIndex":6,"startColumnIndex":0,"endColumnIndex":4},"sortSpecs":[{"dimensionIndex":2,"sortOrder":"DESCENDING"}],"criteria":{"1":{"hiddenValues":["b"]}}}}"#
        );
    }

    // ── find_existing_filter_view ───────────────────────────────────────

    fn workbook_with_filter_views(views: Vec<FilterView>) -> Spreadsheet {
        Spreadsheet {
            sheets: vec![Sheet {
                properties: None,
                filter_views: views,
                ..Default::default()
            }],
            ..Default::default()
        }
    }

    /// Clearing a basic filter that isn't there sends nothing and says so,
    /// in a dry run exactly as in a real run.
    #[tokio::test]
    async fn clearing_an_absent_basic_filter_is_unchanged_in_dry_run_and_real_run() {
        for dry_run in [true, false] {
            let server = wiremock::MockServer::start().await;
            let (drive, sheets) = clients(&server).await;
            mount_file(
                "sheet-1",
                crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
                &["folder-1"],
            )
            .mount(&server)
            .await;
            mount_folder("folder-1").mount(&server).await;
            mount_workbook(serde_json::json!([
                {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
            ]))
            .mount(&server)
            .await;
            let mut opts = dry_run_opts(FilterVerb::ClearBasicFilter {
                sheet: "Q1".to_string(),
            });
            opts.dry_run = dry_run;
            let outcome = filter(&drive, &sheets, &opts, &[allow_rule("folder-1")]).await;
            assert_eq!(
                outcome.result,
                FilterResult::Unchanged {
                    detail: "sheet 'Q1' has no basic filter to clear".to_string()
                },
                "dry_run={dry_run}"
            );
            assert!(describe(&outcome).starts_with("Unchanged: sheet 'Q1' has no basic filter"));
            let requests = server.received_requests().await.unwrap();
            assert!(!requests
                .iter()
                .any(|r| r.url.path().ends_with(":batchUpdate")));
        }
    }

    #[test]
    fn describe_existing_view_names_the_title_and_the_range() {
        let workbook = Spreadsheet {
            sheets: vec![Sheet {
                properties: Some(crate::drive::sheets::types::SheetProperties {
                    sheet_id: Some(0),
                    title: "Q1".to_string(),
                    ..Default::default()
                }),
                ..Default::default()
            }],
            ..Default::default()
        };
        let view = FilterView {
            filter_view_id: Some(7),
            title: Some("Open items".to_string()),
            range: Some(GridRange {
                sheet_id: 0,
                start_row_index: Some(0),
                end_row_index: Some(10),
                start_column_index: Some(0),
                end_column_index: Some(4),
            }),
            ..Default::default()
        };
        assert_eq!(
            describe_existing_view(&workbook, &view),
            " 'Open items' over 'Q1'!A1:D10"
        );
        let open_ended = FilterView {
            title: None,
            range: Some(GridRange {
                sheet_id: 0,
                start_column_index: Some(0),
                end_column_index: Some(1),
                ..Default::default()
            }),
            ..view
        };
        assert_eq!(
            describe_existing_view(&workbook, &open_ended),
            " over sheetId 0, cols 1-1"
        );
        assert_eq!(describe_existing_view(&workbook, &filter_view(7)), "");
    }

    #[test]
    fn find_existing_filter_view_matches_by_id() {
        let workbook = workbook_with_filter_views(vec![filter_view(3), filter_view(5)]);
        let found = find_existing_filter_view(&workbook, 5).unwrap();
        assert_eq!(found.filter_view_id, Some(5));
    }

    #[test]
    fn find_existing_filter_view_refuses_when_absent() {
        let workbook = workbook_with_filter_views(vec![filter_view(3)]);
        let err = find_existing_filter_view(&workbook, 99).unwrap_err();
        assert!(matches!(
            err,
            FilterResult::RefusedFilterViewNotFound { filter_view_id: 99 }
        ));
    }

    // ── write_jsonl / log_status ─────────────────────────────────────────

    #[test]
    fn write_jsonl_emits_one_line_of_json() {
        let outcome = FilterOutcome {
            spreadsheet_id: "sheet-1".to_string(),
            file_name: Some("Budget".to_string()),
            resolved_folder_id: None,
            sheet_id: None,
            verb: FilterVerb::DeleteFilterView { filter_view_id: 3 },
            result: FilterResult::Changed {
                summary: "delete filter view".to_string(),
                filter_view_id: Some(3),
                width_warning: None,
            },
        };
        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"], "changed");
    }

    #[test]
    fn filter_result_log_status_names_every_variant() {
        assert_eq!(
            FilterResult::WouldChange {
                summary: String::new(),
                width_warning: None,
            }
            .log_status(),
            "would-change"
        );
        assert_eq!(
            FilterResult::RefusedFilterViewNotFound { filter_view_id: 1 }.log_status(),
            "refused-filter-view-not-found"
        );
        assert_eq!(
            FilterResult::AppliedReplyUnreadable {
                summary: String::new(),
                filter_view_id: None,
                width_warning: None,
                detail: String::new(),
            }
            .log_status(),
            "applied-reply-unreadable"
        );
        assert_eq!(
            FilterResult::Failed {
                detail: String::new()
            }
            .log_status(),
            "failed"
        );
    }

    // ── end-to-end via wiremock ──────────────────────────────────────────

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

    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 leased_opts_for(spreadsheet_id: &str) -> (Option<String>, std::path::PathBuf) {
        let ledger_path = tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl");
        let token = seed_lease(&ledger_path, spreadsheet_id, "1");
        (Some(token), ledger_path)
    }

    fn allow_rule(folder: &str) -> FolderPermissionRule {
        allow_rule_for(folder, &[DriveOperation::SheetsStructure])
    }

    fn allow_rule_for(folder: &str, operations: &[DriveOperation]) -> FolderPermissionRule {
        FolderPermissionRule {
            folder_id: Some(folder.to_string()),
            file_id: None,
            recursive: true,
            allow: operations.iter().copied().collect(),
            deny: HashSet::default(),
            require_lease: true,
        }
    }

    fn mount_workbook(sheets: serde_json::Value) -> wiremock::Mock {
        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",
                    "properties": {"title": "Budget"},
                    "sheets": sheets,
                })),
            )
    }

    #[tokio::test]
    async fn a_denied_gate_blocks_before_any_read_or_batch_update_call() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        let rules: Vec<FolderPermissionRule> = Vec::new();
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::SetBasicFilter {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                sort_by: Vec::new(),
                hide_values: Vec::new(),
            },
            dry_run: false,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Blocked { .. }));
    }

    // ── set-basic-filter --sort-by's union gate (issue #1940) ────────────

    #[test]
    fn gate_operations_are_the_union_only_for_a_sorting_basic_filter() {
        let sort = || vec!["2:desc".to_string()];
        assert_eq!(
            set_verb(sort(), Vec::new()).gate_operations(),
            GATE_WRITE_AND_STRUCTURE
        );
        assert_eq!(
            set_verb(Vec::new(), vec!["1:Foo".to_string()]).gate_operations(),
            GATE_STRUCTURE_ONLY
        );
        // A filter view's sort is per-view — it reorders nothing.
        for verb in [
            FilterVerb::ClearBasicFilter {
                sheet: "Q1".to_string(),
            },
            FilterVerb::AddFilterView {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                title: None,
                sort_by: sort(),
                hide_values: Vec::new(),
            },
            FilterVerb::UpdateFilterView {
                filter_view_id: 7,
                sheet: None,
                range: None,
                title: None,
                sort_by: sort(),
                hide_values: Vec::new(),
                clear_sort: false,
                clear_criteria: false,
            },
            FilterVerb::DeleteFilterView { filter_view_id: 7 },
        ] {
            assert_eq!(verb.gate_operations(), GATE_STRUCTURE_ONLY, "{verb:?}");
        }
    }

    async fn gated_server() -> (wiremock::MockServer, DriveClient, SheetsClient) {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        (server, drive, sheets)
    }

    fn dry_run_opts(verb: FilterVerb) -> FilterOptions {
        FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb,
            dry_run: true,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        }
    }

    #[tokio::test]
    async fn a_sorting_basic_filter_under_a_structure_only_grant_is_blocked_on_sheets_write() {
        // No workbook or batchUpdate mock: the gate must refuse before
        // either call.
        let (_server, drive, sheets) = gated_server().await;
        let rules = vec![allow_rule("folder-1")];
        let opts = dry_run_opts(set_verb(vec!["2:desc".to_string()], Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert_eq!(
            outcome.result,
            FilterResult::Blocked {
                operation: DriveOperation::SheetsWrite,
                decided_by: None,
            }
        );
    }

    #[tokio::test]
    async fn a_sorting_basic_filter_under_a_write_only_grant_is_blocked_on_sheets_structure() {
        let (_server, drive, sheets) = gated_server().await;
        let rules = vec![allow_rule_for("folder-1", &[DriveOperation::SheetsWrite])];
        let opts = dry_run_opts(set_verb(vec!["2:desc".to_string()], Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert_eq!(
            outcome.result,
            FilterResult::Blocked {
                operation: DriveOperation::SheetsStructure,
                decided_by: None,
            }
        );
    }

    #[tokio::test]
    async fn a_non_sorting_basic_filter_and_a_sorting_filter_view_still_pass_a_structure_only_grant(
    ) {
        let (server, drive, sheets) = gated_server().await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        for verb in [
            set_verb(Vec::new(), vec!["1:Foo".to_string()]),
            FilterVerb::AddFilterView {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                title: None,
                sort_by: vec!["2:desc".to_string()],
                hide_values: Vec::new(),
            },
        ] {
            let outcome = filter(&drive, &sheets, &dry_run_opts(verb), &rules).await;
            assert!(
                matches!(
                    outcome.result,
                    FilterResult::WouldChange {
                        width_warning: None,
                        ..
                    }
                ),
                "{:?}",
                outcome.result
            );
            let text = describe(&outcome);
            assert!(!text.contains("different row"), "{text}");
        }
    }

    #[tokio::test]
    async fn a_sorting_basic_filter_narrower_than_its_sheet_previews_the_reorder_caveats() {
        let (server, drive, sheets) = gated_server().await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0,
                            "gridProperties": {"rowCount": 100, "columnCount": 10}}},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule_for("folder-1", GATE_WRITE_AND_STRUCTURE)];
        let opts = dry_run_opts(set_verb(vec!["2:desc".to_string()], Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert_eq!(
            outcome.result,
            FilterResult::WouldChange {
                summary: "set basic filter (sort 2:desc)".to_string(),
                width_warning: grid_range::width_warning("sorting", 0, 4, Some(10)),
            }
        );
        let lines = describe_lines(&outcome);
        assert_eq!(lines.len(), 3, "{lines:?}");
        assert!(
            lines[0].starts_with("Warning: selected columns 0..4"),
            "{lines:?}"
        );
        assert_eq!(
            lines[1],
            "Would set basic filter (sort 2:desc) in 'sheet-1'"
        );
        assert!(lines[2].ends_with("after sorting"), "{lines:?}");
    }

    #[tokio::test]
    async fn a_sorting_basic_filter_spanning_every_column_has_no_width_warning() {
        let (server, drive, sheets) = gated_server().await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0,
                            "gridProperties": {"rowCount": 100, "columnCount": 4}}},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule_for("folder-1", GATE_WRITE_AND_STRUCTURE)];
        let opts = dry_run_opts(set_verb(vec!["2:desc".to_string()], Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::WouldChange {
                width_warning: None,
                ..
            }
        ));
        let lines = describe_lines(&outcome);
        assert_eq!(lines.len(), 2, "{lines:?}");
    }

    #[test]
    fn reorder_width_warning_handles_open_ended_column_spans() {
        let workbook = |column_count: Option<i64>| -> Spreadsheet {
            serde_json::from_value(serde_json::json!({
                "spreadsheetId": "sheet-1",
                "sheets": [{"properties": {
                    "sheetId": 0, "title": "Q1", "index": 0,
                    "gridProperties": {"columnCount": column_count},
                }}],
            }))
            .unwrap()
        };
        let grid = |start: Option<i64>, end: Option<i64>| GridRange {
            sheet_id: 0,
            start_column_index: start,
            end_column_index: end,
            ..Default::default()
        };
        // Open-ended to the right on a known-width sheet reaches its edge.
        assert_eq!(
            reorder_width_warning(&workbook(Some(5)), &grid(None, None)),
            None
        );
        assert!(reorder_width_warning(&workbook(Some(5)), &grid(Some(1), None)).is_some());
        // Unknown width: only a span starting past column A can warn.
        assert_eq!(
            reorder_width_warning(&workbook(None), &grid(None, None)),
            None
        );
        assert!(reorder_width_warning(&workbook(None), &grid(Some(1), None)).is_some());
        assert!(reorder_width_warning(&workbook(None), &grid(None, Some(3))).is_some());
    }

    #[test]
    fn describe_lines_adds_the_reorder_caveats_only_for_a_sorting_basic_filter() {
        let changed = outcome_with(
            set_verb(vec!["2:desc".to_string()], Vec::new()),
            Some("Budget"),
            FilterResult::Changed {
                summary: "set basic filter (sort 2:desc)".to_string(),
                filter_view_id: None,
                width_warning: Some("narrow".to_string()),
            },
        );
        assert_eq!(
            describe_lines(&changed),
            vec![
                "Warning: narrow".to_string(),
                "Applied: set basic filter (sort 2:desc) in 'Budget'".to_string(),
                "  references outside the range may now observe values from a different row"
                    .to_string(),
            ]
        );

        let would = outcome_with(
            set_verb(vec!["2:desc".to_string()], Vec::new()),
            Some("Budget"),
            FilterResult::WouldChange {
                summary: "set basic filter (sort 2:desc)".to_string(),
                width_warning: None,
            },
        );
        assert_eq!(
            describe_lines(&would),
            vec![
                "Would set basic filter (sort 2:desc) in 'Budget'".to_string(),
                "  references outside the range may observe values from a different row after sorting"
                    .to_string(),
            ]
        );
    }

    #[test]
    fn describe_lines_names_the_denied_operation_and_the_union_grant_hint() {
        let blocked = outcome_with(
            set_verb(vec!["2:desc".to_string()], Vec::new()),
            Some("Budget"),
            FilterResult::Blocked {
                operation: DriveOperation::SheetsWrite,
                decided_by: None,
            },
        );
        let text = describe(&blocked);
        assert!(text.contains("no matching rule for sheets-write"), "{text}");

        let orphan = outcome_with(
            set_verb(vec!["2:desc".to_string()], Vec::new()),
            Some("Budget"),
            FilterResult::RefusedNoVisibleParents,
        );
        let text = describe(&orphan);
        assert!(
            text.contains(r#""allow": ["sheets-write", "sheets-structure"]"#),
            "{text}"
        );
        let orphan = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::RefusedNoVisibleParents,
        );
        let text = describe(&orphan);
        assert!(text.contains(r#""allow": ["sheets-structure"]"#), "{text}");
    }

    #[tokio::test]
    async fn set_basic_filter_sends_a_set_basic_filter_request() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
        ]))
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{}]
                })),
            )
            .mount(&server)
            .await;
        // `--sort-by` physically reorders rows, so it needs the union
        // grant (issue #1940).
        let rules = vec![allow_rule_for("folder-1", GATE_WRITE_AND_STRUCTURE)];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::SetBasicFilter {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                sort_by: vec!["0:asc".to_string()],
                hide_values: vec!["1:Foo,Bar".to_string()],
            },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Changed { .. }));
        // A basic filter has no id of its own to log (unlike a filter
        // view's `filter_view_id`), so the resolved sheet id is the only
        // thing that can tell an audit record which sheet was affected —
        // see `record_attempt`/`docs/log.md`.
        assert_eq!(outcome.sheet_id, Some(0));
    }

    #[tokio::test]
    async fn set_basic_filter_refuses_a_sort_column_outside_the_range() {
        // `--sort-by 9:asc` on a 5-column `A1:E10` range passed dry-run and
        // then 500'd live (`HTTP 500: Internal error encountered`) — refuse
        // it locally, matching `add-pivot-table`'s column-offset check
        // (#1941).
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
        ]))
        .mount(&server)
        .await;
        // `--sort-by` needs the union grant (issue #1940).
        let rules = vec![allow_rule_for("folder-1", GATE_WRITE_AND_STRUCTURE)];
        let opts = unleased_opts(FilterVerb::SetBasicFilter {
            sheet: "Q1".to_string(),
            range: "A1:E10".to_string(),
            sort_by: vec!["9:asc".to_string()],
            hide_values: Vec::new(),
        });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(
                    detail.contains("sort column 9 is outside the selected range's columns 0..5"),
                    "{detail}"
                );
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn clear_basic_filter_reports_the_resolved_sheet_id() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 7, "title": "Q1", "index": 0},
             "basicFilter": {"range": {"sheetId": 7}}},
        ]))
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{}]
                })),
            )
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::ClearBasicFilter {
                sheet: "Q1".to_string(),
            },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Changed { .. }));
        assert_eq!(outcome.sheet_id, Some(7));
    }

    #[tokio::test]
    async fn add_filter_view_reports_the_server_assigned_id() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
        ]))
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{"addFilterView": {"filter": {"filterViewId": 42}}}]
                })),
            )
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::AddFilterView {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                title: Some("Open only".to_string()),
                sort_by: Vec::new(),
                hide_values: Vec::new(),
            },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::Changed { filter_view_id, .. } => {
                assert_eq!(filter_view_id, Some(42));
            }
            other => panic!("expected Changed, got {other:?}"),
        }
        // Unlike the basic filter, a filter view is already identified by
        // its own `filter_view_id`, so `sheet_id` stays unset here — see
        // `clear_basic_filter_reports_the_resolved_sheet_id` for the case
        // that needs it.
        assert_eq!(outcome.sheet_id, None);
    }

    #[tokio::test]
    async fn update_filter_view_refuses_an_unknown_id() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}, "filterViews": []},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::UpdateFilterView {
                filter_view_id: 99,
                sheet: None,
                range: None,
                title: Some("New".to_string()),
                sort_by: Vec::new(),
                hide_values: Vec::new(),
                clear_sort: false,
                clear_criteria: false,
            },
            dry_run: false,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::RefusedFilterViewNotFound { filter_view_id: 99 }
        ));
    }

    /// `update-filter-view --range` with no `--sheet` resolves against the
    /// view's own sheet rather than being refused for naming none (#1941).
    #[tokio::test]
    async fn update_filter_view_range_without_sheet_defaults_to_the_views_sheet() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
            {"properties": {"sheetId": 5, "title": "Q2", "index": 1},
             "filterViews": [{"filterViewId": 7, "range": {"sheetId": 5}}]},
        ]))
        .mount(&server)
        .await;
        let opts = dry_run_opts(FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: None,
            range: Some("A1:C9".to_string()),
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        });
        let outcome = filter(&drive, &sheets, &opts, &[allow_rule("folder-1")]).await;
        assert!(
            matches!(outcome.result, FilterResult::WouldChange { .. }),
            "{:?}",
            outcome.result
        );
    }

    /// An unknown id is reported as not found, not as a bare `--range`
    /// naming no sheet — the sheet defaulting must not mask it.
    #[tokio::test]
    async fn update_filter_view_range_without_sheet_reports_an_unknown_id_as_not_found() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0}},
            {"properties": {"sheetId": 5, "title": "Q2", "index": 1},
             "filterViews": [{"filterViewId": 7, "range": {"sheetId": 5}}]},
        ]))
        .mount(&server)
        .await;
        let opts = dry_run_opts(FilterVerb::UpdateFilterView {
            filter_view_id: 99,
            sheet: None,
            range: Some("A1:C9".to_string()),
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        });
        let outcome = filter(&drive, &sheets, &opts, &[allow_rule("folder-1")]).await;
        assert_eq!(
            outcome.result,
            FilterResult::RefusedFilterViewNotFound { filter_view_id: 99 }
        );
    }

    #[tokio::test]
    async fn delete_filter_view_sends_a_delete_filter_view_request() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0},
             "filterViews": [{"filterViewId": 7}]},
        ]))
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{}]
                })),
            )
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::DeleteFilterView { filter_view_id: 7 },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::Changed { filter_view_id, .. } => {
                assert_eq!(filter_view_id, Some(7));
            }
            other => panic!("expected Changed, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dry_run_makes_no_batch_update_call() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0},
             "basicFilter": {"range": {"sheetId": 0}}},
        ]))
        .mount(&server)
        .await;
        // Deliberately no mock for `POST .../batchUpdate`: a call there
        // would panic the mock server on an unexpected request, proving
        // dry-run never issues one.
        let rules = vec![allow_rule("folder-1")];
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::ClearBasicFilter {
                sheet: "Q1".to_string(),
            },
            dry_run: true,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::WouldChange { .. }));
    }

    // ── refusals before any call ─────────────────────────────────────────

    fn set_verb(sort_by: Vec<String>, hide_values: Vec<String>) -> FilterVerb {
        FilterVerb::SetBasicFilter {
            sheet: "Q1".to_string(),
            range: "A1:D10".to_string(),
            sort_by,
            hide_values,
        }
    }

    fn update_verb() -> FilterVerb {
        FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: None,
            range: None,
            title: Some("Renamed".to_string()),
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        }
    }

    fn unleased_opts(verb: FilterVerb) -> FilterOptions {
        FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb,
            dry_run: false,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        }
    }

    /// A refusal decided from the flags alone must make no HTTP call at
    /// all: the mock server has nothing mounted, so any request would fail
    /// the outcome as `Failed` rather than the refusal asserted here.
    async fn refused_from_flags_alone(verb: FilterVerb) -> FilterOutcome {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let outcome = filter(&drive, &sheets, &unleased_opts(verb), &rules).await;
        assert_eq!(outcome.file_name, None);
        outcome
    }

    #[tokio::test]
    async fn an_invalid_sort_spec_is_refused_before_any_call() {
        let outcome =
            refused_from_flags_alone(set_verb(vec!["0:sideways".to_string()], Vec::new())).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("not 'asc' or 'desc'"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn an_invalid_hidden_values_flag_is_refused_before_any_call() {
        let outcome = refused_from_flags_alone(set_verb(Vec::new(), vec!["1:".to_string()])).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("names no values"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_sheet_beside_a_prefixed_range_is_refused_before_any_call() {
        let verb = FilterVerb::AddFilterView {
            sheet: "Q1".to_string(),
            range: "Q2!A1:D10".to_string(),
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
        };
        let outcome = refused_from_flags_alone(verb).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("already names a sheet"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn update_filter_view_with_nothing_to_change_is_refused_before_any_call() {
        let verb = FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: None,
            range: None,
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        };
        let outcome = refused_from_flags_alone(verb).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("nothing to change"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

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

    #[tokio::test]
    async fn a_metadata_fetch_failure_surfaces_as_failed() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/sheet-1"))
            .respond_with(wiremock::ResponseTemplate::new(404).set_body_string("not found"))
            .mount(&server)
            .await;
        let rules: Vec<FolderPermissionRule> = Vec::new();
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Failed { .. }));
        assert_eq!(outcome.file_name, None);
    }

    #[tokio::test]
    async fn a_shortcut_target_is_refused() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            "application/vnd.google-apps.shortcut",
            &["folder-1"],
        )
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::RefusedShortcut));
        assert_eq!(outcome.file_name.as_deref(), Some("sheet-1"));
    }

    #[tokio::test]
    async fn a_non_spreadsheet_target_is_refused() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            "application/vnd.google-apps.document",
            &["folder-1"],
        )
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(FilterVerb::DeleteFilterView { filter_view_id: 7 });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedNotASpreadsheet { mime_type } => {
                assert_eq!(mime_type, "application/vnd.google-apps.document");
            }
            other => panic!("expected RefusedNotASpreadsheet, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_target_with_no_visible_parents_is_refused() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file("sheet-1", crate::drive::types::GOOGLE_SHEET_MIME_TYPE, &[])
            .mount(&server)
            .await;
        let rules: Vec<FolderPermissionRule> = Vec::new();
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::RefusedNoVisibleParents
        ));
    }

    #[tokio::test]
    async fn a_gate_ancestor_fetch_failure_surfaces_as_failed() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/drive/v3/files/folder-1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Failed { .. }));
        assert_eq!(outcome.file_name.as_deref(), Some("sheet-1"));
    }

    #[tokio::test]
    async fn a_workbook_fetch_failure_after_a_granted_gate_surfaces_as_failed() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-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 rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Failed { .. }));
        assert_eq!(outcome.sheet_id, None);
    }

    #[tokio::test]
    async fn set_basic_filter_refuses_an_unknown_sheet() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q2", "index": 0}},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(set_verb(Vec::new(), Vec::new()));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedSheetNotFound { title, available } => {
                assert_eq!(title, "Q1");
                assert_eq!(available, vec!["Q2".to_string()]);
            }
            other => panic!("expected RefusedSheetNotFound, got {other:?}"),
        }
    }

    // ── update-filter-view end to end ────────────────────────────────────

    fn mount_workbook_with_view_seven() -> wiremock::Mock {
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0},
             "filterViews": [{
                 "filterViewId": 7,
                 "title": "Old",
                 "range": {"sheetId": 0, "startRowIndex": 0, "endRowIndex": 10,
                           "startColumnIndex": 0, "endColumnIndex": 4},
                 "sortSpecs": [{"dimensionIndex": 0, "sortOrder": "ASCENDING"}],
                 "criteria": {"0": {"hiddenValues": ["Old"]}},
             }]},
        ]))
    }

    #[tokio::test]
    async fn update_filter_view_merges_onto_the_existing_view_and_reports_its_id() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        // The `fields` mask names every changed field; the merged
        // `sortSpecs` carry the existing entry alongside the new one, and
        // `criteria` carries only the changed column, since the server
        // merges criteria per column (issue #1931) — a partial body that
        // would 404 (and so fail the outcome) if either were wrong.
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .and(wiremock::matchers::body_partial_json(serde_json::json!({
                "requests": [{"updateFilterView": {
                    "fields": "title,range,sortSpecs,criteria",
                    "filter": {
                        "filterViewId": 7,
                        "title": "Renamed",
                        "range": {"sheetId": 0, "startRowIndex": 0, "endRowIndex": 20,
                                  "startColumnIndex": 0, "endColumnIndex": 4},
                        "sortSpecs": [
                            {"dimensionIndex": 0, "sortOrder": "ASCENDING"},
                            {"dimensionIndex": 2, "sortOrder": "DESCENDING"},
                        ],
                        "criteria": {
                            "1": {"hiddenValues": ["Closed"]},
                        },
                    },
                }}]
            })))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{}]
                })),
            )
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::UpdateFilterView {
                filter_view_id: 7,
                sheet: Some("Q1".to_string()),
                range: Some("A1:D20".to_string()),
                title: Some("Renamed".to_string()),
                sort_by: vec!["2:desc".to_string()],
                hide_values: vec!["1:Closed".to_string()],
                clear_sort: false,
                clear_criteria: false,
            },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::Changed {
                filter_view_id,
                summary,
                ..
            } => {
                assert_eq!(filter_view_id, Some(7));
                assert_eq!(
                    summary,
                    "update filter view (title='Renamed' sort 2:desc hide 1:Closed)"
                );
            }
            other => panic!("expected Changed, got {other:?}"),
        }
        // A filter view is identified by its own id, so `sheet_id` stays
        // unset even though a range was resolved to build the request.
        assert_eq!(outcome.sheet_id, None);
    }

    /// Runs `update-filter-view` against view seven with a mocked
    /// `batchUpdate` answering `reply`, returning the outcome and the one
    /// `batchUpdate` body the server received.
    async fn run_update_on_view_seven(
        verb: FilterVerb,
        reply: serde_json::Value,
    ) -> (FilterOutcome, serde_json::Value) {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(reply))
            .expect(1)
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb,
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        let body = server
            .received_requests()
            .await
            .unwrap()
            .into_iter()
            .find(|request| request.url.path().ends_with(":batchUpdate"))
            .map(|request| serde_json::from_slice(&request.body).unwrap())
            .unwrap();
        (outcome, body)
    }

    fn update_view_seven(
        sort_by: &[&str],
        hide_values: &[&str],
        clear_sort: bool,
        clear_criteria: bool,
    ) -> FilterVerb {
        FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: None,
            range: None,
            title: None,
            sort_by: sort_by.iter().map(|s| (*s).to_string()).collect(),
            hide_values: hide_values.iter().map(|s| (*s).to_string()).collect(),
            clear_sort,
            clear_criteria,
        }
    }

    #[tokio::test]
    async fn update_filter_view_clear_criteria_sends_an_explicit_reset() {
        let (outcome, body) = run_update_on_view_seven(
            update_view_seven(&[], &[], false, true),
            serde_json::json!({"replies": [{}]}),
        )
        .await;
        assert!(
            matches!(outcome.result, FilterResult::Changed { .. }),
            "{:?}",
            outcome.result
        );
        assert_eq!(
            body,
            serde_json::json!({"requests": [{"updateFilterView": {
                "fields": "criteria",
                "filter": {"filterViewId": 7, "criteria": {"0": {}}},
            }}]})
        );
    }

    #[tokio::test]
    async fn update_filter_view_clear_sort_recreates_the_view_under_its_own_id() {
        let (outcome, body) = run_update_on_view_seven(
            update_view_seven(&["2:desc"], &[], true, false),
            serde_json::json!({"replies": [
                {},
                {"addFilterView": {"filter": {"filterViewId": 7}}},
            ]}),
        )
        .await;
        match outcome.result {
            FilterResult::Changed {
                filter_view_id,
                summary,
                ..
            } => {
                assert_eq!(filter_view_id, Some(7));
                assert_eq!(
                    summary,
                    format!("update filter view (clear sort sort 2:desc){REPLACE_NOTE}")
                );
            }
            other => panic!("expected Changed, got {other:?}"),
        }
        assert_eq!(
            body,
            serde_json::json!({"requests": [
                {"deleteFilterView": {"filterId": 7}},
                {"addFilterView": {"filter": {
                    "filterViewId": 7,
                    "title": "Old",
                    "range": {"sheetId": 0, "startRowIndex": 0, "endRowIndex": 10,
                              "startColumnIndex": 0, "endColumnIndex": 4},
                    "sortSpecs": [{"dimensionIndex": 2, "sortOrder": "DESCENDING"}],
                    "criteria": {"0": {"hiddenValues": ["Old"]}},
                }}},
            ]})
        );
    }

    /// Mounts the gate and a workbook whose view 7 is table-bound and
    /// carries an unmodelled `condition` criterion, as a read-back would.
    async fn run_clear_sort_on_bound_view_with_condition() -> (FilterOutcome, serde_json::Value) {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0},
             "filterViews": [{
                 "filterViewId": 7,
                 "title": "Bound",
                 "range": {"sheetId": 0, "startRowIndex": 10, "endRowIndex": 16,
                           "startColumnIndex": 0, "endColumnIndex": 4},
                 "tableId": "454645209",
                 "sortSpecs": [{"dimensionIndex": 2, "sortOrder": "DESCENDING"}],
                 "criteria": {
                     "1": {"hiddenValues": ["b"]},
                     "2": {"condition": {"type": "NUMBER_GREATER",
                                         "values": [{"userEnteredValue": "2"}]}},
                 },
             }]},
        ]))
        .mount(&server)
        .await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "replies": [{}, {"addFilterView": {"filter": {"filterViewId": 7}}}]
                })),
            )
            .expect(1)
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_view_seven(&["0:asc"], &[], true, false),
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        let body = server
            .received_requests()
            .await
            .unwrap()
            .into_iter()
            .find(|request| request.url.path().ends_with(":batchUpdate"))
            .map(|request| serde_json::from_slice(&request.body).unwrap())
            .unwrap();
        (outcome, body)
    }

    #[tokio::test]
    async fn update_filter_view_recreation_carries_unmodelled_criteria_and_the_table_binding() {
        let (outcome, body) = run_clear_sort_on_bound_view_with_condition().await;
        assert!(
            matches!(
                outcome.result,
                FilterResult::Changed {
                    filter_view_id: Some(7),
                    ..
                }
            ),
            "{:?}",
            outcome.result
        );
        assert_eq!(
            body,
            serde_json::json!({"requests": [
                {"deleteFilterView": {"filterId": 7}},
                {"addFilterView": {"filter": {
                    "filterViewId": 7,
                    "title": "Bound",
                    "tableId": "454645209",
                    "sortSpecs": [{"dimensionIndex": 0, "sortOrder": "ASCENDING"}],
                    "criteria": {
                        "1": {"hiddenValues": ["b"]},
                        "2": {"condition": {"type": "NUMBER_GREATER",
                                            "values": [{"userEnteredValue": "2"}]}},
                    },
                }}},
            ]})
        );
    }

    #[tokio::test]
    async fn update_filter_view_dry_run_says_when_it_would_recreate_the_view() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let mut opts = unleased_opts(update_view_seven(&[], &[], true, false));
        opts.dry_run = true;
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::WouldChange { summary, .. } => {
                assert_eq!(
                    summary,
                    format!("update filter view (clear sort){REPLACE_NOTE}")
                );
            }
            other => panic!("expected WouldChange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn update_filter_view_recreation_with_no_range_is_refused_before_any_write() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook(serde_json::json!([
            {"properties": {"sheetId": 0, "title": "Q1", "index": 0},
             "filterViews": [{
                 "filterViewId": 7,
                 "sortSpecs": [{"dimensionIndex": 0, "sortOrder": "ASCENDING"}],
             }]},
        ]))
        .mount(&server)
        .await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(update_view_seven(&[], &[], true, false));
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("re-creates filter view 7"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn update_filter_view_with_only_a_range_change_is_described_bare() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::UpdateFilterView {
                filter_view_id: 7,
                sheet: None,
                range: Some("Q1!A1:D20".to_string()),
                title: None,
                sort_by: Vec::new(),
                hide_values: Vec::new(),
                clear_sort: false,
                clear_criteria: false,
            },
            dry_run: true,
            lease_token: None,
            ledger_path: std::path::PathBuf::new(),
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::WouldChange { summary, .. } => {
                assert_eq!(summary, "update filter view");
            }
            other => panic!("expected WouldChange, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_batch_update_failure_surfaces_as_failed() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_verb(),
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::Failed { detail } => assert!(detail.contains("500"), "{detail}"),
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn update_filter_view_with_an_unreadable_reply_reports_applied_reply_unreadable() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        // 2xx, so the change WAS applied — but the body is not the reply shape.
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
            .expect(1)
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_verb(),
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match &outcome.result {
            // The id resolved before the call is plan-time data, so it survives.
            FilterResult::AppliedReplyUnreadable { filter_view_id, .. } => {
                assert_eq!(*filter_view_id, Some(7));
            }
            other => panic!("expected AppliedReplyUnreadable, got {other:?}"), // omni-dev: coverage ignore-line reason="this arm only fires if the match failed to bind the expected variant; the assertion below it pins the same variant, so the branch never executes"
        }
        assert_eq!(outcome.result.log_status(), "applied-reply-unreadable");
        assert!(
            describe(&outcome).starts_with("Applied, but the reply could not be read"),
            "{}",
            describe(&outcome)
        );
    }

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

    #[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",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(&server)
        .await;
        mount_folder("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        // No batchUpdate mock mounted — a refusal must make zero mutating
        // calls.
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(FilterVerb::DeleteFilterView { filter_view_id: 7 });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::RefusedNoLease));
        assert_eq!(outcome.result.log_status(), "refused-no-lease");
    }

    // ── describe_effect ──────────────────────────────────────────────────

    #[test]
    fn describe_effect_names_the_view_and_its_sort_and_hidden_values() {
        let verb = FilterVerb::AddFilterView {
            sheet: "Q1".to_string(),
            range: "A1:D10".to_string(),
            title: Some("Open only".to_string()),
            sort_by: vec!["0:asc".to_string()],
            hide_values: vec!["1:Closed".to_string()],
        };
        assert_eq!(
            describe_effect(&verb),
            "add filter view 'Open only' (sort 0:asc hide 1:Closed)"
        );
        assert_eq!(
            describe_effect(&set_verb(Vec::new(), Vec::new())),
            "set basic filter"
        );
    }

    #[test]
    fn describe_effect_lists_every_update_filter_view_part_in_order() {
        let verb = FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: None,
            range: None,
            title: Some("Renamed".to_string()),
            sort_by: vec!["0:asc".to_string()],
            hide_values: vec!["1:Closed".to_string()],
            clear_sort: true,
            clear_criteria: true,
        };
        assert_eq!(
            describe_effect(&verb),
            "update filter view (title='Renamed' clear sort sort 0:asc clear criteria hide 1:Closed)"
        );
    }

    // ── describe / describe_lines ────────────────────────────────────────

    fn outcome_with(
        verb: FilterVerb,
        file_name: Option<&str>,
        result: FilterResult,
    ) -> FilterOutcome {
        FilterOutcome {
            spreadsheet_id: "sheet-1".to_string(),
            file_name: file_name.map(str::to_string),
            resolved_folder_id: None,
            sheet_id: None,
            verb,
            result,
        }
    }

    #[test]
    fn describe_lines_renders_would_change() {
        let out = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::WouldChange {
                summary: "set basic filter".to_string(),
                width_warning: None,
            },
        );
        assert_eq!(describe(&out), "Would set basic filter in 'Budget'");
    }

    #[test]
    fn describe_lines_renders_not_a_spreadsheet_with_no_file_name() {
        let out = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            None,
            FilterResult::RefusedNotASpreadsheet {
                mime_type: "text/plain".to_string(),
            },
        );
        let text = describe(&out);
        assert!(text.contains("'sheet-1'"), "{text}");
        assert!(text.contains("set-basic-filter"), "{text}");
        assert!(text.contains("text/plain"), "{text}");
    }

    #[test]
    fn describe_lines_renders_shortcut_and_no_visible_parents() {
        let shortcut = outcome_with(
            FilterVerb::DeleteFilterView { filter_view_id: 7 },
            Some("Budget"),
            FilterResult::RefusedShortcut,
        );
        let text = describe(&shortcut);
        assert!(text.contains("shortcut"), "{text}");
        assert!(text.contains("delete-filter-view"), "{text}");

        let orphan = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::RefusedNoVisibleParents,
        );
        assert!(describe(&orphan).contains("sheets-structure"));
    }

    #[test]
    fn describe_lines_renders_sheet_not_found_with_and_without_available_titles() {
        let none = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::RefusedSheetNotFound {
                title: "Q1".to_string(),
                available: Vec::new(),
            },
        );
        assert!(describe(&none).contains("Available: none"));

        let some = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::RefusedSheetNotFound {
                title: "Q1".to_string(),
                available: vec!["Q2".to_string(), "Q3".to_string()],
            },
        );
        assert!(describe(&some).contains("Available: 'Q2', 'Q3'"));
    }

    #[test]
    fn describe_lines_renders_invalid_range_and_filter_view_not_found() {
        let invalid = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::RefusedInvalidRange {
                detail: "bad range".to_string(),
            },
        );
        assert_eq!(describe(&invalid), "Refused: bad range");

        let missing = outcome_with(
            update_verb(),
            Some("Budget"),
            FilterResult::RefusedFilterViewNotFound { filter_view_id: 7 },
        );
        let text = describe(&missing);
        assert!(text.contains("no filter view with id 7"), "{text}");
        assert!(text.contains("list-filter-views"), "{text}");
    }

    #[test]
    fn describe_lines_renders_blocked_with_and_without_a_deciding_rule() {
        let folder_rule = outcome_with(
            update_verb(),
            Some("Budget"),
            FilterResult::Blocked {
                operation: DriveOperation::SheetsStructure,
                decided_by: Some(DecidingRule::Folder {
                    folder_id: "folder-1".to_string(),
                    depth: 2,
                }),
            },
        );
        let text = describe(&folder_rule);
        assert!(text.contains("update-filter-view"), "{text}");
        assert!(text.contains("folder folder-1 (depth 2)"), "{text}");

        let default_policy = outcome_with(
            FilterVerb::ClearBasicFilter {
                sheet: "Q1".to_string(),
            },
            Some("Budget"),
            FilterResult::Blocked {
                operation: DriveOperation::SheetsStructure,
                decided_by: None,
            },
        );
        let text = describe(&default_policy);
        assert!(text.contains("default policy"), "{text}");
        assert!(text.contains("sheets-structure"), "{text}");
    }

    #[test]
    fn describe_lines_renders_every_lease_refusal_with_the_lease_acquire_hint() {
        for (result, phrase) in [
            (FilterResult::RefusedNoLease, "requires a Drive write lease"),
            (
                FilterResult::RefusedLeaseExpired,
                "expired, released, or unknown",
            ),
            (
                FilterResult::RefusedLeaseWrongFile,
                "acquired for a different file",
            ),
            (
                FilterResult::RefusedLeaseStale,
                "changed since the lease was acquired",
            ),
        ] {
            let out = outcome_with(update_verb(), Some("Budget"), result);
            let text = describe(&out);
            assert!(text.contains(phrase), "{text}");
            assert!(text.contains("drive lease acquire sheet-1"), "{text}");
        }
    }

    #[test]
    fn describe_lines_renders_changed_with_and_without_an_id() {
        let with_id = outcome_with(
            FilterVerb::DeleteFilterView { filter_view_id: 7 },
            Some("Budget"),
            FilterResult::Changed {
                summary: "delete filter view".to_string(),
                filter_view_id: Some(7),
                width_warning: None,
            },
        );
        assert_eq!(
            describe(&with_id),
            "Applied: delete filter view (id 7) in 'Budget'"
        );

        let without_id = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::Changed {
                summary: "set basic filter".to_string(),
                filter_view_id: None,
                width_warning: None,
            },
        );
        assert_eq!(
            describe(&without_id),
            "Applied: set basic filter in 'Budget'"
        );
    }

    #[test]
    fn describe_lines_renders_applied_reply_unreadable_with_an_id_hint_only_for_adds() {
        let added = outcome_with(
            FilterVerb::AddFilterView {
                sheet: "Q1".to_string(),
                range: "A1:D10".to_string(),
                title: None,
                sort_by: Vec::new(),
                hide_values: Vec::new(),
            },
            Some("Budget"),
            FilterResult::AppliedReplyUnreadable {
                summary: "add filter view".to_string(),
                filter_view_id: None,
                width_warning: None,
                detail: "bad reply".to_string(),
            },
        );
        assert_eq!(
            describe_lines(&added),
            vec![
                "Applied, but the reply could not be read: add filter view in 'Budget' — \
                 do not retry; check the spreadsheet first (bad reply)"
                    .to_string(),
                "  the new filter view's id was in the unreadable reply; run \
                 `list-filter-views` to find it"
                    .to_string(),
            ]
        );

        let updated = outcome_with(
            update_verb(),
            Some("Budget"),
            FilterResult::AppliedReplyUnreadable {
                summary: "update filter view 7".to_string(),
                filter_view_id: Some(7),
                width_warning: None,
                detail: "bad reply".to_string(),
            },
        );
        // Names the resolved id exactly as `Changed` does, and adds no hint.
        assert_eq!(
            describe_lines(&updated),
            vec![
                "Applied, but the reply could not be read: update filter view 7 (id 7) in \
                 'Budget' — do not retry; check the spreadsheet first (bad reply)"
                    .to_string()
            ]
        );
    }

    /// Everything `Changed` carries that does not come from the reply — the
    /// width warning and the reorder caveat — survives an unreadable reply.
    #[test]
    fn describe_lines_keeps_the_reorder_caveats_for_an_unreadable_reply() {
        let out = outcome_with(
            set_verb(vec!["0:asc".to_string()], Vec::new()),
            Some("Budget"),
            FilterResult::AppliedReplyUnreadable {
                summary: "set basic filter (sort 0:asc)".to_string(),
                filter_view_id: None,
                width_warning: Some("selected columns 0..4 of 9".to_string()),
                detail: "bad reply".to_string(),
            },
        );
        assert_eq!(
            describe_lines(&out),
            vec![
                "Applied, but the reply could not be read: set basic filter (sort 0:asc) in \
                 'Budget' — do not retry; check the spreadsheet first (bad reply)"
                    .to_string(),
                "Warning: selected columns 0..4 of 9".to_string(),
                "  references outside the range may now observe values from a different row"
                    .to_string(),
            ]
        );
    }

    #[test]
    fn mutation_record_logs_the_same_fields_for_an_unreadable_reply_as_for_changed() {
        let opts = unleased_opts(FilterVerb::DeleteFilterView { filter_view_id: 7 });
        let record = |result| {
            let mut out = outcome_with(opts.verb.clone(), Some("Budget"), result);
            out.sheet_id = Some(3);
            mutation_record(&out, &opts, Duration::ZERO)
        };
        let changed = record(FilterResult::Changed {
            summary: "delete filter view".to_string(),
            filter_view_id: Some(7),
            width_warning: None,
        });
        let unreadable = record(FilterResult::AppliedReplyUnreadable {
            summary: "delete filter view".to_string(),
            filter_view_id: Some(7),
            width_warning: None,
            detail: "bad reply".to_string(),
        });
        assert_eq!(unreadable.sheet_id, Some(3));
        assert_eq!(unreadable.filter_view_id, changed.filter_view_id);
        assert_eq!(unreadable.status, "applied-reply-unreadable");
        assert_eq!(unreadable.error.as_deref(), Some("bad reply"));
        // Every verb records what was applied, not just update-filter-view.
        assert_eq!(
            unreadable.fields_changed.as_deref(),
            Some("delete filter view")
        );
    }

    #[test]
    fn describe_lines_renders_failed() {
        let out = outcome_with(
            set_verb(Vec::new(), Vec::new()),
            Some("Budget"),
            FilterResult::Failed {
                detail: "boom".to_string(),
            },
        );
        assert_eq!(describe(&out), "Failed: boom");
    }

    // ── the other lease refusals ─────────────────────────────────────────

    /// Mounts a granted gate and a workbook holding filter view 7, so the
    /// only thing left to decide the outcome is the lease.
    async fn mount_granted_gate_with_view_seven(server: &wiremock::MockServer) {
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-1"],
        )
        .mount(server)
        .await;
        mount_folder("folder-1").mount(server).await;
        mount_workbook_with_view_seven().mount(server).await;
    }

    fn fresh_ledger_path() -> std::path::PathBuf {
        tempfile::tempdir()
            .unwrap()
            .keep()
            .join("lease-ledger.jsonl")
    }

    #[tokio::test]
    async fn refuses_an_unknown_lease_token() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        // Never seeded — the ledger knows nothing of this token.
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_verb(),
            dry_run: false,
            lease_token: Some("bogus-token".to_string()),
            ledger_path: fresh_ledger_path(),
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::RefusedLeaseExpired));
        assert_eq!(outcome.result.log_status(), "refused-lease-expired");
    }

    #[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_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let ledger_path = fresh_ledger_path();
        // Seeded for a *different* spreadsheet id.
        let token = seed_lease(&ledger_path, "some-other-sheet", "1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_verb(),
            dry_run: false,
            lease_token: Some(token),
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::RefusedLeaseWrongFile
        ));
        assert_eq!(outcome.result.log_status(), "refused-lease-wrong-file");
    }

    #[tokio::test]
    async fn refuses_a_stale_lease_when_the_file_has_moved() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        // `mount_file` always returns version "1"; the lease below was
        // acquired against version "0" — a foreign edit landed since.
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let ledger_path = fresh_ledger_path();
        let token = seed_lease(&ledger_path, "sheet-1", "0");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: update_verb(),
            dry_run: false,
            lease_token: Some(token),
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::RefusedLeaseStale));
        assert_eq!(outcome.result.log_status(), "refused-lease-stale");
    }

    // ── sheet resolution per verb ────────────────────────────────────────

    #[tokio::test]
    async fn clear_basic_filter_refuses_an_unknown_sheet() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(FilterVerb::ClearBasicFilter {
            sheet: "Nope".to_string(),
        });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedSheetNotFound { title, available } => {
                assert_eq!(title, "Nope");
                assert_eq!(available, vec!["Q1".to_string()]);
            }
            other => panic!("expected RefusedSheetNotFound, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn update_filter_view_refuses_a_range_on_an_unknown_sheet() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: Some("Nope".to_string()),
            range: Some("A1:D20".to_string()),
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::RefusedSheetNotFound { .. }
        ));
    }

    #[tokio::test]
    async fn set_basic_filter_refuses_an_unparseable_range() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        // `A1:B` mixes a cell with a whole column on a *different* column,
        // which no supported A1 form spells — so it passes the flag-level
        // `a1::compose` check and only fails once the workbook's sheet is
        // known and the grid is parsed against it.
        let opts = unleased_opts(FilterVerb::SetBasicFilter {
            sheet: "Q1".to_string(),
            range: "A1:B".to_string(),
            sort_by: Vec::new(),
            hide_values: Vec::new(),
        });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        match outcome.result {
            FilterResult::RefusedInvalidRange { detail } => {
                assert!(detail.contains("not a recognised A1 range"), "{detail}");
            }
            other => panic!("expected RefusedInvalidRange, got {other:?}"),
        }
        assert_eq!(outcome.file_name.as_deref(), Some("sheet-1"));
    }

    #[tokio::test]
    async fn update_filter_view_refuses_an_unparseable_range() {
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_granted_gate_with_view_seven(&server).await;
        let rules = vec![allow_rule("folder-1")];
        let opts = unleased_opts(FilterVerb::UpdateFilterView {
            filter_view_id: 7,
            sheet: Some("Q1".to_string()),
            range: Some("A1:B".to_string()),
            title: None,
            sort_by: Vec::new(),
            hide_values: Vec::new(),
            clear_sort: false,
            clear_criteria: false,
        });
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(
            outcome.result,
            FilterResult::RefusedInvalidRange { .. }
        ));
    }

    #[tokio::test]
    async fn a_failed_pre_lease_refetch_is_reported_as_failed_with_no_batch_update_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 change must report `Failed` and never reach
        // `batchUpdate`.
        let server = wiremock::MockServer::start().await;
        let (drive, sheets) = clients(&server).await;
        mount_file(
            "sheet-1",
            crate::drive::types::GOOGLE_SHEET_MIME_TYPE,
            &["folder-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("folder-1").mount(&server).await;
        mount_workbook_with_view_seven().mount(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path(
                "/v4/spreadsheets/sheet-1:batchUpdate",
            ))
            .respond_with(wiremock::ResponseTemplate::new(200))
            .expect(0)
            .mount(&server)
            .await;
        let rules = vec![allow_rule("folder-1")];
        let (lease_token, ledger_path) = leased_opts_for("sheet-1");
        let opts = FilterOptions {
            spreadsheet_id: "sheet-1".to_string(),
            verb: FilterVerb::DeleteFilterView { filter_view_id: 7 },
            dry_run: false,
            lease_token,
            ledger_path,
        };
        let outcome = filter(&drive, &sheets, &opts, &rules).await;
        assert!(matches!(outcome.result, FilterResult::Failed { .. }));
    }
}