omni-dev 0.44.0

//! `gmail sync`'s control flow: backfill vs. incremental, 404-triggered
//! reconciliation, and the throttled fetch fan-out.
//!
//! [`run_sync`] does no direct stdout/stderr I/O itself — it (via
//! [`run_sync_with_progress`]) returns a [`SyncReport`] for the caller
//! (`src/cli/gmail/sync.rs`) to render and turn into a process exit code,
//! mirroring `src/cli/ai/claude/history/sync.rs::run`'s compute → render →
//! decide split. It may optionally emit [`super::progress::SyncProgressEvent`]s
//! over a caller-supplied channel — see ADR-0064's amendment for #1502 —
//! but never touches a terminal itself; only `sync.rs` does that.
//!
//! Presence-on-disk is the real idempotence mechanism (an interrupted
//! backfill needs no cursor to resume correctly): backfill, `--full`, and
//! 404-triggered reconciliation are therefore all the *same* code path,
//! [`run_full_sync`], which lists the whole mailbox and fetches only what's
//! missing on disk — listing and fetching are pipelined (#1502), so the
//! fetch fan-out for early-listed messages starts immediately rather than
//! waiting for the whole mailbox to be listed first. For idempotence to
//! actually hold across a real interruption (not just a clean run), the
//! manifest itself must reach disk periodically during the fetch fan-out,
//! not only once at the very end — see
//! [`fetch_and_archive_messages_streaming`]'s [`MANIFEST_CHECKPOINT_INTERVAL`]
//! (#1467).

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use chrono::Utc;
use futures::stream::{self, FuturesUnordered, StreamExt as _};
use tokio::sync::{mpsc, Semaphore};

use crate::cli::gmail::helpers::label_ids_contain_any;
use crate::gmail::attachments::{extract_attachments, ExtractedAttachment};
use crate::gmail::client::GmailClient;
use crate::gmail::error::GmailError;
use crate::gmail::history_api::HistoryApi;
use crate::gmail::messages_api::{
    ListingProgress, MessageFormat, MessagesApi, GMAIL_QUOTA_UNITS_PER_SECOND, MAX_CONCURRENCY,
    MESSAGES_GET_COST_UNITS,
};
use crate::gmail::profile_api::{Profile, ProfileApi};
use crate::gmail::raw_message::{
    decode_raw_message, extract_attachment_filenames, extract_headers,
};
use crate::utils::rate_limit::TokenBucket;

use super::manifest::{Manifest, ManifestRecord};
use super::progress::SyncProgressEvent;
use super::report::{SyncAction, SyncError, SyncReport};
use super::shard::{attachments_dir, shard_path};
use super::state::{self, ArchiveState, LoadOutcome};

/// Options for one `gmail sync` invocation.
pub(crate) struct SyncOptions {
    pub(crate) output_dir: PathBuf,
    pub(crate) query: Option<String>,
    /// Label ids to exclude from the archive (#1780). Applied fully and
    /// generally on incremental passes (`run_incremental`, via label data
    /// history events already carry — no extra API calls); on full/backfill/
    /// reconciliation passes only the entries with a known `-in:` query
    /// translation take effect (currently `SPAM`/`TRASH`) — see
    /// `full_sync_query`'s doc comment.
    pub(crate) exclude_labels: Vec<String>,
    pub(crate) full: bool,
    pub(crate) concurrency: usize,
    pub(crate) dry_run: bool,
    pub(crate) extract_attachments: bool,
    /// A concurrency cap shared across every account's fetches in one
    /// `gmail sync-all` run (ADR-0068), layered underneath `concurrency`
    /// (each account's own local `buffer_unordered`/`FuturesUnordered`
    /// clamp). `None` for every single-account caller — `gmail sync`'s
    /// behavior is unaffected.
    pub(crate) shared_pool: Option<Arc<Semaphore>>,
}

fn state_path(output_dir: &Path) -> PathBuf {
    output_dir.join("state.json")
}

/// `pub(crate)` — also used by `gmail extract-attachments`
/// (`cli/gmail/extract_attachments/engine.rs`), which reads the same
/// on-disk manifest without going through a full `run_sync`.
pub(crate) fn manifest_path(output_dir: &Path) -> PathBuf {
    output_dir.join("manifest.jsonl")
}

/// Number of successfully-fetched messages between manifest checkpoints
/// during [`fetch_and_archive_messages`]'s fetch fan-out.
///
/// Bounds how much re-fetch work a crash mid-backfill can cost to about one
/// interval's worth (~4s at the documented 50 msg/s quota ceiling for `200`
/// messages), while keeping the number of full-manifest rewrites
/// proportional to `total / interval` rather than `total` (#1467).
const MANIFEST_CHECKPOINT_INTERVAL: usize = 200;

/// Runs one sync: resolves identity, decides backfill vs. incremental (with
/// 404 fallback), fetches whatever's missing, and returns a report. Never
/// panics on per-message failures — see
/// [`fetch_and_archive_messages_streaming`].
///
/// A thin wrapper around [`run_sync_with_progress`] with no progress
/// channel — kept as its own function so every existing caller/test stays
/// unaffected by #1502's progress-reporting addition.
pub(crate) async fn run_sync(client: &GmailClient, opts: &SyncOptions) -> Result<SyncReport> {
    run_sync_with_progress(client, opts, None).await
}

/// [`run_sync`], plus an optional channel to emit
/// [`SyncProgressEvent`]s to during the full-mailbox pass (backfill /
/// `--full` / 404-triggered reconciliation). `run_incremental`'s
/// `history.list` pass is typically a single page already, so it emits no
/// progress events regardless of whether `progress` is set.
///
/// Still performs no direct stdout/stderr I/O — `progress`, if present, is
/// just another channel this function writes structured data to, the same
/// as `report`; only the caller (`src/cli/gmail/sync.rs`) may render
/// anything (ADR-0064's amendment for #1502).
pub(crate) async fn run_sync_with_progress(
    client: &GmailClient,
    opts: &SyncOptions,
    progress: Option<&mpsc::UnboundedSender<SyncProgressEvent>>,
) -> Result<SyncReport> {
    guard_output_dir(&opts.output_dir)?;
    if !opts.dry_run {
        let messages_dir = opts.output_dir.join("messages");
        std::fs::create_dir_all(&messages_dir)
            .with_context(|| format!("Failed to create {}", messages_dir.display()))?;
    }

    let profile = ProfileApi::new(client).get().await?;
    let loaded = state::load(&state_path(&opts.output_dir));
    let mut manifest = Manifest::load(&manifest_path(&opts.output_dir))?;
    let mut report = SyncReport::default();
    let limiter = TokenBucket::new(GMAIL_QUOTA_UNITS_PER_SECOND, GMAIL_QUOTA_UNITS_PER_SECOND);

    let (history_id, carried) = match loaded {
        LoadOutcome::Present(state) if !opts.full => {
            state::validate_identity(&state, &profile.email_address)?;
            match run_incremental(
                client,
                &mut manifest,
                &state,
                opts,
                &limiter,
                &mut report,
                progress,
            )
            .await
            {
                Ok(id) => (id, Vec::new()),
                Err(e) if is_history_not_found(&e) => {
                    report.actions.push(SyncAction::Note {
                        message: "watermark expired (404 on startHistoryId); reconciling"
                            .to_string(),
                    });
                    run_full_sync(
                        client,
                        &mut manifest,
                        &profile,
                        opts,
                        &limiter,
                        &mut report,
                        progress,
                    )
                    .await?
                    .carry_forward(&state.pending_fetch)
                }
                Err(e) => return Err(e),
            }
        }
        LoadOutcome::Present(state) => {
            // `--full` with an otherwise-valid state.json still validates
            // identity — a forced reconciliation is not an excuse to skip
            // the one check that prevents mixing two mailboxes.
            state::validate_identity(&state, &profile.email_address)?;
            run_full_sync(
                client,
                &mut manifest,
                &profile,
                opts,
                &limiter,
                &mut report,
                progress,
            )
            .await?
            .carry_forward(&state.pending_fetch)
        }
        LoadOutcome::Absent => run_full_sync(
            client,
            &mut manifest,
            &profile,
            opts,
            &limiter,
            &mut report,
            progress,
        )
        .await?
        .carry_forward(&[]),
        LoadOutcome::Corrupt(reason) => {
            report.actions.push(SyncAction::Note {
                message: format!("state.json unreadable ({reason}); reconciling"),
            });
            run_full_sync(
                client,
                &mut manifest,
                &profile,
                opts,
                &limiter,
                &mut report,
                progress,
            )
            .await?
            .carry_forward(&[])
        }
    };

    if !opts.dry_run {
        // Always saved: successful per-message mutations (fetches, label
        // deltas, soft-deletes) are independently safe to keep even when
        // this run also hit errors elsewhere.
        manifest.save(&manifest_path(&opts.output_dir))?;
        // The watermark advances even when this run hit errors: the failed
        // ids are carried forward in `pending_fetch` for the next incremental
        // run to retry directly, so no history event needs to stay within
        // Gmail's ~1-week retention window for them to be found again. That
        // window used to be the only way back to a failed id, which meant a
        // chronically rate-limited account could never advance its watermark
        // and was pushed into a full-mailbox reconciliation (#1784).
        // No dedup needed: both fetch paths already fetch each id once, and
        // `carried` only holds ids this run's listing never named, so never
        // an id that could also have failed here.
        let pending_fetch = report
            .errors
            .iter()
            .map(|e| e.id.clone())
            .chain(carried)
            .collect();
        state::save(
            &ArchiveState {
                history_id,
                email_address: profile.email_address,
                last_sync: Utc::now(),
                query: opts.query.clone(),
                pending_fetch,
            },
            &state_path(&opts.output_dir),
        )?;
    }
    Ok(report)
}

/// Backfill, `--full`, and 404-triggered reconciliation are all this same
/// path: list every message currently on the server, fetch whatever's
/// missing on disk (an interrupted prior run's already-archived messages
/// are skipped for free), and soft-delete manifest records for ids that no
/// longer appear.
///
/// Listing and fetching run *concurrently*, joined on this one task via
/// [`tokio::join!`] rather than [`tokio::spawn`] (#1502) — that's what lets
/// both sides keep sharing a single `&TokenBucket` borrow with no `Arc`, and
/// what confines `&mut Manifest` to exactly one side (the fetch consumer,
/// [`fetch_and_archive_messages_streaming`]) with no `Arc<Mutex<_>>` either.
/// The two manifest-mutating passes that need the *complete* listing
/// (undelete-on-reappearance, stale-id soft-deletion) both run after the
/// join — previously undelete ran before the (then-sequential) fetch phase
/// and stale-deletion ran after, so this can change the *order* actions
/// appear in a [`SyncReport`], never which actions occur.
async fn run_full_sync(
    client: &GmailClient,
    manifest: &mut Manifest,
    profile: &Profile,
    opts: &SyncOptions,
    limiter: &TokenBucket,
    report: &mut SyncReport,
    progress: Option<&mpsc::UnboundedSender<SyncProgressEvent>>,
) -> Result<FullListing> {
    let (ids_tx, ids_rx) = mpsc::unbounded_channel::<String>();
    let messages_api = MessagesApi::new(client);

    let (effective_query, translated_exclude_labels) = full_sync_query(opts, report);
    let listing = messages_api.search_all_unbounded_streaming(
        effective_query.as_deref(),
        &[],
        limiter,
        ids_tx,
        |p: ListingProgress| {
            if let Some(tx) = progress {
                let _ = tx.send(SyncProgressEvent::ListingPage {
                    pages: p.page_no,
                    ids_discovered: p.ids_so_far,
                });
            }
        },
    );
    let fetching = fetch_and_archive_messages_streaming(
        client, manifest, ids_rx, limiter, opts, report, progress,
    );

    let (listing_result, fetch_result) = tokio::join!(listing, fetching);
    // A real `messages.list` failure is the more actionable root cause when
    // both sides error (the fetch side would just be draining a channel
    // that stopped growing) — check it first.
    listing_result?;
    let listed_ids = fetch_result?;
    if let Some(tx) = progress {
        let _ = tx.send(SyncProgressEvent::ListingDone);
    }

    for id in &listed_ids {
        let Some(record) = manifest.get(id) else {
            continue;
        };
        if record.deleted_at.is_none() {
            continue;
        }
        // A record whose cached `excluded_labels` names a label this run's
        // query could *not* have filtered server-side (a custom/untranslated
        // one — see `full_sync_query`'s doc comment) must not be resurrected
        // just because it still shows up in an otherwise-unfiltered listing:
        // `run_incremental`'s own labelsRemoved tracking is the only source
        // of truth for that label (#1780). A translated label (SPAM/TRASH)
        // is safe to trust here even if cached: the listing query itself
        // excludes it, so reappearing despite that is authoritative proof
        // it's gone.
        let stuck_on_untranslated_exclusion = record
            .excluded_labels
            .iter()
            .any(|label| !translated_exclude_labels.iter().any(|t| t == label));
        if stuck_on_untranslated_exclusion {
            continue;
        }
        if opts.dry_run {
            report
                .actions
                .push(SyncAction::WouldUndelete { id: id.clone() });
        } else {
            manifest.undelete(id);
            report
                .actions
                .push(SyncAction::Undeleted { id: id.clone() });
        }
    }

    let stale: Vec<String> = manifest
        .ids_not_deleted()
        .filter(|id| !listed_ids.contains(*id))
        .map(str::to_string)
        .collect();
    for id in stale {
        if opts.dry_run {
            report.actions.push(SyncAction::WouldDelete { id });
        } else {
            // Plain, reason-less soft-delete: this pass has no way to know
            // *which* configured label (if any) actually explains a given
            // id's absence — `--query` can narrow the listing for entirely
            // unrelated reasons too — so it must never guess and tag
            // `excluded_labels` (#1780). Leaving it untouched (cleared, via
            // `mark_deleted`) means only `run_incremental`'s own precise,
            // event-driven tracking ever populates that field, which is
            // exactly what keeps the undelete-on-reappearance check above
            // safe.
            manifest.mark_deleted(&id, Utc::now());
            report.actions.push(SyncAction::Deleted { id });
        }
    }

    // Snapshotted *before* the listing began (by the caller, via `profile`)
    // — mail arriving mid-listing is simply re-observed as an ordinary
    // `messagesAdded` event on the next incremental run, a self-healing
    // race window rather than a gap.
    Ok(FullListing {
        history_id: profile.history_id.clone(),
        scoped: effective_query.is_some_and(|q| !q.trim().is_empty()),
        listed_ids,
    })
}

/// What [`run_full_sync`] listed, for deciding which of a previous run's
/// `pending_fetch` ids survive it.
struct FullListing {
    history_id: String,
    /// Whether the listing was narrowed by `--query` or a translated
    /// `--exclude-label` clause, so an id's absence from it doesn't mean the
    /// message is gone.
    scoped: bool,
    listed_ids: HashSet<String>,
}

impl FullListing {
    /// Returns the watermark plus the `previous` pending ids to carry forward
    /// (#1784). A listed id was already retried by the listing itself, so it
    /// is never carried. An unlisted one is dropped after an unscoped
    /// listing, where absence means the message is gone, but carried after a
    /// scoped one, which never looked for it.
    fn carry_forward(self, previous: &[String]) -> (String, Vec<String>) {
        let carried = if self.scoped {
            previous
                .iter()
                .filter(|id| !self.listed_ids.contains(*id))
                .cloned()
                .collect()
        } else {
            Vec::new()
        };
        (self.history_id, carried)
    }
}

/// Label ids with a well-known, safe `-in:<keyword>` exclusion translation
/// for [`full_sync_query`]. Deliberately just these two, not every Gmail
/// system label: `UNREAD`/`STARRED`/`IMPORTANT` are matched via `is:`, not
/// `in:`, so guessing wrong would silently exclude nothing rather than
/// erroring — `SPAM`/`TRASH` are what #1780 and docs/gmail.md's existing
/// `--query "-in:spam"` example already name as the motivating case.
const FULL_SYNC_EXCLUDABLE_LABELS: &[(&str, &str)] = &[("SPAM", "spam"), ("TRASH", "trash")];

/// Folds `opts.exclude_labels` into the query used for [`run_full_sync`]'s
/// listing pass, for whichever entries have a known translation (see
/// [`FULL_SYNC_EXCLUDABLE_LABELS`]). An entry with no known translation (a
/// custom/user label, or a system label outside the allow-list) can't be
/// reliably excluded server-side by id alone, so it's left out of the query
/// and a [`SyncAction::Note`] is pushed instead: `run_incremental` still
/// filters it going forward (it works generally, off `label_ids` rather
/// than a query translation), but this pass won't retroactively exclude it
/// unless the caller also sets `--query`.
///
/// A caller-supplied `--query` is parenthesized before any exclusion
/// clause is appended — `q OR r -in:spam` would parse as `q OR (r AND
/// -in:spam)` (Gmail's implicit AND-by-juxtaposition binds tighter than an
/// explicit `OR`), silently narrowing the exclusion to only the right-hand
/// side of the caller's own query; `(q OR r) -in:spam` applies it to the
/// whole thing, which is what `--exclude-label` promises regardless of
/// what `--query` happens to contain (#1780). Parenthesizing is skipped
/// when no exclusion clause ends up being appended, so a caller who never
/// uses `--exclude-label` sees `--query` passed through byte-for-byte,
/// unchanged from before this flag existed.
///
/// Also returns the subset of `exclude_labels` that *did* get a
/// translation, deduplicated — [`run_full_sync`]'s undelete-on-reappearance
/// pass uses it to tell a translated (trustworthy, since the listing query
/// itself enforces it) cached exclusion reason from an untranslated
/// (custom-label) one it must not casually undo.
fn full_sync_query(opts: &SyncOptions, report: &mut SyncReport) -> (Option<String>, Vec<String>) {
    // Tokens already present in the caller's own `--query`, so a repeated
    // `--exclude-label` value that duplicates `--query`'s own text doesn't
    // fold in the same `-in:` token twice.
    let existing_tokens: HashSet<&str> = opts
        .query
        .as_deref()
        .unwrap_or_default()
        .split_whitespace()
        .collect();
    let mut translated = Vec::new();
    let mut exclusion_tokens: Vec<String> = Vec::new();
    let mut seen_tokens: HashSet<String> = HashSet::new();
    for label in &opts.exclude_labels {
        match FULL_SYNC_EXCLUDABLE_LABELS
            .iter()
            .find(|(id, _)| *id == label)
        {
            Some((_, keyword)) => {
                if !translated.contains(label) {
                    translated.push(label.clone());
                }
                let token = format!("-in:{keyword}");
                if !existing_tokens.contains(token.as_str()) && seen_tokens.insert(token.clone()) {
                    exclusion_tokens.push(token);
                }
            }
            None => report.actions.push(SyncAction::Note {
                message: format!(
                    "--exclude-label {label} has no full-sync query translation; it only \
                     filters incremental syncs going forward — add it to --query yourself to \
                     also exclude it on this pass"
                ),
            }),
        }
    }
    if exclusion_tokens.is_empty() {
        return (opts.query.clone(), translated);
    }
    let mut query = match opts.query.as_deref().filter(|q| !q.is_empty()) {
        Some(user_query) => format!("({user_query})"),
        None => String::new(),
    };
    for token in exclusion_tokens {
        if !query.is_empty() {
            query.push(' ');
        }
        query.push_str(&token);
    }
    (Some(query), translated)
}

/// Applies `messagesAdded`/`messagesDeleted`/`labelsAdded`/`labelsRemoved`
/// history events since `previous.history_id`. A 404 (watermark past Gmail's
/// retention window) propagates unchanged for [`run_sync`] to catch.
///
/// A message can be added and deleted again within the same history
/// window (routine server-side churn — an auto-filtered message, a sent
/// mail immediately recalled) — every deleted id is pre-scanned across the
/// *whole* response first specifically so `to_fetch` can exclude them:
/// fetching one would just 404 (it's already gone by the time we'd ask),
/// and that 404 has nothing to do with a real failure. Symmetrically,
/// [`SyncAction::Deleted`] is only reported — and the manifest only
/// touched — for an id [`Manifest::mark_deleted`] actually had a record
/// for; a same-window churn id never got archived, so there's nothing to
/// report deleting.
///
/// `history.list`'s own pagination isn't streamed (unlike
/// [`run_full_sync`]'s listing — see its module-level rationale): this path
/// is typically a single page already (`docs/gmail.md`'s Sync section), so
/// this only emits one before/after [`SyncProgressEvent::ListingPage`]/
/// [`SyncProgressEvent::ListingDone`] pair around the whole call rather than
/// per-page updates — enough that `sync`'s progress bars don't sit frozen
/// at their initial state for an incremental run's entire duration, without
/// a second streaming primitive to get there.
///
/// `opts.exclude_labels` (#1780) filters this path fully and generally,
/// off the label data history events already carry for free: a
/// `messagesAdded` event for an excluded id is never fetched at all (no
/// manifest record is ever created for it — see [`SyncAction::Excluded`]);
/// a `labelsAdded`/`labelsRemoved` event that changes an already-archived
/// record's current label set across the excluded boundary
/// soft-deletes/undeletes it via [`Manifest::mark_excluded`]/
/// [`Manifest::undelete`] — see [`reconcile_label_exclusion`], which this
/// only calls *after* the fetch phase below (not inline in the event
/// loop), so a message added and given an excluded label within the same
/// `history.list` response still has a manifest record to reconcile
/// against by the time its exclusion state is decided. One known gap, not
/// solved here: a message that arrives already excluded is never archived,
/// so if the excluded label is later removed there is no manifest record
/// for the `labelsRemoved` event to act on — it stays un-archived until
/// the next `--full`/reconciliation pass re-lists and fetches it fresh,
/// the same "self-healing on the next full pass" framing this module
/// already uses for a listing race (see [`run_full_sync`]'s doc comment).
///
/// `previous.pending_fetch` is the previous run's failed-fetch ids (#1784),
/// retried here alongside this window's `messagesAdded` ids through the
/// same presence-checked fetch: one that is
/// already archived is skipped, one that 404s is [`SyncAction::Vanished`],
/// and one deleted in this window is not fetched at all. Its exclusion state
/// is decided from the labels the fetch itself returns, since the
/// `messagesAdded` event that carried its labels was consumed by an earlier
/// run.
async fn run_incremental(
    client: &GmailClient,
    manifest: &mut Manifest,
    previous: &ArchiveState,
    opts: &SyncOptions,
    limiter: &TokenBucket,
    report: &mut SyncReport,
    progress: Option<&mpsc::UnboundedSender<SyncProgressEvent>>,
) -> Result<String> {
    let history = HistoryApi::new(client)
        .list_all_unbounded(&previous.history_id, &[], limiter)
        .await?;

    let deleted_ids: HashSet<String> = history
        .history
        .iter()
        .flat_map(|record| &record.messages_deleted)
        .map(|deleted| deleted.message.id.clone())
        .collect();

    let mut seen = HashSet::new();
    let mut to_fetch = Vec::new();
    let mut pending_retried = Vec::new();
    for id in &previous.pending_fetch {
        if !deleted_ids.contains(id) && seen.insert(id.clone()) {
            to_fetch.push(id.clone());
            pending_retried.push(id.clone());
        }
    }
    if !pending_retried.is_empty() {
        report.actions.push(SyncAction::Note {
            message: format!(
                "retrying {} message(s) that failed on a previous run",
                pending_retried.len()
            ),
        });
    }
    // Every id touched by a labelsAdded/labelsRemoved event this window,
    // in first-seen order — reconciled against `--exclude-label` only
    // *after* the fetch phase (see this function's doc comment).
    let mut labels_touched = Vec::new();
    let mut labels_touched_seen = HashSet::new();
    for record in &history.history {
        for added in &record.messages_added {
            if !deleted_ids.contains(&added.message.id) && seen.insert(added.message.id.clone()) {
                if label_ids_contain_any(&added.message.label_ids, &opts.exclude_labels) {
                    report.actions.push(SyncAction::Excluded {
                        id: added.message.id.clone(),
                    });
                    continue;
                }
                to_fetch.push(added.message.id.clone());
            }
        }
        for deleted in &record.messages_deleted {
            if manifest.get(&deleted.message.id).is_some() {
                manifest.mark_deleted(&deleted.message.id, Utc::now());
                report.actions.push(SyncAction::Deleted {
                    id: deleted.message.id.clone(),
                });
            }
        }
        for change in &record.labels_added {
            manifest.add_labels(&change.message.id, &change.label_ids);
            report.actions.push(SyncAction::LabelsUpdated {
                id: change.message.id.clone(),
                added: change.label_ids.clone(),
                removed: Vec::new(),
            });
            if labels_touched_seen.insert(change.message.id.clone()) {
                labels_touched.push(change.message.id.clone());
            }
        }
        for change in &record.labels_removed {
            manifest.remove_labels(&change.message.id, &change.label_ids);
            report.actions.push(SyncAction::LabelsUpdated {
                id: change.message.id.clone(),
                added: Vec::new(),
                removed: change.label_ids.clone(),
            });
            if labels_touched_seen.insert(change.message.id.clone()) {
                labels_touched.push(change.message.id.clone());
            }
        }
    }

    if let Some(tx) = progress {
        let _ = tx.send(SyncProgressEvent::ListingPage {
            pages: 1,
            ids_discovered: to_fetch.len(),
        });
        let _ = tx.send(SyncProgressEvent::ListingDone);
    }

    fetch_and_archive_messages(client, manifest, &to_fetch, limiter, opts, report, progress)
        .await?;

    for id in labels_touched.iter().chain(
        pending_retried
            .iter()
            .filter(|id| !labels_touched_seen.contains(*id)),
    ) {
        reconcile_label_exclusion(manifest, report, id, &opts.exclude_labels);
    }

    Ok(history
        .history_id
        .unwrap_or_else(|| previous.history_id.clone()))
}

/// Recomputes and applies `--exclude-label` state for `id` from its
/// manifest record's *current* `label_ids` (#1780) — the one place that
/// current set is known precisely, since it's built by replaying real
/// `labelsAdded`/`labelsRemoved` events rather than guessed. A no-op if
/// `id` has no manifest record (e.g. it was itself excluded at add-time, so
/// never archived — see [`run_incremental`]'s "known gap" doc note).
///
/// Three outcomes, decided by comparing the freshly-recomputed exclusion
/// set against the record's current state:
/// - not deleted → now excluded: soft-deletes via
///   [`Manifest::mark_excluded`], recording exactly which label(s) caused
///   it.
/// - deleted *because of* a previously-tracked exclusion (i.e.
///   [`ManifestRecord::excluded_labels`] was non-empty) → no longer
///   excluded: undeletes. This is the one branch [`Manifest::mark_deleted`]
///   exists specifically to keep safe: a record soft-deleted for *any other
///   reason* (a real `messagesDeleted` event, or `run_full_sync`'s
///   listing-absence pass) always has an empty `excluded_labels`, so this
///   condition can never fire for it — an unrelated label change on such a
///   record does not resurrect it.
/// - anything else (still excluded by a possibly-different label subset,
///   or deleted for an unrelated reason) → only the cached reason is kept
///   current via [`Manifest::set_excluded_labels`], `deleted_at` untouched.
fn reconcile_label_exclusion(
    manifest: &mut Manifest,
    report: &mut SyncReport,
    id: &str,
    exclude_labels: &[String],
) {
    let Some(record) = manifest.get(id) else {
        return;
    };
    let now_excluded: Vec<String> = exclude_labels
        .iter()
        .filter(|label| record.label_ids.iter().any(|existing| existing == *label))
        .cloned()
        .collect();
    let was_tracked_excluded = !record.excluded_labels.is_empty();
    let currently_deleted = record.deleted_at.is_some();

    if !currently_deleted && !now_excluded.is_empty() {
        manifest.mark_excluded(id, Utc::now(), now_excluded);
        report
            .actions
            .push(SyncAction::Deleted { id: id.to_string() });
    } else if currently_deleted && was_tracked_excluded && now_excluded.is_empty() {
        manifest.undelete(id);
        report
            .actions
            .push(SyncAction::Undeleted { id: id.to_string() });
    } else if was_tracked_excluded {
        manifest.set_excluded_labels(id, now_excluded);
    }
}

/// [`run_full_sync`]'s fetch/consumer side of the listing+fetch pipeline
/// (#1502): the presence-on-disk filter, the bounded/throttled fan-out, and
/// the manifest checkpointing all still work exactly as
/// [`fetch_and_archive_messages`] describes below — the only thing that
/// changed is that ids now arrive one at a time from `ids_rx` instead of as
/// a pre-collected `Vec`.
///
/// A plain `stream::iter(..).buffer_unordered(..)` (as
/// [`fetch_and_archive_messages`] uses) can't work here: that combinator
/// needs a complete `Vec` of ids *before* it borrows `manifest` for the
/// drain loop, so the presence-check filter and the drain loop never borrow
/// `manifest` at the same time. Once ids stream in instead, that filter has
/// to run per-id, interleaved with the drain loop — both wanting
/// `&mut Manifest` at once, which the borrow checker rejects as chained
/// stream combinators. Instead this is a manual pump loop: `tokio::select!`
/// alternates between pulling the next id (and synchronously filtering it
/// against `manifest`) and draining the next completed fetch (and
/// synchronously applying its result to `manifest`) — every manifest touch
/// is a synchronous statement inside a `select!` arm, never inside a future
/// stored in `in_flight`, so only one borrow of `manifest` is ever live.
///
/// Also returns every id seen on `ids_rx` (regardless of whether it needed
/// fetching) — [`run_full_sync`] needs that complete set, once listing
/// finishes, for its undelete/stale-deletion passes.
async fn fetch_and_archive_messages_streaming(
    client: &GmailClient,
    manifest: &mut Manifest,
    mut ids_rx: mpsc::UnboundedReceiver<String>,
    limiter: &TokenBucket,
    opts: &SyncOptions,
    report: &mut SyncReport,
    progress: Option<&mpsc::UnboundedSender<SyncProgressEvent>>,
) -> Result<HashSet<String>> {
    let output_dir = &opts.output_dir;
    let concurrency = opts.concurrency.clamp(1, MAX_CONCURRENCY);
    let mut seen = HashSet::new();
    let mut listed_ids = HashSet::new();
    let mut in_flight = FuturesUnordered::new();
    let mut since_checkpoint = 0usize;
    let mut ids_open = true;

    loop {
        tokio::select! {
            maybe_id = ids_rx.recv(), if ids_open && in_flight.len() < concurrency => {
                match maybe_id {
                    Some(id) => {
                        listed_ids.insert(id.clone());
                        if !seen.insert(id.clone()) {
                            continue;
                        }
                        // Not `shard_path(output_dir, id).exists()`: a
                        // not-yet-fetched message's shard depends on its
                        // `internal_date`, which isn't known until after
                        // it's fetched. The manifest's already-recorded
                        // `path` is the only presence check available
                        // before a fetch happens.
                        let already_archived = manifest
                            .get(&id)
                            .is_some_and(|record| output_dir.join(&record.path).exists());
                        if already_archived {
                            continue;
                        }
                        if opts.dry_run {
                            report.actions.push(SyncAction::WouldFetch { id });
                            continue;
                        }
                        let extract_attachments_flag = opts.extract_attachments;
                        let shared_pool = opts.shared_pool.clone();
                        in_flight.push(async move {
                            let _permit = match &shared_pool {
                                Some(pool) => match pool.acquire().await {
                                    Ok(permit) => Some(permit),
                                    Err(e) => {
                                        return (
                                            id,
                                            Err(anyhow::anyhow!(
                                                "sync-all's shared semaphore closed \
                                                 unexpectedly: {e}"
                                            )),
                                        );
                                    }
                                },
                                None => None,
                            };
                            limiter.acquire(MESSAGES_GET_COST_UNITS).await;
                            let result = fetch_and_write_one(
                                client,
                                output_dir,
                                &id,
                                extract_attachments_flag,
                            )
                            .await;
                            (id, result)
                        });
                        if let Some(tx) = progress {
                            let _ = tx.send(SyncProgressEvent::FetchQueued);
                        }
                    }
                    None => ids_open = false,
                }
            }
            Some((id, result)) = in_flight.next(), if !in_flight.is_empty() => {
                let mut failed = false;
                match result {
                    Ok(record) => {
                        report.actions.push(SyncAction::Fetched {
                            id,
                            path: record.path.clone(),
                            bytes: record.size,
                        });
                        manifest.upsert(record);
                    }
                    Err(e) if is_message_not_found(&e) => {
                        report.actions.push(SyncAction::Vanished { id });
                    }
                    Err(e) => {
                        failed = true;
                        report.errors.push(SyncError {
                            id,
                            reason: format!("{e:#}"),
                        });
                    }
                }
                since_checkpoint += 1;
                if since_checkpoint >= MANIFEST_CHECKPOINT_INTERVAL {
                    manifest.save(&manifest_path(output_dir))?;
                    since_checkpoint = 0;
                }
                if let Some(tx) = progress {
                    let _ = tx.send(SyncProgressEvent::FetchCompleted { failed });
                }
            }
            else => break,
        }
    }
    Ok(listed_ids)
}

/// The one place presence-on-disk is checked before any fetch, and the
/// bounded, throttled fan-out over whatever's actually missing.
///
/// A per-message failure is pushed to `report.errors` and never aborts the
/// batch — the opposite of the Facebook harvester's whole-run-fatal
/// `?`-propagation. Concurrency is bounded by `buffer_unordered` (order
/// doesn't matter for archival, unlike `search`'s `buffered`), and each
/// future acquires its slot in that bound before drawing from `limiter` —
/// debiting quota units before a concurrency slot is actually available
/// would desync the bucket's notion of "spent" from real request issuance.
///
/// Results are applied as each fetch *completes* (not batched through an
/// intermediate `Vec` after the whole fan-out finishes), and the manifest is
/// checkpointed to disk every [`MANIFEST_CHECKPOINT_INTERVAL`] completions —
/// see the module doc's interruption note. The final partial interval is
/// still covered by `run_sync`'s own unconditional save once this function
/// returns, so no extra flush is needed here on the way out.
///
/// Used by [`run_incremental`] only — [`run_full_sync`]'s pipelined listing
/// uses [`fetch_and_archive_messages_streaming`] instead (#1502).
async fn fetch_and_archive_messages(
    client: &GmailClient,
    manifest: &mut Manifest,
    ids: &[String],
    limiter: &TokenBucket,
    opts: &SyncOptions,
    report: &mut SyncReport,
    progress: Option<&mpsc::UnboundedSender<SyncProgressEvent>>,
) -> Result<()> {
    let output_dir = &opts.output_dir;
    let mut seen = HashSet::new();
    let mut to_fetch = Vec::new();
    for id in ids {
        if !seen.insert(id.clone()) {
            continue;
        }
        // Not `shard_path(output_dir, id).exists()`: a not-yet-fetched
        // message's shard depends on its `internal_date`, which isn't known
        // until after it's fetched. The manifest's already-recorded `path`
        // (the same check `run_full_sync` uses) is the only presence check
        // available before a fetch happens.
        let already_archived = manifest
            .get(id)
            .is_some_and(|record| output_dir.join(&record.path).exists());
        if !already_archived {
            to_fetch.push(id.clone());
        }
    }
    if to_fetch.is_empty() {
        return Ok(());
    }

    if opts.dry_run {
        for id in to_fetch {
            report.actions.push(SyncAction::WouldFetch { id });
        }
        return Ok(());
    }

    // The whole batch is already known (unlike run_full_sync's live-streamed
    // listing), so the fetch bar's total is knowable up front — one
    // `FetchQueued` per item gives a fully determinate bar from the start
    // rather than one that grows as ids trickle in.
    if let Some(tx) = progress {
        for _ in 0..to_fetch.len() {
            let _ = tx.send(SyncProgressEvent::FetchQueued);
        }
    }

    let concurrency = opts.concurrency.clamp(1, MAX_CONCURRENCY);
    let mut fetches = stream::iter(to_fetch)
        .map(|id| async move {
            let _permit = match opts.shared_pool.as_ref() {
                Some(pool) => match pool.acquire().await {
                    Ok(permit) => Some(permit),
                    Err(e) => {
                        return (
                            id,
                            Err(anyhow::anyhow!(
                                "sync-all's shared semaphore closed unexpectedly: {e}"
                            )),
                        );
                    }
                },
                None => None,
            };
            limiter.acquire(MESSAGES_GET_COST_UNITS).await;
            let result =
                fetch_and_write_one(client, output_dir, &id, opts.extract_attachments).await;
            (id, result)
        })
        .buffer_unordered(concurrency);

    let mut since_checkpoint = 0usize;
    while let Some((id, result)) = fetches.next().await {
        let mut failed = false;
        match result {
            Ok(record) => {
                report.actions.push(SyncAction::Fetched {
                    id,
                    path: record.path.clone(),
                    bytes: record.size,
                });
                manifest.upsert(record);
            }
            Err(e) if is_message_not_found(&e) => {
                report.actions.push(SyncAction::Vanished { id });
            }
            Err(e) => {
                failed = true;
                report.errors.push(SyncError {
                    id,
                    reason: format!("{e:#}"),
                });
            }
        }
        since_checkpoint += 1;
        if since_checkpoint >= MANIFEST_CHECKPOINT_INTERVAL {
            manifest.save(&manifest_path(output_dir))?;
            since_checkpoint = 0;
        }
        if let Some(tx) = progress {
            let _ = tx.send(SyncProgressEvent::FetchCompleted { failed });
        }
    }
    Ok(())
}

/// Fetches one message as `format=raw`, decodes it, writes the byte-exact
/// `.eml`, and builds its manifest record — `Subject`/`From`/`Message-Id`
/// come from scanning the already-decoded bytes ([`extract_headers`]), not
/// a second `format=metadata` round-trip, since a second fetch would double
/// the request volume for data already in hand.
///
/// When `extract_attachments` is set, also writes each `Content-Disposition:
/// attachment` part's decoded bytes into the message's `attachments/`
/// directory (see [`extract_attachments`] / ADR-0065) — a purely additive
/// step that never touches the manifest record built here, which keeps
/// coming from the cheap [`extract_attachment_filenames`] heuristic exactly
/// as before.
async fn fetch_and_write_one(
    client: &GmailClient,
    output_dir: &Path,
    id: &str,
    extract_attachments_flag: bool,
) -> Result<ManifestRecord> {
    let message = MessagesApi::new(client)
        .get(id, MessageFormat::Raw, &[])
        .await?;
    let bytes = decode_raw_message(&message)?;
    let headers = extract_headers(
        &bytes,
        &[
            "Subject",
            "From",
            "To",
            "Message-Id",
            "In-Reply-To",
            "References",
        ],
    );
    let attachments = extract_attachment_filenames(&bytes);
    let date = message.internal_date_utc();

    let path = shard_path(output_dir, id, date);
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("Failed to create {}", parent.display()))?;
    }
    write_atomic(&path, &bytes)?;
    let relative = path.strip_prefix(output_dir).unwrap_or(&path).to_path_buf();

    if extract_attachments_flag {
        let extracted = extract_attachments(&bytes);
        if !extracted.is_empty() {
            let dir = attachments_dir(output_dir, id, date);
            write_attachments_atomically(&dir, &extracted)?;
        }
    }

    Ok(ManifestRecord {
        id: id.to_string(),
        thread_id: message.thread_id,
        label_ids: message.label_ids,
        internal_date: message.internal_date,
        subject: headers.get("Subject").cloned(),
        from: headers.get("From").cloned(),
        to: headers.get("To").cloned(),
        rfc822_msgid: headers.get("Message-Id").cloned(),
        in_reply_to: headers.get("In-Reply-To").cloned(),
        references: headers.get("References").cloned(),
        attachment_count: attachments.count as u32,
        attachment_filenames: attachments.filenames,
        path: relative,
        size: bytes.len() as u64,
        history_id: message.history_id,
        deleted_at: None,
        excluded_labels: Vec::new(),
    })
}

/// Writes `contents` to `path` via a sibling dotfile + rename, so a crash
/// mid-write can never leave a partial `.eml` that a later run's
/// presence-on-disk check would mistake for a complete archive.
///
/// `pub(crate)` — also used by `gmail extract-attachments`
/// (`cli/gmail/extract_attachments/engine.rs`) to write extracted
/// attachment files, the exact same durability requirement this was
/// written for.
pub(crate) fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> {
    let file_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("out");
    let tmp = path.with_file_name(format!(".{file_name}.tmp"));
    std::fs::write(&tmp, contents).with_context(|| format!("Failed to write {}", tmp.display()))?;
    std::fs::rename(&tmp, path).with_context(|| format!("Failed to finalise {}", path.display()))
}

/// Writes every attachment in `attachments` into `dir`, publishing the
/// whole set in one atomic step — the directory-level counterpart of
/// [`write_atomic`]. Builds the files in a hidden sibling directory first
/// and only `rename`s it to `dir` once every file has landed, so a
/// mid-batch failure (disk full, a permission error, ...) never leaves
/// `dir` existing-but-partial: either `dir` ends up fully populated, or it
/// never exists at all. That is load-bearing for callers that treat
/// `dir.exists()` as "already extracted" — `gmail extract-attachments`
/// (`cli/gmail/extract_attachments/engine.rs`) relies on exactly that to
/// decide whether a message is safe to skip on a re-run, and a
/// partially-written `dir` would otherwise be skipped forever, never
/// retried.
///
/// `pub(crate)` — shared by [`fetch_and_write_one`] and
/// `extract_attachments::engine::run_extract_attachments`, which both
/// write a `Vec<ExtractedAttachment>` into a message's `attachments/`
/// directory.
pub(crate) fn write_attachments_atomically(
    dir: &Path,
    attachments: &[ExtractedAttachment],
) -> Result<()> {
    let dir_name = dir.file_name().and_then(|n| n.to_str()).unwrap_or("out");
    let tmp_dir = dir.with_file_name(format!(".{dir_name}.tmp"));
    if tmp_dir.exists() {
        // Left behind by an earlier run that failed partway through this
        // same message — clear it before rebuilding from scratch.
        std::fs::remove_dir_all(&tmp_dir)
            .with_context(|| format!("Failed to clear stale {}", tmp_dir.display()))?;
    }
    std::fs::create_dir_all(&tmp_dir)
        .with_context(|| format!("Failed to create {}", tmp_dir.display()))?;
    for attachment in attachments {
        write_atomic(&tmp_dir.join(&attachment.filename), &attachment.contents)?;
    }
    std::fs::rename(&tmp_dir, dir).with_context(|| format!("Failed to finalise {}", dir.display()))
}

/// A 404 on `history.list` is Google's documented signal for a
/// `startHistoryId` older than the mailbox's retention window. Treated as
/// history-not-found when the error's `reason` is absent/unparseable (fails
/// toward reconciling, matching `state.rs`'s disposable-state posture) or
/// equals `"notFound"`; a 404 with a *different*, present reason is
/// something else and must not silently trigger a full reconciliation —
/// mirroring how [`crate::gmail::client`]'s 403 quota check requires a
/// specific reason rather than matching on status alone.
fn is_history_not_found(err: &anyhow::Error) -> bool {
    match err.downcast_ref::<GmailError>() {
        Some(e @ GmailError::ApiRequestFailed { status: 404, .. }) => {
            e.reason().is_none_or(|r| r == "notFound")
        }
        _ => false,
    }
}

/// A 404 on `messages.get` for an id `history.list` just reported as added
/// means the message was permanently deleted from the server in the window
/// between the two calls — a real race under concurrent fetch fan-out, and
/// never going to succeed on retry (#1509). Unlike [`is_history_not_found`],
/// this does *not* fail open on an absent/unparseable reason: a message
/// fetch 404 for any other reason is a real failure that must still surface
/// as a [`SyncError`].
fn is_message_not_found(err: &anyhow::Error) -> bool {
    match err.downcast_ref::<GmailError>() {
        Some(e @ GmailError::ApiRequestFailed { status: 404, .. }) => {
            e.reason() == Some("notFound")
        }
        _ => false,
    }
}

/// Refuses an obviously-wrong `--output-dir`: `$HOME` itself, or anywhere
/// inside omni-dev's own settings directory.
///
/// `output_dir` may not exist yet (sync creates it), so ancestors are
/// canonicalised walking up to the deepest existing one first — the same
/// approach `src/cli/ai/claude/history/common.rs::is_inside` uses, kept as
/// a small local copy here rather than a new shared extraction (sync's
/// "source" is the network, not a local tree, so this guards a different
/// thing: an obviously-wrong destination, not source/target overlap).
fn guard_output_dir(output_dir: &Path) -> Result<()> {
    let Some(home) = dirs::home_dir() else {
        return Ok(());
    };
    let canon_output = canonicalize_best_effort(output_dir);
    if canon_output == canonicalize_best_effort(&home) {
        anyhow::bail!(
            "refusing to sync directly into your home directory ({}); use a dedicated \
             subdirectory",
            home.display()
        );
    }
    let omni_dev_dir = canonicalize_best_effort(&home.join(".omni-dev"));
    if canon_output == omni_dev_dir || canon_output.starts_with(&omni_dev_dir) {
        anyhow::bail!(
            "refusing to sync into {}: it is inside omni-dev's own settings directory",
            output_dir.display()
        );
    }
    Ok(())
}

fn canonicalize_best_effort(path: &Path) -> PathBuf {
    let mut existing = path;
    let mut tail: Vec<&std::ffi::OsStr> = Vec::new();
    loop {
        if let Ok(mut canon) = existing.canonicalize() {
            for component in tail.into_iter().rev() {
                canon.push(component);
            }
            return canon;
        }
        match existing.parent() {
            Some(parent) => {
                if let Some(name) = existing.file_name() {
                    tail.push(name);
                }
                existing = parent;
            }
            None => return path.to_path_buf(),
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use base64::Engine as _;
    use chrono::DateTime;
    use std::sync::atomic::Ordering;

    use crate::gmail::auth::{GmailCredentials, GmailScope};
    use crate::utils::secret::Secret;

    fn test_credentials() -> GmailCredentials {
        GmailCredentials {
            client_id: "client-1".to_string(),
            client_secret: Secret::new("secret-1"),
            refresh_token: Secret::new("refresh-1"),
            scope: GmailScope::ReadOnly,
        }
    }

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

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

    async fn mount_profile(server: &wiremock::MockServer, email: &str, history_id: &str) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "emailAddress": email,
                    "messagesTotal": 1,
                    "threadsTotal": 1,
                    "historyId": history_id,
                })),
            )
            .mount(server)
            .await;
    }

    async fn mount_message_list(server: &wiremock::MockServer, ids: &[&str]) {
        let messages: Vec<serde_json::Value> = ids
            .iter()
            .map(|id| serde_json::json!({"id": id, "threadId": "t1"}))
            .collect();
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": messages})),
            )
            .mount(server)
            .await;
    }

    fn raw_message_body(id: &str, subject: &str) -> String {
        let source = format!("Subject: {subject}\r\nFrom: a@example.com\r\n\r\nBody of {id}.");
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(source)
    }

    async fn mount_raw_get(server: &wiremock::MockServer, id: &str, subject: &str) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path(format!(
                "/gmail/v1/users/me/messages/{id}"
            )))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": id,
                    "threadId": "t1",
                    "labelIds": ["INBOX"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": raw_message_body(id, subject),
                })),
            )
            .mount(server)
            .await;
    }

    /// The date `mount_raw_get`'s `internalDate` ("1700000000000") resolves
    /// to — shared so tests asserting a fetched message's on-disk shard
    /// location don't hardcode a second, independently-computed date.
    fn mock_internal_date() -> DateTime<Utc> {
        DateTime::from_timestamp_millis(1_700_000_000_000).unwrap()
    }

    fn opts(output_dir: PathBuf) -> SyncOptions {
        SyncOptions {
            output_dir,
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 4,
            dry_run: false,
            extract_attachments: false,
            shared_pool: None,
        }
    }

    // ── write_attachments_atomically ────────────────────────────────

    #[test]
    fn write_attachments_atomically_writes_every_file_and_leaves_no_tmp_dir() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("m1").join("attachments");
        let attachments = vec![
            ExtractedAttachment {
                filename: "a.txt".to_string(),
                contents: b"A".to_vec(),
            },
            ExtractedAttachment {
                filename: "b.txt".to_string(),
                contents: b"B".to_vec(),
            },
        ];

        write_attachments_atomically(&dir, &attachments).unwrap();

        assert_eq!(std::fs::read(dir.join("a.txt")).unwrap(), b"A");
        assert_eq!(std::fs::read(dir.join("b.txt")).unwrap(), b"B");
        assert!(!dir.with_file_name(".attachments.tmp").exists());
    }

    #[test]
    fn write_attachments_atomically_never_leaves_a_partial_dir_on_failure() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("m1").join("attachments");
        let attachments = vec![
            ExtractedAttachment {
                filename: "a.txt".to_string(),
                contents: b"A".to_vec(),
            },
            // A NUL byte is never a filename `extract_attachments` would
            // actually produce (its `attachment_filename` sanitiser rules
            // it out), but it's a deterministic, portable way to force
            // `write_atomic` to fail partway through the batch here.
            ExtractedAttachment {
                filename: "b\0.txt".to_string(),
                contents: b"B".to_vec(),
            },
        ];

        assert!(write_attachments_atomically(&dir, &attachments).is_err());

        // The whole point: a failure partway through must never leave
        // `dir` existing-but-partial, since callers key their
        // already-extracted skip on `dir.exists()` alone. A stray hidden
        // tmp dir is fine — the next attempt clears it before rebuilding.
        assert!(!dir.exists());
    }

    #[test]
    fn write_attachments_atomically_recovers_from_a_stale_tmp_dir() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("m1").join("attachments");
        let tmp_dir = dir.with_file_name(".attachments.tmp");
        std::fs::create_dir_all(&tmp_dir).unwrap();
        std::fs::write(tmp_dir.join("leftover-from-a-crash.bin"), b"stale").unwrap();

        let attachments = vec![ExtractedAttachment {
            filename: "a.txt".to_string(),
            contents: b"A".to_vec(),
        }];

        write_attachments_atomically(&dir, &attachments).unwrap();

        assert_eq!(std::fs::read(dir.join("a.txt")).unwrap(), b"A");
        // The stale leftover from the aborted attempt must not survive
        // into the finalised directory.
        assert!(!dir.join("leftover-from-a-crash.bin").exists());
    }

    // ── backfill ─────────────────────────────────────────────────────

    #[tokio::test]
    async fn run_sync_backfills_on_first_run_and_persists_the_watermark() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        mount_message_list(&server, &["m1"]).await;
        mount_raw_get(&server, "m1", "Hello").await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
        assert!(shard_path(&output_dir, "m1", Some(mock_internal_date())).exists());

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => {
                assert_eq!(s.history_id, "500");
                assert_eq!(s.email_address, "user@example.com");
            }
            _ => panic!("expected a persisted state after a clean backfill"),
        }

        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert_eq!(record.subject.as_deref(), Some("Hello"));
        assert_eq!(record.from.as_deref(), Some("a@example.com"));
    }

    #[tokio::test]
    async fn run_sync_records_threading_headers_and_attachment_metadata() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        mount_message_list(&server, &["m1"]).await;

        let source = "Subject: Report\r\n\
From: a@example.com\r\n\
To: b@example.com\r\n\
Message-Id: <m1@example.com>\r\n\
In-Reply-To: <parent@example.com>\r\n\
References: <parent@example.com>\r\n\
MIME-Version: 1.0\r\n\
Content-Type: multipart/mixed; boundary=\"BOUNDARY\"\r\n\
\r\n\
--BOUNDARY\r\n\
Content-Type: text/plain\r\n\
\r\n\
Hello\r\n\
--BOUNDARY\r\n\
Content-Type: application/pdf\r\n\
Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
\r\n\
not-really-a-pdf\r\n\
--BOUNDARY--\r\n";
        let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(source);
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "m1",
                    "threadId": "t1",
                    "labelIds": ["INBOX"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": encoded,
                })),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();
        assert!(report.errors.is_empty());

        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert_eq!(record.to.as_deref(), Some("b@example.com"));
        assert_eq!(record.in_reply_to.as_deref(), Some("<parent@example.com>"));
        assert_eq!(record.references.as_deref(), Some("<parent@example.com>"));
        assert_eq!(record.attachment_count, 1);
        assert_eq!(record.attachment_filenames, vec!["report.pdf".to_string()]);
    }

    // ── 404 → reconciliation ─────────────────────────────────────────

    #[tokio::test]
    async fn run_sync_404_on_watermark_triggers_reconciliation_not_a_gap() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "1".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"message": "Not Found", "errors": [{"reason": "notFound"}]}
                })),
            )
            .mount(&server)
            .await;
        mount_message_list(&server, &["m1"]).await;
        mount_raw_get(&server, "m1", "Hello").await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Note { message } if message.contains("reconciling"))));
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => assert_eq!(s.history_id, "999"),
            _ => panic!("expected reconciliation to persist the profile's current historyId"),
        }
    }

    // ── valid watermark applies all 4 event types ─────────────────────

    #[tokio::test]
    async fn run_sync_applies_added_deleted_and_label_events_from_a_valid_watermark() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        // Pre-existing m2 (to be deleted) and m3 (to have labels changed) —
        // both already archived, neither should be re-fetched or rewritten.
        let mut manifest = Manifest::default();
        for id in ["m2", "m3"] {
            let path = shard_path(&output_dir, id, None);
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            std::fs::write(&path, format!("From: a@example.com\r\n\r\n{id} body")).unwrap();
            manifest.upsert(ManifestRecord {
                id: id.to_string(),
                thread_id: Some("t1".to_string()),
                label_ids: vec!["INBOX".to_string(), "UNREAD".to_string()],
                internal_date: None,
                subject: None,
                from: None,
                to: None,
                rfc822_msgid: None,
                in_reply_to: None,
                references: None,
                attachment_count: 0,
                attachment_filenames: Vec::new(),
                path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
                size: std::fs::metadata(&path).unwrap().len(),
                history_id: Some("50".to_string()),
                deleted_at: None,
                excluded_labels: Vec::new(),
            });
        }
        manifest.save(&manifest_path(&output_dir)).unwrap();
        let m3_bytes_before = std::fs::read(shard_path(&output_dir, "m3", None)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .and(wiremock::matchers::query_param("startHistoryId", "100"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "history": [{
                    "id": "150",
                    "messagesAdded": [{"message": {"id": "m1", "threadId": "t1", "labelIds": ["INBOX"]}}],
                    "messagesDeleted": [{"message": {"id": "m2", "threadId": "t1"}}],
                    "labelsAdded": [{"message": {"id": "m3", "threadId": "t1"}, "labelIds": ["IMPORTANT"]}],
                    "labelsRemoved": [{"message": {"id": "m3", "threadId": "t1"}, "labelIds": ["UNREAD"]}],
                }],
                "historyId": "300",
            })))
            .mount(&server)
            .await;
        mount_raw_get(&server, "m1", "New message").await;
        // Deliberately no mock for GET .../messages/m3 — a label-only change
        // must never fetch the message it applies to.

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();
        assert!(report.errors.is_empty());

        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").is_some(), "m1 should have been fetched");
        assert!(
            manifest.get("m2").unwrap().deleted_at.is_some(),
            "m2 should be soft-deleted"
        );
        assert!(
            shard_path(&output_dir, "m2", None).exists(),
            "m2's .eml must survive a soft-delete"
        );
        let m3 = manifest.get("m3").unwrap();
        assert!(m3.label_ids.contains(&"IMPORTANT".to_string()));
        assert!(!m3.label_ids.contains(&"UNREAD".to_string()));
        assert_eq!(
            std::fs::read(shard_path(&output_dir, "m3", None)).unwrap(),
            m3_bytes_before,
            "a label-only change must never rewrite the .eml"
        );

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => assert_eq!(s.history_id, "300"),
            _ => panic!("expected the new historyId to be persisted"),
        }
    }

    #[tokio::test]
    async fn run_sync_ignores_a_message_added_and_deleted_within_the_same_history_window() {
        // Routine server-side churn (an auto-filtered message, a sent mail
        // immediately recalled): a message can be added and deleted again
        // before the next sync ever sees it. Fetching it would just 404,
        // and that isn't a real failure — see `run_incremental`'s doc
        // comment.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [{"message": {"id": "churn1", "threadId": "t1"}}],
                        "messagesDeleted": [{"message": {"id": "churn1", "threadId": "t1"}}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        // Deliberately no mock for GET .../messages/churn1 — fetching it at
        // all (not just erroring) is the bug this test guards against.

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Deleted { id } if id == "churn1")),
            "a message never archived shouldn't be reported as deleted"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(
            manifest.get("churn1").is_none(),
            "a same-window add+delete should never create a manifest record"
        );
    }

    // ── --exclude-label (#1780) ─────────────────────────────────────────

    #[tokio::test]
    async fn run_incremental_never_undeletes_a_record_with_no_tracked_exclusion_reason_by_default()
    {
        // Regression test: a user who never passes --exclude-label at all
        // must see run_incremental behave exactly as it did before this
        // flag existed — never undeleting anything. Before the fix, this
        // failed because `!label_ids_excluded(labels, [])` was vacuously
        // `true`, collapsing the undelete guard to just "is it currently
        // soft-deleted" for every default caller, not only --exclude-label
        // adopters.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        // m1 is already soft-deleted for a reason with nothing to do with
        // labels (a real messagesDeleted event, or run_full_sync's
        // listing-absence pass) — plain mark_deleted, no exclusion
        // tracking.
        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.mark_deleted("m1", Utc::now());
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "labelsRemoved": [
                            {"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["IMPORTANT"]},
                        ],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;

        // Default opts(): exclude_labels is empty, i.e. --exclude-label was
        // never passed.
        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m1")),
            "an unrelated labelsRemoved event must never undelete a message soft-deleted for a \
             reason unrelated to label exclusion, even when --exclude-label was never configured"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").unwrap().deleted_at.is_some());
    }

    #[tokio::test]
    async fn run_incremental_excludes_a_message_labeled_within_the_same_history_window_it_was_added_in(
    ) {
        // Regression test: a message added via messagesAdded (not yet
        // excluded per that event's own label set) and then given an
        // excluded label via labelsAdded, both within the *same*
        // history.list response, must still end up excluded — not archived
        // despite matching --exclude-label within this very run. Before the
        // fix, the exclusion recheck ran inline in the event-scanning loop,
        // before the batched fetch (which only happens after that loop)
        // had created a manifest record for the message, so the recheck was
        // always a no-op for a same-window id.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [
                            {"message": {"id": "m1", "threadId": "t1", "labelIds": ["INBOX"]}},
                        ],
                    }, {
                        "id": "151",
                        "labelsAdded": [
                            {"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["SPAM"]},
                        ],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        // Not `mount_raw_get` (hardcoded to `labelIds: ["INBOX"]`): the
        // live fetch is authoritative for the message's label state *at
        // fetch time*, which is after the labelsAdded event above already
        // happened server-side — so the mock must reflect that SPAM is
        // already present, exactly as the real Gmail API would return.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "m1",
                    "threadId": "t1",
                    "labelIds": ["INBOX", "SPAM"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": raw_message_body("m1", "New but spammed"),
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(
            report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Deleted { id } if id == "m1")),
            "a message excluded within the same window it was added in must still be excluded"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_some());
        assert_eq!(record.excluded_labels, vec!["SPAM".to_string()]);
    }

    #[tokio::test]
    async fn run_incremental_skips_fetching_a_new_message_with_an_excluded_label() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [{"message": {"id": "spam1", "threadId": "t1", "labelIds": ["SPAM"]}}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        // Deliberately no mock for GET .../messages/spam1 — fetching an
        // excluded message at all is the bug this test guards against.

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Excluded { id } if id == "spam1")));
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(
            manifest.get("spam1").is_none(),
            "an excluded new message should never be archived"
        );
    }

    #[tokio::test]
    async fn run_incremental_soft_deletes_an_archived_message_labeled_with_an_excluded_label() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "labelsAdded": [{"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["SPAM"]}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Deleted { id } if id == "m1")));
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").unwrap().deleted_at.is_some());
        assert!(
            shard_path(&output_dir, "m1", None).exists(),
            "the .eml must survive a label-triggered soft-delete"
        );
    }

    #[tokio::test]
    async fn run_incremental_undeletes_only_once_every_excluded_label_is_removed() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        let mut manifest = Manifest::default();
        for id in ["m1", "m2"] {
            let path = shard_path(&output_dir, id, None);
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            std::fs::write(&path, format!("From: a@example.com\r\n\r\n{id} body")).unwrap();
            manifest.upsert(ManifestRecord {
                id: id.to_string(),
                thread_id: Some("t1".to_string()),
                label_ids: vec!["SPAM".to_string(), "TRASH".to_string()],
                internal_date: None,
                subject: None,
                from: None,
                to: None,
                rfc822_msgid: None,
                in_reply_to: None,
                references: None,
                attachment_count: 0,
                attachment_filenames: Vec::new(),
                path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
                size: std::fs::metadata(&path).unwrap().len(),
                history_id: Some("50".to_string()),
                deleted_at: None,
                excluded_labels: Vec::new(),
            });
            // Simulate an earlier incremental run's own tracked exclusion
            // (via `mark_excluded`, not a raw `deleted_at`) — only that
            // precise tracking makes an automatic undelete safe (#1780).
            manifest.mark_excluded(
                id,
                Utc::now(),
                vec!["SPAM".to_string(), "TRASH".to_string()],
            );
        }
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        // m1 loses both excluded labels; m2 only loses SPAM
                        // and still carries TRASH.
                        "labelsRemoved": [
                            {"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["SPAM", "TRASH"]},
                            {"message": {"id": "m2", "threadId": "t1"}, "labelIds": ["SPAM"]},
                        ],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string(), "TRASH".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m1")));
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m2")),
            "m2 still carries an excluded label (TRASH) and must stay soft-deleted"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").unwrap().deleted_at.is_none());
        assert!(manifest.get("m2").unwrap().deleted_at.is_some());
    }

    #[tokio::test]
    async fn run_full_sync_folds_known_system_labels_into_the_listing_query() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param("q", "-in:spam -in:trash"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string(), "TRASH".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
    }

    #[tokio::test]
    async fn run_full_sync_does_not_duplicate_an_in_token_already_named_by_query() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        // `--query "-in:spam"` and `--exclude-label SPAM` both name spam —
        // the folded query must not repeat the `-in:spam` token. The
        // caller's query is still parenthesized before `-in:trash` is
        // appended (#1780) even though nothing here actually needs the
        // protection — parenthesizing is unconditional once at least one
        // exclusion clause is appended, not just when the query contains an
        // `OR` that needs it.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param("q", "(-in:spam) -in:trash"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let opts = SyncOptions {
            query: Some("-in:spam".to_string()),
            exclude_labels: vec!["SPAM".to_string(), "TRASH".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
    }

    #[tokio::test]
    async fn run_full_sync_parenthesizes_an_or_query_before_appending_an_exclusion() {
        // Gmail's implicit AND-by-juxtaposition binds tighter than an
        // explicit `OR` — appending `-in:spam` as a bare trailing token to
        // `from:x OR from:y` would parse as `from:x OR (from:y AND
        // -in:spam)`, silently narrowing the exclusion to only the
        // right-hand side. Parenthesizing the caller's query first
        // (`(from:x OR from:y) -in:spam`) applies it to the whole
        // expression instead (#1780).
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param(
                "q",
                "(from:x OR from:y) -in:spam",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let opts = SyncOptions {
            query: Some("from:x OR from:y".to_string()),
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
    }

    #[tokio::test]
    async fn run_full_sync_leaves_query_untouched_when_no_exclude_label_translates() {
        // Parenthesizing is conditional on actually appending an exclusion
        // clause — a caller who passes `--query` alone (no
        // `--exclude-label`, or one with no known translation) must see it
        // reach the API byte-for-byte, unchanged from before this flag
        // existed.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param("q", "from:x OR from:y"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let opts = SyncOptions {
            query: Some("from:x OR from:y".to_string()),
            exclude_labels: vec!["Label_16".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
    }

    #[tokio::test]
    async fn run_full_sync_notes_an_exclude_label_it_cannot_translate_into_a_query() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let opts = SyncOptions {
            exclude_labels: vec!["Label_16".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report.actions.iter().any(|a| matches!(
            a,
            SyncAction::Note { message } if message.contains("Label_16") && message.contains("--query")
        )));
    }

    #[tokio::test]
    async fn run_full_sync_vanished_record_is_never_bogus_undeleted_by_an_unrelated_label_change() {
        // A record `run_full_sync` soft-deletes via listing-absence never
        // learns *why* it vanished (`--query` can narrow a listing for
        // reasons that have nothing to do with `--exclude-label`, so
        // guessing would be unsound — see `run_full_sync`'s stale-deletion
        // doc comment). `excluded_labels` therefore stays empty for it, and
        // that's what a later unrelated `labelsRemoved` event must respect
        // (#1780): `reconcile_label_exclusion` only ever undeletes a record
        // whose `excluded_labels` shows it was *this* mechanism that
        // deleted it.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        // m1 is already archived, with label_ids that have never recorded
        // SPAM — the state a mailbox's existing archive is in the moment
        // `--exclude-label SPAM` first gets configured.
        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "100").await;
        // m1 no longer appears in the `-in:spam`-filtered listing — it's
        // been spammed on the server, though this pass never fetches it to
        // confirm that.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param("q", "-in:spam"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Deleted { id } if id == "m1")));
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_some());
        assert!(
            record.excluded_labels.is_empty(),
            "run_full_sync doesn't know *why* a listing-absent record vanished, so it must \
             never guess and tag excluded_labels (#1780)"
        );
        assert_eq!(
            record.label_ids,
            vec!["INBOX".to_string()],
            "a plain stale-deletion must never rewrite label_ids"
        );

        // An incremental run now observes a totally unrelated label change
        // on the same message. run_incremental must not resurrect it: it
        // has no tracked exclusion reason for m1 at all, so
        // reconcile_label_exclusion's undelete branch can't fire regardless
        // of what changed.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "labelsRemoved": [
                            {"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["IMPORTANT"]},
                        ],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m1")),
            "m1 has no tracked exclusion reason; an unrelated label change must not resurrect it"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").unwrap().deleted_at.is_some());
    }

    #[tokio::test]
    async fn run_full_sync_reappearance_does_not_undelete_a_record_excluded_by_a_custom_label() {
        // A record run_incremental soft-deleted via a custom (untranslated)
        // `--exclude-label` must not be resurrected just because it still
        // appears in a full-sync listing that couldn't have filtered that
        // label out server-side (#1780) — see run_full_sync's
        // undelete-on-reappearance doc comment.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string(), "Label_16".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        // Simulate a prior incremental run's tracked exclusion via the
        // custom label.
        manifest.mark_excluded("m1", Utc::now(), vec!["Label_16".to_string()]);
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "100").await;
        // No `-in:` clause can express `Label_16`, so the listing is
        // unfiltered — m1 still appears in it even though it's still
        // excluded per our own tracking.
        mount_message_list(&server, &["m1"]).await;

        let opts = SyncOptions {
            exclude_labels: vec!["Label_16".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m1")),
            "m1 is still tracked as excluded by a custom label the listing query couldn't \
             filter; merely reappearing in it must not resurrect it"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_some());
        assert_eq!(record.excluded_labels, vec!["Label_16".to_string()]);
    }

    #[tokio::test]
    async fn run_full_sync_reappearance_undeletes_a_record_once_a_translated_label_is_confirmed_gone(
    ) {
        // Unlike a custom label, a translated one (SPAM/TRASH) reappearing
        // in a listing this run's own query excludes it from is
        // authoritative proof it's gone — the reappearance-undelete must
        // still fire even though excluded_labels is non-empty (#1780).
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.mark_excluded("m1", Utc::now(), vec!["SPAM".to_string()]);
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "100").await;
        // m1 reappears in the `-in:spam`-filtered listing: confirmation
        // it's no longer spam on the server.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .and(wiremock::matchers::query_param("q", "-in:spam"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "messages": [{"id": "m1", "threadId": "t1"}]
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "m1")));
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_none());
        assert!(record.excluded_labels.is_empty());
    }

    #[tokio::test]
    async fn run_incremental_via_run_sync_with_progress_emits_fetch_events() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [{"message": {"id": "m1", "threadId": "t1"}}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        mount_raw_get(&server, "m1", "New message").await;

        let (tx, mut rx) = mpsc::unbounded_channel();
        let report = run_sync_with_progress(&client, &opts(output_dir), Some(&tx))
            .await
            .unwrap();
        drop(tx);

        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }

        assert!(report.errors.is_empty());
        // The bars must not sit frozen at their initial state for an
        // incremental run's whole duration (the bug this test guards
        // against): at least one real listing update plus a matched
        // queued/completed pair for the one message fetched.
        assert!(events.iter().any(|e| matches!(
            e,
            SyncProgressEvent::ListingPage {
                ids_discovered: 1,
                ..
            }
        )));
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::ListingDone))
                .count(),
            1
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::FetchQueued))
                .count(),
            1
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::FetchCompleted { failed: false }))
                .count(),
            1
        );
    }

    // ── account-identity validation ────────────────────────────────────

    #[tokio::test]
    async fn run_sync_rejects_a_state_json_for_a_different_account() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "1".to_string(),
                email_address: "wrong@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();
        let manifest_before = b"".to_vec();
        std::fs::write(manifest_path(&output_dir), &manifest_before).unwrap();

        mount_profile(&server, "user@example.com", "999").await;

        let err = run_sync(&client, &opts(output_dir.clone()))
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("wrong@example.com"));
        assert!(msg.contains("user@example.com"));

        // Neither file was touched by the rejected run.
        assert_eq!(
            std::fs::read(manifest_path(&output_dir)).unwrap(),
            manifest_before
        );
        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => assert_eq!(s.email_address, "wrong@example.com"),
            _ => panic!("expected the original (mismatched) state to survive untouched"),
        }
    }

    // ── idempotence ────────────────────────────────────────────────────

    #[tokio::test]
    async fn run_sync_rerun_with_no_server_side_changes_fetches_and_writes_nothing_new() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        mount_message_list(&server, &["m1"]).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "id": "m1", "threadId": "t1", "labelIds": ["INBOX"],
                "internalDate": "1700000000000", "historyId": "500",
                "raw": raw_message_body("m1", "Hello"),
            })))
            .expect(1) // exactly once across BOTH runs below
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let o = opts(output_dir.clone());
        let first = run_sync(&client, &o).await.unwrap();
        assert!(first.errors.is_empty());
        let manifest_before = std::fs::read(manifest_path(&output_dir)).unwrap();

        // Second run takes the incremental path against the watermark the
        // first run just persisted; an empty history means nothing to do.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [], "historyId": "500",
                })),
            )
            .mount(&server)
            .await;

        let second = run_sync(&client, &o).await.unwrap();
        assert!(second
            .actions
            .iter()
            .all(|a| !matches!(a, SyncAction::Fetched { .. })));
        let manifest_after = std::fs::read(manifest_path(&output_dir)).unwrap();
        assert_eq!(manifest_before, manifest_after);
    }

    // ── interruption / the Facebook-harvester regression ────────────────

    #[tokio::test]
    async fn run_sync_survives_a_missing_state_json_without_refetching_or_truncating() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");

        // An existing archive from a prior, already-completed run.
        let eml_path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(eml_path.parent().unwrap()).unwrap();
        let original_bytes = b"From: a@example.com\r\n\r\nOriginal body.".to_vec();
        std::fs::write(&eml_path, &original_bytes).unwrap();
        let mut manifest = Manifest::default();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: eml_path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: original_bytes.len() as u64,
            history_id: Some("1".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();
        // `state.json` is deliberately absent — as if it were deleted, or
        // this is the very first run after `preload_prior`/`open_sink`-style
        // machinery would otherwise have truncated an existing archive.

        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &["m1"]).await;
        // Deliberately no mock for GET .../messages/m1 — a re-fetch here
        // would mean the regression reappeared.

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report
            .actions
            .iter()
            .all(|a| !matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
        assert_eq!(
            std::fs::read(&eml_path).unwrap(),
            original_bytes,
            "the pre-existing archive must survive a missing state.json intact"
        );
        let manifest_after = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert_eq!(
            manifest_after.get("m1").unwrap().size,
            original_bytes.len() as u64
        );
    }

    // ── per-item failure ────────────────────────────────────────────────

    #[tokio::test]
    async fn run_sync_one_unfetchable_message_yields_an_error_and_the_run_completes() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &["m1", "m2"]).await;
        mount_raw_get(&server, "m1", "Good").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m2"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert_eq!(report.errors.len(), 1);
        assert_eq!(report.errors[0].id, "m2");
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));

        // A backfill with an error still records its watermark, carrying
        // the failed id forward, so the next run is incremental rather than
        // another full-mailbox listing (#1784).
        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => {
                assert_eq!(s.history_id, "999");
                assert_eq!(s.pending_fetch, ["m2"]);
            }
            _ => panic!("expected state.json to be written"),
        }
        // And the successfully-fetched message's manifest entry survives.
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").is_some());
    }

    #[tokio::test]
    async fn run_sync_advances_the_watermark_and_carries_an_incremental_failure_forward() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "history": [{"id": "150", "messagesAdded": [{"message": {"id": "m1", "threadId": "t1"}}]}],
                "historyId": "200",
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();
        assert_eq!(report.errors.len(), 1);

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => {
                assert_eq!(s.history_id, "200", "watermark must advance");
                assert_eq!(s.pending_fetch, ["m1"], "the failure is carried forward");
            }
            _ => panic!("expected state.json to be written"),
        }
    }

    // ── vanished message during fetch (#1509) ───────────────────────────

    #[tokio::test]
    async fn run_sync_treats_a_vanished_message_404_as_benign_not_an_error() {
        // A message reported by history.list can be permanently deleted from
        // the server before its own messages.get call runs — real once you
        // add concurrency to the fetch fan-out. Unlike an ordinary
        // per-message failure, this can never succeed on retry, so it must
        // not be carried forward in `pending_fetch` (unlike
        // `run_sync_advances_the_watermark_and_carries_an_incremental_failure_forward`
        // above, whose 500 genuinely is worth retrying).
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [{"message": {"id": "vanished1", "threadId": "t1"}}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path(
                "/gmail/v1/users/me/messages/vanished1",
            ))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"message": "Not Found", "errors": [{"reason": "notFound"}]}
                })),
            )
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(
            report.errors.is_empty(),
            "a vanished message must not be a SyncError"
        );
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Vanished { id } if id == "vanished1")));

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => assert_eq!(
                s.history_id, "300",
                "watermark must advance past a vanished-message 404"
            ),
            _ => panic!("expected the watermark to be saved"),
        }
    }

    #[tokio::test]
    async fn run_sync_message_404_with_a_different_reason_is_still_an_error() {
        // The reason check must not fail open the way `is_history_not_found`
        // deliberately does for `history.list` — a 404 on messages.get for
        // any reason other than `notFound` is a real failure and must still
        // be carried forward for retry.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "messagesAdded": [{"message": {"id": "m1", "threadId": "t1"}}],
                    }],
                    "historyId": "300",
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"message": "Backend error", "errors": [{"reason": "backendError"}]}
                })),
            )
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert_eq!(report.errors.len(), 1);
        assert_eq!(report.errors[0].id, "m1");
        assert!(!report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Vanished { .. })));

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => {
                assert_eq!(s.history_id, "300", "watermark must advance");
                assert_eq!(
                    s.pending_fetch,
                    ["m1"],
                    "a non-notFound 404 is retried next run"
                );
            }
            _ => panic!("expected state.json to be written"),
        }
    }

    // ── pending_fetch carry-forward (#1784) ─────────────────────────────

    fn save_state_with_pending(output_dir: &Path, pending: &[&str]) {
        std::fs::create_dir_all(output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "100".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: pending.iter().map(|id| (*id).to_string()).collect(),
            },
            &state_path(output_dir),
        )
        .unwrap();
    }

    async fn mount_history(server: &wiremock::MockServer, body: serde_json::Value) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body))
            .mount(server)
            .await;
    }

    async fn mount_get_never_called(server: &wiremock::MockServer, id: &str) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path(format!(
                "/gmail/v1/users/me/messages/{id}"
            )))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .expect(0)
            .mount(server)
            .await;
    }

    fn load_present(output_dir: &Path) -> ArchiveState {
        match state::load(&state_path(output_dir)) {
            LoadOutcome::Present(s) => s,
            _ => panic!("expected state.json to be written"),
        }
    }

    #[tokio::test]
    async fn run_sync_retries_a_pending_fetch_even_with_an_empty_history_window() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_history(&server, serde_json::json!({"historyId": "200"})).await;
        mount_raw_get(&server, "m1", "Retried").await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
        assert!(report.actions.iter().any(|a| matches!(
            a,
            SyncAction::Note { message } if message.contains("retrying 1 message")
        )));
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(manifest.get("m1").is_some());
        let s = load_present(&output_dir);
        assert_eq!(s.history_id, "200");
        assert!(s.pending_fetch.is_empty());
    }

    #[tokio::test]
    async fn run_sync_keeps_a_pending_fetch_that_fails_again_and_still_advances() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_history(&server, serde_json::json!({"historyId": "200"})).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert_eq!(report.errors.len(), 1);
        let s = load_present(&output_dir);
        assert_eq!(s.history_id, "200");
        assert_eq!(s.pending_fetch, ["m1"]);
    }

    #[tokio::test]
    async fn run_sync_drops_a_pending_fetch_that_has_since_vanished() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_history(&server, serde_json::json!({"historyId": "200"})).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"message": "Not Found", "errors": [{"reason": "notFound"}]}
                })),
            )
            .mount(&server)
            .await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Vanished { id } if id == "m1")));
        assert!(load_present(&output_dir).pending_fetch.is_empty());
    }

    #[tokio::test]
    async fn run_sync_does_not_refetch_a_pending_id_already_archived_on_disk() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);

        let mut manifest = Manifest::default();
        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, "From: a@example.com\r\n\r\nm1 body").unwrap();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: std::fs::metadata(&path).unwrap().len(),
            history_id: Some("50".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        mount_history(&server, serde_json::json!({"historyId": "200"})).await;
        mount_get_never_called(&server, "m1").await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(load_present(&output_dir).pending_fetch.is_empty());
    }

    #[tokio::test]
    async fn run_sync_does_not_fetch_a_pending_id_deleted_in_this_window() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_history(
            &server,
            serde_json::json!({
                "history": [{
                    "id": "150",
                    "messagesDeleted": [{"message": {"id": "m1", "threadId": "t1"}}],
                }],
                "historyId": "200",
            }),
        )
        .await;
        mount_get_never_called(&server, "m1").await;

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(load_present(&output_dir).pending_fetch.is_empty());
    }

    #[tokio::test]
    async fn run_sync_soft_deletes_a_retried_pending_fetch_that_now_carries_an_excluded_label() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_history(&server, serde_json::json!({"historyId": "200"})).await;
        // Not `mount_raw_get` (hardcoded to `labelIds: ["INBOX"]`): the
        // label that moved it into SPAM arrived in a window an earlier run
        // consumed, so only the fetch itself can tell this run about it.
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "m1",
                    "threadId": "t1",
                    "labelIds": ["INBOX", "SPAM"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": raw_message_body("m1", "Spammy"),
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_some());
        assert_eq!(record.excluded_labels, ["SPAM"]);
    }

    #[tokio::test]
    async fn run_sync_full_replaces_pending_fetch_with_only_this_runs_errors() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        // "gone" no longer exists server-side, so a full listing drops it;
        // "m2" fails on this run and becomes the new pending set.
        save_state_with_pending(&output_dir, &["gone"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &["m1", "m2"]).await;
        mount_raw_get(&server, "m1", "Good").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m2"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;
        mount_get_never_called(&server, "gone").await;

        let opts = SyncOptions {
            full: true,
            ..opts(output_dir.clone())
        };
        run_sync(&client, &opts).await.unwrap();

        let s = load_present(&output_dir);
        assert_eq!(s.history_id, "999");
        assert_eq!(s.pending_fetch, ["m2"]);
    }

    #[tokio::test]
    async fn run_sync_full_with_query_carries_forward_a_pending_id_it_did_not_list() {
        // A `--query`-scoped listing never looked for "outside", so its
        // absence proves nothing: it must stay pending for the next
        // incremental run rather than be dropped. "m1" was listed, so the
        // listing itself retried it and it is not carried.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1", "outside"]);
        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &["m1"]).await;
        mount_raw_get(&server, "m1", "Listed").await;
        mount_get_never_called(&server, "outside").await;

        let opts = SyncOptions {
            full: true,
            query: Some("from:boss@example.com".to_string()),
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        let s = load_present(&output_dir);
        assert_eq!(s.history_id, "999");
        assert_eq!(s.pending_fetch, ["outside"]);
    }

    #[tokio::test]
    async fn run_sync_excludes_a_pending_retry_also_labeled_this_window() {
        // A pending-retry id can also be touched by a labelsAdded/labelsRemoved
        // event in the very window that retries it, so it appears in both
        // `pending_retried` and `labels_touched`. It must still have its
        // exclusion reconciled against the labels the retry fetch returns.
        // This does not pin that the reconcile runs only once:
        // `reconcile_label_exclusion` is idempotent (a second call finds the
        // record already excluded for the same labels and pushes no action),
        // so a duplicate call is unobservable here.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        save_state_with_pending(&output_dir, &["m1"]);
        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "history": [{
                        "id": "150",
                        "labelsAdded": [
                            {"message": {"id": "m1", "threadId": "t1"}, "labelIds": ["SPAM"]},
                        ],
                    }],
                    "historyId": "200",
                })),
            )
            .mount(&server)
            .await;
        // Not `mount_raw_get` (hardcoded to `labelIds: ["INBOX"]`): the
        // retry fetch is what tells this run m1 now carries SPAM — the
        // labelsAdded event above only proves the label was added, not
        // what else is currently set (m1 was never archived by an earlier
        // run, so there's no prior manifest label set to merge onto).
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "m1",
                    "threadId": "t1",
                    "labelIds": ["INBOX", "SPAM"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": raw_message_body("m1", "Retried and spammed"),
                })),
            )
            .mount(&server)
            .await;

        let opts = SyncOptions {
            exclude_labels: vec!["SPAM".to_string()],
            ..opts(output_dir.clone())
        };
        let report = run_sync(&client, &opts).await.unwrap();

        assert!(report.errors.is_empty());
        let deleted_count = report
            .actions
            .iter()
            .filter(|a| matches!(a, SyncAction::Deleted { id } if id == "m1"))
            .count();
        assert_eq!(
            deleted_count, 1,
            "the retried id must be soft-deleted as excluded"
        );
        let manifest = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let record = manifest.get("m1").unwrap();
        assert!(record.deleted_at.is_some());
        assert_eq!(record.excluded_labels, ["SPAM"]);
        assert!(load_present(&output_dir).pending_fetch.is_empty());
    }

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

    #[tokio::test]
    async fn run_sync_dry_run_reports_planned_actions_and_touches_no_files() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &["m1"]).await;
        // Deliberately no mock for GET .../messages/m1 — dry-run must never
        // fetch a message body.

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        let report = run_sync(
            &client,
            &SyncOptions {
                output_dir: output_dir.clone(),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                concurrency: 4,
                dry_run: true,
                extract_attachments: false,
                shared_pool: None,
            },
        )
        .await
        .unwrap();

        assert!(matches!(
            report.actions.as_slice(),
            [SyncAction::WouldFetch { id }] if id == "m1"
        ));
        assert!(
            !output_dir.exists(),
            "dry-run must not create the output directory"
        );
    }

    // ── guard_output_dir ─────────────────────────────────────────────────

    #[test]
    fn guard_output_dir_rejects_home_directory_itself() {
        let _guard = crate::gmail::test_support::EnvGuard::take();
        let home = tempfile::tempdir().unwrap();
        std::env::set_var("HOME", home.path());
        let err = guard_output_dir(home.path()).unwrap_err();
        assert!(err.to_string().contains("home directory"));
    }

    #[test]
    fn guard_output_dir_rejects_inside_omni_dev_settings_dir() {
        let _guard = crate::gmail::test_support::EnvGuard::take();
        let home = tempfile::tempdir().unwrap();
        std::env::set_var("HOME", home.path());
        let target = home.path().join(".omni-dev").join("mail-archive");
        std::fs::create_dir_all(&target).unwrap();
        let err = guard_output_dir(&target).unwrap_err();
        assert!(err.to_string().contains("settings directory"));
    }

    #[test]
    fn guard_output_dir_accepts_an_ordinary_subdirectory() {
        let _guard = crate::gmail::test_support::EnvGuard::take();
        let home = tempfile::tempdir().unwrap();
        std::env::set_var("HOME", home.path());
        assert!(guard_output_dir(&home.path().join("mail-archive")).is_ok());
    }

    // ── corrupt manifest is a hard failure ──────────────────────────────

    #[tokio::test]
    async fn run_sync_propagates_a_corrupt_manifest_as_a_hard_error() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "1").await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        std::fs::write(manifest_path(&output_dir), "not json\n").unwrap();

        let err = run_sync(&client, &opts(output_dir)).await.unwrap_err();
        assert!(err.to_string().contains("Failed to parse manifest"));
    }

    // ── reconciliation must not truncate at HARD_CAP (#1467) ────────────

    #[tokio::test]
    async fn run_sync_full_reconciliation_does_not_mislabel_mail_past_hard_cap() {
        use crate::gmail::messages_api::HARD_CAP;

        /// Serves `ids` back across as many `messages.list` pages as it
        /// takes, terminating only when exhausted — mirrors real Gmail
        /// pagination rather than hand-mounting one `Mock` per page.
        struct SequentialIdPages {
            ids: Vec<String>,
            calls: std::sync::atomic::AtomicUsize,
        }
        impl wiremock::Respond for SequentialIdPages {
            fn respond(&self, _req: &wiremock::Request) -> wiremock::ResponseTemplate {
                let call = self.calls.fetch_add(1, Ordering::SeqCst);
                let page_size = 500usize;
                let start = call * page_size;
                let page: Vec<serde_json::Value> = self
                    .ids
                    .iter()
                    .skip(start)
                    .take(page_size)
                    .map(|id| serde_json::json!({"id": id, "threadId": "t1"}))
                    .collect();
                let mut body = serde_json::json!({"messages": page});
                if start + page_size < self.ids.len() {
                    body["nextPageToken"] = serde_json::json!(format!("token-{}", call + 1));
                }
                wiremock::ResponseTemplate::new(200).set_body_json(body)
            }
        }

        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        // One more id than `HARD_CAP`. Before the fix, `run_full_sync`'s
        // listing truncated to `HARD_CAP`, so every id past that point
        // would have been wrongly soft-deleted below, even though the
        // listing (mocked here to return every one of them) says otherwise.
        let ids: Vec<String> = (0..HARD_CAP + 10).map(|i| format!("m{i}")).collect();
        let mut manifest = Manifest::default();
        for id in &ids {
            let path = shard_path(&output_dir, id, None);
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            std::fs::write(&path, b"From: a@example.com\r\n\r\nbody").unwrap();
            manifest.upsert(ManifestRecord {
                id: id.clone(),
                thread_id: Some("t1".to_string()),
                label_ids: vec!["INBOX".to_string()],
                internal_date: None,
                subject: None,
                from: None,
                to: None,
                rfc822_msgid: None,
                in_reply_to: None,
                references: None,
                attachment_count: 0,
                attachment_filenames: Vec::new(),
                path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
                size: 4,
                history_id: Some("1".to_string()),
                deleted_at: None,
                excluded_labels: Vec::new(),
            });
        }
        manifest.save(&manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(SequentialIdPages {
                ids: ids.clone(),
                calls: std::sync::atomic::AtomicUsize::new(0),
            })
            .mount(&server)
            .await;
        // Deliberately no mock for any `messages/<id>?format=raw` fetch —
        // every id is already archived on disk, so a re-fetch here would
        // mean the presence-on-disk check regressed too.

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty(), "no id should need re-fetching");
        assert!(
            !report
                .actions
                .iter()
                .any(|a| matches!(a, SyncAction::Deleted { .. })),
            "no already-archived message should be marked deleted just because it fell \
             outside a truncated listing"
        );
        let manifest_after = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert_eq!(
            manifest_after.ids_not_deleted().count(),
            ids.len(),
            "every id should still be present and not soft-deleted"
        );
    }

    // ── manifest checkpointing during the fetch loop (#1467) ─────────────

    #[tokio::test]
    async fn fetch_and_archive_messages_checkpoints_the_manifest_across_multiple_intervals() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let total = MANIFEST_CHECKPOINT_INTERVAL * 2 + 5;
        let ids: Vec<String> = (0..total).map(|i| format!("m{i}")).collect();
        for id in &ids {
            mount_raw_get(&server, id, "Hello").await;
        }

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let opts = SyncOptions {
            output_dir: output_dir.clone(),
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 20,
            dry_run: false,
            extract_attachments: false,
            shared_pool: None,
        };

        fetch_and_archive_messages(
            &client,
            &mut manifest,
            &ids,
            &limiter,
            &opts,
            &mut report,
            None,
        )
        .await
        .unwrap();
        assert!(report.errors.is_empty());

        // Two checkpoint boundaries' worth of records should already have
        // landed on disk *before* any final save — proving multiple
        // checkpoints accumulate correctly rather than each one clobbering
        // the last. `buffer_unordered` completes out of dispatch order, so
        // this checks a count, not which specific ids made it.
        let checkpointed = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let expected_checkpointed =
            (total / MANIFEST_CHECKPOINT_INTERVAL) * MANIFEST_CHECKPOINT_INTERVAL;
        assert_eq!(
            checkpointed.ids_not_deleted().count(),
            expected_checkpointed,
            "expected exactly two checkpoints' worth of records on disk before any final save"
        );

        // The final partial interval (the last 5 ids) is only on disk once
        // the caller does its own final save — the same thing `run_sync`
        // always does after this function returns.
        manifest.save(&manifest_path(&output_dir)).unwrap();
        let final_on_disk = Manifest::load(&manifest_path(&output_dir)).unwrap();
        for id in &ids {
            assert!(
                final_on_disk.get(id).is_some(),
                "missing {id} after the final save"
            );
        }
    }

    #[tokio::test]
    async fn fetch_and_archive_messages_checkpoints_before_the_whole_batch_completes() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let fast_ids: Vec<String> = (0..MANIFEST_CHECKPOINT_INTERVAL)
            .map(|i| format!("fast{i}"))
            .collect();
        let slow_ids: Vec<String> = (0..5).map(|i| format!("slow{i}")).collect();
        for id in &fast_ids {
            mount_raw_get(&server, id, "Hello").await;
        }
        for id in &slow_ids {
            wiremock::Mock::given(wiremock::matchers::method("GET"))
                .and(wiremock::matchers::path(format!(
                    "/gmail/v1/users/me/messages/{id}"
                )))
                .and(wiremock::matchers::query_param("format", "raw"))
                .respond_with(
                    wiremock::ResponseTemplate::new(200)
                        .set_body_json(serde_json::json!({
                            "id": id, "threadId": "t1", "labelIds": ["INBOX"],
                            "internalDate": "1700000000000", "historyId": "500",
                            "raw": raw_message_body(id, "Slow"),
                        }))
                        .set_delay(std::time::Duration::from_secs(3600)),
                )
                .mount(&server)
                .await;
        }

        let mut ids = fast_ids.clone();
        ids.extend(slow_ids.clone());
        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let opts = SyncOptions {
            output_dir: output_dir.clone(),
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 20,
            dry_run: false,
            extract_attachments: false,
            shared_pool: None,
        };

        // The slow ids' 3600s delay never elapses within this test, so the
        // only way `poll_checkpoint` can win this race is if the manifest
        // was actually flushed to disk *during* the fetch loop — proving
        // checkpointing isn't just an artifact of the final save `run_sync`
        // does once this function returns.
        let poll_checkpoint = async {
            loop {
                if let Ok(on_disk) = Manifest::load(&manifest_path(&output_dir)) {
                    if fast_ids.iter().all(|id| on_disk.get(id).is_some()) {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            }
        };

        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            tokio::select! {
                _ = fetch_and_archive_messages(
                    &client, &mut manifest, &ids, &limiter, &opts, &mut report, None,
                ) => {
                    panic!(
                        "fetch_and_archive_messages returned before the slow ids' delay could \
                         possibly elapse — checkpointing must have regressed"
                    );
                }
                () = poll_checkpoint => {}
            }
        })
        .await
        .expect("manifest was never checkpointed for the fast batch within 10s");

        let on_disk = Manifest::load(&manifest_path(&output_dir)).unwrap();
        for id in &fast_ids {
            assert!(
                on_disk.get(id).is_some(),
                "checkpoint should have covered {id}"
            );
        }
    }

    // ── pipelined listing+fetch and progress events (#1502) ──────────────

    #[tokio::test]
    async fn fetch_and_archive_messages_streaming_checkpoints_the_manifest_across_multiple_intervals(
    ) {
        // Mirrors `fetch_and_archive_messages_checkpoints_the_manifest_across_multiple_intervals`
        // above, but feeds ids through a channel instead of a slice — the
        // highest-risk piece of new async control flow (the `select!` pump)
        // must preserve the exact same checkpointing behavior.
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let total = MANIFEST_CHECKPOINT_INTERVAL * 2 + 5;
        let ids: Vec<String> = (0..total).map(|i| format!("m{i}")).collect();
        for id in &ids {
            mount_raw_get(&server, id, "Hello").await;
        }

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let opts = SyncOptions {
            output_dir: output_dir.clone(),
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 20,
            dry_run: false,
            extract_attachments: false,
            shared_pool: None,
        };

        let (ids_tx, ids_rx) = mpsc::unbounded_channel();
        for id in &ids {
            ids_tx.send(id.clone()).unwrap();
        }
        drop(ids_tx);

        let listed_ids = fetch_and_archive_messages_streaming(
            &client,
            &mut manifest,
            ids_rx,
            &limiter,
            &opts,
            &mut report,
            None,
        )
        .await
        .unwrap();
        assert!(report.errors.is_empty());
        assert_eq!(listed_ids.len(), total);

        let checkpointed = Manifest::load(&manifest_path(&output_dir)).unwrap();
        let expected_checkpointed =
            (total / MANIFEST_CHECKPOINT_INTERVAL) * MANIFEST_CHECKPOINT_INTERVAL;
        assert_eq!(
            checkpointed.ids_not_deleted().count(),
            expected_checkpointed,
            "expected exactly two checkpoints' worth of records on disk before any final save"
        );

        manifest.save(&manifest_path(&output_dir)).unwrap();
        let final_on_disk = Manifest::load(&manifest_path(&output_dir)).unwrap();
        for id in &ids {
            assert!(
                final_on_disk.get(id).is_some(),
                "missing {id} after the final save"
            );
        }
    }

    #[tokio::test]
    async fn fetch_and_archive_messages_streaming_emits_queued_and_completed_progress_events() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        mount_raw_get(&server, "ok1", "Hello").await;
        mount_raw_get(&server, "ok2", "Hello").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/bad1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let opts = opts(output_dir.clone());

        let (ids_tx, ids_rx) = mpsc::unbounded_channel();
        for id in ["ok1", "ok2", "bad1"] {
            ids_tx.send(id.to_string()).unwrap();
        }
        drop(ids_tx);

        let (progress_tx, mut progress_rx) = mpsc::unbounded_channel();
        fetch_and_archive_messages_streaming(
            &client,
            &mut manifest,
            ids_rx,
            &limiter,
            &opts,
            &mut report,
            Some(&progress_tx),
        )
        .await
        .unwrap();
        drop(progress_tx);

        let mut events = Vec::new();
        while let Some(event) = progress_rx.recv().await {
            events.push(event);
        }

        assert_eq!(report.errors.len(), 1);
        let queued = events
            .iter()
            .filter(|e| matches!(e, SyncProgressEvent::FetchQueued))
            .count();
        let completed_ok = events
            .iter()
            .filter(|e| matches!(e, SyncProgressEvent::FetchCompleted { failed: false }))
            .count();
        let completed_failed = events
            .iter()
            .filter(|e| matches!(e, SyncProgressEvent::FetchCompleted { failed: true }))
            .count();
        assert_eq!(queued, 3, "one FetchQueued per dispatched fetch");
        assert_eq!(completed_ok, 2);
        assert_eq!(completed_failed, 1);
    }

    #[tokio::test]
    async fn run_sync_with_progress_emits_listing_events_ending_in_listing_done() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "500").await;
        mount_message_list(&server, &["m1", "m2"]).await;
        mount_raw_get(&server, "m1", "Hello").await;
        mount_raw_get(&server, "m2", "Hello").await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");

        let (tx, mut rx) = mpsc::unbounded_channel();
        let report = run_sync_with_progress(&client, &opts(output_dir), Some(&tx))
            .await
            .unwrap();
        drop(tx);

        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }

        assert!(report.errors.is_empty());
        assert!(
            events
                .iter()
                .any(|e| matches!(e, SyncProgressEvent::ListingPage { .. })),
            "expected at least one ListingPage event"
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::ListingDone))
                .count(),
            1,
            "expected exactly one ListingDone event"
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::FetchQueued))
                .count(),
            2
        );
        assert_eq!(
            events
                .iter()
                .filter(|e| matches!(e, SyncProgressEvent::FetchCompleted { failed: false }))
                .count(),
            2
        );
        // Both sides of the `tokio::join!` have already finished by the
        // time `run_full_sync` sends `ListingDone` — it's always the last
        // event this run emits.
        assert!(matches!(
            events.last(),
            Some(SyncProgressEvent::ListingDone)
        ));
    }

    #[tokio::test]
    async fn run_sync_full_reconciliation_undeletes_reappeared_mail_and_deletes_vanished_mail() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_profile(&server, "user@example.com", "999").await;
        // Only "back" is listed as still present on the server; "gone" no
        // longer appears at all.
        mount_message_list(&server, &["back"]).await;
        mount_raw_get(&server, "back", "Hello").await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let back_path = shard_path(&output_dir, "back", None);
        std::fs::create_dir_all(back_path.parent().unwrap()).unwrap();
        std::fs::write(&back_path, b"From: a@example.com\r\n\r\nbody").unwrap();
        let gone_path = shard_path(&output_dir, "gone", None);
        std::fs::write(&gone_path, b"From: a@example.com\r\n\r\nbody").unwrap();

        let mut manifest = Manifest::default();
        manifest.upsert(ManifestRecord {
            id: "back".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: back_path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: 4,
            history_id: Some("1".to_string()),
            // Previously soft-deleted; the server lists it again this run.
            deleted_at: Some(Utc::now()),
            excluded_labels: Vec::new(),
        });
        manifest.upsert(ManifestRecord {
            id: "gone".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: gone_path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: 4,
            history_id: Some("1".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();

        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

        assert!(report.errors.is_empty());
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Undeleted { id } if id == "back")));
        assert!(report
            .actions
            .iter()
            .any(|a| matches!(a, SyncAction::Deleted { id } if id == "gone")));

        let on_disk = Manifest::load(&manifest_path(&output_dir)).unwrap();
        assert!(on_disk.get("back").unwrap().deleted_at.is_none());
        assert!(on_disk.get("gone").unwrap().deleted_at.is_some());
    }

    // ── --dry-run reconciliation reports planned actions (#1467) ────────

    #[tokio::test]
    async fn run_sync_dry_run_reconciliation_reports_would_delete_without_mutating_manifest() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let path = shard_path(&output_dir, "m1", None);
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(&path, b"From: a@example.com\r\n\r\nbody").unwrap();
        let mut manifest = Manifest::default();
        manifest.upsert(ManifestRecord {
            id: "m1".to_string(),
            thread_id: Some("t1".to_string()),
            label_ids: vec!["INBOX".to_string()],
            internal_date: None,
            subject: None,
            from: None,
            to: None,
            rfc822_msgid: None,
            in_reply_to: None,
            references: None,
            attachment_count: 0,
            attachment_filenames: Vec::new(),
            path: path.strip_prefix(&output_dir).unwrap().to_path_buf(),
            size: 4,
            history_id: Some("1".to_string()),
            deleted_at: None,
            excluded_labels: Vec::new(),
        });
        manifest.save(&manifest_path(&output_dir)).unwrap();
        let manifest_bytes_before = std::fs::read(manifest_path(&output_dir)).unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        mount_message_list(&server, &[]).await; // m1 no longer appears server-side

        let report = run_sync(
            &client,
            &SyncOptions {
                output_dir: output_dir.clone(),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                concurrency: 4,
                dry_run: true,
                extract_attachments: false,
                shared_pool: None,
            },
        )
        .await
        .unwrap();

        assert!(matches!(
            report.actions.as_slice(),
            [SyncAction::WouldDelete { id }] if id == "m1"
        ));
        assert_eq!(
            std::fs::read(manifest_path(&output_dir)).unwrap(),
            manifest_bytes_before,
            "dry-run must not mutate the manifest, in memory or on disk"
        );
    }

    // ── 404 reason-check tightening (#1467) ──────────────────────────────

    #[tokio::test]
    async fn run_sync_404_with_a_different_reason_is_not_treated_as_history_not_found() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &ArchiveState {
                history_id: "1".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: Utc::now(),
                query: None,
                pending_fetch: Vec::new(),
            },
            &state_path(&output_dir),
        )
        .unwrap();

        mount_profile(&server, "user@example.com", "999").await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(404).set_body_json(serde_json::json!({
                    "error": {"message": "Backend Error", "errors": [{"reason": "backendError"}]}
                })),
            )
            .mount(&server)
            .await;
        // Deliberately no mock for messages.list — a wrongly-triggered
        // reconciliation would attempt one and fail differently than
        // asserted below.

        let err = run_sync(&client, &opts(output_dir.clone()))
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("404"));
        assert!(msg.contains("backendError"));

        match state::load(&state_path(&output_dir)) {
            LoadOutcome::Present(s) => {
                assert_eq!(
                    s.history_id, "1",
                    "never treated as a reconciliation trigger"
                );
            }
            _ => panic!("expected the original state to survive"),
        }
    }

    // ── shared concurrency pool for `sync-all` (ADR-0068) ────────────────

    #[tokio::test]
    async fn shared_pool_of_one_never_lets_two_fetches_be_in_flight_at_once() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        // Both ids' responses never arrive within this test — the only way
        // a second `messages.get` request could ever reach the server is if
        // the shared semaphore (sized to 1) incorrectly let a second fetch
        // start before the first one's permit was released.
        let slow_ids = ["slow0", "slow1"];
        for id in slow_ids {
            wiremock::Mock::given(wiremock::matchers::method("GET"))
                .and(wiremock::matchers::path(format!(
                    "/gmail/v1/users/me/messages/{id}"
                )))
                .and(wiremock::matchers::query_param("format", "raw"))
                .respond_with(
                    wiremock::ResponseTemplate::new(200)
                        .set_body_json(serde_json::json!({
                            "id": id, "threadId": "t1", "labelIds": ["INBOX"],
                            "internalDate": "1700000000000", "historyId": "500",
                            "raw": raw_message_body(id, "Slow"),
                        }))
                        .set_delay(std::time::Duration::from_secs(3600)),
                )
                .mount(&server)
                .await;
        }

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let ids: Vec<String> = slow_ids.iter().copied().map(str::to_string).collect();
        let opts = SyncOptions {
            output_dir: output_dir.clone(),
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            // A generous *local* concurrency: only the shared pool should
            // be the thing capping in-flight requests to 1 here.
            concurrency: 20,
            dry_run: false,
            extract_attachments: false,
            shared_pool: Some(Arc::new(Semaphore::new(1))),
        };

        let never_more_than_one_in_flight = async {
            // Wait for the one fetch the semaphore should admit.
            loop {
                let received = server.received_requests().await.unwrap_or_default();
                if received
                    .iter()
                    .any(|r| r.url.path().contains("/messages/slow"))
                {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            }
            // Give a broken (unbounded) implementation ample opportunity to
            // also have started the second fetch by now.
            tokio::time::sleep(std::time::Duration::from_millis(300)).await;
            let received = server.received_requests().await.unwrap_or_default();
            let in_flight = received
                .iter()
                .filter(|r| r.url.path().contains("/messages/slow"))
                .count();
            assert_eq!(
                in_flight, 1,
                "shared_pool sized to 1 should never admit a second concurrent fetch"
            );
        };

        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            tokio::select! {
                _ = fetch_and_archive_messages(
                    &client, &mut manifest, &ids, &limiter, &opts, &mut report, None,
                ) => {
                    panic!(
                        "fetch_and_archive_messages returned before the slow ids' 3600s delay \
                         could possibly elapse"
                    );
                }
                () = never_more_than_one_in_flight => {}
            }
        })
        .await
        .expect("the single admitted fetch was never observed within 5s");
    }

    // A closed `Semaphore` can't happen through any code path today (nothing
    // ever calls `Semaphore::close`) — these two tests exercise the
    // defensive `pool.acquire()` error arms directly, the same way a real
    // `AcquireError` would surface, rather than leaving them unreachable in
    // practice and untested in principle.

    #[tokio::test]
    async fn fetch_and_archive_messages_streaming_reports_error_when_shared_pool_closed() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let pool = Arc::new(Semaphore::new(1));
        pool.close();
        let opts = SyncOptions {
            output_dir,
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 4,
            dry_run: false,
            extract_attachments: false,
            shared_pool: Some(pool),
        };

        let (ids_tx, ids_rx) = mpsc::unbounded_channel();
        ids_tx.send("id1".to_string()).unwrap();
        drop(ids_tx);

        fetch_and_archive_messages_streaming(
            &client,
            &mut manifest,
            ids_rx,
            &limiter,
            &opts,
            &mut report,
            None,
        )
        .await
        .unwrap();

        assert_eq!(report.errors.len(), 1);
        assert!(report.errors[0]
            .reason
            .contains("sync-all's shared semaphore closed unexpectedly"));
    }

    #[tokio::test]
    async fn fetch_and_archive_messages_reports_error_when_shared_pool_closed() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();

        let mut manifest = Manifest::default();
        let limiter = TokenBucket::new(1_000_000, 1_000_000);
        let mut report = SyncReport::default();
        let pool = Arc::new(Semaphore::new(1));
        pool.close();
        let opts = SyncOptions {
            output_dir,
            query: None,
            exclude_labels: Vec::new(),
            full: false,
            concurrency: 4,
            dry_run: false,
            extract_attachments: false,
            shared_pool: Some(pool),
        };

        fetch_and_archive_messages(
            &client,
            &mut manifest,
            &["id1".to_string()],
            &limiter,
            &opts,
            &mut report,
            None,
        )
        .await
        .unwrap();

        assert_eq!(report.errors.len(), 1);
        assert!(report.errors[0]
            .reason
            .contains("sync-all's shared semaphore closed unexpectedly"));
    }
}