use std::{
    path::PathBuf,
    sync::Arc,
    time::{Duration, SystemTime},
};

use async_trait::async_trait;
use crank_artifacts::{
    ArtifactError, ArtifactStore, ReconciliationCursor, ReconciliationMutation,
    ReconciliationNamespace, ReconciliationPresence, ReconciliationRegistration,
    ReconciliationScanStop, TempScanCursor,
};
use crank_registry::{
    ArtifactClaimFinalization, ArtifactClaimFinalizeOutcome, ArtifactClaimOutcome,
    ArtifactClaimRecheckOutcome, ArtifactClaimToken, ArtifactExpiredClaimProbe,
    ClaimArtifactReconciliationRequest, ClaimExpiredArtifactReconciliationRequest,
    PostgresRegistry,
};
use time::OffsetDateTime;
use tokio::time::MissedTickBehavior;
use tracing::{info, warn};

pub const RECONCILIATION_INTERVAL: Duration = Duration::from_secs(15 * 60);
pub const RECONCILIATION_GRACE: Duration = Duration::from_secs(24 * 60 * 60);
pub const RECONCILIATION_LEASE: Duration = Duration::from_secs(5 * 60);
pub const RECONCILIATION_TRAVERSAL_LIMIT: usize = 4_096;
pub const RECONCILIATION_CANDIDATE_LIMIT: usize = 32;
pub const RECONCILIATION_MUTATION_LIMIT: usize = 32;
pub const RECONCILIATION_TICK_DEADLINE: Duration = Duration::from_secs(30);
pub const TEMP_CLEANUP_GRACE: Duration = Duration::from_secs(60 * 60);
pub const TEMP_CLEANUP_SCAN_LIMIT: usize = 512;
pub const TEMP_CLEANUP_DELETE_LIMIT: usize = 16;
pub const IMPORT_JOB_CLEANUP_LIMIT: u32 = 128;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Phase {
    Recovery,
    Sweep,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct StepLimits {
    traversal_syscalls: usize,
    candidates: usize,
    mutations: usize,
}

#[derive(Debug)]
struct Step<C> {
    cursor: Option<C>,
    scan_complete: bool,
    traversal_syscalls: usize,
    candidates: usize,
    mutation_attempts: usize,
    classifications: ReconciliationClassificationCounters,
    physical_mutation_possible: bool,
    retryable: bool,
}

/// Bounded, identity-free classifications observed during one or more scans.
///
/// The fixed fields are the complete telemetry vocabulary: no digest, path,
/// token, source or workspace value can be attached to an item classification.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ReconciliationClassificationCounters {
    pub scanner_malformed: usize,
    pub scanner_unsafe: usize,
    pub recovery_present: usize,
    pub recovery_safety: usize,
    pub recovery_retryable: usize,
    pub registration_integrity: usize,
    pub registration_safety: usize,
    pub registration_retryable: usize,
    pub claim_metadata_conflict: usize,
    pub mutation_integrity: usize,
    pub mutation_safety: usize,
    pub mutation_retryable: usize,
}

impl ReconciliationClassificationCounters {
    fn checked_add(self, other: Self) -> Option<Self> {
        Some(Self {
            scanner_malformed: self
                .scanner_malformed
                .checked_add(other.scanner_malformed)?,
            scanner_unsafe: self.scanner_unsafe.checked_add(other.scanner_unsafe)?,
            recovery_present: self.recovery_present.checked_add(other.recovery_present)?,
            recovery_safety: self.recovery_safety.checked_add(other.recovery_safety)?,
            recovery_retryable: self
                .recovery_retryable
                .checked_add(other.recovery_retryable)?,
            registration_integrity: self
                .registration_integrity
                .checked_add(other.registration_integrity)?,
            registration_safety: self
                .registration_safety
                .checked_add(other.registration_safety)?,
            registration_retryable: self
                .registration_retryable
                .checked_add(other.registration_retryable)?,
            claim_metadata_conflict: self
                .claim_metadata_conflict
                .checked_add(other.claim_metadata_conflict)?,
            mutation_integrity: self
                .mutation_integrity
                .checked_add(other.mutation_integrity)?,
            mutation_safety: self.mutation_safety.checked_add(other.mutation_safety)?,
            mutation_retryable: self
                .mutation_retryable
                .checked_add(other.mutation_retryable)?,
        })
    }

    fn record_registration_error(&mut self, error: ArtifactError) -> bool {
        match error {
            ArtifactError::Integrity
            | ArtifactError::EmptySource
            | ArtifactError::SourceTooLarge => {
                self.registration_integrity += 1;
                false
            }
            ArtifactError::UnsafeRoot
            | ArtifactError::InvalidReference
            | ArtifactError::NotFound => {
                self.registration_safety += 1;
                false
            }
            ArtifactError::Storage => {
                self.registration_retryable += 1;
                true
            }
        }
    }

    fn record_mutation_error(&mut self, error: ArtifactError) -> bool {
        match error {
            ArtifactError::Integrity
            | ArtifactError::EmptySource
            | ArtifactError::SourceTooLarge => self.mutation_integrity += 1,
            ArtifactError::UnsafeRoot
            | ArtifactError::InvalidReference
            | ArtifactError::NotFound => self.mutation_safety += 1,
            ArtifactError::Storage => self.mutation_retryable += 1,
        }
        true
    }

    fn record_recovery_presence(&mut self, presence: ReconciliationPresence) -> bool {
        match presence {
            ReconciliationPresence::Final | ReconciliationPresence::Quarantine => {
                self.recovery_present += 1;
                false
            }
            ReconciliationPresence::Unsafe => {
                self.recovery_safety += 1;
                false
            }
            ReconciliationPresence::Retryable => {
                self.recovery_retryable += 1;
                true
            }
            ReconciliationPresence::Absent => false,
        }
    }

    fn record_mutation_presence(&mut self, presence: ReconciliationPresence) -> bool {
        match presence {
            ReconciliationPresence::Absent => false,
            ReconciliationPresence::Unsafe => {
                self.mutation_safety += 1;
                true
            }
            ReconciliationPresence::Final
            | ReconciliationPresence::Quarantine
            | ReconciliationPresence::Retryable => {
                self.mutation_retryable += 1;
                true
            }
        }
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ReconciliationTickReport {
    pub traversal_syscalls: usize,
    pub candidates: usize,
    pub mutation_attempts: usize,
    pub classifications: ReconciliationClassificationCounters,
    pub recovery_completed: bool,
    pub cycle_completed: bool,
    pub retryable: bool,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
pub enum ReconciliationCoordinatorError {
    #[error("artifact reconciliation backend is unavailable")]
    Backend,
    #[error("artifact reconciliation backend exceeded its bounded contract")]
    Bounds,
    #[error("artifact reconciliation backend exceeded its deadline")]
    Deadline,
}

#[async_trait]
trait ReconciliationBackend: Send + Sync {
    type Cursor: Send;

    async fn step(
        &self,
        phase: Phase,
        cursor: Option<Self::Cursor>,
        limits: StepLimits,
    ) -> Result<Step<Self::Cursor>, ReconciliationCoordinatorError>;
}

struct ReconciliationCoordinator<B: ReconciliationBackend> {
    backend: B,
    phase: Phase,
    cursor: Option<B::Cursor>,
    tick_limits: StepLimits,
}

#[derive(Clone)]
struct PostgresReconciliationBackend {
    registry: PostgresRegistry,
    store: Arc<ArtifactStore>,
}

enum PostgresReconciliationCursor {
    ExpiredClaim(ArtifactExpiredClaimProbe),
    Filesystem(ReconciliationCursor),
}

impl std::fmt::Debug for PostgresReconciliationCursor {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("PostgresReconciliationCursor(..)")
    }
}

impl PostgresReconciliationBackend {
    fn new(registry: PostgresRegistry, store: Arc<ArtifactStore>) -> Self {
        Self { registry, store }
    }

    async fn database_now(&self) -> Result<OffsetDateTime, ReconciliationCoordinatorError> {
        self.registry
            .artifact_reconciliation_now()
            .await
            .map_err(|_| ReconciliationCoordinatorError::Backend)
    }

    async fn presence(
        &self,
        artifact_ref: crank_artifacts::ArtifactRef,
    ) -> Result<ReconciliationPresence, ReconciliationCoordinatorError> {
        let store = Arc::clone(&self.store);
        let result =
            tokio::task::spawn_blocking(move || store.reconciliation_presence(&artifact_ref))
                .await
                .map_err(|_| ReconciliationCoordinatorError::Backend)?;
        Ok(match result {
            Ok(presence) => presence,
            Err(ArtifactError::Storage) => ReconciliationPresence::Retryable,
            Err(_) => ReconciliationPresence::Unsafe,
        })
    }
}

fn reconciliation_lease_expires_at(claimed_at: OffsetDateTime) -> OffsetDateTime {
    claimed_at
        + time::Duration::seconds(
            i64::try_from(RECONCILIATION_LEASE.as_secs())
                .expect("the fixed reconciliation lease fits i64"),
        )
}

impl<B: ReconciliationBackend> ReconciliationCoordinator<B> {
    fn new(backend: B) -> Self {
        Self {
            backend,
            phase: Phase::Recovery,
            cursor: None,
            tick_limits: StepLimits {
                traversal_syscalls: RECONCILIATION_TRAVERSAL_LIMIT,
                candidates: RECONCILIATION_CANDIDATE_LIMIT,
                mutations: RECONCILIATION_MUTATION_LIMIT,
            },
        }
    }

    #[cfg(test)]
    fn with_tick_limits(backend: B, tick_limits: StepLimits) -> Self {
        Self {
            backend,
            phase: Phase::Recovery,
            cursor: None,
            tick_limits,
        }
    }

    async fn tick(&mut self) -> Result<ReconciliationTickReport, ReconciliationCoordinatorError> {
        let mut report = ReconciliationTickReport::default();

        loop {
            let limits = StepLimits {
                traversal_syscalls: self
                    .tick_limits
                    .traversal_syscalls
                    .saturating_sub(report.traversal_syscalls),
                candidates: self
                    .tick_limits
                    .candidates
                    .saturating_sub(report.candidates),
                mutations: self
                    .tick_limits
                    .mutations
                    .saturating_sub(report.mutation_attempts),
            };
            if limits.traversal_syscalls == 0 || limits.candidates == 0 || limits.mutations == 0 {
                break;
            }

            let step = match self
                .backend
                .step(self.phase, self.cursor.take(), limits)
                .await
            {
                Ok(step) => step,
                Err(error) => {
                    self.cursor = None;
                    self.phase = Phase::Recovery;
                    return Err(error);
                }
            };
            if step.traversal_syscalls > limits.traversal_syscalls
                || step.candidates > limits.candidates
                || step.mutation_attempts > limits.mutations
                || step.mutation_attempts > step.candidates
            {
                self.cursor = None;
                self.phase = Phase::Recovery;
                return Err(ReconciliationCoordinatorError::Bounds);
            }

            report.traversal_syscalls += step.traversal_syscalls;
            report.candidates += step.candidates;
            report.mutation_attempts += step.mutation_attempts;
            report.classifications = report
                .classifications
                .checked_add(step.classifications)
                .ok_or_else(|| {
                    self.cursor = None;
                    self.phase = Phase::Recovery;
                    ReconciliationCoordinatorError::Bounds
                })?;
            report.retryable |= step.retryable;

            let made_progress =
                step.traversal_syscalls != 0 || step.candidates != 0 || step.mutation_attempts != 0;
            if step.physical_mutation_possible {
                // A cursor is valid only while the namespace is unchanged.  A
                // possible rename/unlink includes ambiguous fsync outcomes.
                drop(step.cursor);
                self.cursor = None;
                self.phase = Phase::Recovery;
            } else {
                self.cursor = step.cursor;
                if step.scan_complete {
                    self.cursor = None;
                    match self.phase {
                        Phase::Recovery => {
                            report.recovery_completed = true;
                            self.phase = Phase::Sweep;
                        }
                        Phase::Sweep => {
                            report.cycle_completed = true;
                            self.phase = Phase::Recovery;
                            break;
                        }
                    }
                }
            }

            if step.retryable || (!made_progress && !step.scan_complete) {
                break;
            }
        }

        Ok(report)
    }
}

#[async_trait]
impl ReconciliationBackend for PostgresReconciliationBackend {
    type Cursor = PostgresReconciliationCursor;

    async fn step(
        &self,
        phase: Phase,
        cursor: Option<Self::Cursor>,
        limits: StepLimits,
    ) -> Result<Step<Self::Cursor>, ReconciliationCoordinatorError> {
        let mut recovered_candidates = 0;
        let mut classifications = ReconciliationClassificationCounters::default();
        let mut recovery_retryable = false;
        let mut filesystem_cursor = None;
        if phase == Phase::Recovery {
            let after = match cursor {
                Some(PostgresReconciliationCursor::ExpiredClaim(probe)) => Some(probe),
                Some(PostgresReconciliationCursor::Filesystem(cursor)) => {
                    filesystem_cursor = Some(cursor);
                    None
                }
                None => None,
            };
            if filesystem_cursor.is_none() {
                let database_now = self.database_now().await?;
                let probe_limit = u32::try_from(limits.candidates).unwrap_or(u32::MAX);
                let probes = self
                    .registry
                    .list_expired_artifact_reconciliation_probes_after(
                        database_now,
                        after.as_ref(),
                        probe_limit,
                    )
                    .await
                    .map_err(|_| ReconciliationCoordinatorError::Backend)?;
                let page_full = probes.len() == probe_limit as usize;
                let mut last_processed = after;
                for probe in probes {
                    recovered_candidates += 1;
                    last_processed = Some(probe.clone());
                    let presence = self.presence(probe.artifact_ref().clone()).await?;
                    if presence != ReconciliationPresence::Absent {
                        recovery_retryable |= classifications.record_recovery_presence(presence);
                        continue;
                    }

                    let claimed_at = self.database_now().await?;
                    let claim = self
                        .registry
                        .claim_expired_artifact_reconciliation(
                            &probe,
                            ClaimExpiredArtifactReconciliationRequest {
                                token: ArtifactClaimToken::generate(),
                                claimed_at,
                                lease_expires_at: reconciliation_lease_expires_at(claimed_at),
                            },
                        )
                        .await
                        .map_err(|_| ReconciliationCoordinatorError::Backend)?;
                    let Some(claim) = claim else {
                        continue;
                    };

                    // Recheck after the DB CAS. A concurrent filesystem put
                    // does not consult PostgreSQL, so the first probe alone
                    // cannot prove absence at finalization time.
                    let presence = self.presence(claim.artifact_ref().clone()).await?;
                    if presence != ReconciliationPresence::Absent {
                        classifications.record_recovery_presence(presence);
                        recovery_retryable = true;
                        continue;
                    }

                    let recovered_at = self.database_now().await?;
                    let result = self
                        .registry
                        .recover_artifact_reconciliation_claim(
                            &claim,
                            ArtifactClaimFinalization::AlreadyAbsent,
                            recovered_at,
                        )
                        .await;
                    let recovered = matches!(result, Ok(ArtifactClaimFinalizeOutcome::Unavailable));
                    if recovered {
                        let presence = self.presence(claim.artifact_ref().clone()).await?;
                        if presence != ReconciliationPresence::Absent {
                            classifications.record_recovery_presence(presence);
                            recovery_retryable = true;
                        }
                    }
                    return Ok(Step {
                        cursor: last_processed.map(PostgresReconciliationCursor::ExpiredClaim),
                        scan_complete: false,
                        traversal_syscalls: 0,
                        candidates: recovered_candidates,
                        mutation_attempts: 1,
                        classifications,
                        physical_mutation_possible: false,
                        retryable: recovery_retryable || !recovered,
                    });
                }

                if page_full {
                    return Ok(Step {
                        cursor: last_processed.map(PostgresReconciliationCursor::ExpiredClaim),
                        scan_complete: false,
                        traversal_syscalls: 0,
                        candidates: recovered_candidates,
                        mutation_attempts: 0,
                        classifications,
                        physical_mutation_possible: false,
                        retryable: recovery_retryable,
                    });
                }
            }
        } else if let Some(PostgresReconciliationCursor::Filesystem(cursor)) = cursor {
            filesystem_cursor = Some(cursor);
        }

        let scan_candidate_limit = limits.candidates.saturating_sub(recovered_candidates);
        if scan_candidate_limit == 0 {
            return Ok(Step {
                cursor: filesystem_cursor.map(PostgresReconciliationCursor::Filesystem),
                scan_complete: false,
                traversal_syscalls: 0,
                candidates: recovered_candidates,
                mutation_attempts: 0,
                classifications,
                physical_mutation_possible: false,
                retryable: recovery_retryable,
            });
        }
        let namespace = match phase {
            Phase::Recovery => ReconciliationNamespace::Quarantine,
            Phase::Sweep => ReconciliationNamespace::Final,
        };
        let store = Arc::clone(&self.store);
        let (mut scan, candidates) = tokio::task::spawn_blocking(move || {
            store.scan_reconciliation_namespace(
                namespace,
                filesystem_cursor,
                limits.traversal_syscalls,
                scan_candidate_limit,
            )
        })
        .await
        .map_err(|_| ReconciliationCoordinatorError::Backend)?
        .map_err(|_| ReconciliationCoordinatorError::Backend)?;

        let mut step = Step {
            cursor: scan
                .continuation
                .take()
                .map(PostgresReconciliationCursor::Filesystem),
            scan_complete: scan.stop == ReconciliationScanStop::Complete,
            traversal_syscalls: scan.traversal_syscalls,
            candidates: recovered_candidates,
            mutation_attempts: 0,
            classifications: classifications
                .checked_add(ReconciliationClassificationCounters {
                    scanner_malformed: scan.malformed,
                    scanner_unsafe: scan.unsafe_entries,
                    ..ReconciliationClassificationCounters::default()
                })
                .ok_or(ReconciliationCoordinatorError::Bounds)?,
            physical_mutation_possible: false,
            retryable: recovery_retryable || scan.stop == ReconciliationScanStop::Retryable,
        };

        for candidate in candidates {
            if step.candidates == limits.candidates || step.mutation_attempts == limits.mutations {
                break;
            }
            step.candidates += 1;
            let registration_grace = match phase {
                Phase::Recovery => Duration::ZERO,
                Phase::Sweep => RECONCILIATION_GRACE,
            };
            let store = Arc::clone(&self.store);
            let registration = tokio::task::spawn_blocking(move || {
                let registration = store.register_reconciliation(
                    &candidate,
                    registration_grace,
                    SystemTime::now(),
                );
                (candidate, registration)
            })
            .await
            .map_err(|_| ReconciliationCoordinatorError::Backend)?;
            let (candidate, registration) = registration;
            let artifact = match registration {
                Ok(ReconciliationRegistration::Registered(artifact)) => artifact,
                Ok(ReconciliationRegistration::NotEligible) => continue,
                Err(error) => {
                    step.retryable |= step.classifications.record_registration_error(error);
                    continue;
                }
            };

            let claimed_at = self.database_now().await?;
            let lease_expires_at = reconciliation_lease_expires_at(claimed_at);
            let detached_grace = time::Duration::seconds(
                i64::try_from(RECONCILIATION_GRACE.as_secs())
                    .expect("the fixed reconciliation grace fits i64"),
            );
            let claim = self
                .registry
                .claim_artifact_reconciliation(ClaimArtifactReconciliationRequest {
                    artifact: &artifact,
                    token: ArtifactClaimToken::generate(),
                    claimed_at,
                    lease_expires_at,
                    detached_grace,
                })
                .await
                .map_err(|_| ReconciliationCoordinatorError::Backend)?;
            let claim = match claim {
                ArtifactClaimOutcome::Claimed(claim) => claim,
                ArtifactClaimOutcome::HeldByOther
                | ArtifactClaimOutcome::ActiveReference
                | ArtifactClaimOutcome::DetachedReferenceInGrace => continue,
                ArtifactClaimOutcome::MetadataConflict => {
                    step.classifications.claim_metadata_conflict += 1;
                    continue;
                }
            };

            let rechecked_at = self.database_now().await?;
            let recheck = self
                .registry
                .recheck_artifact_reconciliation_claim(&claim, rechecked_at, detached_grace)
                .await
                .map_err(|_| ReconciliationCoordinatorError::Backend)?;
            if recheck != ArtifactClaimRecheckOutcome::Mutate {
                continue;
            }

            step.mutation_attempts += 1;
            step.physical_mutation_possible = true;
            let store = Arc::clone(&self.store);
            let mutation = match tokio::task::spawn_blocking(move || match phase {
                Phase::Recovery => store.delete_quarantined_reconciliation(candidate),
                Phase::Sweep => store.quarantine_reconciliation(candidate),
            })
            .await
            {
                Ok(mutation) => mutation,
                Err(_) => Err(ArtifactError::Storage),
            };
            let mut observation = match mutation {
                Ok(ReconciliationMutation::Quarantined)
                | Ok(ReconciliationMutation::AlreadyQuarantined) => {
                    ArtifactClaimFinalization::Quarantined
                }
                Ok(ReconciliationMutation::Deleted) => ArtifactClaimFinalization::Deleted,
                Ok(ReconciliationMutation::AlreadyAbsent) => {
                    ArtifactClaimFinalization::AlreadyAbsent
                }
                Ok(ReconciliationMutation::Retryable) => {
                    step.classifications.mutation_retryable += 1;
                    step.retryable = true;
                    ArtifactClaimFinalization::Retryable
                }
                Err(error) => {
                    step.retryable |= step.classifications.record_mutation_error(error);
                    ArtifactClaimFinalization::Retryable
                }
            };
            if matches!(
                observation,
                ArtifactClaimFinalization::Deleted | ArtifactClaimFinalization::AlreadyAbsent
            ) {
                // `unavailable` means both canonical locations are confirmed
                // absent. A concurrent republish can recreate final bytes
                // without consulting PostgreSQL, so unlink alone is not proof.
                let presence = self.presence(claim.artifact_ref().clone()).await?;
                observation = match presence {
                    ReconciliationPresence::Absent => ArtifactClaimFinalization::AlreadyAbsent,
                    presence => {
                        step.retryable |= step.classifications.record_mutation_presence(presence);
                        ArtifactClaimFinalization::Retryable
                    }
                };
            }
            let observed_at = self.database_now().await?;
            let finalization = match phase {
                Phase::Recovery => {
                    self.registry
                        .recover_artifact_reconciliation_claim(&claim, observation, observed_at)
                        .await
                }
                Phase::Sweep => {
                    self.registry
                        .finalize_artifact_reconciliation_claim(&claim, observation, observed_at)
                        .await
                }
            };
            let finalized_as_expected =
                match observation {
                    ArtifactClaimFinalization::Deleted
                    | ArtifactClaimFinalization::AlreadyAbsent => matches!(
                        finalization.as_ref(),
                        Ok(ArtifactClaimFinalizeOutcome::Unavailable)
                    ),
                    ArtifactClaimFinalization::Quarantined
                    | ArtifactClaimFinalization::Retryable => matches!(
                        finalization.as_ref(),
                        Ok(ArtifactClaimFinalizeOutcome::Retained)
                    ),
                };
            if !finalized_as_expected {
                step.retryable = true;
            }
            if matches!(
                finalization.as_ref(),
                Ok(ArtifactClaimFinalizeOutcome::Unavailable)
            ) {
                let presence = self.presence(claim.artifact_ref().clone()).await?;
                step.retryable |= step.classifications.record_mutation_presence(presence);
            }

            // The scan cursor and every remaining observation came from the
            // pre-mutation namespace. Return immediately so the coordinator
            // drops them and restarts a recovery cycle from `None`.
            break;
        }

        Ok(step)
    }
}

/// Opens the protected root without running filesystem syscalls on Tokio's
/// async executor.
pub async fn open_reconciliation_store(
    root: PathBuf,
) -> Result<Arc<ArtifactStore>, ReconciliationCoordinatorError> {
    tokio::task::spawn_blocking(move || ArtifactStore::open(root).map(Arc::new))
        .await
        .map_err(|_| ReconciliationCoordinatorError::Backend)?
        .map_err(|_| ReconciliationCoordinatorError::Backend)
}

/// Runs the immediate bounded startup cycle, then schedules 15-minute ticks.
/// Recovery remains ahead of the final sweep even when it spans several ticks.
pub async fn spawn_artifact_reconciliation(registry: PostgresRegistry, store: Arc<ArtifactStore>) {
    let cleanup_registry = registry.clone();
    let backend = PostgresReconciliationBackend::new(registry, Arc::clone(&store));
    let mut coordinator = ReconciliationCoordinator::new(backend);
    let mut temp_cursor = None;
    let startup_maintenance = async {
        observe_import_job_cleanup(&cleanup_registry).await;
        observe_tick(run_bounded_reconciliation_tick(&mut coordinator).await);
        temp_cursor = observe_stale_temp_cleanup(Arc::clone(&store), None).await;
    };
    if tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, startup_maintenance)
        .await
        .is_err()
    {
        warn!(
            name: "admin.artifact_maintenance.startup_deadline",
            "startup maintenance exceeded its shared deadline; remaining work will resume in background"
        );
    }

    tokio::spawn(async move {
        let mut interval = tokio::time::interval(RECONCILIATION_INTERVAL);
        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
        // The immediate tick was run above as part of startup composition.
        interval.tick().await;
        loop {
            interval.tick().await;
            observe_import_job_cleanup(&cleanup_registry).await;
            observe_tick(run_bounded_reconciliation_tick(&mut coordinator).await);
            temp_cursor = observe_stale_temp_cleanup(Arc::clone(&store), temp_cursor).await;
        }
    });
}

async fn observe_import_job_cleanup(registry: &PostgresRegistry) {
    let cleanup = async {
        let mut deleted_jobs = 0_u64;
        let mut detached_sources = 0_u64;
        loop {
            let report = registry
                .cleanup_expired_import_jobs(IMPORT_JOB_CLEANUP_LIMIT)
                .await?;
            deleted_jobs = deleted_jobs.saturating_add(report.deleted_jobs);
            detached_sources = detached_sources.saturating_add(report.detached_sources);
            if !report.more_work {
                return Ok::<_, crank_registry::RegistryError>((deleted_jobs, detached_sources));
            }
            // Each pass commits independently. Yield before requesting the
            // next lock batch so normal import traffic can make progress.
            tokio::task::yield_now().await;
        }
    };
    match tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, cleanup).await {
        Ok(Ok((deleted_jobs, detached_sources))) => info!(
            name: "admin.openapi_import_cleanup.tick",
            deleted_jobs,
            detached_sources,
            more_work = false,
            "OpenAPI import cleanup tick completed"
        ),
        Ok(Err(_)) => warn!(
            name: "admin.openapi_import_cleanup.failed",
            error_category = "registry",
            "OpenAPI import cleanup tick will be retried"
        ),
        Err(_) => warn!(
            name: "admin.openapi_import_cleanup.failed",
            error_category = "deadline",
            "OpenAPI import cleanup tick exceeded its deadline"
        ),
    }
}

async fn run_bounded_reconciliation_tick<B: ReconciliationBackend>(
    coordinator: &mut ReconciliationCoordinator<B>,
) -> Result<ReconciliationTickReport, ReconciliationCoordinatorError> {
    tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, coordinator.tick())
        .await
        .map_err(|_| ReconciliationCoordinatorError::Deadline)?
}

async fn observe_stale_temp_cleanup(
    store: Arc<ArtifactStore>,
    cursor: Option<TempScanCursor>,
) -> Option<TempScanCursor> {
    let progress = Arc::new(std::sync::Mutex::new(cursor));
    let cleanup_progress = Arc::clone(&progress);
    let cleanup = async move {
        let mut scanned = 0_usize;
        let mut omitted = 0_usize;
        let mut deleted = 0_usize;
        let mut retryable = 0_usize;
        loop {
            let scanner = Arc::clone(&store);
            let cursor = cleanup_progress
                .lock()
                .map_err(|_| ArtifactError::Storage)?
                .clone();
            let (scan, candidates, next_cursor) = tokio::task::spawn_blocking(move || {
                scanner.scan_stale_temps_after(
                    TEMP_CLEANUP_GRACE,
                    cursor,
                    TEMP_CLEANUP_SCAN_LIMIT,
                    TEMP_CLEANUP_DELETE_LIMIT,
                )
            })
            .await
            .map_err(|_| ArtifactError::Storage)??;
            scanned = scanned.saturating_add(scan.scanned);
            omitted = omitted.saturating_add(scan.omitted);
            *cleanup_progress
                .lock()
                .map_err(|_| ArtifactError::Storage)? = Some(next_cursor);

            for candidate in candidates {
                let store = Arc::clone(&store);
                match tokio::task::spawn_blocking(move || store.delete_stale_temp(candidate)).await
                {
                    Ok(Ok(())) => deleted += 1,
                    Ok(Err(_)) | Err(_) => retryable += 1,
                }
            }
            if scan.complete {
                return Ok::<_, ArtifactError>((
                    scanned,
                    omitted,
                    deleted,
                    retryable,
                    cleanup_progress
                        .lock()
                        .map_err(|_| ArtifactError::Storage)?
                        .clone(),
                ));
            }
            tokio::task::yield_now().await;
        }
    };

    match tokio::time::timeout(RECONCILIATION_TICK_DEADLINE, cleanup).await {
        Ok(Ok((scanned, omitted, deleted, retryable, next_cursor))) => {
            info!(
                name: "admin.artifact_temp_cleanup.tick",
                scanned,
                omitted,
                deleted,
                retryable,
                "artifact temporary-file cleanup tick completed"
            );
            next_cursor
        }
        Ok(Err(_)) => {
            warn!(
                name: "admin.artifact_temp_cleanup.failed",
                error_category = "backend",
                "artifact temporary-file cleanup tick will be retried"
            );
            progress.lock().ok().and_then(|cursor| cursor.clone())
        }
        Err(_) => {
            warn!(
                name: "admin.artifact_temp_cleanup.failed",
                error_category = "deadline",
                "artifact temporary-file cleanup tick exceeded its deadline"
            );
            progress.lock().ok().and_then(|cursor| cursor.clone())
        }
    }
}

fn observe_tick(result: Result<ReconciliationTickReport, ReconciliationCoordinatorError>) {
    match result {
        Ok(report) => info!(
            name: "admin.artifact_reconciliation.tick",
            traversal_syscalls = report.traversal_syscalls,
            candidates = report.candidates,
            mutation_attempts = report.mutation_attempts,
            scanner_malformed = report.classifications.scanner_malformed,
            scanner_unsafe = report.classifications.scanner_unsafe,
            recovery_present = report.classifications.recovery_present,
            recovery_safety = report.classifications.recovery_safety,
            recovery_retryable = report.classifications.recovery_retryable,
            registration_integrity = report.classifications.registration_integrity,
            registration_safety = report.classifications.registration_safety,
            registration_retryable = report.classifications.registration_retryable,
            claim_metadata_conflict = report.classifications.claim_metadata_conflict,
            mutation_integrity = report.classifications.mutation_integrity,
            mutation_safety = report.classifications.mutation_safety,
            mutation_retryable = report.classifications.mutation_retryable,
            recovery_completed = report.recovery_completed,
            cycle_completed = report.cycle_completed,
            retryable = report.retryable,
            "artifact reconciliation tick completed"
        ),
        Err(error) => warn!(
            name: "admin.artifact_reconciliation.failed",
            error_category = match error {
                ReconciliationCoordinatorError::Backend => "backend",
                ReconciliationCoordinatorError::Bounds => "bounds",
                ReconciliationCoordinatorError::Deadline => "deadline",
            },
            "artifact reconciliation tick will be retried"
        ),
    }
}

#[cfg(test)]
#[path = "reconciliation/tests.rs"]
mod tests;