//! Single-cycle sync runner.
//!
//! This module owns the per-cycle boundary that sits between the outer
//! process/watch loop and the download engine: mode choice, enumeration config
//! hash handling, grouping preload, download dispatch, zone-token writes, and
//! the aggregate cycle result.

use std::sync::Arc;

use tokio_util::sync::CancellationToken;

use crate::auth;
use crate::config;
use crate::download;
use crate::state;

/// Per-library state: zone name, sync token key, and resolved album plan.
pub(crate) struct LibraryState {
    pub(crate) library: crate::icloud::photos::PhotoLibrary,
    pub(crate) cross_zone_libraries: Vec<crate::icloud::photos::PhotoLibrary>,
    pub(crate) pass_scope: crate::commands::PassScope,
    pub(crate) zone_name: String,
    pub(crate) sync_token_key: String,
    /// Ordered list of download passes. Each pass carries its own
    /// exclude-asset-ids set. See [`crate::commands::AlbumPlan`].
    pub(crate) plan: crate::commands::AlbumPlan,
    /// True when `resolve_passes` failed at the end of the prior cycle and
    /// the plan above is the previous cycle's stale snapshot. Album
    /// membership data captured under a stale plan can route assets to the
    /// wrong pass (e.g. an asset added to a newly-created album shows up in
    /// the unfiled pass), so any cycle that consumes a stale plan must not
    /// advance the sync token for any zone -- doing so would skip the
    /// change events those assets generated and leave `asset_albums`
    /// permanently incomplete.
    pub(crate) plan_is_stale: bool,
    /// True after an idle watch sleep. Refreshing only when a later
    /// `changes/database` pre-check finds relevant work avoids burning album
    /// listing calls on quiet watch cycles.
    pub(crate) plan_needs_refresh: bool,
}

/// Metadata key holding the SHA-256 of the enumeration-affecting subset of
/// the user's download config. Distinct from the path-affecting
/// `config_hash` consumed by the download pipeline; using a single key for
/// both would cause each cycle to overwrite the other's value and
/// permanently invalidate incremental sync.
pub(crate) const ENUM_CONFIG_HASH_KEY: &str = "enum_config_hash";
pub(crate) const PENDING_ENUM_CONFIG_HASH_KEY: &str = "pending_enum_config_hash";
pub(crate) const PENDING_DOWNLOAD_CONFIG_HASH_KEY: &str = "pending_download_config_hash";
const PENDING_ZONE_TOKEN_PREFIX: &str = "pending_sync_token:";
const LAST_CHECKPOINT_STATUS_KEY: &str = "last_checkpoint_status";
const LAST_RECOVERY_ACTION_KEY: &str = "last_recovery_action";
const LAST_FULL_ENUMERATION_REASON_KEY: &str = "last_full_enumeration_reason";

/// Prefix for every per-zone CloudKit sync token row in the metadata table.
pub(crate) const SYNC_TOKEN_PREFIX: &str = "sync_token:";
// Legacy inventory_anchor: rows lack coverage proof and are never read or migrated.
const INVENTORY_ANCHOR_PREFIX: &str = "inventory_anchor_v1:";
const INVENTORY_ANCHOR_VERSION: u8 = 1;
/// Presence is a durable obligation, including unknown future payloads. Only a
/// proven FINISHED full inventory may remove it with checkpoint promotion.
pub(crate) const INDEXING_RECONCILIATION_PREFIX: &str = "indexing_reconciliation_required:";

pub(crate) fn indexing_reconciliation_key(zone: &str) -> String {
    format!("{INDEXING_RECONCILIATION_PREFIX}{zone}")
}

/// A qualified inventory retains the old cursor and schedules its replay in a
/// later trusted cycle. No deletion-capable bridge runs inside the held phase.
pub(crate) const INDEXING_REPLAY_PREFIX: &str = "indexing_replay_required:";

pub(crate) fn indexing_replay_key(zone: &str) -> String {
    format!("{INDEXING_REPLAY_PREFIX}{zone}")
}

/// Managed qualification did not queue files. This obligation forces a full
/// managed publication pass before retained-cursor replay may discharge it.
pub(crate) fn indexing_materialization_key(zone: &str) -> String {
    format!("indexing_materialization_required:{zone}")
}

/// Observe readiness and durably retain every RUNNING observation, even if a
/// later readiness request fails before enumeration starts.
pub(crate) async fn prepare_indexing_reconciliation(
    library: &crate::icloud::photos::PhotoLibrary,
    db: Option<&dyn download::DownloadStore>,
    dry_run: bool,
) -> anyhow::Result<(bool, Option<String>)> {
    let key = indexing_reconciliation_key(library.zone_name());
    let mut hold = match db {
        Some(db) => db.get_metadata(&key).await?,
        None => None,
    };
    if !library.last_indexing_finished() && !dry_run {
        let db = db.ok_or_else(|| {
            anyhow::anyhow!("Provisional indexing downloads require durable state")
        })?;
        if hold.is_none() {
            db.set_metadata(&key, "1").await?;
            hold = Some("1".to_owned());
        }
    }
    let finished = library.check_indexing().await?;
    if !finished && !dry_run {
        let db = db.ok_or_else(|| {
            anyhow::anyhow!("Provisional indexing downloads require durable state")
        })?;
        if hold.is_none() {
            db.set_metadata(&key, "1").await?;
            hold = Some("1".to_owned());
        }
    }
    Ok((finished, hold))
}

/// Cross-zone album hydration contributes to the selected listing too. Any
/// unfinished source makes the selected zone provisional until all are ready.
pub(crate) async fn prepare_library_indexing(
    library: &LibraryState,
    db: Option<&dyn download::DownloadStore>,
    dry_run: bool,
) -> anyhow::Result<(bool, Option<String>)> {
    let (mut finished, mut hold) =
        prepare_indexing_reconciliation(&library.library, db, dry_run).await?;
    for source in &library.cross_zone_libraries {
        if source.zone_name() == library.zone_name {
            continue;
        }
        let source_was_running = !source.last_indexing_finished();
        let (source_finished, _) = prepare_indexing_reconciliation(source, db, dry_run).await?;
        finished &= source_finished;
        if (source_was_running || !source_finished) && hold.is_none() && !dry_run {
            let db = db.ok_or_else(|| {
                anyhow::anyhow!("Provisional indexing downloads require durable state")
            })?;
            db.set_metadata(&indexing_reconciliation_key(&library.zone_name), "1")
                .await?;
            hold = Some("1".to_owned());
        }
    }
    Ok((finished, hold))
}

pub(crate) const RETAINED_CHECKPOINT_HOLD_PREFIX: &str = "retained_checkpoint_hold:";
const RETAINED_CHECKPOINT_RETRY_SECONDS: i64 = 60 * 60;
const RETAINED_CHECKPOINT_MAX_ATTEMPTS: u8 = 3;

const fn first_retained_checkpoint_attempt() -> u8 {
    1
}

/// A delay is bound to the retained boundary and the requested policy. It never
/// certifies inventory coverage or historical identity/deletion resolution.
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct RetainedCheckpointHold {
    version: u8,
    prior_cursor_sha256: String,
    enum_config_hash: String,
    preservation_config_hash: Option<String>,
    next_attempt_at: i64,
    #[serde(default = "first_retained_checkpoint_attempt")]
    attempt_count: u8,
}

impl RetainedCheckpointHold {
    fn new(prior_cursor: &str, enum_hash: &str, preservation_hash: Option<&str>) -> Self {
        use sha2::{Digest, Sha256};
        Self {
            version: 1,
            attempt_count: 1,
            prior_cursor_sha256: data_encoding::HEXLOWER.encode(&Sha256::digest(prior_cursor)),
            enum_config_hash: enum_hash.to_owned(),
            preservation_config_hash: preservation_hash.map(str::to_owned),
            next_attempt_at: chrono::Utc::now()
                .timestamp()
                .saturating_add(RETAINED_CHECKPOINT_RETRY_SECONDS),
        }
    }

    fn matches(&self, current: &Self) -> bool {
        self.version == 1
            && self.attempt_count > 0
            && self.prior_cursor_sha256 == current.prior_cursor_sha256
            && self.enum_config_hash == current.enum_config_hash
            && self.preservation_config_hash == current.preservation_config_hash
    }

    fn deferred(&self, now: i64) -> bool {
        self.next_attempt_at > now
            && self.next_attempt_at <= now.saturating_add(RETAINED_CHECKPOINT_RETRY_SECONDS)
    }
}

pub(crate) fn retained_checkpoint_hold_action(encoded: &str) -> &'static str {
    match serde_json::from_str::<RetainedCheckpointHold>(encoded) {
        Ok(hold) if hold.version == 1 && hold.attempt_count > 0 => {
            if hold.attempt_count >= RETAINED_CHECKPOINT_MAX_ATTEMPTS {
                RecoveryAction::RetainedCheckpointRetryExhausted.as_str()
            } else {
                RecoveryAction::AwaitRetainedCheckpointEvidence.as_str()
            }
        }
        _ => RecoveryAction::RepairRetainedCheckpointEvidence.as_str(),
    }
}

fn retained_checkpoint_hold_key(zone: &str) -> String {
    format!("{RETAINED_CHECKPOINT_HOLD_PREFIX}{zone}")
}

fn block_for_retained_checkpoint(stats: &mut download::SyncStats) {
    stats.sync_token_blocked = true;
    stats.sync_token_blocked_reason = Some("retained_checkpoint_expired");
    stats.sync_token_blocked_source = Some("kei");
    stats.sync_token_blocked_explanation = Some(download::sync_token_blocked_explanation(
        "retained_checkpoint_expired",
    ));
}

async fn persist_retained_checkpoint_hold(
    db: &dyn download::DownloadStore,
    key: String,
    hold: &RetainedCheckpointHold,
    expected_metadata: Vec<(String, Option<String>)>,
) -> anyhow::Result<()> {
    db.commit_checkpoint_transition(state::CheckpointTransition {
        expected_metadata,
        legacy_preservation_proofs: Vec::new(),
        legacy_config_hash: None,
        sparse_identity_proofs: Vec::new(),
        metadata_updates: vec![
            (key, serde_json::to_string(hold)?),
            (
                LAST_CHECKPOINT_STATUS_KEY.to_owned(),
                "preserved".to_owned(),
            ),
            (
                LAST_RECOVERY_ACTION_KEY.to_owned(),
                RecoveryAction::AwaitRetainedCheckpointEvidence
                    .as_str()
                    .to_owned(),
            ),
        ],
        metadata_deletes: Vec::new(),
    })
    .await?;
    Ok(())
}
const INVENTORY_DROP_THRESHOLD_PERCENT: f64 = 5.0;

#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
struct InventoryAnchor {
    version: u8,
    scope: download::InventoryScope,
    api_total_at_start: u64,
    assets_seen: u64,
    completed_at: i64,
}

#[derive(Debug, Clone, Copy, PartialEq)]
struct InventoryDrop {
    previous_total: u64,
    current_total: u64,
    drop_assets: u64,
    drop_percent: f64,
}

/// Return the metadata key that stores the CloudKit sync token for a zone.
pub(crate) fn sync_token_key(zone_name: &str) -> String {
    format!("{SYNC_TOKEN_PREFIX}{zone_name}")
}

fn inventory_anchor_key(
    enum_config_hash: &str,
    zone_name: &str,
    scope: &download::InventoryScope,
) -> String {
    use sha2::{Digest, Sha256};
    let scope_hash = data_encoding::HEXLOWER.encode(&Sha256::digest(scope.0.as_bytes()));
    format!("{INVENTORY_ANCHOR_PREFIX}{enum_config_hash}:{zone_name}:{scope_hash}")
}

pub(crate) fn pending_zone_token_key(enum_config_hash: &str, zone_name: &str) -> String {
    format!("{PENDING_ZONE_TOKEN_PREFIX}{enum_config_hash}:{zone_name}")
}

#[allow(
    clippy::cast_precision_loss,
    reason = "inventory totals are operator diagnostics; f64 percent is only for warning/report context"
)]
fn classify_inventory_drop(previous_total: u64, current_total: u64) -> Option<InventoryDrop> {
    if previous_total == 0 || current_total >= previous_total {
        return None;
    }

    let drop_assets = previous_total - current_total;
    let drop_percent = (drop_assets as f64 / previous_total as f64) * 100.0;
    if drop_percent < INVENTORY_DROP_THRESHOLD_PERCENT {
        return None;
    }

    Some(InventoryDrop {
        previous_total,
        current_total,
        drop_assets,
        drop_percent,
    })
}

/// Outcome of a single sync cycle across all libraries.
#[derive(Debug)]
pub(crate) struct CycleResult {
    pub(crate) failed_count: usize,
    pub(crate) session_expired: bool,
    pub(crate) stats: download::SyncStats,
    pub(crate) db_sync_token_advance_safe: bool,
}

impl CycleResult {
    /// The database pre-check needs a clean aggregate cycle as well as safe
    /// selected-zone checkpoints. A durable transfer failure can satisfy the
    /// zone gate without satisfying this broader gate.
    #[must_use]
    pub(crate) fn can_advance_database_checkpoint(&self) -> bool {
        !self.session_expired
            && self.failed_count == 0
            && !self.stats.interrupted
            && self.db_sync_token_advance_safe
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointBasis {
    IncrementalDelta,
    CompleteInventory,
    InventoryWithDeltaBridge,
}

impl CheckpointBasis {
    const fn as_str(self) -> &'static str {
        match self {
            Self::IncrementalDelta => "incremental_delta",
            Self::CompleteInventory => "complete_inventory",
            Self::InventoryWithDeltaBridge => "inventory_with_delta_bridge",
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointHoldReason {
    DryRun,
    StalePassPlan,
    Interrupted,
    SessionExpired,
    EnumerationIncomplete,
    StateNotDurable,
    TokenProofIncomplete,
    LegacyPreservationIncomplete,
}

impl CheckpointHoldReason {
    const fn as_str(self) -> &'static str {
        match self {
            Self::DryRun => "dry_run",
            Self::StalePassPlan => "stale_pass_plan",
            Self::Interrupted => "interrupted",
            Self::SessionExpired => "session_expired",
            Self::EnumerationIncomplete => "enumeration_incomplete",
            Self::StateNotDurable => "state_not_durable",
            Self::TokenProofIncomplete => "token_proof_incomplete",
            Self::LegacyPreservationIncomplete => "legacy_preservation_incomplete",
        }
    }
}

pub(crate) use download::RecoveryAction;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SourceCheckpointDecision {
    Advance {
        token: String,
        basis: CheckpointBasis,
    },
    Preserve {
        reason: CheckpointHoldReason,
        recovery: RecoveryAction,
    },
}

/// Capture the owning decision without serializing its provider token.
pub(crate) fn checkpoint_support_fields(decision: &SourceCheckpointDecision) -> serde_json::Value {
    match decision {
        SourceCheckpointDecision::Advance { basis, .. } => serde_json::json!({
            "decision":"advanced", "basis":basis.as_str(), "recovery":"none"
        }),
        SourceCheckpointDecision::Preserve { reason, recovery } => serde_json::json!({
            "decision":"preserved", "reason":reason.as_str(), "recovery":recovery.as_str()
        }),
    }
}

fn observe_sync_component(
    zone: &str,
    component: &'static str,
    result: &download::SyncResult,
    passes: usize,
) {
    crate::support::observe_scoped(
        zone,
        None,
        "support_pass_completion_v1",
        serde_json::json!({
            "component":component, "count_scope":"selected_invocation", "passes":passes,
            "phase":if result.full_enumeration_ran {"full"} else {"incremental"},
            // These totals describe this invocation, before any bridge composition.
            // Query EOF evidence is recorded by the provider fetcher, not inferred
            // from absence of reported errors or widened into inventory coverage.
            "inventory_complete":serde_json::Value::Null,
            "selection_comparable":serde_json::Value::Null,
            "enumeration_incomplete":result.checkpoint.enumeration_incomplete,
            "enumeration_errors":result.checkpoint.enumeration_errors,
            "identity_incomplete":result.checkpoint.identity_incomplete,
            "interrupted":result.checkpoint.interrupted,
            "api_total":result.stats.api_total_at_start, "assets_seen":result.stats.assets_seen,
            "outcome":match result.outcome {
                download::DownloadOutcome::Success => "success",
                download::DownloadOutcome::SessionExpired { .. } => "session_expired",
                download::DownloadOutcome::PartialFailure { .. } => "partial_failure",
            },
            "token_present":result.sync_token.is_some(),
            "token_usable":result.sync_token.as_ref().is_some_and(|token| !token.trim().is_empty()),
            "sync_token_blocked":result.checkpoint.sync_token_blocked,
            "sync_token_blocked_reason":result.stats.sync_token_blocked_reason,
            "sync_token_blocked_source":result.stats.sync_token_blocked_source,
            "sync_token_expected_receivers":result.stats.sync_token_expected_receivers,
            "sync_token_receivers_with_token":result.stats.sync_token_receivers_with_token,
            "sync_token_receivers_missing":result.stats.sync_token_receivers_missing,
            "sync_token_receivers_blank":result.stats.sync_token_receivers_blank,
            "sync_token_receivers_dropped":result.stats.sync_token_receivers_dropped,
            "sync_token_unique_values":result.stats.sync_token_unique_values,
            "same_cycle_recovery_attempts":result.stats.same_cycle_recovery_attempts,
            "same_cycle_recovery_successes":result.stats.same_cycle_recovery_successes,
            "completed_delta_replay":result.checkpoint.completed_delta_replay,
            "retained_token_rejected":result.checkpoint.retained_token_rejected,
        }),
    );
}

fn source_checkpoint_decision(
    result: &download::SyncResult,
    dry_run: bool,
    stale_pass_plan: bool,
    basis: CheckpointBasis,
) -> SourceCheckpointDecision {
    // CONTRACT: SOURCE_CHECKPOINT_REQUIRES_DURABLE_RECOVERY
    let preserve = |reason, recovery| SourceCheckpointDecision::Preserve { reason, recovery };
    if dry_run {
        return preserve(CheckpointHoldReason::DryRun, RecoveryAction::Stop);
    }
    if stale_pass_plan {
        return preserve(
            CheckpointHoldReason::StalePassPlan,
            RecoveryAction::ReplayFromPriorToken,
        );
    }
    if matches!(
        result.outcome,
        download::DownloadOutcome::SessionExpired { .. }
    ) {
        return preserve(
            CheckpointHoldReason::SessionExpired,
            RecoveryAction::Reauthenticate,
        );
    }
    if result.checkpoint.interrupted {
        return preserve(
            CheckpointHoldReason::Interrupted,
            RecoveryAction::ReplayFromPriorToken,
        );
    }
    if result.checkpoint.enumeration_errors > 0 || result.checkpoint.enumeration_incomplete {
        return preserve(
            CheckpointHoldReason::EnumerationIncomplete,
            RecoveryAction::ContinueTail,
        );
    }
    if result.checkpoint.state_write_failures > 0 {
        return preserve(
            CheckpointHoldReason::StateNotDurable,
            RecoveryAction::ReplayFromPriorToken,
        );
    }
    let Some(token) = result
        .sync_token
        .as_ref()
        .filter(|token| !token.trim().is_empty())
    else {
        let recovery = if !result.checkpoint.retry_passes.is_empty() {
            RecoveryAction::RetryPasses(result.checkpoint.retry_passes.clone())
        } else if !result.checkpoint.revalidate_records.is_empty() {
            RecoveryAction::RevalidateRecords(result.checkpoint.revalidate_records.clone())
        } else {
            RecoveryAction::ReplayFromPriorToken
        };
        return preserve(CheckpointHoldReason::TokenProofIncomplete, recovery);
    };
    if result.checkpoint.sync_token_blocked || result.checkpoint.identity_incomplete {
        let recovery = if result.checkpoint.retry_passes.is_empty() {
            RecoveryAction::ReplayFromPriorToken
        } else {
            RecoveryAction::RetryPasses(result.checkpoint.retry_passes.clone())
        };
        return preserve(CheckpointHoldReason::TokenProofIncomplete, recovery);
    }
    SourceCheckpointDecision::Advance {
        token: token.clone(),
        basis,
    }
}

fn merge_download_outcomes(
    first: &download::DownloadOutcome,
    second: &download::DownloadOutcome,
) -> download::DownloadOutcome {
    match (first, second) {
        (download::DownloadOutcome::SessionExpired { auth_error_count }, _)
        | (_, download::DownloadOutcome::SessionExpired { auth_error_count }) => {
            download::DownloadOutcome::SessionExpired {
                auth_error_count: *auth_error_count,
            }
        }
        (
            download::DownloadOutcome::PartialFailure {
                failed_count: first,
            },
            download::DownloadOutcome::PartialFailure {
                failed_count: second,
            },
        ) => download::DownloadOutcome::PartialFailure {
            failed_count: first.saturating_add(*second),
        },
        (download::DownloadOutcome::PartialFailure { failed_count }, _)
        | (_, download::DownloadOutcome::PartialFailure { failed_count }) => {
            download::DownloadOutcome::PartialFailure {
                failed_count: *failed_count,
            }
        }
        (download::DownloadOutcome::Success, download::DownloadOutcome::Success) => {
            download::DownloadOutcome::Success
        }
    }
}

/// Decide whether the per-zone `sync_token` should be persisted to the state
/// DB after a download pass.
///
/// The contract is "advance only on full success and not in dry-run":
/// - On `PartialFailure`, a stored token would skip the failed assets on the
///   next incremental sync (they'd never appear in the delta again -- silent
///   data loss).
/// - On `SessionExpired`, the cycle aborts mid-stream; the token may be
///   stale or only reflect a subset of the work.
/// - In `--dry-run`, we promise to make no DB writes that survive the run
///   (apart from the `sync_runs` ledger). Advancing the token here would
///   silently break the next real sync.
///
/// The returned bool is the gate: callers still check that `sync_token` is
/// `Some(_)` and that a state DB is configured before persisting.
#[cfg(test)]
pub(crate) fn should_store_sync_token(outcome: &download::DownloadOutcome, dry_run: bool) -> bool {
    matches!(outcome, download::DownloadOutcome::Success) && !dry_run
}

/// Library-level gate that combines the per-library outcome check with the
/// stale-plan flag for the same zone.
///
/// A reused album plan can route assets created or moved between cycles to
/// the wrong pass, so the affected zone must not advance. Unaffected zones
/// can still store their own zone tokens; the broader database pre-check
/// token remains conservative until every selected zone is clean.
#[cfg(test)]
pub(crate) fn should_store_sync_token_for_cycle(
    outcome: &download::DownloadOutcome,
    dry_run: bool,
    library_plan_is_stale: bool,
) -> bool {
    should_store_sync_token(outcome, dry_run) && !library_plan_is_stale
}

fn has_active_passes(lib_state: &LibraryState) -> bool {
    !lib_state.plan.passes.is_empty()
}

fn should_warn_zero_assets(
    sync_result: &download::SyncResult,
    library_completed_without_errors: bool,
    run_mode: download::DownloadRunMode,
    is_retry_failed: bool,
    lib_state: &LibraryState,
) -> bool {
    sync_result.full_enumeration_ran
        && library_completed_without_errors
        && run_mode.downloads_files()
        && !is_retry_failed
        && sync_result.stats.assets_seen == 0
        && has_active_passes(lib_state)
}

async fn update_inventory_anchor_for_cycle<D>(
    db: &D,
    enum_config_hash: &str,
    zone_name: &str,
    inventory: &download::InventoryMeasurement,
    stats: &mut download::SyncStats,
) where
    D: state::SyncTokenStore + ?Sized,
{
    let current_total = inventory.total;
    let key = inventory_anchor_key(enum_config_hash, zone_name, &inventory.scope);
    match db.get_metadata(&key).await {
        Ok(Some(raw)) => match serde_json::from_str::<InventoryAnchor>(&raw) {
            Ok(anchor)
                if anchor.version == INVENTORY_ANCHOR_VERSION
                    && anchor.scope == inventory.scope =>
            {
                if let Some(drop) =
                    classify_inventory_drop(anchor.api_total_at_start, current_total)
                {
                    tracing::warn!(
                        library = zone_name,
                        previous_api_total = drop.previous_total,
                        current_api_total = drop.current_total,
                        drop_assets = drop.drop_assets,
                        drop_percent = drop.drop_percent,
                        threshold_percent = INVENTORY_DROP_THRESHOLD_PERCENT,
                        strict_inventory = false,
                        enum_config_hash,
                        "Inventory dropped below previous comparable full run"
                    );
                    stats.inventory_drop_warnings = stats.inventory_drop_warnings.saturating_add(1);
                    stats.inventory_drop_assets = drop.drop_assets;
                    stats.inventory_drop_percent = Some(drop.drop_percent);
                    stats.inventory_drop_previous_total = Some(drop.previous_total);
                    stats.inventory_drop_current_total = Some(drop.current_total);
                    stats.inventory_drop_library = Some(zone_name.to_string());
                }
            }
            Ok(_) => {}
            Err(e) => {
                tracing::debug!(
                    error = %e,
                    library = zone_name,
                    "Ignoring unreadable inventory anchor"
                );
            }
        },
        Ok(None) => {}
        Err(e) => {
            tracing::debug!(
                error = %e,
                library = zone_name,
                "Failed to read inventory anchor"
            );
            return;
        }
    }

    let anchor = InventoryAnchor {
        version: INVENTORY_ANCHOR_VERSION,
        scope: inventory.scope.clone(),
        api_total_at_start: current_total,
        assets_seen: inventory.assets_seen,
        completed_at: chrono::Utc::now().timestamp(),
    };
    match serde_json::to_string(&anchor) {
        Ok(value) => {
            if let Err(e) = db.set_metadata(&key, &value).await {
                tracing::debug!(
                    error = %e,
                    library = zone_name,
                    "Failed to persist inventory anchor"
                );
            }
        }
        Err(e) => {
            tracing::debug!(
                error = %e,
                library = zone_name,
                "Failed to serialize inventory anchor"
            );
        }
    }
}

/// Closure shape used to derive a per-library `DownloadConfig` from the
/// shared base config. Boxed dyn so `run_cycle` can accept a single
/// reference instead of a generic parameter (avoids reuse-by-monomorphization
/// blow-up in error messages).
pub(crate) type BuildDownloadConfigFn<'a> = dyn Fn(
        download::SyncMode,
        Arc<rustc_hash::FxHashSet<String>>,
        Arc<download::AssetGroupings>,
        Arc<str>,
    ) -> Arc<download::DownloadConfig>
    + 'a;

/// Outcome of [`check_and_persist_enum_config_hash`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum EnumConfigHashOutcome {
    /// No prior hash; current hash persisted. Sync tokens left alone:
    /// a first-run DB must not invalidate tokens another process may
    /// have written.
    Initial,
    Unchanged,
    /// Hash drifted; the candidate hash was staged while active tokens and
    /// the active hash remain unchanged until reconciliation succeeds.
    Changed,
    /// Hash drift was detected, but the pending reconciliation hash could
    /// not be staged.
    ChangedTokenPurgeFailed,
    /// The stored hash could not be read, so the cycle cannot prove whether
    /// existing sync tokens still match the current config.
    ReadFailed,
}

impl EnumConfigHashOutcome {
    fn must_force_full_sync(self) -> bool {
        matches!(
            self,
            Self::Changed | Self::ChangedTokenPurgeFailed | Self::ReadFailed
        )
    }
}

/// State of the path-affecting download config hash before a cycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DownloadConfigHashOutcome {
    Unchanged,
    Changed,
    ReadFailed,
    TokenPurgeFailed,
}

impl DownloadConfigHashOutcome {
    const fn must_force_full_sync(self) -> bool {
        matches!(self, Self::ReadFailed | Self::TokenPurgeFailed)
    }

    const fn token_purge_failed(self) -> bool {
        matches!(self, Self::TokenPurgeFailed)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SyncModeDecision {
    pub(crate) mode: download::SyncMode,
    pub(crate) full_enumeration_reason: Option<download::FullEnumerationReason>,
}

fn metadata_refresh_mode_decision(refresh_metadata: bool) -> Option<SyncModeDecision> {
    refresh_metadata.then_some(SyncModeDecision {
        mode: download::SyncMode::Full,
        full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
    })
}

/// Compare the current enumeration-config hash against the active value.
/// Drift is staged without deleting the last safe provider checkpoint.
pub(crate) async fn check_and_persist_enum_config_hash<D>(
    db: &D,
    current_hash: &str,
) -> EnumConfigHashOutcome
where
    D: state::SyncTokenStore + ?Sized,
{
    let stored_hash = match db.get_metadata(ENUM_CONFIG_HASH_KEY).await {
        Ok(hash) => hash,
        Err(e) => {
            tracing::warn!(error = %e, "Failed to read enum_config_hash");
            return EnumConfigHashOutcome::ReadFailed;
        }
    };
    let outcome = match stored_hash.as_deref() {
        Some(h) if h == current_hash => {
            if let Err(e) = db
                .delete_metadata_by_prefix(PENDING_ENUM_CONFIG_HASH_KEY)
                .await
            {
                tracing::debug!(error = %e, "Failed to clear reverted pending enum config hash");
            }
            if let Err(e) = db
                .delete_metadata_by_prefix(PENDING_ZONE_TOKEN_PREFIX)
                .await
            {
                tracing::debug!(error = %e, "Failed to clear reverted pending zone checkpoints");
            }
            return EnumConfigHashOutcome::Unchanged;
        }
        Some(_) => EnumConfigHashOutcome::Changed,
        None => EnumConfigHashOutcome::Initial,
    };

    let key = if matches!(outcome, EnumConfigHashOutcome::Changed) {
        tracing::info!(
            "Enumeration config changed since last sync; preserving active checkpoints and staging reconciliation"
        );
        match db.get_metadata(PENDING_ENUM_CONFIG_HASH_KEY).await {
            Ok(Some(pending_hash)) if pending_hash == current_hash => {}
            Ok(_) => {
                if let Err(e) = db
                    .delete_metadata_by_prefix(PENDING_ZONE_TOKEN_PREFIX)
                    .await
                {
                    tracing::warn!(error = %e, "Failed to clear checkpoints from superseded config reconciliation");
                    return EnumConfigHashOutcome::ChangedTokenPurgeFailed;
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "Failed to read pending enum config hash");
                return EnumConfigHashOutcome::ReadFailed;
            }
        }
        PENDING_ENUM_CONFIG_HASH_KEY
    } else {
        ENUM_CONFIG_HASH_KEY
    };
    if let Err(e) = db.set_metadata(key, current_hash).await {
        tracing::warn!(error = %e, key, "Failed to persist enum config hash state");
        if matches!(outcome, EnumConfigHashOutcome::Changed) {
            return EnumConfigHashOutcome::ChangedTokenPurgeFailed;
        }
    }
    outcome
}

/// Check the path-affecting download config hash without deleting provider
/// cursors. Reconciliation promotes the pending hash only after it succeeds.
///
/// This hash does not prove CloudKit cursor safety. It proves local
/// path/state trust: after a directory or filename-template change, an
/// incremental `/changes/zone` cycle would only see new deltas and could miss
/// already-known assets that must be written to the new local paths.
pub(crate) async fn check_download_config_hash_for_cycle<D>(
    db: &D,
    current_hash: &str,
    legacy_hash: &str,
) -> DownloadConfigHashOutcome
where
    D: state::SyncTokenStore + ?Sized,
{
    match db.get_metadata(download::DOWNLOAD_CONFIG_HASH_KEY).await {
        Ok(Some(stored)) if stored == current_hash => {
            if let Err(e) = db
                .delete_metadata_by_prefix(PENDING_DOWNLOAD_CONFIG_HASH_KEY)
                .await
            {
                tracing::debug!(error = %e, "Failed to clear reverted pending download config hash");
            }
            DownloadConfigHashOutcome::Unchanged
        }
        Ok(Some(stored)) if stored == legacy_hash => {
            if let Err(e) = db
                .commit_checkpoint_transition(state::CheckpointTransition {
                    expected_metadata: Vec::new(),
                    legacy_preservation_proofs: Vec::new(),
                    legacy_config_hash: None,
                    sparse_identity_proofs: Vec::new(),
                    metadata_updates: vec![(
                        download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
                        current_hash.to_owned(),
                    )],
                    metadata_deletes: vec![PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned()],
                })
                .await
            {
                tracing::warn!(error = %e, "Failed to migrate legacy download config_hash");
                return DownloadConfigHashOutcome::ReadFailed;
            }
            tracing::info!("Migrated legacy download path config hash");
            DownloadConfigHashOutcome::Unchanged
        }
        Ok(None) => {
            if let Err(e) = db
                .set_metadata(download::DOWNLOAD_CONFIG_HASH_KEY, current_hash)
                .await
            {
                tracing::warn!(error = %e, "Failed to persist download config_hash");
                return DownloadConfigHashOutcome::ReadFailed;
            }
            DownloadConfigHashOutcome::Unchanged
        }
        Ok(Some(_stored)) => {
            tracing::info!(
                "Download path config changed since last sync; preserving provider checkpoints and staging local reconciliation"
            );
            if let Err(e) = db
                .set_metadata(PENDING_DOWNLOAD_CONFIG_HASH_KEY, current_hash)
                .await
            {
                tracing::warn!(error = %e, "Failed to stage pending download config_hash");
                DownloadConfigHashOutcome::TokenPurgeFailed
            } else {
                if let Err(e) = db
                    .set_metadata(LAST_RECOVERY_ACTION_KEY, "reconcile_local_paths")
                    .await
                {
                    tracing::debug!(error = %e, "Failed to persist local path reconciliation state");
                }
                DownloadConfigHashOutcome::Changed
            }
        }
        Err(e) => {
            tracing::warn!(error = %e, "Failed to read download config_hash");
            DownloadConfigHashOutcome::ReadFailed
        }
    }
}

/// Run one sync cycle: iterate all libraries, download photos, store sync tokens.
pub(crate) async fn run_cycle(
    library_states: &[&LibraryState],
    config: &config::Config,
    state_db: Option<&dyn download::DownloadStore>,
    is_retry_failed: bool,
    build_download_config: &BuildDownloadConfigFn<'_>,
    download_controls: download::DownloadControls,
    shared_session: &auth::SharedSession,
    shutdown_token: &CancellationToken,
) -> anyhow::Result<CycleResult> {
    let mut cycle_failed_count = 0usize;
    let mut cycle_session_expired = false;
    let mut cycle_stats = download::SyncStats::default();
    let mut sparse_identity_proofs = Vec::new();

    let cycle_has_stale_plan = library_states.iter().any(|s| s.plan_is_stale);
    if cycle_has_stale_plan {
        tracing::warn!(
            "One or more libraries are running on a stale album plan; sync \
             database pre-check token will not advance this cycle"
        );
    }
    // CONTRACT: SYNC_TOKEN_ADVANCE_REQUIRES_CLEAN_CYCLE
    let mut db_sync_token_advance_safe = !config.runtime.dry_run && !cycle_has_stale_plan;
    let mut force_full_for_config_hash = false;
    let mut force_full_for_download_config_hash = false;
    let mut checkpoint_transition_state_safe = true;
    let mut enum_config_hash_outcome = EnumConfigHashOutcome::Unchanged;
    let mut pending_download_config_hash = None;
    let mut path_reconciliation_complete = true;
    let mut path_reconciliation_cycle_safe = true;
    let path_reconciliation_has_smart_folders = library_states.iter().any(|state| {
        state
            .plan
            .passes
            .iter()
            .any(|pass| pass.kind == crate::commands::PassKind::SmartFolder)
    });
    let mut checkpoint_hold_action: Option<&'static str> = None;
    let enum_config_hash = download::compute_config_hash(config);

    // Check if token-unsafe eligibility config changed since last sync. If
    // so, clear sync tokens and force full enumeration for this cycle -- the
    // stored incremental token would miss assets that are newly eligible
    // under the changed config (e.g. a user switching [photos].resolution or
    // changing [filters].media). The hash is
    // cycle-invariant across libraries,
    // so this runs once per cycle, not once per library.
    //
    // The metadata key `enum_config_hash` is distinct from the download
    // pipeline's `config_hash` (which tracks path-affecting fields only).
    // Using a single key for both would cause the two hashes to overwrite
    // each other every cycle, permanently preventing incremental sync.
    if !config.runtime.dry_run
        && let Some(db) = state_db
    {
        enum_config_hash_outcome = check_and_persist_enum_config_hash(db, &enum_config_hash).await;
        force_full_for_config_hash = enum_config_hash_outcome.must_force_full_sync();
        if matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed) {
            let recovery = RecoveryAction::ReconcileInventory(
                download::FullEnumerationReason::EnumConfigHashDrift,
            );
            if let Err(e) = db
                .set_metadata(LAST_RECOVERY_ACTION_KEY, recovery.as_str())
                .await
            {
                tracing::debug!(error = %e, "Failed to persist inventory reconciliation action");
            }
        }
        if matches!(
            enum_config_hash_outcome,
            EnumConfigHashOutcome::ChangedTokenPurgeFailed | EnumConfigHashOutcome::ReadFailed
        ) {
            checkpoint_transition_state_safe = false;
            db_sync_token_advance_safe = false;
        }
    }

    let mut preservation_config_hash = None;

    // A path-affecting config drift does not make the CloudKit token itself
    // unsafe, but it does make an incremental cycle insufficient: with no
    // changed assets, `/changes/zone` would never re-plan already-known media
    // into the new directory/template. Detect that before choosing
    // incremental mode. Path-only drift keeps source tracking incremental;
    // the pending marker drives separate catalog/targeted rehydration work.
    if !config.runtime.dry_run
        && let (Some(db), Some(first_library)) = (state_db, library_states.first())
    {
        let probe_config = build_download_config(
            download::SyncMode::Full,
            Arc::new(rustc_hash::FxHashSet::default()),
            Arc::new(download::AssetGroupings::default()),
            Arc::from(first_library.zone_name.as_str()),
        );
        let download_config_hash = download::hash_download_config(&probe_config);
        {
            let mut zones = library_states
                .iter()
                .map(|library| library.zone_name.clone())
                .collect::<Vec<_>>();
            zones.sort();
            zones.dedup();
            let coverage = download::sync_coverage_fingerprint_json(
                config,
                "icloud",
                1,
                &zones,
                &enum_config_hash,
                &download_config_hash,
            )?;
            let policy = serde_json::json!([
                1,
                state::METADATA_CAPTURE_REVISION,
                coverage,
                format!("{:?}", probe_config.metadata),
                config.runtime.refresh_metadata,
                config.runtime.repair_capture_timestamps,
                config.runtime.repair_truncated
            ]);
            preservation_config_hash = Some(hash_legacy_preservation_policy(
                &policy,
                probe_config.legacy_preservation_allow_hardlinks,
            ));
        }

        let legacy_download_config_hash = download::hash_legacy_download_config(&probe_config);
        let outcome = check_download_config_hash_for_cycle(
            db,
            &download_config_hash,
            &legacy_download_config_hash,
        )
        .await;
        force_full_for_download_config_hash = outcome.must_force_full_sync();
        if matches!(outcome, DownloadConfigHashOutcome::Changed) {
            pending_download_config_hash = Some(download_config_hash);
        }
        if outcome.token_purge_failed() {
            checkpoint_transition_state_safe = false;
            db_sync_token_advance_safe = false;
        }
    }

    let path_metadata_at_start = if let Some(db) = state_db {
        vec![
            (
                download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
                db.get_metadata(download::DOWNLOAD_CONFIG_HASH_KEY).await?,
            ),
            (
                PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
                pending_download_config_hash.clone(),
            ),
        ]
    } else {
        Vec::new()
    };

    let enum_metadata_at_start = if let Some(db) = state_db {
        vec![
            (
                ENUM_CONFIG_HASH_KEY.to_owned(),
                db.get_metadata(ENUM_CONFIG_HASH_KEY).await?,
            ),
            (
                PENDING_ENUM_CONFIG_HASH_KEY.to_owned(),
                db.get_metadata(PENDING_ENUM_CONFIG_HASH_KEY).await?,
            ),
        ]
    } else {
        Vec::new()
    };
    let mut active_cursor_fences = Vec::new();

    for lib_state in library_states.iter().copied() {
        if shutdown_token.is_cancelled() {
            break;
        }

        let indexing_key = indexing_reconciliation_key(&lib_state.zone_name);
        let (indexing_finished, indexing_hold_at_start) =
            prepare_library_indexing(lib_state, state_db, config.runtime.dry_run).await?;
        let indexing_reconciliation = indexing_hold_at_start.is_some() || !indexing_finished;
        if !has_active_passes(lib_state) {
            db_sync_token_advance_safe &= !indexing_reconciliation;
            continue;
        }
        if !indexing_finished {
            db_sync_token_advance_safe = false;
            path_reconciliation_complete = false;
            tracing::warn!(
                "Photo library indexing is RUNNING; downloads are provisional, deletions and checkpoint promotion are disabled"
            );
        }
        let retained_cursor_at_start = if let Some(db) = state_db {
            db.get_metadata(&lib_state.sync_token_key).await?
        } else {
            None
        };
        let mut retained_checkpoint_attempt = 1;
        let hold_key = retained_checkpoint_hold_key(&lib_state.zone_name);
        let retained_hold_at_start = if let Some(db) = state_db {
            db.get_metadata(&hold_key).await?
        } else {
            None
        };
        let replay_key = indexing_replay_key(&lib_state.zone_name);
        let replay_at_start = match state_db {
            Some(db) => db.get_metadata(&replay_key).await?,
            None => None,
        };
        let indexing_replay_pending = replay_at_start.is_some();
        let materialization_key = indexing_materialization_key(&lib_state.zone_name);
        let materialization_at_start = match state_db {
            Some(db) => db.get_metadata(&materialization_key).await?,
            None => None,
        };
        let indexing_materialization_pending = materialization_at_start.is_some();
        let mut zone_metadata_fence = enum_metadata_at_start.clone();
        zone_metadata_fence.push((
            materialization_key.clone(),
            materialization_at_start.clone(),
        ));
        active_cursor_fences.push((materialization_key.clone(), materialization_at_start));
        zone_metadata_fence.push((indexing_key.clone(), indexing_hold_at_start.clone()));
        zone_metadata_fence.push((replay_key.clone(), replay_at_start.clone()));
        active_cursor_fences.push((replay_key.clone(), replay_at_start));
        active_cursor_fences.push((indexing_key.clone(), indexing_hold_at_start.clone()));
        zone_metadata_fence.push((
            lib_state.sync_token_key.clone(),
            retained_cursor_at_start.clone(),
        ));
        active_cursor_fences.push((
            lib_state.sync_token_key.clone(),
            retained_cursor_at_start.clone(),
        ));
        zone_metadata_fence.push((hold_key.clone(), retained_hold_at_start.clone()));
        // Successful zone staging removes its hold before aggregate activation.
        active_cursor_fences.push((hold_key.clone(), None));

        if !config.runtime.dry_run
            && retained_hold_at_start.as_deref().is_some_and(|encoded| {
                retained_checkpoint_hold_action(encoded)
                    == RecoveryAction::RepairRetainedCheckpointEvidence.as_str()
            })
        {
            cycle_stats.sync_token_blocked = true;
            cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_hold_invalid");
            cycle_stats.sync_token_blocked_source = Some("kei");
            cycle_stats.sync_token_blocked_explanation = Some(
                download::sync_token_blocked_explanation("retained_checkpoint_hold_invalid"),
            );
            cycle_stats
                .sync_token_blocked_zone
                .get_or_insert_with(|| lib_state.zone_name.clone());
            cycle_failed_count = cycle_failed_count.saturating_add(1);
            db_sync_token_advance_safe = false;
            checkpoint_hold_action =
                Some(RecoveryAction::RepairRetainedCheckpointEvidence.as_str());
            tracing::warn!(
                diagnostic = "retained_checkpoint_hold_invalid",
                "Retained checkpoint evidence requires validation; preserving the durable hold"
            );
            continue;
        }

        // A rejected retained boundary cannot be repaired by repeating the same
        // rank inventory. Reserve each due attempt durably before provider or
        // legacy work, including attempts later cancelled or found incomplete.
        if !config.runtime.dry_run
            && let Some(db) = state_db
            && let Some(encoded) = retained_hold_at_start
            && let Ok(hold) = serde_json::from_str::<RetainedCheckpointHold>(&encoded)
            && let Some(prior) = retained_cursor_at_start.as_deref()
        {
            let mut current = RetainedCheckpointHold::new(
                prior,
                &enum_config_hash,
                preservation_config_hash.as_deref(),
            );
            if hold.matches(&current) {
                let exhausted = hold.attempt_count >= RETAINED_CHECKPOINT_MAX_ATTEMPTS;
                current.attempt_count = hold.attempt_count.saturating_add(1);
                retained_checkpoint_attempt = current.attempt_count;
                let deferred = exhausted || hold.deferred(chrono::Utc::now().timestamp());
                let reservation_failed = if deferred {
                    false
                } else {
                    persist_retained_checkpoint_hold(
                        db,
                        hold_key.clone(),
                        &current,
                        zone_metadata_fence.clone(),
                    )
                    .await
                    .is_err()
                };
                if !deferred
                    && !reservation_failed
                    && let Some((_, expected)) = zone_metadata_fence.last_mut()
                {
                    *expected = Some(serde_json::to_string(&current)?);
                }
                if deferred || reservation_failed {
                    block_for_retained_checkpoint(&mut cycle_stats);
                    cycle_stats
                        .sync_token_blocked_zone
                        .get_or_insert_with(|| lib_state.zone_name.clone());
                    if reservation_failed {
                        cycle_stats.state_write_failures =
                            cycle_stats.state_write_failures.saturating_add(1);
                    }
                    cycle_failed_count = cycle_failed_count.saturating_add(1);
                    db_sync_token_advance_safe = false;
                    checkpoint_hold_action = Some(if exhausted {
                        RecoveryAction::RetainedCheckpointRetryExhausted.as_str()
                    } else {
                        RecoveryAction::AwaitRetainedCheckpointEvidence.as_str()
                    });
                    tracing::warn!(
                        diagnostic = "retained_checkpoint_expired",
                        retry_deferred = deferred,
                        state_write_failed = reservation_failed,
                        retry_exhausted = exhausted,
                        "Retained checkpoint evidence is unavailable; preserving history and deferring recovery"
                    );
                    continue;
                }
            }
        }

        let mut legacy_cycle = download::legacy_preservation::LegacyCycle::default();
        if indexing_finished
            && !indexing_reconciliation
            && download_controls.run_mode.downloads_files()
            && !lib_state.plan_is_stale
            && let Some(db) = state_db
            && db
                .get_summary()
                .await?
                .metadata_capture
                .iter()
                .any(|capture| {
                    capture.library == lib_state.zone_name
                        && (capture.unresolved_assets > 0 || capture.unattributed_legacy_assets > 0)
                })
        {
            let legacy_probe = build_download_config(
                download::SyncMode::Full,
                Arc::new(rustc_hash::FxHashSet::default()),
                Arc::new(download::AssetGroupings::default()),
                Arc::from(lib_state.zone_name.as_str()),
            );
            legacy_cycle = download::legacy_preservation::LegacyCycle::begin(
                db,
                &legacy_probe,
                &lib_state.library.all(),
                preservation_config_hash.as_deref().unwrap_or(""),
                shutdown_token,
            )
            .await?;
        }

        // Determine source-enumeration mode per library. Failed transfers are
        // rehydrated from durable pending state by the download engine and do
        // not require replaying the provider inventory.
        let pending_zone_token =
            if matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed) {
                if let Some(db) = state_db {
                    db.get_metadata(&pending_zone_token_key(
                        &enum_config_hash,
                        &lib_state.zone_name,
                    ))
                    .await?
                    .filter(|token| !token.trim().is_empty())
                } else {
                    None
                }
            } else {
                None
            };
        let sync_mode_decision = if indexing_reconciliation || indexing_materialization_pending {
            SyncModeDecision {
                mode: download::SyncMode::Full,
                full_enumeration_reason: Some(download::FullEnumerationReason::OtherStaticReason),
            }
        } else if let Some(refresh_decision) =
            metadata_refresh_mode_decision(config.runtime.refresh_metadata)
        {
            refresh_decision
        } else if legacy_cycle.requires_inventory() {
            SyncModeDecision {
                mode: download::SyncMode::Full,
                full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
            }
        } else if let Some(zone_sync_token) = pending_zone_token {
            tracing::debug!(
                zone = %lib_state.zone_name,
                "Continuing config reconciliation from the completed zone checkpoint"
            );
            SyncModeDecision {
                mode: download::SyncMode::Incremental { zone_sync_token },
                full_enumeration_reason: None,
            }
        } else if force_full_for_config_hash {
            let reason = download::FullEnumerationReason::EnumConfigHashDrift;
            tracing::debug!(
                zone = %lib_state.zone_name,
                full_enumeration_reason = reason.as_str(),
                "Forcing full sync because config-hash validation invalidated stored sync tokens"
            );
            SyncModeDecision {
                mode: download::SyncMode::Full,
                full_enumeration_reason: Some(reason),
            }
        } else if force_full_for_download_config_hash {
            let reason = download::FullEnumerationReason::DownloadConfigHashDrift;
            tracing::debug!(
                zone = %lib_state.zone_name,
                full_enumeration_reason = reason.as_str(),
                "Forcing full sync because download config hash drift requires local path reconciliation"
            );
            SyncModeDecision {
                mode: download::SyncMode::Full,
                full_enumeration_reason: Some(reason),
            }
        } else {
            determine_sync_mode_decision(
                is_retry_failed,
                library_states.len(),
                state_db,
                &lib_state.sync_token_key,
                &lib_state.zone_name,
            )
            .await
        };
        let sync_mode = sync_mode_decision.mode;

        let sync_mode_label = match &sync_mode {
            download::SyncMode::Full => "full",
            download::SyncMode::Incremental { .. } => "incremental",
        };
        if let Some(reason) = sync_mode_decision.full_enumeration_reason {
            if let Some(db) = state_db
                && let Err(e) = db
                    .set_metadata(LAST_FULL_ENUMERATION_REASON_KEY, reason.as_str())
                    .await
            {
                tracing::debug!(error = %e, "Failed to persist full-enumeration reason");
            }
            tracing::info!(
                sync_mode = sync_mode_label,
                zone = %lib_state.zone_name,
                full_enumeration_reason = reason.as_str(),
                "Starting full-enumeration sync cycle"
            );
        } else {
            tracing::debug!(sync_mode = sync_mode_label, zone = %lib_state.zone_name, "Starting sync cycle");
        }

        // Skip the DB scan entirely when nothing downstream will read it.
        #[cfg(feature = "xmp")]
        let asset_groupings = if config.metadata.embed_xmp || config.metadata.xmp_sidecar {
            preload_asset_groupings(state_db, &lib_state.zone_name).await
        } else {
            Arc::new(download::AssetGroupings::default())
        };
        #[cfg(not(feature = "xmp"))]
        let asset_groupings = Arc::new(download::AssetGroupings::default());
        // Each pass carries its own exclude-asset-ids, so the config built
        // here starts with an empty set; download_photos_with_sync derives
        // per-pass configs internally via `with_exclude_ids`.
        let mut download_config = build_download_config(
            sync_mode,
            Arc::new(rustc_hash::FxHashSet::default()),
            Arc::clone(&asset_groupings),
            Arc::from(lib_state.zone_name.as_str()),
        );
        // FINISHED is a prerequisite, not proof that this inventory remained
        // complete. Deletion-capable maintenance stays deferred until the
        // full scope and its final readiness check have qualified.
        Arc::make_mut(&mut download_config).provisional =
            indexing_reconciliation || !indexing_finished;
        let managed_indexing_recovery = indexing_reconciliation
            && (download_config.edited_naming == crate::types::EditedNaming::Primary
                || match state_db {
                    Some(db) => db.has_primary_layouts(&lib_state.zone_name).await?,
                    None => false,
                });
        Arc::make_mut(&mut download_config).indexing_inventory_only =
            managed_indexing_recovery && indexing_finished;
        if pending_download_config_hash.is_some() && path_reconciliation_has_smart_folders {
            Arc::make_mut(&mut download_config).complete_path_selection = true;
        }
        let download_client = shared_session.read().await.download_client().clone();
        // Only an unrestricted current full-library selection can admit an exact
        // sparse target without inventing album/rank membership.
        let mut sparse_targets = Vec::new();
        if download_controls.run_mode.downloads_files() && !is_retry_failed {
            for target in library_states.iter().copied().filter(|target| {
                target.zone_name != lib_state.zone_name
                    && !target.plan_is_stale
                    && target.plan.passes.len() == 1
            }) {
                let Some(pass) = target.plan.passes.first() else {
                    continue;
                };
                if pass.kind != crate::commands::PassKind::Unfiled || !pass.exclude_ids.is_empty() {
                    continue;
                }
                #[cfg(feature = "xmp")]
                let target_groupings = if config.metadata.embed_xmp || config.metadata.xmp_sidecar {
                    preload_asset_groupings(state_db, &target.zone_name).await
                } else {
                    Arc::new(download::AssetGroupings::default())
                };
                #[cfg(not(feature = "xmp"))]
                let target_groupings = Arc::new(download::AssetGroupings::default());
                let target_config = build_download_config(
                    download::SyncMode::Full,
                    Arc::clone(&pass.exclude_ids),
                    target_groupings,
                    Arc::from(target.zone_name.as_str()),
                );
                if target_config.recent.is_some()
                    || target_config.skip_created_before.is_some()
                    || target_config.skip_created_after.is_some()
                    || target_config.retry_only
                    || target_config.edited_naming == crate::types::EditedNaming::Primary
                {
                    continue;
                }
                let mut target_config = target_config.with_pass(pass);
                // A full-library path label is not evidence of album membership.
                target_config.album_name = None;
                sparse_targets.push(download::sparse_target::SelectedSparseTarget {
                    library: target.library.clone(),
                    album: pass.album.clone(),
                    config: Arc::new(target_config),
                    client: download_client.clone(),
                    controls: download_controls,
                });
            }
        }
        Arc::make_mut(&mut download_config).sparse_targets = sparse_targets.into();

        let mut path_reconciliation_requires_smart_query = false;
        if indexing_finished
            && !indexing_reconciliation
            && pending_download_config_hash.is_some()
            && download_controls.run_mode.downloads_files()
        {
            let reconciliation = download::reconcile_catalog_paths_with_client(
                &download_client,
                &lib_state.plan.passes,
                Arc::clone(&download_config),
                shutdown_token.clone(),
            )
            .await?;
            path_reconciliation_requires_smart_query = lib_state
                .plan
                .passes
                .iter()
                .any(|pass| pass.kind == crate::commands::PassKind::SmartFolder);
            path_reconciliation_complete &=
                reconciliation.complete || reconciliation.complete_after_smart_query;
            let reconciliation_failures = reconciliation
                .stats
                .failed
                .saturating_add(reconciliation.stats.exif_failures)
                .saturating_add(reconciliation.stats.state_write_failures);
            cycle_failed_count = cycle_failed_count.saturating_add(reconciliation_failures);
            cycle_stats.accumulate(&reconciliation.stats);
            if reconciliation_failures > 0 {
                // A normal download/adoption pass must not move state past a
                // rejected reconciliation destination or failed finalization.
                db_sync_token_advance_safe = false;
                continue;
            }
        }
        let mut sync_result = download::download_photos_with_sync(
            &download_client,
            &lib_state.plan.passes,
            Arc::clone(&download_config),
            download_controls,
            shutdown_token.clone(),
        )
        .await?;

        // Capture only the initial complete selected scope, before a bridge
        // adds repeated count observations to the public activity statistics.
        let inventory = sync_result.inventory.take().filter(|measurement| {
            measurement.scope == download::inventory_scope(&lib_state.plan.passes, &download_config)
        });

        observe_sync_component(
            &lib_state.zone_name,
            "initial",
            &sync_result,
            lib_state.plan.passes.len(),
        );

        if indexing_finished
            && legacy_cycle.requires_inventory()
            && sync_result.full_enumeration_ran
            && matches!(
                source_checkpoint_decision(
                    &sync_result,
                    config.runtime.dry_run,
                    lib_state.plan_is_stale,
                    CheckpointBasis::CompleteInventory
                ),
                SourceCheckpointDecision::Advance { .. }
            )
            && let Some(db) = state_db
        {
            let prior = db
                .get_metadata(&lib_state.sync_token_key)
                .await?
                .unwrap_or_default();
            if let Err(error) = legacy_cycle
                .certify(
                    db,
                    &download_config,
                    &lib_state.library.all(),
                    preservation_config_hash.as_deref().unwrap_or(""),
                    &prior,
                    shutdown_token,
                )
                .await
            {
                download::legacy_preservation::log_preservation_hold("certification", &error);
            }
        }

        let mut retained_checkpoint_expired = false;
        let mut checkpoint_basis =
            if sync_result.full_enumeration_ran && !sync_result.checkpoint.completed_delta_replay {
                CheckpointBasis::CompleteInventory
            } else {
                CheckpointBasis::IncrementalDelta
            };

        // Eligibility reconciliation begins from a trusted prior cursor when
        // one exists. The inventory covers unchanged newly eligible assets;
        // replaying the prior cursor then bridges changes that occurred while
        // that inventory was running. Only the bridged token may replace the
        // active checkpoint.
        let unresolved_identity = if let Some(db) = state_db {
            db.get_metadata(&state::unresolved_identity_key(&lib_state.zone_name))
                .await?
                .is_some()
        } else {
            false
        };
        if (matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed)
            || unresolved_identity
            || legacy_cycle.requires_inventory()
            || ((indexing_replay_pending || indexing_materialization_pending)
                && !is_retry_failed
                && !download_config.retry_only))
            && indexing_finished
            && !indexing_reconciliation
            && sync_result.full_enumeration_ran
            && !sync_result.checkpoint.completed_delta_replay
            && checkpoint_transition_state_safe
            && (!indexing_materialization_pending
                || (matches!(sync_result.outcome, download::DownloadOutcome::Success)
                    && sync_result.stats.failed == 0
                    && sync_result.stats.exif_failures == 0
                    && download_config.recent.is_none()
                    && download_config.skip_created_before.is_none()
                    && download_config.skip_created_after.is_none()))
        {
            let prior_token = retained_cursor_at_start.clone();
            if let Some(prior_token) = prior_token.filter(|token| !token.trim().is_empty()) {
                let inventory_decision = source_checkpoint_decision(
                    &sync_result,
                    config.runtime.dry_run,
                    lib_state.plan_is_stale,
                    CheckpointBasis::CompleteInventory,
                );
                if matches!(inventory_decision, SourceCheckpointDecision::Advance { .. }) {
                    let mut bridge_config = build_download_config(
                        download::SyncMode::Incremental {
                            zone_sync_token: prior_token.clone(),
                        },
                        Arc::new(rustc_hash::FxHashSet::default()),
                        Arc::clone(&asset_groupings),
                        Arc::from(lib_state.zone_name.as_str()),
                    );
                    tracing::info!(
                        zone = %lib_state.zone_name,
                        "Inventory reconciliation complete; bridging changes from the preserved provider checkpoint"
                    );
                    Arc::make_mut(&mut bridge_config).sparse_targets =
                        Arc::clone(&download_config.sparse_targets);
                    let bridge_result = download::download_photos_with_sync(
                        &download_client,
                        &lib_state.plan.passes,
                        bridge_config,
                        download_controls,
                        shutdown_token.clone(),
                    )
                    .await?;
                    observe_sync_component(
                        &lib_state.zone_name,
                        "delta_bridge",
                        &bridge_result,
                        lib_state.plan.passes.len(),
                    );
                    let bridge_decision = source_checkpoint_decision(
                        &bridge_result,
                        config.runtime.dry_run,
                        lib_state.plan_is_stale,
                        CheckpointBasis::IncrementalDelta,
                    );
                    sync_result.outcome =
                        merge_download_outcomes(&sync_result.outcome, &bridge_result.outcome);
                    sync_result.accumulate(&bridge_result);
                    if bridge_result.checkpoint.retained_token_rejected
                        && !bridge_result.checkpoint.completed_delta_replay
                        && let Some(db) = state_db
                    {
                        retained_checkpoint_expired = true;
                        let mut hold = RetainedCheckpointHold::new(
                            &prior_token,
                            &enum_config_hash,
                            preservation_config_hash.as_deref(),
                        );
                        hold.attempt_count = retained_checkpoint_attempt;
                        if persist_retained_checkpoint_hold(
                            db,
                            retained_checkpoint_hold_key(&lib_state.zone_name),
                            &hold,
                            zone_metadata_fence.clone(),
                        )
                        .await
                        .is_err()
                        {
                            sync_result.checkpoint.state_write_failures = sync_result
                                .checkpoint
                                .state_write_failures
                                .saturating_add(1);
                            sync_result.stats.state_write_failures =
                                sync_result.stats.state_write_failures.saturating_add(1);
                        }
                        block_for_retained_checkpoint(&mut sync_result.stats);
                    }
                    if !bridge_result.full_enumeration_ran
                        || bridge_result.checkpoint.completed_delta_replay
                    {
                        if let SourceCheckpointDecision::Advance { token, .. } = bridge_decision {
                            sync_result.sync_token = Some(token);
                            checkpoint_basis = CheckpointBasis::InventoryWithDeltaBridge;
                        } else {
                            sync_result.sync_token = None;
                            sync_result.checkpoint.sync_token_blocked = true;
                            sync_result.checkpoint.project(&mut sync_result.stats);
                            sync_result.stats.sync_token_blocked_reason =
                                Some("inventory_delta_bridge_failed");
                            sync_result.stats.sync_token_blocked_source = Some("kei");
                            sync_result.stats.sync_token_blocked_explanation = Some(
                                "the inventory completed, but replay from the preserved provider checkpoint did not finish safely",
                            );
                        }
                    } else {
                        sync_result.sync_token = None;
                        sync_result.checkpoint.sync_token_blocked = true;
                        sync_result.checkpoint.project(&mut sync_result.stats);
                        if bridge_result.checkpoint.retained_token_rejected {
                            block_for_retained_checkpoint(&mut sync_result.stats);
                        } else {
                            sync_result.stats.sync_token_blocked_reason =
                                Some("inventory_delta_bridge_failed");
                            sync_result.stats.sync_token_blocked_source = Some("kei");
                            sync_result.stats.sync_token_blocked_explanation = Some(
                                "the prior provider checkpoint could not be replayed incrementally after inventory reconciliation",
                            );
                        }
                    }
                }
            }
        }

        if unresolved_identity
            && !indexing_reconciliation
            && matches!(checkpoint_basis, CheckpointBasis::CompleteInventory)
        {
            // An inventory cannot prove that the previously ambiguous deltas
            // were resolved. Require replay from the retained cursor.
            sync_result.block_for_unresolved_identity();
        }

        if retained_checkpoint_expired {
            block_for_retained_checkpoint(&mut sync_result.stats);
        }

        if sync_result.full_enumeration_ran && sync_result.stats.full_enumeration_reason.is_none() {
            sync_result.stats.full_enumeration_reason = sync_mode_decision.full_enumeration_reason;
        }

        // --refresh-metadata bypasses automatic capture repair. Inspect the
        // retained queue before inventory anchors or provider cursors commit,
        // not just when reporting aggregate cycle health afterward.
        if let Some(db) = state_db
            && let Some(capture) = db
                .get_summary()
                .await?
                .metadata_capture
                .into_iter()
                .find(|capture| capture.library == lib_state.zone_name)
        {
            sync_result.stats.unattributed_legacy_assets = capture.unattributed_legacy_assets;
            sync_result.stats.unattributed_legacy_pending = capture.unattributed_legacy_pending;
            sync_result.stats.metadata_capture_remaining = capture.remaining_assets;
            sync_result.stats.metadata_capture_unresolved = capture.unresolved_assets;
            sync_result.stats.metadata_capture_deferred = capture.deferred_assets;
            if capture.unresolved_assets > 0 {
                sync_result.block_for_metadata_capture();
            }
        }

        let library_completed_without_errors =
            matches!(&sync_result.outcome, download::DownloadOutcome::Success)
                && !sync_result.checkpoint.identity_incomplete
                && !sync_result.checkpoint.interrupted
                && sync_result.checkpoint.enumeration_errors == 0
                && !shutdown_token.is_cancelled();
        if path_reconciliation_requires_smart_query {
            // Historical smart membership cannot certify the current selection.
            // Full query proof is separate from a bounded source cursor hold.
            path_reconciliation_complete &= library_completed_without_errors
                && download_controls.run_mode.downloads_files()
                && sync_result.full_enumeration_ran;
        }
        if should_warn_zero_assets(
            &sync_result,
            library_completed_without_errors,
            download_controls.run_mode,
            is_retry_failed,
            lib_state,
        ) {
            tracing::warn!(
                library = %lib_state.zone_name,
                library_count = library_states.len(),
                assets_seen = sync_result.stats.assets_seen,
                "Sync completed after enumerating zero assets; check iCloud library \
                 access and filters if this was unexpected"
            );
        }
        if indexing_finished
            && !indexing_reconciliation
            && library_completed_without_errors
            && !sync_result.checkpoint.enumeration_incomplete
            && sync_result.checkpoint.state_write_failures == 0
            && !cycle_has_stale_plan
            && !is_retry_failed
            && download_controls.run_mode.downloads_files()
            && let Some(inventory) = inventory.as_ref()
            && let Some(db) = state_db
        {
            update_inventory_anchor_for_cycle(
                db,
                &enum_config_hash,
                &lib_state.zone_name,
                inventory,
                &mut sync_result.stats,
            )
            .await;
        }

        // Readiness can regress during a long inventory. Recheck at the
        // checkpoint boundary; even a failed probe leaves durable retry debt.
        let readiness_after = tokio::select! {
            biased;
            () = shutdown_token.cancelled() => None,
            result = async {
                let mut finished = lib_state.library.check_indexing().await?;
                for source in &lib_state.cross_zone_libraries {
                    if source.zone_name() != lib_state.zone_name {
                        finished &= source.check_indexing().await?;
                    }
                }
                Ok::<_, crate::icloud::error::ICloudError>(finished)
            } => Some(result),
        };
        // Cancellation is not an indexing observation. Preserve existing debt,
        // but do not turn ordinary interrupted sync into a new indexing hold.
        let indexing_hold_required =
            !indexing_finished || matches!(readiness_after, Some(Ok(false)) | Some(Err(_)));
        if indexing_hold_required {
            if !config.runtime.dry_run
                && let Some(db) = state_db
            {
                db.set_metadata(&indexing_key, "1").await?;
                // This cycle must not stage or activate a token, even if all
                // selected records happened to carry one unanimous candidate.
            }
            sync_result.checkpoint.sync_token_blocked = true;
            sync_result.checkpoint.project(&mut sync_result.stats);
            sync_result.stats.sync_token_blocked_reason = Some("indexing_reconciliation_required");
            sync_result.stats.sync_token_blocked_source = Some("icloud");
            sync_result.stats.sync_token_blocked_explanation = Some(
                "indexing results are provisional; a FINISHED full reconciliation is required",
            );
            db_sync_token_advance_safe = false;
            path_reconciliation_complete = false;
        }
        // Retry-only work cannot discharge either phase: new assets may be
        // skipped before they acquire durable retry rows. Keep their cursor.
        let recovery_retry_only = (indexing_reconciliation
            || indexing_replay_pending
            || indexing_materialization_pending)
            && (is_retry_failed || download_config.retry_only);
        if recovery_retry_only
            || (indexing_reconciliation
                && (!sync_result.full_enumeration_ran
                    || !library_completed_without_errors
                    || sync_result.stats.failed > 0
                    || sync_result.stats.exif_failures > 0))
        {
            sync_result.checkpoint.sync_token_blocked = true;
            sync_result.checkpoint.project(&mut sync_result.stats);
            if recovery_retry_only && !indexing_reconciliation {
                sync_result.stats.sync_token_blocked_reason = Some("indexing_replay_required");
                sync_result.stats.sync_token_blocked_source = Some("kei");
                sync_result.stats.sync_token_blocked_explanation =
                    Some("retry-only processing cannot discharge retained-cursor replay");
            }
        }

        // An inventory-only pass queued nothing. Unlike ordinary durable
        // transfer retries, managed recovery cannot release its full-work debt
        // until every selected publication and its retained bridge succeeds.
        if indexing_materialization_pending
            && !indexing_reconciliation
            && (!library_completed_without_errors
                || !sync_result.full_enumeration_ran
                || sync_result.stats.failed > 0
                || sync_result.stats.exif_failures > 0
                || download_config.recent.is_some()
                || download_config.skip_created_before.is_some()
                || download_config.skip_created_after.is_some()
                || (retained_cursor_at_start
                    .as_ref()
                    .is_some_and(|s| !s.trim().is_empty())
                    && !matches!(checkpoint_basis, CheckpointBasis::InventoryWithDeltaBridge)))
        {
            sync_result.checkpoint.sync_token_blocked = true;
            sync_result.checkpoint.project(&mut sync_result.stats);
        }

        // Provider cursor safety is independent from transfer completion.
        // The download pipeline persists every planned work item before this
        // boundary, so failed transfers may advance while state, identity,
        // enumeration, auth, and token-proof failures preserve the old cursor.
        let mut checkpoint_decision =
            if checkpoint_transition_state_safe && !shutdown_token.is_cancelled() {
                source_checkpoint_decision(
                    &sync_result,
                    config.runtime.dry_run,
                    lib_state.plan_is_stale,
                    checkpoint_basis,
                )
            } else if shutdown_token.is_cancelled() {
                SourceCheckpointDecision::Preserve {
                    reason: CheckpointHoldReason::Interrupted,
                    recovery: RecoveryAction::ReplayFromPriorToken,
                }
            } else {
                SourceCheckpointDecision::Preserve {
                    reason: CheckpointHoldReason::StateNotDurable,
                    recovery: RecoveryAction::ReplayFromPriorToken,
                }
            };

        if sync_result.stats.sync_token_blocked_reason == Some("retained_checkpoint_expired")
            && let SourceCheckpointDecision::Preserve { reason, .. } = checkpoint_decision
        {
            checkpoint_decision = SourceCheckpointDecision::Preserve {
                reason,
                recovery: RecoveryAction::AwaitRetainedCheckpointEvidence,
            };
        }

        if indexing_reconciliation {
            let mut qualified = false;
            if matches!(
                checkpoint_decision,
                SourceCheckpointDecision::Advance { .. }
            ) && !is_retry_failed
                && !download_config.retry_only
                && !indexing_hold_required
                && indexing_finished
                && sync_result.full_enumeration_ran
                && library_completed_without_errors
                && (!managed_indexing_recovery
                    || (matches!(readiness_after, Some(Ok(true)))
                        && download_config.recent.is_none()
                        && download_config.skip_created_before.is_none()
                        && download_config.skip_created_after.is_none()))
                && let Some(db) = state_db
            {
                // Qualify coverage without adopting its cursor. Managed
                // inventory queued no files, so retain separate full-work debt.
                let mut obligations = vec![(replay_key.clone(), "1".to_owned())];
                if managed_indexing_recovery {
                    obligations.push((materialization_key.clone(), "1".to_owned()));
                }
                match db
                    .commit_checkpoint_transition(state::CheckpointTransition {
                        expected_metadata: zone_metadata_fence.clone(),
                        legacy_preservation_proofs: Vec::new(),
                        legacy_config_hash: None,
                        sparse_identity_proofs: Vec::new(),
                        metadata_updates: obligations,
                        metadata_deletes: vec![indexing_key.clone()],
                    })
                    .await
                {
                    Ok(()) => qualified = true,
                    Err(error) => {
                        tracing::warn!(error = %error, "Failed to qualify indexing inventory; retaining recovery hold");
                        sync_result.checkpoint.state_write_failures += 1;
                        sync_result.stats.state_write_failures += 1;
                    }
                }
            }
            if matches!(
                checkpoint_decision,
                SourceCheckpointDecision::Advance { .. }
            ) {
                checkpoint_decision = SourceCheckpointDecision::Preserve {
                    reason: CheckpointHoldReason::TokenProofIncomplete,
                    recovery: RecoveryAction::ReplayFromPriorToken,
                };
            }
            sync_result.checkpoint.sync_token_blocked = true;
            sync_result.checkpoint.project(&mut sync_result.stats);
            sync_result.stats.sync_token_blocked_reason = Some(if qualified {
                "indexing_inventory_qualified"
            } else {
                "indexing_reconciliation_required"
            });
            sync_result.stats.sync_token_blocked_source = Some("kei");
            sync_result.stats.sync_token_blocked_explanation = Some(
                if qualified && managed_indexing_recovery {
                    "managed inventory qualified; full protected materialization and retained-cursor replay are still required"
                } else if qualified {
                    "full inventory qualified; the old trusted cursor is retained for the next incremental cycle"
                } else {
                    "a complete non-retry inventory from a FINISHED plan is still required"
                },
            );
            db_sync_token_advance_safe = false;
            path_reconciliation_complete = false;
            path_reconciliation_cycle_safe = false;
        }

        let mut legacy_proofs = Vec::new();
        if let SourceCheckpointDecision::Advance { token, .. } = &checkpoint_decision
            && let Some(db) = state_db
        {
            let qualified = if legacy_cycle.requires_inventory()
                && (!matches!(sync_result.outcome, download::DownloadOutcome::Success)
                    || sync_result.stats.failed > 0
                    || sync_result.stats.exif_failures > 0)
            {
                Err(anyhow::anyhow!("Legacy current work is incomplete"))
            } else {
                legacy_cycle
                    .finish(db, &download_config, token, shutdown_token)
                    .await
            };
            match qualified {
                Ok(proofs) => legacy_proofs = proofs,
                Err(error) => {
                    download::legacy_preservation::log_preservation_hold("checkpoint", &error);
                    checkpoint_decision = SourceCheckpointDecision::Preserve {
                        reason: CheckpointHoldReason::LegacyPreservationIncomplete,
                        recovery: RecoveryAction::ReplayFromPriorToken,
                    };
                    sync_result.stats.sync_token_blocked = true;
                    sync_result.stats.sync_token_blocked_reason =
                        Some("legacy_preservation_incomplete");
                    sync_result.stats.sync_token_blocked_source = Some("kei");
                    sync_result.stats.sync_token_blocked_explanation = Some(
                        "legacy evidence is protected, but independent current coverage is incomplete",
                    );
                }
            }
        }

        path_reconciliation_cycle_safe &= library_completed_without_errors
            && checkpoint_transition_state_safe
            && !retained_checkpoint_expired
            && (matches!(
                checkpoint_decision,
                SourceCheckpointDecision::Advance { .. }
            ) || (matches!(
                checkpoint_decision,
                SourceCheckpointDecision::Preserve {
                    reason: CheckpointHoldReason::TokenProofIncomplete,
                    ..
                }
            ) && sync_result.checkpoint.bounded_path_selection_complete
                && !sync_result.checkpoint.enumeration_incomplete
                && sync_result.checkpoint.state_write_failures == 0
                && !legacy_cycle.requires_inventory()));

        let mut checkpoint_evidence = checkpoint_support_fields(&checkpoint_decision);
        let mut checkpoint_persistence = "no_state_db";
        match checkpoint_decision {
            SourceCheckpointDecision::Advance { token, basis } => {
                crate::metrics::record_checkpoint_decision("advanced", basis.as_str());
                crate::metrics::record_deferred_transfers(
                    sync_result
                        .stats
                        .failed
                        .saturating_add(sync_result.stats.exif_failures),
                );
                if let Some(db) = state_db {
                    let reconciliation_active = force_full_for_config_hash;
                    if reconciliation_active {
                        sparse_identity_proofs.extend(
                            sync_result
                                .checkpoint
                                .sparse_identity_proofs
                                .iter()
                                .cloned(),
                        );
                        let candidate_key =
                            pending_zone_token_key(&enum_config_hash, &lib_state.zone_name);
                        if let Err(e) = db
                            .commit_checkpoint_transition(state::CheckpointTransition {
                                expected_metadata: zone_metadata_fence.clone(),
                                legacy_preservation_proofs: legacy_proofs,
                                legacy_config_hash: preservation_config_hash.clone(),
                                sparse_identity_proofs: Vec::new(),
                                metadata_updates: vec![(candidate_key, token)],
                                metadata_deletes: vec![retained_checkpoint_hold_key(
                                    &lib_state.zone_name,
                                )],
                            })
                            .await
                        {
                            checkpoint_hold_action =
                                Some(RecoveryAction::ReplayFromPriorToken.as_str());
                            db_sync_token_advance_safe = false;
                            path_reconciliation_cycle_safe = false;
                            tracing::warn!(
                                zone = %lib_state.zone_name,
                                error = %e,
                                "Failed to retain completed zone reconciliation checkpoint"
                            );
                            checkpoint_persistence = "failed";
                        } else {
                            checkpoint_persistence = "staged_reconciliation";
                        }
                    } else {
                        let mut metadata_updates = vec![(lib_state.sync_token_key.clone(), token)];
                        if let Some(recovery_action) = checkpoint_hold_action {
                            metadata_updates.push((
                                LAST_CHECKPOINT_STATUS_KEY.to_owned(),
                                "preserved".to_owned(),
                            ));
                            metadata_updates.push((
                                LAST_RECOVERY_ACTION_KEY.to_owned(),
                                recovery_action.to_owned(),
                            ));
                        } else {
                            metadata_updates.push((
                                LAST_CHECKPOINT_STATUS_KEY.to_owned(),
                                "current".to_owned(),
                            ));
                            metadata_updates
                                .push((LAST_RECOVERY_ACTION_KEY.to_owned(), "none".to_owned()));
                        }
                        if let Err(e) = db
                            .commit_checkpoint_transition(state::CheckpointTransition {
                                expected_metadata: zone_metadata_fence.clone(),
                                legacy_preservation_proofs: legacy_proofs,
                                legacy_config_hash: preservation_config_hash.clone(),
                                sparse_identity_proofs: sync_result
                                    .checkpoint
                                    .sparse_identity_proofs
                                    .clone(),
                                metadata_updates,
                                metadata_deletes: vec![
                                    state::unresolved_identity_key(&lib_state.zone_name),
                                    retained_checkpoint_hold_key(&lib_state.zone_name),
                                    indexing_key.clone(),
                                    replay_key.clone(),
                                    materialization_key.clone(),
                                ],
                            })
                            .await
                        {
                            checkpoint_hold_action =
                                Some(RecoveryAction::ReplayFromPriorToken.as_str());
                            db_sync_token_advance_safe = false;
                            path_reconciliation_cycle_safe = false;
                            tracing::warn!(error = %e, "Failed to store provider checkpoint");
                            checkpoint_persistence = "failed";
                        } else {
                            checkpoint_persistence = "stored";
                            if cycle_has_stale_plan && !lib_state.plan_is_stale {
                                tracing::warn!(
                                    zone = %lib_state.zone_name,
                                    diagnostic = "stale_plan_unaffected_zone",
                                    "Stored clean zone checkpoint despite another selected zone's stale plan"
                                );
                            }
                            if sync_result.stats.failed > 0 || sync_result.stats.exif_failures > 0 {
                                tracing::info!(
                                    zone = %lib_state.zone_name,
                                    basis = ?basis,
                                    deferred_transfers = sync_result.stats.failed,
                                    metadata_failures = sync_result.stats.exif_failures,
                                    "Provider checkpoint advanced: incomplete local work is durably queued for targeted retry"
                                );
                            } else {
                                tracing::debug!(zone = %lib_state.zone_name, basis = ?basis, "Stored provider checkpoint for next incremental sync");
                            }
                        }
                    }
                } else {
                    db_sync_token_advance_safe = false;
                    tracing::debug!(
                        zone = %lib_state.zone_name,
                        "Provider checkpoint available but no state DB is configured"
                    );
                }
            }
            SourceCheckpointDecision::Preserve { reason, recovery } => {
                checkpoint_hold_action = Some(recovery.as_str());
                crate::metrics::record_checkpoint_decision("preserved", reason.as_str());
                db_sync_token_advance_safe = false;
                if let Some(db) = state_db
                    && let Err(e) = db
                        .commit_checkpoint_transition(state::CheckpointTransition {
                            expected_metadata: Vec::new(),
                            legacy_preservation_proofs: Vec::new(),
                            legacy_config_hash: None,
                            sparse_identity_proofs: Vec::new(),
                            metadata_updates: vec![
                                (
                                    LAST_CHECKPOINT_STATUS_KEY.to_owned(),
                                    "preserved".to_owned(),
                                ),
                                (
                                    LAST_RECOVERY_ACTION_KEY.to_owned(),
                                    recovery.as_str().to_owned(),
                                ),
                            ],
                            metadata_deletes: Vec::new(),
                        })
                        .await
                {
                    path_reconciliation_cycle_safe = false;
                    tracing::debug!(error = %e, "Failed to persist checkpoint hold status");
                    checkpoint_persistence = "failed";
                } else if state_db.is_some() {
                    checkpoint_persistence = "stored_hold";
                }
                let diagnostic = sync_result
                    .stats
                    .sync_token_blocked_reason
                    .unwrap_or("provider_checkpoint_preserved");
                let explanation = sync_result
                    .stats
                    .sync_token_blocked_explanation
                    .unwrap_or_else(|| download::sync_token_blocked_explanation(diagnostic));
                tracing::warn!(
                    zone = %lib_state.zone_name,
                    reason = ?reason,
                    recovery = ?recovery,
                    diagnostic,
                    explanation,
                    "Provider checkpoint preserved; recovery will use the narrowest safe action"
                );
            }
        }

        if let Some(fields) = checkpoint_evidence.as_object_mut() {
            fields.insert(
                "persistence".into(),
                serde_json::json!(checkpoint_persistence),
            );
            fields.insert(
                "token_present".into(),
                serde_json::json!(sync_result.sync_token.is_some()),
            );
            fields.insert(
                "full_enumeration".into(),
                serde_json::json!(sync_result.full_enumeration_ran),
            );
            fields.insert(
                "identity_incomplete".into(),
                serde_json::json!(sync_result.checkpoint.identity_incomplete),
            );
            fields.insert(
                "state_write_failures".into(),
                serde_json::json!(sync_result.checkpoint.state_write_failures),
            );
            fields.insert(
                "enumeration_errors".into(),
                serde_json::json!(sync_result.checkpoint.enumeration_errors),
            );
            if checkpoint_persistence == "failed" {
                fields.insert(
                    "recovery".into(),
                    serde_json::json!("replay_from_prior_token"),
                );
            }
        }
        crate::support::observe_scoped(
            &lib_state.zone_name,
            None,
            "support_checkpoint_v1",
            checkpoint_evidence,
        );
        observe_sync_component(
            &lib_state.zone_name,
            "combined",
            &sync_result,
            lib_state.plan.passes.len(),
        );
        if sync_result.stats.sync_token_blocked
            && sync_result.stats.sync_token_blocked_zone.is_none()
        {
            sync_result.stats.sync_token_blocked_zone = Some(lib_state.zone_name.clone());
        }

        if sync_result.checkpoint.identity_incomplete {
            cycle_failed_count += 1;
        }

        if sync_result.stats.unattributed_legacy_assets > 0
            && let Some(db) = state_db
            && let Some(capture) = db
                .get_summary()
                .await?
                .metadata_capture
                .into_iter()
                .find(|capture| capture.library == lib_state.zone_name)
        {
            sync_result.stats.unattributed_legacy_assets = capture.unattributed_legacy_assets;
            sync_result.stats.unattributed_legacy_pending = capture.unattributed_legacy_pending;
        }

        // Accumulate stats across libraries.
        cycle_stats.accumulate(&sync_result.stats);

        match sync_result.outcome {
            download::DownloadOutcome::Success => {}
            download::DownloadOutcome::SessionExpired { auth_error_count } => {
                tracing::warn!(
                    auth_error_count,
                    zone = %lib_state.zone_name,
                    "Session expired during library sync"
                );
                cycle_session_expired = true;
                break; // Stop iterating libraries -- need re-auth
            }
            download::DownloadOutcome::PartialFailure { failed_count } => {
                cycle_failed_count += failed_count;
            }
        }
    }

    let reconciliation_active = force_full_for_config_hash;
    if reconciliation_active
        && db_sync_token_advance_safe
        && !cycle_session_expired
        && !shutdown_token.is_cancelled()
        && let Some(db) = state_db
    {
        let mut metadata_updates = Vec::new();
        let mut metadata_deletes = Vec::new();
        if force_full_for_config_hash {
            for lib_state in library_states
                .iter()
                .copied()
                .filter(|state| has_active_passes(state))
            {
                let candidate_key = pending_zone_token_key(&enum_config_hash, &lib_state.zone_name);
                let Some(token) = db
                    .get_metadata(&candidate_key)
                    .await?
                    .filter(|token| !token.trim().is_empty())
                else {
                    db_sync_token_advance_safe = false;
                    path_reconciliation_cycle_safe = false;
                    tracing::warn!(
                        zone = %lib_state.zone_name,
                        "Config reconciliation has no completed checkpoint for selected zone"
                    );
                    break;
                };
                active_cursor_fences.push((candidate_key.clone(), Some(token.clone())));
                metadata_updates.push((lib_state.sync_token_key.clone(), token));
                metadata_deletes.push(candidate_key);
                metadata_deletes.push(state::unresolved_identity_key(&lib_state.zone_name));
                metadata_deletes.push(indexing_reconciliation_key(&lib_state.zone_name));
                metadata_deletes.push(indexing_replay_key(&lib_state.zone_name));
                metadata_deletes.push(indexing_materialization_key(&lib_state.zone_name));
            }
            if db_sync_token_advance_safe {
                metadata_updates.push((ENUM_CONFIG_HASH_KEY.to_owned(), enum_config_hash));
                metadata_deletes.push(PENDING_ENUM_CONFIG_HASH_KEY.to_owned());
            }
        }
        if db_sync_token_advance_safe {
            metadata_updates.push((LAST_CHECKPOINT_STATUS_KEY.to_owned(), "current".to_owned()));
            metadata_updates.push((LAST_RECOVERY_ACTION_KEY.to_owned(), "none".to_owned()));
            if let Err(e) = db
                .commit_checkpoint_transition(state::CheckpointTransition {
                    expected_metadata: enum_metadata_at_start
                        .iter()
                        .cloned()
                        .chain(active_cursor_fences)
                        .collect(),
                    legacy_preservation_proofs: Vec::new(),
                    legacy_config_hash: preservation_config_hash.clone(),
                    sparse_identity_proofs,
                    metadata_updates,
                    metadata_deletes,
                })
                .await
            {
                db_sync_token_advance_safe = false;
                path_reconciliation_cycle_safe = false;
                tracing::warn!(error = %e, "Failed to atomically promote checkpoint reconciliation");
            }
        }
    }

    if path_reconciliation_complete
        && (!path_reconciliation_has_smart_folders
            || (cycle_failed_count == 0 && path_reconciliation_cycle_safe && !cycle_has_stale_plan))
        && !cycle_session_expired
        && !shutdown_token.is_cancelled()
        && let (Some(db), Some(download_config_hash)) = (state_db, pending_download_config_hash)
        && let Err(e) = db
            .commit_checkpoint_transition(state::CheckpointTransition {
                expected_metadata: path_metadata_at_start,
                legacy_preservation_proofs: Vec::new(),
                legacy_config_hash: None,
                sparse_identity_proofs: Vec::new(),
                metadata_updates: vec![(
                    download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
                    download_config_hash,
                )],
                metadata_deletes: vec![PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned()],
            })
            .await
    {
        tracing::warn!(error = %e, "Failed to promote completed local path reconciliation");
    }

    if let Some(db) = state_db {
        let summary = db.get_summary().await?;
        if summary.unresolved_identity_zones > 0 {
            download::block_sync_token_for_unresolved_identity(&mut cycle_stats);
            cycle_failed_count = cycle_failed_count.max(1);
            db_sync_token_advance_safe = false;
        }
        if summary
            .metadata_capture
            .iter()
            .any(|capture| capture.unattributed_legacy_pending > 0)
        {
            // Protected history no longer appears in the capture retry count,
            // but pending current-work proof is still incomplete cycle work.
            cycle_failed_count = cycle_failed_count.max(1);
            db_sync_token_advance_safe = false;
        }
        if summary
            .metadata_capture
            .iter()
            .any(|capture| capture.unresolved_assets > 0)
        {
            download::block_sync_token_for_metadata_capture(&mut cycle_stats);
            cycle_failed_count = cycle_failed_count.max(1);
            db_sync_token_advance_safe = false;
        }
        if summary
            .last_recovery_action
            .as_deref()
            .is_some_and(|action| {
                matches!(
                    action,
                    "await_retained_checkpoint_evidence"
                        | "retained_checkpoint_retry_exhausted"
                        | "repair_retained_checkpoint_evidence"
                )
            })
        {
            block_for_retained_checkpoint(&mut cycle_stats);
            if summary.last_recovery_action.as_deref()
                == Some("retained_checkpoint_retry_exhausted")
            {
                cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_retry_exhausted");
                cycle_stats.sync_token_blocked_explanation = Some(
                    download::sync_token_blocked_explanation("retained_checkpoint_retry_exhausted"),
                );
            }
            if summary.last_recovery_action.as_deref()
                == Some("repair_retained_checkpoint_evidence")
            {
                cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_hold_invalid");
                cycle_stats.sync_token_blocked_explanation = Some(
                    download::sync_token_blocked_explanation("retained_checkpoint_hold_invalid"),
                );
            }
            cycle_failed_count = cycle_failed_count.max(1);
            db_sync_token_advance_safe = false;
        }
    }

    #[cfg(all(test, target_os = "linux"))]
    crate::test_helpers::process_death_point("checkpoint-persisted");

    Ok(CycleResult {
        failed_count: cycle_failed_count,
        session_expired: cycle_session_expired,
        stats: cycle_stats,
        db_sync_token_advance_safe,
    })
}

/// Bulk-load `asset_albums` + `asset_people` into an in-memory index so the
/// filter phase can enrich payloads without per-asset DB hits. Scoped to a
/// single library so multi-library accounts don't cross-attribute album /
/// person memberships across zones (the v9 schema scopes both join tables
/// per library; this reader honours that scope).
#[cfg(feature = "xmp")]
pub(crate) async fn preload_asset_groupings<D>(
    state_db: Option<&D>,
    library: &str,
) -> Arc<download::AssetGroupings>
where
    D: state::MembershipStore + ?Sized,
{
    let Some(db) = state_db else {
        return Arc::new(download::AssetGroupings::default());
    };
    let albums = db.get_all_asset_albums(library).await;
    let people = db.get_all_asset_people(library).await;
    let mut groupings = download::AssetGroupings::default();
    match albums {
        Ok(rows) => {
            for (asset_id, album) in rows {
                groupings.albums.entry(asset_id).or_default().push(album);
            }
        }
        Err(e) => tracing::warn!(error = %e, library, "Failed to preload asset_albums"),
    }
    match people {
        Ok(rows) => {
            for (asset_id, person) in rows {
                groupings.people.entry(asset_id).or_default().push(person);
            }
        }
        Err(e) => tracing::warn!(error = %e, library, "Failed to preload asset_people"),
    }
    Arc::new(groupings)
}

/// Determine the sync mode for a library: full enumeration or incremental.
pub(crate) async fn determine_sync_mode_decision<D>(
    _is_retry_failed: bool,
    _library_count: usize,
    state_db: Option<&D>,
    sync_token_key: &str,
    zone_name: &str,
) -> SyncModeDecision
where
    D: state::SyncTokenStore + ?Sized,
{
    if let Some(db) = state_db {
        match db.get_metadata(sync_token_key).await {
            Ok(Some(ref token)) if !token.is_empty() => {
                tracing::debug!(zone = %zone_name, "Stored sync token found, using incremental sync");
                SyncModeDecision {
                    mode: download::SyncMode::Incremental {
                        zone_sync_token: token.clone(),
                    },
                    full_enumeration_reason: None,
                }
            }
            Ok(_) => {
                let reason = download::FullEnumerationReason::NoStoredToken;
                tracing::debug!(
                    zone = %zone_name,
                    full_enumeration_reason = reason.as_str(),
                    "No sync token found, performing full enumeration"
                );
                SyncModeDecision {
                    mode: download::SyncMode::Full,
                    full_enumeration_reason: Some(reason),
                }
            }
            Err(e) => {
                let reason = download::FullEnumerationReason::OtherStaticReason;
                tracing::warn!(
                    error = %e,
                    full_enumeration_reason = reason.as_str(),
                    "Failed to load sync token, falling back to full enumeration"
                );
                SyncModeDecision {
                    mode: download::SyncMode::Full,
                    full_enumeration_reason: Some(reason),
                }
            }
        }
    } else {
        SyncModeDecision {
            mode: download::SyncMode::Full,
            full_enumeration_reason: Some(download::FullEnumerationReason::NoStoredToken),
        }
    }
}

#[cfg(test)]
pub(crate) async fn determine_sync_mode<D>(
    is_retry_failed: bool,
    library_count: usize,
    state_db: Option<&D>,
    sync_token_key: &str,
    zone_name: &str,
) -> download::SyncMode
where
    D: state::SyncTokenStore + ?Sized,
{
    determine_sync_mode_decision(
        is_retry_failed,
        library_count,
        state_db,
        sync_token_key,
        zone_name,
    )
    .await
    .mode
}

/// Keep strict-policy hashes byte-compatible with existing preservation proofs.
/// The opt-in affects only legacy evidence qualification, not download provenance.
pub(crate) fn hash_legacy_preservation_policy(
    policy: &serde_json::Value,
    allow_hardlinks: bool,
) -> String {
    use sha2::{Digest, Sha256};
    let mut hash = Sha256::new();
    hash.update(policy.to_string().as_bytes());
    if allow_hardlinks {
        hash.update(b"\0legacy-preservation-allow-hardlinks:v1");
    }
    data_encoding::HEXLOWER.encode(&hash.finalize())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::commands::PassKind;

    #[test]
    fn support_component_totals_keep_inventory_and_bridge_separate() {
        use tracing_subscriber::layer::SubscriberExt;
        let (layer, drain) = crate::support::history::test_layer();
        tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), || {
            let mut inventory = download::SyncResult::from_execution(
                download::DownloadOutcome::Success,
                Some("PRIVATE_TOKEN".into()),
                download::SyncStats {
                    assets_seen: 3899,
                    api_total_at_start: Some(3899),
                    ..Default::default()
                },
            );
            inventory.full_enumeration_ran = true;
            let bridge = download::SyncResult::from_execution(
                download::DownloadOutcome::Success,
                Some("PRIVATE_SUCCESSOR".into()),
                download::SyncStats {
                    assets_seen: 238,
                    api_total_at_start: Some(238),
                    ..Default::default()
                },
            );
            observe_sync_component("PRIVATE_ZONE", "initial", &inventory, 1);
            observe_sync_component("PRIVATE_ZONE", "delta_bridge", &bridge, 1);
            inventory.accumulate(&bridge);
            observe_sync_component("PRIVATE_ZONE", "combined", &inventory, 1);
        });
        let evidence = drain();
        assert_eq!(evidence.len(), 3);
        assert_eq!(evidence[0]["fields"]["assets_seen"], 3899);
        assert_eq!(evidence[1]["fields"]["assets_seen"], 238);
        assert_eq!(evidence[2]["fields"]["assets_seen"], 4137);
        assert_eq!(evidence[0]["fields"]["component"], "initial");
        assert!(evidence[2]["fields"]["inventory_complete"].is_null());
        assert!(
            !serde_json::to_string(&evidence)
                .unwrap()
                .contains("PRIVATE")
        );
    }

    #[test]
    fn retained_checkpoint_hold_round_trip_and_version_one_migration() {
        let hold = RetainedCheckpointHold::new(
            "private-cursor",
            "enum-policy",
            Some("preservation-policy"),
        );
        let encoded = serde_json::to_string(&hold).expect("encode receipt");
        assert!(!encoded.contains("private-cursor"));
        let decoded: RetainedCheckpointHold =
            serde_json::from_str(&encoded).expect("decode receipt");
        assert!(decoded.matches(&hold));
        assert_eq!(decoded.attempt_count, 1);
        assert_eq!(decoded.next_attempt_at, hold.next_attempt_at);
        assert_eq!(serde_json::to_string(&decoded).unwrap(), encoded);

        let mut version_one: serde_json::Value = serde_json::from_str(&encoded).unwrap();
        version_one.as_object_mut().unwrap().remove("attempt_count");
        let migrated: RetainedCheckpointHold = serde_json::from_value(version_one).unwrap();
        assert!(migrated.matches(&hold));
        assert_eq!(
            migrated.attempt_count, 1,
            "an existing receipt already consumed its first attempt"
        );
        let migrated_encoded = serde_json::to_string(&migrated).unwrap();
        assert_eq!(
            retained_checkpoint_hold_action(&migrated_encoded),
            "await_retained_checkpoint_evidence"
        );
        let mut exhausted = migrated;
        exhausted.attempt_count = RETAINED_CHECKPOINT_MAX_ATTEMPTS;
        assert_eq!(
            retained_checkpoint_hold_action(&serde_json::to_string(&exhausted).unwrap()),
            "retained_checkpoint_retry_exhausted"
        );
    }

    #[test]
    fn legacy_preservation_trust_hash_preserves_strict_compatibility() {
        let policy = serde_json::json!([1, 1, "coverage", "None", false, false, false]);
        // Historical SHA-256 of the exact serialized strict policy, independently computed.
        let strict = "6f307ea388dd8281b2db2d31c9a8ba37f3c910027bbdb47e95c014477ddf358f";
        assert_eq!(hash_legacy_preservation_policy(&policy, false), strict);
        assert_ne!(hash_legacy_preservation_policy(&policy, true), strict);
        assert_eq!(hash_legacy_preservation_policy(&policy, false), strict);
    }

    fn make_library_state(has_passes: bool) -> LibraryState {
        LibraryState {
            library: crate::icloud::photos::PhotoLibrary::new_stub_with_zone(
                Box::new(crate::test_helpers::MockPhotosSession::new()),
                "PrimarySync",
            ),
            cross_zone_libraries: Vec::new(),
            pass_scope: crate::commands::PassScope {
                include_albums: false,
                include_smart_folders: false,
                include_unfiled: has_passes,
            },
            zone_name: "PrimarySync".to_string(),
            sync_token_key: "sync_token:PrimarySync".to_string(),
            plan: crate::commands::AlbumPlan {
                passes: if has_passes {
                    vec![crate::commands::AlbumPass {
                        kind: PassKind::Unfiled,
                        album: crate::icloud::photos::PhotoAlbum::stub_for_test(Arc::from(
                            "PrimarySync",
                        )),
                        exclude_ids: Arc::new(rustc_hash::FxHashSet::default()),
                    }]
                } else {
                    Vec::new()
                },
            },
            plan_is_stale: false,
            plan_needs_refresh: false,
        }
    }

    #[test]
    fn should_warn_zero_assets_requires_active_passes() {
        let sync_result = download::SyncResult {
            full_enumeration_ran: true,
            ..download::SyncResult::from_execution(
                download::DownloadOutcome::Success,
                Some("zone-token".to_string()),
                download::SyncStats {
                    assets_seen: 0,
                    ..download::SyncStats::default()
                },
            )
        };
        assert!(!should_warn_zero_assets(
            &sync_result,
            true,
            download::DownloadRunMode::Download,
            false,
            &make_library_state(false),
        ));
    }

    #[test]
    fn should_warn_zero_assets_when_all_gates_are_true() {
        let sync_result = download::SyncResult {
            full_enumeration_ran: true,
            ..download::SyncResult::from_execution(
                download::DownloadOutcome::Success,
                Some("zone-token".to_string()),
                download::SyncStats {
                    assets_seen: 0,
                    ..download::SyncStats::default()
                },
            )
        };
        assert!(should_warn_zero_assets(
            &sync_result,
            true,
            download::DownloadRunMode::Download,
            false,
            &make_library_state(true),
        ));
    }

    #[test]
    fn durable_transfer_failure_can_advance_source_checkpoint() {
        let result = download::SyncResult::from_execution(
            download::DownloadOutcome::PartialFailure { failed_count: 1 },
            Some("zone-token-next".to_string()),
            download::SyncStats {
                failed: 1,
                ..download::SyncStats::default()
            },
        );

        assert_eq!(
            source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta,),
            SourceCheckpointDecision::Advance {
                token: "zone-token-next".to_string(),
                basis: CheckpointBasis::IncrementalDelta,
            }
        );
    }

    #[test]
    fn metadata_capture_hold_survives_report_changes_and_result_composition() {
        let mut result = download::SyncResult::from_execution(
            download::DownloadOutcome::Success,
            Some("zone-token-next".to_string()),
            download::SyncStats::default(),
        );
        result.block_for_metadata_capture();
        result.stats = download::SyncStats::default();
        result.accumulate(&download::SyncResult::from_execution(
            download::DownloadOutcome::Success,
            None,
            download::SyncStats::default(),
        ));
        assert!(result.checkpoint.identity_incomplete);
        assert!(result.stats.identity_incomplete);
        assert_eq!(
            source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta),
            SourceCheckpointDecision::Preserve {
                reason: CheckpointHoldReason::TokenProofIncomplete,
                recovery: RecoveryAction::ReplayFromPriorToken,
            }
        );
    }

    #[test]
    fn failed_retry_state_write_preserves_source_checkpoint() {
        let result = download::SyncResult::from_execution(
            download::DownloadOutcome::PartialFailure { failed_count: 1 },
            Some("zone-token-next".to_string()),
            download::SyncStats {
                failed: 1,
                state_write_failures: 1,
                ..download::SyncStats::default()
            },
        );

        assert_eq!(
            source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta,),
            SourceCheckpointDecision::Preserve {
                reason: CheckpointHoldReason::StateNotDurable,
                recovery: RecoveryAction::ReplayFromPriorToken,
            }
        );
    }

    #[test]
    fn incomplete_token_proof_preserves_concrete_retry_passes() {
        let pass = download::PassKey {
            index: 2,
            kind: PassKind::Unfiled,
            label: "unfiled".to_string(),
        };
        let mut result = download::SyncResult {
            full_enumeration_ran: true,
            ..download::SyncResult::from_execution(
                download::DownloadOutcome::Success,
                None,
                download::SyncStats {
                    sync_token_blocked: true,
                    ..download::SyncStats::default()
                },
            )
        };

        result.checkpoint.retry_passes = vec![pass.clone()];
        assert_eq!(
            source_checkpoint_decision(&result, false, false, CheckpointBasis::CompleteInventory),
            SourceCheckpointDecision::Preserve {
                reason: CheckpointHoldReason::TokenProofIncomplete,
                recovery: RecoveryAction::RetryPasses(vec![pass]),
            }
        );
    }

    #[test]
    fn contract_source_checkpoint_requires_durable_recovery_exhaustive() {
        let tokens = [Some("zone-token"), None, Some("   ")];
        for mask in 0u16..1024 {
            let dry_run = mask & 1 != 0;
            let stale_pass_plan = mask & 2 != 0;
            let session_expired = mask & 4 != 0;
            let interrupted = mask & 8 != 0;
            let enumeration_error = mask & 16 != 0;
            let enumeration_incomplete = mask & 32 != 0;
            let state_not_durable = mask & 64 != 0;
            let token_proof_blocked = mask & 128 != 0;
            let durable_transfer_failure = mask & 256 != 0;
            let identity_incomplete = mask & 512 != 0;
            let has_safety_blocker = mask & 255 != 0 || identity_incomplete;
            for token in tokens {
                let mut result = download::SyncResult::from_execution(
                    if session_expired {
                        download::DownloadOutcome::SessionExpired {
                            auth_error_count: 1,
                        }
                    } else if durable_transfer_failure {
                        download::DownloadOutcome::PartialFailure { failed_count: 1 }
                    } else {
                        download::DownloadOutcome::Success
                    },
                    token.map(str::to_string),
                    download::SyncStats {
                        interrupted,
                        enumeration_errors: usize::from(enumeration_error),
                        enumeration_incomplete,
                        state_write_failures: usize::from(state_not_durable),
                        sync_token_blocked: token_proof_blocked,
                        identity_incomplete,
                        ..download::SyncStats::default()
                    },
                );
                let should_advance = !has_safety_blocker && token == Some("zone-token");
                let decision = source_checkpoint_decision(
                    &result,
                    dry_run,
                    stale_pass_plan,
                    CheckpointBasis::IncrementalDelta,
                );
                assert_eq!(
                    matches!(decision, SourceCheckpointDecision::Advance { .. }),
                    should_advance,
                    "unexpected checkpoint decision for mask {mask:#010b} and token {token:?}"
                );
                // Reporting snapshots cannot remove or introduce safety evidence.
                result.stats = download::SyncStats {
                    state_write_failures: usize::from(!state_not_durable),
                    enumeration_errors: usize::from(!enumeration_error),
                    enumeration_incomplete: !enumeration_incomplete,
                    interrupted: !interrupted,
                    sync_token_blocked: !token_proof_blocked,
                    identity_incomplete: !identity_incomplete,
                    ..download::SyncStats::default()
                };
                let reported_decision = source_checkpoint_decision(
                    &result,
                    dry_run,
                    stale_pass_plan,
                    CheckpointBasis::IncrementalDelta,
                );
                assert_eq!(
                    reported_decision, decision,
                    "report mutation changed mask {mask:#010b}"
                );
            }
        }
    }

    #[test]
    fn inventory_drop_classifier_uses_five_percent_threshold() {
        assert_eq!(classify_inventory_drop(100, 96), None);
        assert_eq!(
            classify_inventory_drop(100, 95),
            Some(InventoryDrop {
                previous_total: 100,
                current_total: 95,
                drop_assets: 5,
                drop_percent: 5.0,
            })
        );
        assert_eq!(classify_inventory_drop(0, 0), None);
        assert_eq!(classify_inventory_drop(100, 101), None);
    }

    #[tokio::test]
    async fn inventory_anchor_warns_and_updates_on_comparable_drop() {
        let db = state::SqliteStateDb::open_in_memory().expect("state db");
        let inventory = download::InventoryMeasurement {
            scope: download::InventoryScope("test complete scope".to_owned()),
            total: 80,
            assets_seen: 80,
        };
        let key = inventory_anchor_key("hash-a", "PrimarySync", &inventory.scope);
        let prior = InventoryAnchor {
            version: INVENTORY_ANCHOR_VERSION,
            scope: inventory.scope.clone(),
            api_total_at_start: 100,
            assets_seen: 100,
            completed_at: 1_700_000_000,
        };
        db.set_metadata(&key, &serde_json::to_string(&prior).unwrap())
            .await
            .expect("seed anchor");

        let mut stats = download::SyncStats {
            api_total_at_start: Some(80),
            assets_seen: 80,
            ..download::SyncStats::default()
        };
        update_inventory_anchor_for_cycle(&db, "hash-a", "PrimarySync", &inventory, &mut stats)
            .await;

        assert_eq!(stats.inventory_drop_warnings, 1);
        assert_eq!(stats.inventory_drop_assets, 20);
        assert_eq!(stats.inventory_drop_previous_total, Some(100));
        assert_eq!(stats.inventory_drop_current_total, Some(80));
        assert_eq!(
            stats.inventory_drop_library,
            Some("PrimarySync".to_string())
        );
        let raw = db
            .get_metadata(&key)
            .await
            .expect("read anchor")
            .expect("anchor exists");
        let updated: InventoryAnchor = serde_json::from_str(&raw).expect("anchor json");
        assert_eq!(updated.api_total_at_start, 80);
    }

    #[tokio::test]
    async fn inventory_anchor_is_scoped_by_enum_config_hash() {
        let db = state::SqliteStateDb::open_in_memory().expect("state db");
        let inventory = download::InventoryMeasurement {
            scope: download::InventoryScope("test complete scope".to_owned()),
            total: 80,
            assets_seen: 80,
        };
        let old_key = inventory_anchor_key("old-hash", "PrimarySync", &inventory.scope);
        let prior = InventoryAnchor {
            version: INVENTORY_ANCHOR_VERSION,
            scope: inventory.scope.clone(),
            api_total_at_start: 100,
            assets_seen: 100,
            completed_at: 1_700_000_000,
        };
        db.set_metadata(&old_key, &serde_json::to_string(&prior).unwrap())
            .await
            .expect("seed anchor");

        let mut stats = download::SyncStats {
            api_total_at_start: Some(80),
            assets_seen: 80,
            ..download::SyncStats::default()
        };
        update_inventory_anchor_for_cycle(&db, "new-hash", "PrimarySync", &inventory, &mut stats)
            .await;

        assert_eq!(stats.inventory_drop_warnings, 0);
        assert!(
            db.get_metadata(&inventory_anchor_key(
                "new-hash",
                "PrimarySync",
                &inventory.scope
            ))
            .await
            .expect("read new anchor")
            .is_some(),
            "new hash should get an independent anchor"
        );
    }

    #[tokio::test]
    async fn determine_sync_mode_two_normal_syncs_reuse_stored_token() {
        let db = state::SqliteStateDb::open_in_memory().expect("state db");
        let sync_token_key = sync_token_key("PrimarySync");
        db.set_metadata(&sync_token_key, "stored-token-abc")
            .await
            .expect("set token");

        for cycle in 1..=2 {
            let mode =
                determine_sync_mode(false, 1, Some(&db), &sync_token_key, "PrimarySync").await;
            assert!(
                matches!(mode, download::SyncMode::Incremental { ref zone_sync_token } if zone_sync_token == "stored-token-abc"),
                "normal sync cycle {cycle} should use the stored token, got {mode:?}"
            );
        }

        assert_eq!(
            db.get_metadata(&sync_token_key).await.expect("read token"),
            Some("stored-token-abc".to_string()),
            "mode selection must not consume or clear the stored token"
        );
    }

    #[test]
    fn metadata_refresh_forces_full_enumeration() {
        assert_eq!(
            metadata_refresh_mode_decision(true),
            Some(SyncModeDecision {
                mode: download::SyncMode::Full,
                full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
            })
        );
        assert_eq!(metadata_refresh_mode_decision(false), None);
    }

    #[tokio::test]
    async fn determine_sync_mode_decision_records_explicit_full_reasons() {
        let db = state::SqliteStateDb::open_in_memory().expect("state db");
        let sync_token_key = sync_token_key("PrimarySync");
        db.set_metadata(&sync_token_key, "stored-token-abc")
            .await
            .expect("set token");

        let retry =
            determine_sync_mode_decision(true, 1, Some(&db), &sync_token_key, "PrimarySync").await;
        assert!(matches!(
            retry.mode,
            download::SyncMode::Incremental { ref zone_sync_token }
                if zone_sync_token == "stored-token-abc"
        ));
        assert_eq!(retry.full_enumeration_reason, None);

        db.delete_metadata_by_prefix(SYNC_TOKEN_PREFIX)
            .await
            .expect("clear token");
        let no_token =
            determine_sync_mode_decision(false, 1, Some(&db), &sync_token_key, "PrimarySync").await;
        assert!(matches!(no_token.mode, download::SyncMode::Full));
        assert_eq!(
            no_token.full_enumeration_reason,
            Some(download::FullEnumerationReason::NoStoredToken)
        );
    }
}