cargo-rail 0.29.0

//! Exact machine installation for transparent local compiler reuse.

use std::ffi::OsStr;
use std::fs::{self, File};
use std::io::{Read as _, Seek as _, Write as _};
use std::path::{Path, PathBuf};
use std::process::Command;

use rscrypto::Sha256;
use serde::{Deserialize, Serialize};
use toml_edit::{DocumentMut, Item, Table, Value};

use crate::cache::cas::{LocalCacheSelection, LocalCas};
use crate::cargo::CargoConfigSnapshot;
use crate::error::{RailError, RailResult};
use crate::source::ContentDigest;

const INSTALLATION_VERSION: u32 = 5;
const INSTALLATION_DIRECTORY: &str = "compiler-cache-v1";
const INSTALLATION_LOCK_FILE: &str = "compiler-cache-v1.lock";
const RECEIPT_FILE: &str = "setup.json";
const SESSION_MEMO_FILE: &str = "session.json";
const SESSION_LOCK_FILE: &str = "session.lock";
const USAGE_FILE: &str = "usage-v1.log";
const EARLY_BYPASS_FILE: &str = "early-bypass-v1.log";
const FAILURE_COUNTERS_FILE: &str = "failure-counters-v1.json";
const FAILURE_COUNTERS_LOCK_FILE: &str = "failure-counters-v1.lock";
const INSTALLATION_QUARANTINE_DIRECTORY: &str = "installation-quarantine-v1";
pub(crate) const READINESS_WORKSPACE_ENV: &str = "CARGO_RAIL_CACHE_READINESS_WORKSPACE";
const COORDINATOR_STATE_PREFIX: &str = "remote-coordinator-v1-";
const MAX_RECEIPT_BYTES: u64 = 64 * 1024;
const MAX_SESSION_MEMO_BYTES: u64 = 256 * 1024;
const MAX_USAGE_BYTES: u64 = 64 * 1024;
const MAX_EARLY_BYPASS_BYTES: u64 = 64 * 1024;
const MAX_EARLY_BYPASS_REASON_BYTES: usize = 96;
const MAX_FAILURE_COUNTERS_BYTES: u64 = 4 * 1024;
const MAX_DISTRIBUTED_PLACEMENT_HISTORY_BYTES: u64 = 64 * 1024;
#[cfg(not(windows))]
const WRAPPER_FILE: &str = "cargo-rail-native-rustc-wrapper";
#[cfg(windows)]
const WRAPPER_FILE: &str = "cargo-rail-native-rustc-wrapper.exe";
#[cfg(not(windows))]
const WORKER_FILE: &str = "cargo-rail-native-rustc-worker";
#[cfg(windows)]
const WORKER_FILE: &str = "cargo-rail-native-rustc-worker.exe";
#[cfg(not(windows))]
const DISTRIBUTED_WORKER_FILE: &str = "cargo-rail-distributed-worker";
#[cfg(windows)]
const DISTRIBUTED_WORKER_FILE: &str = "cargo-rail-distributed-worker.exe";
const DISTRIBUTED_AUTHORITY_FILE: &str = "distributed-authority.pem";
const DISTRIBUTED_CLIENT_CERTIFICATE_FILE: &str = "distributed-client.pem";
const DISTRIBUTED_CLIENT_PRIVATE_KEY_FILE: &str = "distributed-client.key";
const DISTRIBUTED_PLACEMENT_HISTORY_FILE: &str = "distributed-placement-v1.json";
const DISTRIBUTED_PLACEMENT_LOCK_FILE: &str = "distributed-placement-v1.lock";
const DIRECT_LAUNCHER_ENV: &str = "CARGO_RAIL_DIRECT_CACHE_LAUNCHER";

/// Requested machine policy. Omitted values preserve an existing installation.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct SetupRequest {
    pub(crate) profile_id: Option<String>,
    pub(crate) distributed_local: bool,
    pub(crate) distributed_endpoint: Option<String>,
    pub(crate) distributed_server_name: Option<String>,
    pub(crate) distributed_capability: Option<String>,
    pub(crate) distributed_authority: Option<PathBuf>,
    pub(crate) distributed_client_certificate: Option<PathBuf>,
    pub(crate) distributed_client_private_key: Option<PathBuf>,
    pub(crate) distributed_policy: Option<String>,
    pub(crate) local_dir: Option<PathBuf>,
    pub(crate) max_bytes: Option<u64>,
    pub(crate) remote_url: Option<String>,
    pub(crate) remote_mode: Option<String>,
    pub(crate) remote_environment: Vec<String>,
    pub(crate) root_portability: Option<String>,
    pub(crate) local_only: bool,
}

/// One lossless Cargo configuration and private-state mutation preview.
pub(crate) struct SetupPlan {
    cargo_home: PathBuf,
    config_path: PathBuf,
    config_before: Option<Vec<u8>>,
    config_after: Vec<u8>,
    receipt_before: Option<Vec<u8>>,
    stale_receipt_quarantine: Option<PathBuf>,
    receipt: InstallationReceipt,
    wrapper: ExecutableSetup,
    worker: ExecutableSetup,
    compiler_components: Vec<CompilerComponentSetup>,
    retired_compiler_components: Vec<(InstalledCompilerComponent, Option<String>)>,
    distributed_worker: Option<ExecutableSetup>,
    source_distributed_identity: Option<SourceDistributedIdentity>,
    profile: crate::cache::profile::ProfileSetupPlan,
    pending: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct InstalledExecutable {
    digest: String,
    generation: Vec<u8>,
}

struct ExecutableSetup {
    source: PathBuf,
    source_digest: String,
    before: Option<InstalledExecutable>,
}

struct CompilerComponentSetup {
    file: ExecutableSetup,
    executable: bool,
    mode_matches: bool,
}

impl CompilerComponentSetup {
    fn requires_installation(&self) -> bool {
        self.file.requires_installation() || !self.mode_matches
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct InstalledCompilerComponent {
    path: PathBuf,
    digest: String,
    generation: Vec<u8>,
}

impl ExecutableSetup {
    fn requires_installation(&self) -> bool {
        self.before
            .as_ref()
            .is_none_or(|before| before.digest != self.source_digest)
    }

    fn preserved_generation(&self) -> Option<&[u8]> {
        match &self.before {
            Some(before) if before.digest == self.source_digest => Some(&before.generation),
            _ => None,
        }
    }
}

struct SourceDistributedIdentity {
    authority: PathBuf,
    authority_digest: String,
    client_certificate: PathBuf,
    client_certificate_digest: String,
    client_private_key: PathBuf,
    client_private_key_digest: String,
}

/// One exact uninstall preview. The local compiler-result cache is deliberately
/// outside this authority and is never removed here.
pub(crate) struct RemovalPlan {
    config_path: PathBuf,
    config_before: Option<Vec<u8>>,
    config_after: Option<Vec<u8>>,
    receipt_before: Option<Vec<u8>>,
    receipt: Option<InstallationReceipt>,
    wrapper_before_digest: Option<String>,
    worker_before_digest: Option<String>,
    compiler_component_before_digests: Vec<Option<String>>,
    distributed_worker_before_digest: Option<String>,
    distributed_identity_before_digests: Vec<Option<String>>,
}

impl RemovalPlan {
    pub(crate) fn pending(&self) -> bool {
        self.receipt.is_some()
    }

    pub(crate) fn config_path(&self) -> &Path {
        &self.config_path
    }

    pub(crate) fn wrapper_path(&self) -> Option<&Path> {
        self.receipt.as_ref().map(InstallationReceipt::wrapper_path)
    }

    pub(crate) fn receipt_path(&self) -> Option<PathBuf> {
        self.receipt
            .as_ref()
            .and_then(|receipt| receipt.installation_directory().ok())
            .map(|directory| directory.join(RECEIPT_FILE))
    }

    pub(crate) fn config_action(&self) -> &'static str {
        match (&self.config_before, &self.config_after) {
            (None, _) => "unchanged",
            (Some(_), None) => "remove_file",
            (Some(before), Some(after)) if before == after => "unchanged",
            (Some(_), Some(_)) => "remove_field",
        }
    }
}

impl SetupPlan {
    pub(crate) const fn pending(&self) -> bool {
        self.pending
    }

    pub(crate) fn config_path(&self) -> &Path {
        &self.config_path
    }

    pub(crate) fn wrapper_path(&self) -> &Path {
        &self.receipt.wrapper_path
    }

    pub(crate) fn cache_base(&self) -> &Path {
        self.profile.profile().cache().base()
    }

    pub(crate) fn max_bytes(&self) -> u64 {
        self.profile.profile().cache().max_bytes()
    }

    pub(crate) fn remote_selection(&self) -> RailResult<Option<crate::remote_cache::RemoteCacheSelection>> {
        self.profile.profile().remote_selection()
    }

    pub(crate) fn distributed_mode(&self) -> Option<&str> {
        self.receipt
            .distributed
            .as_ref()
            .map(|distributed| distributed.mode.as_str())
    }

    pub(crate) fn distributed_policy(&self) -> Option<&'static str> {
        self.receipt
            .distributed
            .as_ref()
            .and_then(|distributed| distributed.placement)
            .map(DistributedPlacementPolicy::as_str)
    }

    pub(crate) fn root_portability(&self) -> &'static str {
        self.profile.profile().root_portability().as_str()
    }

    pub(crate) fn profile_id(&self) -> &str {
        self.profile.profile().profile_id()
    }

    pub(crate) fn receipt_path(&self) -> RailResult<PathBuf> {
        Ok(self.receipt.installation_directory()?.join(RECEIPT_FILE))
    }

    pub(crate) fn config_action(&self) -> &'static str {
        match self.config_before.as_deref() {
            None => "create_file_and_field",
            Some(before) if before == self.config_after => "unchanged",
            Some(_) => "set_field",
        }
    }

    pub(crate) const fn private_state_action(&self) -> &'static str {
        if self.stale_receipt_quarantine.is_some() {
            "quarantine_and_reinstall"
        } else if self.pending {
            "install_or_repair"
        } else {
            "verify"
        }
    }

    pub(crate) fn quarantine_receipt_path(&self) -> Option<&Path> {
        self.stale_receipt_quarantine.as_deref()
    }
}

/// Private authority loaded by the installed wrapper after acquisition-free gates.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct InstallationReceipt {
    version: u32,
    authority: String,
    cargo_home: PathBuf,
    config_path: PathBuf,
    config_created: bool,
    build_table_created: bool,
    wrapper_path: PathBuf,
    wrapper_digest: String,
    wrapper_generation: Vec<u8>,
    worker_path: PathBuf,
    worker_digest: String,
    worker_generation: Vec<u8>,
    compiler_components: Vec<InstalledCompilerComponent>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    distributed: Option<InstalledDistributedQualification>,
    #[serde(skip)]
    active_profile: Option<crate::cache::profile::InstalledCacheProfile>,
    #[serde(skip)]
    active_profile_lock: Option<crate::cache::profile::ProfileLifecycleLock>,
    #[serde(skip)]
    readiness_local_only: bool,
}

#[derive(Deserialize)]
struct LegacyInstallationReceipt {
    version: u32,
    authority: String,
    cargo_home: PathBuf,
    config_path: PathBuf,
    wrapper_path: PathBuf,
    #[serde(default)]
    cache: Option<LocalCacheSelection>,
    #[serde(default)]
    root_portability: Option<InstalledRootPortability>,
}

/// Explicit machine authority for sharing remapped compiler results across checkout roots.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub(crate) enum InstalledRootPortability {
    #[default]
    Physical,
    Remap,
}

impl InstalledRootPortability {
    pub(crate) const fn is_physical(&self) -> bool {
        matches!(self, Self::Physical)
    }

    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Physical => "physical",
            Self::Remap => "remap",
        }
    }
}

impl From<InstalledRootPortability> for crate::cache::profile::RootPortability {
    fn from(value: InstalledRootPortability) -> Self {
        match value {
            InstalledRootPortability::Physical => Self::Physical,
            InstalledRootPortability::Remap => Self::Remap,
        }
    }
}

impl From<crate::cache::profile::RootPortability> for InstalledRootPortability {
    fn from(value: crate::cache::profile::RootPortability) -> Self {
        match value {
            crate::cache::profile::RootPortability::Physical => Self::Physical,
            crate::cache::profile::RootPortability::Remap => Self::Remap,
        }
    }
}

/// Optional machine-owned execution authority installed beside the wrapper.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct InstalledDistributedQualification {
    mode: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    placement: Option<DistributedPlacementPolicy>,
    worker_path: PathBuf,
    worker_digest: String,
    worker_generation: Vec<u8>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    mutual_tls: Option<InstalledMutualTlsDirect>,
}

/// Machine-owned decision boundary for one direct mTLS worker.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DistributedPlacementPolicy {
    /// Delegate only when bounded retained observations predict a material win.
    Automatic,
    /// Collect explicit qualification observations even when no cost history exists.
    Qualification,
}

impl DistributedPlacementPolicy {
    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Automatic => "automatic",
            Self::Qualification => "qualification",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct InstalledMutualTlsDirect {
    endpoint: String,
    server_name: String,
    worker_capability_id: String,
    authority_certificate: PathBuf,
    authority_digest: String,
    authority_generation: Vec<u8>,
    client_certificate: PathBuf,
    client_certificate_digest: String,
    client_certificate_generation: Vec<u8>,
    client_private_key: PathBuf,
    client_private_key_digest: String,
    client_private_key_generation: Vec<u8>,
}

pub(crate) struct InstallationSessionLock {
    _file: File,
}

pub(super) struct InstallationMutationLock {
    _file: File,
}

#[derive(Debug, Clone)]
pub(crate) struct InstalledLocalSelection {
    cache: LocalCacheSelection,
    remote: Option<crate::remote_cache::InstalledRemoteCache>,
    _profile: crate::cache::profile::ProfileLifecycleLock,
}

impl InstalledLocalSelection {
    pub(crate) fn cache(&self) -> &LocalCacheSelection {
        &self.cache
    }

    pub(crate) fn remote(&self) -> Option<&crate::remote_cache::InstalledRemoteCache> {
        self.remote.as_ref()
    }
}

pub(crate) struct InstalledRemoteSelection {
    remote: Option<crate::remote_cache::InstalledRemoteCache>,
    _profile: crate::cache::profile::ProfileLifecycleLock,
}

impl InstalledRemoteSelection {
    pub(crate) fn remote(&self) -> Option<&crate::remote_cache::InstalledRemoteCache> {
        self.remote.as_ref()
    }
}

/// Stable fail-open classes retained by ordinary cache status.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum NativeCacheFailureReason {
    ActionCapture,
    ActionIdentity,
    PostExecutionWitness,
}

impl NativeCacheFailureReason {
    const ALL: [Self; 3] = [Self::ActionCapture, Self::ActionIdentity, Self::PostExecutionWitness];

    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::ActionCapture => "complete_action_capture_unavailable",
            Self::ActionIdentity => "complete_action_identity_unavailable",
            Self::PostExecutionWitness => "post_execution_witness_validation_unavailable",
        }
    }

    pub(crate) fn from_reason(reason: &str) -> Option<Self> {
        Self::ALL.into_iter().find(|candidate| candidate.as_str() == reason)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct NativeCacheFailureCounters {
    schema_version: u32,
    complete_action_capture_unavailable: u64,
    complete_action_identity_unavailable: u64,
    post_execution_witness_validation_unavailable: u64,
}

impl Default for NativeCacheFailureCounters {
    fn default() -> Self {
        Self {
            schema_version: 1,
            complete_action_capture_unavailable: 0,
            complete_action_identity_unavailable: 0,
            post_execution_witness_validation_unavailable: 0,
        }
    }
}

impl NativeCacheFailureCounters {
    fn increment(&mut self, reason: NativeCacheFailureReason) {
        let counter = match reason {
            NativeCacheFailureReason::ActionCapture => &mut self.complete_action_capture_unavailable,
            NativeCacheFailureReason::ActionIdentity => &mut self.complete_action_identity_unavailable,
            NativeCacheFailureReason::PostExecutionWitness => &mut self.post_execution_witness_validation_unavailable,
        };
        *counter = counter.saturating_add(1);
    }

    fn status(&self) -> std::collections::BTreeMap<String, u64> {
        NativeCacheFailureReason::ALL
            .into_iter()
            .map(|reason| {
                let count = match reason {
                    NativeCacheFailureReason::ActionCapture => self.complete_action_capture_unavailable,
                    NativeCacheFailureReason::ActionIdentity => self.complete_action_identity_unavailable,
                    NativeCacheFailureReason::PostExecutionWitness => {
                        self.post_execution_witness_validation_unavailable
                    }
                };
                (reason.as_str().to_string(), count)
            })
            .collect()
    }
}

/// Bounded observational counts. These never authorize a restore.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub(crate) struct InstallationUsageStatus {
    pub(crate) recorded_events: u64,
    pub(crate) hits: u64,
    pub(crate) misses: u64,
    pub(crate) bypasses: u64,
    pub(crate) failures: u64,
    pub(crate) ledger_full: bool,
    pub(crate) early_bypasses: u64,
    pub(crate) early_bypass_reasons: std::collections::BTreeMap<String, u64>,
    pub(crate) early_bypass_ledger_full: bool,
    pub(crate) early_bypass_incomplete_tail: bool,
    pub(crate) failure_reason_counts_available: bool,
    pub(crate) failure_reasons: std::collections::BTreeMap<String, u64>,
}

impl InstallationReceipt {
    pub(crate) fn authority(&self) -> &str {
        &self.authority
    }

    pub(crate) fn cache(&self) -> RailResult<&LocalCacheSelection> {
        self.active_profile
            .as_ref()
            .map(crate::cache::profile::InstalledCacheProfile::cache)
            .ok_or_else(|| RailError::message("no workspace cache profile is selected"))
    }

    pub(crate) fn wrapper_path(&self) -> &Path {
        &self.wrapper_path
    }

    pub(crate) fn worker_path(&self) -> &Path {
        &self.worker_path
    }

    pub(crate) fn remote(&self) -> Option<&crate::remote_cache::InstalledRemoteCache> {
        if self.readiness_local_only {
            return None;
        }
        self.active_profile
            .as_ref()
            .and_then(crate::cache::profile::InstalledCacheProfile::remote)
    }

    pub(crate) fn root_portability(&self) -> InstalledRootPortability {
        self.active_profile
            .as_ref()
            .map_or(InstalledRootPortability::Physical, |profile| {
                profile.root_portability().into()
            })
    }

    pub(crate) fn profile(&self) -> RailResult<&crate::cache::profile::InstalledCacheProfile> {
        self.active_profile
            .as_ref()
            .ok_or_else(|| RailError::message("no workspace cache profile is selected"))
    }

    pub(crate) fn attach_profile(&mut self, profile: crate::cache::profile::InstalledCacheProfile) {
        self.active_profile = Some(profile);
        self.active_profile_lock = None;
    }

    fn attach_locked_profile(
        &mut self,
        profile: crate::cache::profile::InstalledCacheProfile,
        lock: crate::cache::profile::ProfileLifecycleLock,
    ) {
        self.active_profile = Some(profile);
        self.active_profile_lock = Some(lock);
    }

    pub(crate) fn local_distributed_worker(&self) -> Option<&Path> {
        self.distributed
            .as_ref()
            .filter(|distributed| distributed.mode == "local_process_qualification_v1")
            .filter(|distributed| {
                crate::utils::stable_file_generation(&distributed.worker_path).as_ref()
                    == Some(&distributed.worker_generation)
            })
            .map(|distributed| distributed.worker_path.as_path())
    }

    pub(crate) fn mutual_tls_distributed_worker(
        &self,
    ) -> Option<(
        crate::compiler::distributed::MutualTlsClientIdentity<'_>,
        DistributedPlacementPolicy,
    )> {
        let distributed = self.distributed.as_ref()?;
        let mutual_tls = distributed.mutual_tls.as_ref()?;
        if distributed.mode != "mutual_tls_direct_v1"
            || crate::utils::stable_file_generation(&distributed.worker_path).as_ref()
                != Some(&distributed.worker_generation)
            || crate::utils::stable_file_generation(&mutual_tls.authority_certificate).as_ref()
                != Some(&mutual_tls.authority_generation)
            || crate::utils::stable_file_generation(&mutual_tls.client_certificate).as_ref()
                != Some(&mutual_tls.client_certificate_generation)
            || crate::utils::stable_file_generation(&mutual_tls.client_private_key).as_ref()
                != Some(&mutual_tls.client_private_key_generation)
        {
            return None;
        }
        Some((
            crate::compiler::distributed::MutualTlsClientIdentity {
                endpoint: &mutual_tls.endpoint,
                server_name: &mutual_tls.server_name,
                worker_capability_id: &mutual_tls.worker_capability_id,
                authority_certificate: &mutual_tls.authority_certificate,
                client_certificate: &mutual_tls.client_certificate,
                client_private_key: &mutual_tls.client_private_key,
            },
            distributed.placement?,
        ))
    }

    pub(crate) fn installation_directory(&self) -> RailResult<&Path> {
        self.wrapper_path
            .parent()
            .ok_or_else(|| RailError::message("installed compiler wrapper has no parent"))
    }

    pub(crate) fn worker_digest(&self) -> &str {
        &self.worker_digest
    }

    fn session_memo_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(SESSION_MEMO_FILE))
    }

    fn session_lock_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(SESSION_LOCK_FILE))
    }

    fn usage_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(USAGE_FILE))
    }

    fn early_bypass_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(EARLY_BYPASS_FILE))
    }

    fn failure_counters_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(FAILURE_COUNTERS_FILE))
    }

    fn failure_counters_lock_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(FAILURE_COUNTERS_LOCK_FILE))
    }

    fn distributed_placement_history_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(DISTRIBUTED_PLACEMENT_HISTORY_FILE))
    }

    fn distributed_placement_lock_path(&self) -> RailResult<PathBuf> {
        Ok(self.profile_state_directory()?.join(DISTRIBUTED_PLACEMENT_LOCK_FILE))
    }

    pub(crate) fn profile_state_directory(&self) -> RailResult<PathBuf> {
        self.active_profile.as_ref().map_or_else(
            || self.installation_directory().map(Path::to_path_buf),
            |profile| crate::cache::profile::state_directory(&self.cargo_home, profile.profile_id()),
        )
    }

    fn validate(&self) -> RailResult<()> {
        if self.version != INSTALLATION_VERSION
            || !valid_hex_digest(&self.authority)
            || !valid_sha256(&self.wrapper_digest)
            || !valid_sha256(&self.worker_digest)
            || self.wrapper_generation.is_empty()
            || self.wrapper_generation.len() > 512
            || self.worker_generation.is_empty()
            || self.worker_generation.len() > 512
            || !self.cargo_home.is_absolute()
            || !self.config_path.is_absolute()
            || !self.wrapper_path.is_absolute()
            || !self.worker_path.is_absolute()
            || self.config_path.parent() != Some(self.cargo_home.as_path())
            || self.wrapper_path.parent()
                != Some(
                    self.cargo_home
                        .join("cargo-rail")
                        .join(INSTALLATION_DIRECTORY)
                        .as_path(),
                )
            || self.worker_path.parent() != self.wrapper_path.parent()
            || self.wrapper_path.file_name() != Some(std::ffi::OsStr::new(WRAPPER_FILE))
            || self.worker_path.file_name() != Some(std::ffi::OsStr::new(WORKER_FILE))
        {
            return Err(RailError::message(
                "transparent compiler-cache installation receipt is invalid",
            ));
        }
        if let Some(distributed) = &self.distributed
            && (!matches!(
                distributed.mode.as_str(),
                "local_process_qualification_v1" | "mutual_tls_direct_v1"
            ) || !valid_sha256(&distributed.worker_digest)
                || distributed.worker_generation.is_empty()
                || distributed.worker_generation.len() > 512
                || !distributed.worker_path.is_absolute()
                || distributed.worker_path.parent() != self.wrapper_path.parent()
                || distributed.worker_path.file_name() != Some(std::ffi::OsStr::new(DISTRIBUTED_WORKER_FILE))
                || (distributed.mode == "local_process_qualification_v1" && distributed.mutual_tls.is_some())
                || (distributed.mode == "local_process_qualification_v1" && distributed.placement.is_some())
                || (distributed.mode == "mutual_tls_direct_v1"
                    && (distributed.placement.is_none()
                        || distributed.mutual_tls.as_ref().is_none_or(|mutual_tls| {
                            !valid_installed_mutual_tls(mutual_tls, &distributed.worker_path)
                        }))))
        {
            return Err(RailError::message(
                "transparent compiler-cache distributed qualification receipt is invalid",
            ));
        }
        if self.compiler_components.len() > 2
            || self
                .compiler_components
                .windows(2)
                .any(|pair| pair[0].path >= pair[1].path)
            || self.compiler_components.iter().any(|component| {
                component.path.parent() != self.wrapper_path.parent()
                    || !matches!(
                        component.path.file_name().and_then(OsStr::to_str),
                        Some(name) if name == crate::compiler::driver::compiler_driver_file_name()
                            || name == crate::compiler::driver::COMPILER_DRIVER_SOURCE_FILE_NAME
                    )
                    || !valid_sha256(&component.digest)
                    || component.generation.is_empty()
                    || component.generation.len() > 512
            })
        {
            return Err(RailError::message("installed compiler component receipt is invalid"));
        }
        Ok(())
    }
}

// A stored pin remains removable/repairable after the execution protocol changes.
// Setup and runtime activation separately require the current worker capability.
fn valid_installed_worker_capability(value: &str) -> bool {
    let Some((version, digest)) = value
        .strip_prefix("worker-capability-v")
        .and_then(|value| value.split_once(':'))
    else {
        return false;
    };
    version
        .parse::<u32>()
        .ok()
        .is_some_and(|number| number > 0 && number.to_string() == version)
        && valid_sha256(digest)
}

fn valid_installed_mutual_tls(mutual_tls: &InstalledMutualTlsDirect, worker_path: &Path) -> bool {
    let Some(directory) = worker_path.parent() else {
        return false;
    };
    mutual_tls.endpoint.parse::<std::net::SocketAddr>().is_ok()
        && rustls::pki_types::ServerName::try_from(mutual_tls.server_name.clone()).is_ok()
        && valid_installed_worker_capability(&mutual_tls.worker_capability_id)
        && [
            (
                &mutual_tls.authority_certificate,
                DISTRIBUTED_AUTHORITY_FILE,
                &mutual_tls.authority_digest,
                &mutual_tls.authority_generation,
            ),
            (
                &mutual_tls.client_certificate,
                DISTRIBUTED_CLIENT_CERTIFICATE_FILE,
                &mutual_tls.client_certificate_digest,
                &mutual_tls.client_certificate_generation,
            ),
            (
                &mutual_tls.client_private_key,
                DISTRIBUTED_CLIENT_PRIVATE_KEY_FILE,
                &mutual_tls.client_private_key_digest,
                &mutual_tls.client_private_key_generation,
            ),
        ]
        .into_iter()
        .all(|(path, file_name, digest, generation)| {
            path.is_absolute()
                && path.parent() == Some(directory)
                && path.file_name() == Some(std::ffi::OsStr::new(file_name))
                && valid_sha256(digest)
                && !generation.is_empty()
                && generation.len() <= 512
        })
}

pub(crate) fn load_session_memo(receipt: &InstallationReceipt) -> RailResult<Option<Vec<u8>>> {
    read_optional_regular(&receipt.session_memo_path()?, MAX_SESSION_MEMO_BYTES)
}

pub(crate) fn lock_session(receipt: &InstallationReceipt) -> RailResult<InstallationSessionLock> {
    let path = receipt.session_lock_path()?;
    let file = crate::utils::open_cache_lock_file(&path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        file.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&file, &path, 0)? {
        return Err(RailError::message(
            "transparent compiler session lock is not a private regular file",
        ));
    }
    file.lock()?;
    if !crate::utils::private_file_matches_path(&file, &path, 0)? {
        return Err(RailError::message(
            "transparent compiler session lock changed while it was acquired",
        ));
    }
    Ok(InstallationSessionLock { _file: file })
}

pub(super) fn lock_installation(cargo_home: &Path) -> RailResult<InstallationMutationLock> {
    ensure_real_directory(cargo_home)?;
    let owner = cargo_home.join("cargo-rail");
    create_private_directory(&owner)?;
    let path = owner.join(INSTALLATION_LOCK_FILE);
    let file = crate::utils::open_cache_lock_file(&path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        file.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&file, &path, 0)? {
        return Err(RailError::message(
            "transparent compiler installation lock is not a private regular file",
        ));
    }
    file.lock()?;
    if !crate::utils::private_file_matches_path(&file, &path, 0)? {
        return Err(RailError::message(
            "transparent compiler installation lock changed while it was acquired",
        ));
    }
    Ok(InstallationMutationLock { _file: file })
}

pub(crate) fn store_session_memo(receipt: &InstallationReceipt, bytes: &[u8]) -> RailResult<()> {
    if bytes.len() as u64 > MAX_SESSION_MEMO_BYTES {
        return Err(RailError::message(
            "transparent compiler session memo exceeds its bound",
        ));
    }
    write_private_atomic(&receipt.session_memo_path()?, bytes)
}

/// Read bounded observational placement history. It never grants result authority.
pub(crate) fn read_distributed_placement_history(receipt: &InstallationReceipt) -> RailResult<Option<Vec<u8>>> {
    read_optional_regular(
        &receipt.distributed_placement_history_path()?,
        MAX_DISTRIBUTED_PLACEMENT_HISTORY_BYTES,
    )
}

/// Serialize one bounded placement-history update across concurrent Cargo wrappers.
pub(crate) fn update_distributed_placement_history(
    receipt: &InstallationReceipt,
    update: impl FnOnce(Option<&[u8]>) -> RailResult<Vec<u8>>,
) -> RailResult<()> {
    if receipt
        .distributed
        .as_ref()
        .is_none_or(|distributed| distributed.mutual_tls.is_none())
    {
        return Err(RailError::message(
            "distributed placement history has no installed mTLS authority",
        ));
    }
    let lock_path = receipt.distributed_placement_lock_path()?;
    let lock = crate::utils::open_cache_lock_file(&lock_path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        lock.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "distributed placement lock is not a private regular file",
        ));
    }
    lock.lock()?;
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "distributed placement lock changed while it was acquired",
        ));
    }
    let current = read_distributed_placement_history(receipt)?;
    let next = update(current.as_deref())?;
    if next.len() as u64 > MAX_DISTRIBUTED_PLACEMENT_HISTORY_BYTES {
        return Err(RailError::message("distributed placement history exceeds its bound"));
    }
    write_private_atomic(&receipt.distributed_placement_history_path()?, &next)
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct InstallationStatus {
    pub(crate) state: &'static str,
    pub(crate) healthy: bool,
    pub(crate) installation_integrity: &'static str,
    pub(crate) component_authentication: &'static str,
    pub(crate) workspace_enrollment: &'static str,
    pub(crate) observed_reuse: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) receipt_version: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) recovery_action: Option<&'static str>,
    pub(crate) cargo_home: String,
    pub(crate) config_path: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) wrapper_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) profile_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) bound_workspace_root: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) trust_domain: Option<String>,
    pub(crate) selection_source: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) cache_base: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) max_bytes: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) root_portability: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) distributed: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) distributed_policy: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) distributed_placement_history: Option<crate::compiler::distributed::PlacementHistoryStatus>,
    pub(crate) cargo_l0: &'static str,
    pub(crate) owned_bytes: u64,
    pub(crate) required_bytes: u64,
    pub(crate) reclaimable_bytes: u64,
    pub(crate) quarantined_receipts: u64,
    pub(crate) usage: InstallationUsageStatus,
    pub(crate) issues: Vec<String>,
}

/// Record one post-context wrapper outcome in a bounded private log.
///
/// Ordinary cache execution deliberately discards this result. Explicit
/// benchmark coverage instead surfaces it so its usage and action ledgers
/// cannot silently diverge.
pub(crate) fn record_usage(
    receipt: &InstallationReceipt,
    outcome: u8,
    failure_reason: Option<NativeCacheFailureReason>,
) -> RailResult<()> {
    if !matches!(outcome, b'H' | b'M' | b'B' | b'F') {
        return Err(RailError::message("compiler-cache usage outcome is invalid"));
    }
    let usage = record_usage_event(receipt, outcome);
    let failure = failure_reason.map(|reason| record_failure_reason(receipt, reason));
    match (usage, failure) {
        (Err(error), _) | (Ok(()), Some(Err(error))) => Err(error),
        (Ok(()), None | Some(Ok(()))) => Ok(()),
    }
}

fn record_usage_event(receipt: &InstallationReceipt, outcome: u8) -> RailResult<()> {
    let path = receipt.usage_path()?;
    let mut file = crate::utils::open_cache_lock_file(&path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        file.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    file.lock()?;
    let metadata = file.metadata()?;
    if metadata.len() >= MAX_USAGE_BYTES {
        return Err(RailError::message("compiler-cache usage log reached its size bound"));
    }
    if !crate::utils::private_file_matches_path(&file, &path, metadata.len())? {
        return Err(RailError::message(
            "compiler-cache usage log is not the retained bounded private file",
        ));
    }
    file.seek(std::io::SeekFrom::End(0))?;
    file.write_all(&[outcome])?;
    Ok(())
}

fn record_failure_reason(receipt: &InstallationReceipt, reason: NativeCacheFailureReason) -> RailResult<()> {
    let lock_path = receipt.failure_counters_lock_path()?;
    let lock = crate::utils::open_cache_lock_file(&lock_path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        lock.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "compiler-cache failure counter lock is not a private regular file",
        ));
    }
    lock.lock()?;
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "compiler-cache failure counter lock changed while it was acquired",
        ));
    }
    let mut counters = read_failure_counters(receipt)?.unwrap_or_default();
    counters.increment(reason);
    write_private_atomic(&receipt.failure_counters_path()?, &encode_failure_counters(&counters)?)
}

fn read_failure_counters(receipt: &InstallationReceipt) -> RailResult<Option<NativeCacheFailureCounters>> {
    let Some(bytes) = read_optional_regular(&receipt.failure_counters_path()?, MAX_FAILURE_COUNTERS_BYTES)? else {
        return Ok(None);
    };
    let counters: NativeCacheFailureCounters = serde_json::from_slice(&bytes)?;
    if counters.schema_version != 1 || encode_failure_counters(&counters)? != bytes {
        return Err(RailError::message(
            "compiler-cache failure counters are not canonical version 1 state",
        ));
    }
    Ok(Some(counters))
}

fn encode_failure_counters(counters: &NativeCacheFailureCounters) -> RailResult<Vec<u8>> {
    let mut bytes = serde_json::to_vec_pretty(counters)?;
    bytes.push(b'\n');
    if bytes.len() as u64 > MAX_FAILURE_COUNTERS_BYTES {
        return Err(RailError::message("compiler-cache failure counters exceed their bound"));
    }
    Ok(bytes)
}

/// Record a pre-cache bypass in the exact selected workspace profile.
///
/// This path selects and locks the profile but deliberately does not open the
/// compiler session, local CAS, or remote. The reason is a classifier-owned
/// constant and never contains compiler arguments or environment values.
pub(crate) fn record_early_bypass(source_root: &Path, reason: &'static str) -> RailResult<()> {
    if reason.is_empty()
        || reason.len() > MAX_EARLY_BYPASS_REASON_BYTES
        || !reason
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
    {
        return Err(RailError::message("compiler-cache early bypass reason is invalid"));
    }
    let executable = std::env::current_exe()?;
    let executable_name = executable.file_name();
    if !matches!(
        executable_name,
        Some(name) if name == std::ffi::OsStr::new(WRAPPER_FILE) || name == std::ffi::OsStr::new(WORKER_FILE)
    ) {
        return Err(RailError::message(
            "compiler-cache early bypass did not originate from an installed wrapper",
        ));
    }
    let receipt = load_for_wrapper(&executable, source_root)?;
    let session_lock_path = receipt.session_lock_path()?;
    let session_lock = crate::utils::open_cache_lock_file(&session_lock_path, false)?;
    if !crate::utils::private_file_matches_path(&session_lock, &session_lock_path, 0)? {
        return Err(RailError::message(
            "transparent compiler session lock is not a private regular file",
        ));
    }
    session_lock.lock()?;
    if !crate::utils::private_file_matches_path(&session_lock, &session_lock_path, 0)? {
        return Err(RailError::message(
            "transparent compiler session lock changed while recording an early bypass",
        ));
    }
    let path = receipt.early_bypass_path()?;
    let mut file = crate::utils::open_cache_lock_file(&path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        file.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    file.lock()?;
    let metadata = file.metadata()?;
    let record_len = u64::try_from(reason.len().saturating_add(1)).unwrap_or(u64::MAX);
    if metadata.len().saturating_add(record_len) > MAX_EARLY_BYPASS_BYTES {
        return Err(RailError::message(
            "compiler-cache early bypass log reached its size bound",
        ));
    }
    if !crate::utils::private_file_matches_path(&file, &path, metadata.len())? {
        return Err(RailError::message(
            "compiler-cache early bypass log is not the retained bounded private file",
        ));
    }
    file.seek(std::io::SeekFrom::End(0))?;
    file.write_all(&[u8::try_from(reason.len())
        .map_err(|_| RailError::message("compiler-cache early bypass reason exceeded its encoded bound"))?])?;
    file.write_all(reason.as_bytes())?;
    drop(session_lock);
    Ok(())
}

fn usage_status(receipt: &InstallationReceipt) -> RailResult<InstallationUsageStatus> {
    let mut usage = InstallationUsageStatus::default();
    if let Some(bytes) = read_bounded_private_log(&receipt.usage_path()?, MAX_USAGE_BYTES)? {
        usage.recorded_events = bytes.len() as u64;
        usage.ledger_full = bytes.len() as u64 == MAX_USAGE_BYTES;
        for byte in bytes {
            match byte {
                b'H' => usage.hits = usage.hits.saturating_add(1),
                b'M' => usage.misses = usage.misses.saturating_add(1),
                b'B' => usage.bypasses = usage.bypasses.saturating_add(1),
                b'F' => usage.failures = usage.failures.saturating_add(1),
                _ => return Err(RailError::message("transparent compiler-cache usage log is malformed")),
            }
        }
    }
    if let Some(bytes) = read_bounded_private_log(&receipt.early_bypass_path()?, MAX_EARLY_BYPASS_BYTES)? {
        usage.early_bypass_ledger_full = bytes.len() as u64 == MAX_EARLY_BYPASS_BYTES;
        let mut remaining = bytes.as_slice();
        while let Some((&length, tail)) = remaining.split_first() {
            let length = usize::from(length);
            if length == 0 || length > MAX_EARLY_BYPASS_REASON_BYTES || tail.len() < length {
                usage.early_bypass_incomplete_tail = true;
                break;
            }
            let reason = std::str::from_utf8(&tail[..length])?;
            if !reason
                .bytes()
                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
            {
                return Err(RailError::message(
                    "transparent compiler-cache early bypass log is malformed",
                ));
            }
            usage.early_bypasses = usage.early_bypasses.saturating_add(1);
            let count = usage.early_bypass_reasons.entry(reason.to_string()).or_default();
            *count = count.saturating_add(1);
            remaining = &tail[length..];
        }
    }
    Ok(usage)
}

fn populate_failure_reason_status(receipt: &InstallationReceipt, usage: &mut InstallationUsageStatus) {
    usage.failure_reasons = NativeCacheFailureCounters::default().status();
    match read_failure_counters(receipt) {
        Ok(Some(counters)) => {
            usage.failure_reasons = counters.status();
            usage.failure_reason_counts_available = true;
        }
        Ok(None) => usage.failure_reason_counts_available = true,
        Err(_) => usage.failure_reason_counts_available = false,
    }
}

fn empty_usage_status() -> InstallationUsageStatus {
    InstallationUsageStatus {
        failure_reason_counts_available: true,
        failure_reasons: NativeCacheFailureCounters::default().status(),
        ..InstallationUsageStatus::default()
    }
}

fn read_bounded_private_log(path: &Path, maximum: u64) -> RailResult<Option<Vec<u8>>> {
    let mut file = match crate::utils::open_cache_lock_file(path, false) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    file.lock()?;
    let metadata = file.metadata()?;
    if metadata.len() > maximum || !crate::utils::private_file_matches_path(&file, path, metadata.len())? {
        return Err(RailError::message(
            "transparent compiler-cache observational log is not a bounded private regular file",
        ));
    }
    let mut bytes = Vec::with_capacity(usize::try_from(metadata.len()).unwrap_or(0));
    file.read_to_end(&mut bytes)?;
    if bytes.len() as u64 != metadata.len() {
        return Err(RailError::message(
            "transparent compiler-cache observational log changed while it was read",
        ));
    }
    Ok(Some(bytes))
}

/// Build a no-write setup plan from Cargo's current hierarchy and machine state.
pub(crate) fn plan_setup(current_dir: &Path, request: &SetupRequest) -> RailResult<SetupPlan> {
    ensure_supported_installation_platform()?;
    let cargo_home = resolve_cargo_home(current_dir)?;
    let install_directory = cargo_home.join("cargo-rail").join(INSTALLATION_DIRECTORY);
    let receipt_path = install_directory.join(RECEIPT_FILE);
    let receipt_before = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)?;
    let receipt_version = receipt_before
        .as_deref()
        .map(|bytes| parse_receipt_version(bytes, &receipt_path))
        .transpose()?;
    let existing = match (receipt_before.as_deref(), receipt_version) {
        (Some(bytes), Some(INSTALLATION_VERSION)) => Some(parse_receipt(bytes, &receipt_path)?),
        _ => None,
    };
    let stale_receipt_version = receipt_version.filter(|version| *version != INSTALLATION_VERSION);
    let config_path = selected_user_config(&cargo_home);
    let legacy = stale_receipt_version
        .zip(receipt_before.as_deref())
        .map(|(_, bytes)| parse_legacy_receipt(bytes, &cargo_home, &config_path))
        .transpose()?;
    let stale_receipt_quarantine = stale_receipt_version
        .zip(receipt_before.as_deref())
        .map(|(version, bytes)| {
            cargo_home
                .join("cargo-rail")
                .join(INSTALLATION_QUARANTINE_DIRECTORY)
                .join(format!("setup-v{version}-{}.json", ContentDigest::sha256(bytes)))
        });
    if let Some(existing) = &existing
        && (existing.cargo_home != cargo_home || existing.config_path != config_path)
    {
        return Err(RailError::message(
            "transparent compiler-cache setup authority does not match the selected Cargo home",
        ));
    }
    reject_shadowing_global_wrapper(current_dir, &config_path)?;

    let config_before = read_optional_regular(&config_path, 16 * 1024 * 1024)?;
    let original = config_before
        .as_deref()
        .map(|bytes| {
            std::str::from_utf8(bytes)
                .map(str::to_string)
                .map_err(|_| RailError::message("Cargo user configuration is not valid UTF-8"))
        })
        .transpose()?
        .unwrap_or_default();
    let wrapper_path = install_directory.join(WRAPPER_FILE);
    let worker_path = install_directory.join(WORKER_FILE);
    let distributed_worker_path = install_directory.join(DISTRIBUTED_WORKER_FILE);
    let distributed_authority_path = install_directory.join(DISTRIBUTED_AUTHORITY_FILE);
    let distributed_client_certificate_path = install_directory.join(DISTRIBUTED_CLIENT_CERTIFICATE_FILE);
    let distributed_client_private_key_path = install_directory.join(DISTRIBUTED_CLIENT_PRIVATE_KEY_FILE);
    let owned_wrapper = existing
        .as_ref()
        .map(InstallationReceipt::wrapper_path)
        .or_else(|| stale_receipt_version.map(|_| wrapper_path.as_path()));
    let (config_after, build_table_created) = install_wrapper_value(&original, &wrapper_path, owned_wrapper)?;
    let wrapper_source = crate::compiler::native_cache::direct_wrapper_executable()?;
    let worker_source = crate::compiler::native_cache::direct_worker_executable()?;
    require_consistent_component_versions(&[
        ("cargo-rail-native-rustc-wrapper", &wrapper_source),
        ("cargo-rail-native-rustc-worker", &worker_source),
    ])?;
    let wrapper = plan_executable_setup(wrapper_source, &wrapper_path, "installed compiler wrapper")?;
    let worker = plan_executable_setup(worker_source, &worker_path, "installed compiler worker")?;
    let sources =
        crate::compiler::driver::CompilerFactDriverAuthority::installation_components(&std::env::current_exe()?)?;
    let mut compiler_components = Vec::with_capacity(sources.len());
    let mut installed_compiler_components = Vec::with_capacity(sources.len());
    for source in sources {
        let name = source
            .source_path
            .file_name()
            .ok_or_else(|| RailError::message("compiler component has no file name"))?;
        let path = install_directory.join(name);
        let file = plan_executable_setup(source.source_path, &path, "installed compiler component")?;
        if file.source_digest != source.content_digest {
            return Err(RailError::message("compiler component changed after authentication"));
        }
        let mode_matches = file.before.is_none() || compiler_component_mode_matches(&path, source.executable)?;
        let generation = if mode_matches {
            file.preserved_generation().map_or_else(|| vec![0], <[u8]>::to_vec)
        } else {
            vec![0]
        };
        installed_compiler_components.push(InstalledCompilerComponent {
            path,
            digest: source.content_digest,
            generation,
        });
        compiler_components.push(CompilerComponentSetup {
            file,
            executable: source.executable,
            mode_matches,
        });
    }
    let retired_compiler_components = existing
        .as_ref()
        .into_iter()
        .flat_map(|receipt| &receipt.compiler_components)
        .filter(|component| {
            !installed_compiler_components
                .iter()
                .any(|selected| selected.path == component.path)
        })
        .map(|component| Ok((component.clone(), compiler_component_removal_digest(component)?)))
        .collect::<RailResult<Vec<_>>>()?;
    let requested_mutual_tls = match (
        request.distributed_endpoint.as_deref(),
        request.distributed_server_name.as_deref(),
        request.distributed_capability.as_deref(),
        request.distributed_authority.as_deref(),
        request.distributed_client_certificate.as_deref(),
        request.distributed_client_private_key.as_deref(),
    ) {
        (None, None, None, None, None, None) => None,
        (
            Some(endpoint),
            Some(server_name),
            Some(worker_capability_id),
            Some(authority),
            Some(client_certificate),
            Some(client_private_key),
        ) => {
            if request.distributed_local {
                return Err(RailError::message(
                    "local-process and mutually authenticated distributed modes are exclusive",
                ));
            }
            let endpoint = endpoint
                .parse::<std::net::SocketAddr>()
                .map_err(|_| RailError::message("distributed endpoint must be an explicit socket address"))?
                .to_string();
            rustls::pki_types::ServerName::try_from(server_name.to_string())
                .map_err(|_| RailError::message("distributed TLS server name is invalid"))?;
            let authority = resolve_requested_path(current_dir, authority)?;
            let client_certificate = resolve_requested_path(current_dir, client_certificate)?;
            let client_private_key = resolve_requested_path(current_dir, client_private_key)?;
            crate::compiler::distributed::validate_mutual_tls_client_identity(
                server_name,
                worker_capability_id,
                &authority,
                &client_certificate,
                &client_private_key,
            )?;
            Some((
                endpoint,
                server_name.to_string(),
                worker_capability_id.to_string(),
                authority,
                client_certificate,
                client_private_key,
            ))
        }
        _ => {
            return Err(RailError::message(
                "distributed mTLS setup requires endpoint, server name, worker capability, authority, client certificate, and client key",
            ));
        }
    };
    if request.distributed_local
        && existing
            .as_ref()
            .and_then(|receipt| receipt.distributed.as_ref())
            .is_some_and(|distributed| distributed.mutual_tls.is_some())
    {
        return Err(RailError::with_help(
            "cannot replace an mTLS distributed installation with local qualification during setup",
            "run `cargo rail cache uninstall`, then rerun setup with `--distributed-local`",
        ));
    }
    let requested_placement = request
        .distributed_policy
        .as_deref()
        .map(|policy| match policy {
            "automatic" => Ok(DistributedPlacementPolicy::Automatic),
            "qualification" => Ok(DistributedPlacementPolicy::Qualification),
            _ => Err(RailError::message("distributed placement policy is invalid")),
        })
        .transpose()?;
    let distributed_enabled = request.distributed_local
        || requested_mutual_tls.is_some()
        || requested_placement.is_some()
        || existing.as_ref().is_some_and(|receipt| receipt.distributed.is_some());
    let distributed_worker = distributed_enabled
        .then(|| {
            let source = crate::compiler::native_cache::direct_distributed_worker_executable()?;
            require_consistent_component_versions(&[("cargo-rail-distributed-worker", &source)])?;
            plan_executable_setup(
                source,
                &distributed_worker_path,
                "installed distributed compiler worker",
            )
        })
        .transpose()?;
    let source_distributed_identity = requested_mutual_tls
        .as_ref()
        .map(
            |(_, _, _, authority, client_certificate, client_private_key)| -> RailResult<SourceDistributedIdentity> {
                Ok(SourceDistributedIdentity {
                    authority: authority.clone(),
                    authority_digest: installation_source_digest(authority)?,
                    client_certificate: client_certificate.clone(),
                    client_certificate_digest: installation_source_digest(client_certificate)?,
                    client_private_key: client_private_key.clone(),
                    client_private_key_digest: installation_source_digest(client_private_key)?,
                })
            },
        )
        .transpose()?;
    let wrapper_generation = wrapper.preserved_generation().map_or_else(|| vec![0], <[u8]>::to_vec);
    let worker_generation = worker.preserved_generation().map_or_else(|| vec![0], <[u8]>::to_vec);
    let distributed_worker_generation = distributed_worker
        .as_ref()
        .and_then(ExecutableSetup::preserved_generation)
        .map_or_else(|| vec![0], <[u8]>::to_vec);
    let mutual_tls = if let (Some((endpoint, server_name, worker_capability_id, ..)), Some(source)) =
        (requested_mutual_tls.as_ref(), source_distributed_identity.as_ref())
    {
        let installed_generation = |path: &Path, digest: &str| -> RailResult<Vec<u8>> {
            if optional_file_digest(path)?.as_deref() == Some(digest) {
                crate::utils::stable_file_generation(path).ok_or_else(|| {
                    RailError::message("installed distributed TLS identity has no stable file generation")
                })
            } else {
                Ok(vec![0])
            }
        };
        Some(InstalledMutualTlsDirect {
            endpoint: endpoint.clone(),
            server_name: server_name.clone(),
            worker_capability_id: worker_capability_id.clone(),
            authority_certificate: distributed_authority_path.clone(),
            authority_digest: source.authority_digest.clone(),
            authority_generation: installed_generation(&distributed_authority_path, &source.authority_digest)?,
            client_certificate: distributed_client_certificate_path.clone(),
            client_certificate_digest: source.client_certificate_digest.clone(),
            client_certificate_generation: installed_generation(
                &distributed_client_certificate_path,
                &source.client_certificate_digest,
            )?,
            client_private_key: distributed_client_private_key_path.clone(),
            client_private_key_digest: source.client_private_key_digest.clone(),
            client_private_key_generation: installed_generation(
                &distributed_client_private_key_path,
                &source.client_private_key_digest,
            )?,
        })
    } else {
        existing
            .as_ref()
            .and_then(|receipt| receipt.distributed.as_ref())
            .and_then(|distributed| distributed.mutual_tls.clone())
    };
    if mutual_tls.as_ref().is_some_and(|identity| {
        !crate::compiler::distributed::worker_capability_identity_is_valid(&identity.worker_capability_id)
    }) {
        return Err(RailError::with_help(
            "installed distributed worker pin uses an incompatible protocol",
            "run cache setup with the current worker capability and complete TLS inputs, or preview cache uninstall --check before removing this installation",
        ));
    }
    let placement = if mutual_tls.is_some() {
        requested_placement
            .or_else(|| {
                existing
                    .as_ref()
                    .and_then(|receipt| receipt.distributed.as_ref())
                    .and_then(|distributed| distributed.placement)
            })
            .or(Some(DistributedPlacementPolicy::Automatic))
    } else if requested_placement.is_some() {
        return Err(RailError::message(
            "distributed placement policy requires an installed or requested mTLS worker",
        ));
    } else {
        None
    };
    let authority = existing
        .as_ref()
        .map(|receipt| receipt.authority.clone())
        .or_else(|| legacy.as_ref().map(|receipt| receipt.authority.clone()))
        .map_or_else(random_authority, Ok)?;
    let mut receipt = InstallationReceipt {
        version: INSTALLATION_VERSION,
        authority,
        cargo_home: cargo_home.clone(),
        config_path: config_path.clone(),
        config_created: existing
            .as_ref()
            .map_or_else(|| config_before.is_none(), |receipt| receipt.config_created),
        build_table_created: existing
            .as_ref()
            .map_or(build_table_created, |receipt| receipt.build_table_created),
        wrapper_path,
        wrapper_digest: wrapper.source_digest.clone(),
        wrapper_generation,
        worker_path,
        worker_digest: worker.source_digest.clone(),
        worker_generation,
        compiler_components: installed_compiler_components,
        distributed: distributed_worker
            .as_ref()
            .map(|worker| InstalledDistributedQualification {
                mode: if mutual_tls.is_some() {
                    "mutual_tls_direct_v1".to_string()
                } else {
                    "local_process_qualification_v1".to_string()
                },
                placement,
                worker_path: distributed_worker_path,
                worker_digest: worker.source_digest.clone(),
                worker_generation: distributed_worker_generation,
                mutual_tls,
            }),
        active_profile: None,
        active_profile_lock: None,
        readiness_local_only: false,
    };
    let profile = crate::cache::profile::plan_setup(
        &cargo_home,
        current_dir,
        &receipt.authority,
        crate::cache::profile::ProfileSetupRequest {
            requested_profile: request.profile_id.as_deref(),
            local_dir: request.local_dir.as_deref().or_else(|| {
                legacy
                    .as_ref()
                    .and_then(|receipt| receipt.cache.as_ref())
                    .map(LocalCacheSelection::base)
            }),
            max_bytes: request.max_bytes.or_else(|| {
                legacy
                    .as_ref()
                    .and_then(|receipt| receipt.cache.as_ref())
                    .map(LocalCacheSelection::max_bytes)
            }),
            trust_domain: legacy
                .as_ref()
                .and_then(|receipt| receipt.cache.as_ref())
                .and_then(LocalCacheSelection::trust_domain),
            remote_url: request.remote_url.as_deref(),
            remote_mode: request.remote_mode.as_deref(),
            remote_environment: &request.remote_environment,
            root_portability: request.root_portability.as_deref().or_else(|| {
                legacy
                    .as_ref()
                    .and_then(|receipt| receipt.root_portability)
                    .map(InstalledRootPortability::as_str)
            }),
            local_only: request.local_only,
        },
    )?;
    receipt.attach_profile(profile.profile().clone());
    receipt.validate()?;
    let encoded_receipt = encode_receipt(&receipt)?;
    let cache_ready = profile.profile().cache().configured_root()?.is_some();
    let distributed_identity_current = receipt
        .distributed
        .as_ref()
        .and_then(|distributed| distributed.mutual_tls.as_ref())
        .map(|mutual_tls| -> RailResult<[Option<String>; 3]> {
            Ok([
                optional_file_digest(&mutual_tls.authority_certificate)?,
                optional_file_digest(&mutual_tls.client_certificate)?,
                optional_file_digest(&mutual_tls.client_private_key)?,
            ])
        })
        .transpose()?;
    let pending = config_before.as_deref() != Some(config_after.as_bytes())
        || receipt_before.as_deref() != Some(encoded_receipt.as_slice())
        || wrapper.requires_installation()
        || worker.requires_installation()
        || compiler_components
            .iter()
            .any(CompilerComponentSetup::requires_installation)
        || !retired_compiler_components.is_empty()
        || distributed_worker
            .as_ref()
            .is_some_and(ExecutableSetup::requires_installation)
        || receipt
            .distributed
            .as_ref()
            .and_then(|distributed| distributed.mutual_tls.as_ref())
            .zip(distributed_identity_current.as_ref())
            .is_some_and(|(mutual_tls, current)| {
                current[0].as_deref() != Some(mutual_tls.authority_digest.as_str())
                    || current[1].as_deref() != Some(mutual_tls.client_certificate_digest.as_str())
                    || current[2].as_deref() != Some(mutual_tls.client_private_key_digest.as_str())
            })
        || !cache_ready
        || profile.pending();
    Ok(SetupPlan {
        cargo_home,
        config_path,
        config_before,
        config_after: config_after.into_bytes(),
        receipt_before,
        stale_receipt_quarantine,
        receipt,
        wrapper,
        worker,
        compiler_components,
        retired_compiler_components,
        distributed_worker,
        source_distributed_identity,
        profile,
        pending,
    })
}

#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
fn ensure_supported_installation_platform() -> RailResult<()> {
    if let Some(reason) = crate::compiler::native_cache::unsupported_native_cache_host_reason() {
        return Err(unsupported_native_cache_host_error(reason));
    }
    Ok(())
}

#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
fn ensure_supported_installation_platform() -> RailResult<()> {
    Err(RailError::message(
        "transparent compiler-cache setup is unsupported on this platform",
    ))
}

pub(crate) fn unsupported_native_cache_host_error(reason: &'static str) -> RailError {
    RailError::with_help(
        format!("transparent compiler-cache setup is unsupported on this unqualified host ({reason})"),
        "continue using Cargo normally; compiler-result reuse remains disabled",
    )
}

/// Apply one setup plan after revalidating every opened input.
pub(crate) fn apply_setup(mut plan: SetupPlan) -> RailResult<()> {
    if !plan.pending {
        return Ok(());
    }
    revalidate_optional(&plan.config_path, plan.config_before.as_deref(), 16 * 1024 * 1024)?;
    let receipt_path = plan
        .receipt
        .wrapper_path
        .parent()
        .ok_or_else(|| RailError::message("installed compiler wrapper has no parent"))?
        .join(RECEIPT_FILE);
    revalidate_optional(&receipt_path, plan.receipt_before.as_deref(), MAX_RECEIPT_BYTES)?;
    if installation_source_digest(&plan.wrapper.source)? != plan.wrapper.source_digest {
        return Err(RailError::message(
            "compiler wrapper executable changed after setup planning",
        ));
    }
    if installation_source_digest(&plan.worker.source)? != plan.worker.source_digest {
        return Err(RailError::message(
            "compiler worker executable changed after setup planning",
        ));
    }
    for component in &plan.compiler_components {
        if installation_source_digest(&component.file.source)? != component.file.source_digest {
            return Err(RailError::message("compiler component changed after setup planning"));
        }
    }
    if let Some(distributed_worker) = &plan.distributed_worker
        && installation_source_digest(&distributed_worker.source)? != distributed_worker.source_digest
    {
        return Err(RailError::message(
            "distributed compiler worker executable changed after setup planning",
        ));
    }
    if let Some(source) = &plan.source_distributed_identity {
        for (path, expected) in [
            (&source.authority, &source.authority_digest),
            (&source.client_certificate, &source.client_certificate_digest),
            (&source.client_private_key, &source.client_private_key_digest),
        ] {
            if installation_source_digest(path)? != *expected {
                return Err(RailError::message(
                    "distributed TLS identity changed after setup planning",
                ));
            }
        }
    } else if let Some(mutual_tls) = plan
        .receipt
        .distributed
        .as_ref()
        .and_then(|distributed| distributed.mutual_tls.as_ref())
    {
        for (path, expected, generation) in [
            (
                &mutual_tls.authority_certificate,
                &mutual_tls.authority_digest,
                &mutual_tls.authority_generation,
            ),
            (
                &mutual_tls.client_certificate,
                &mutual_tls.client_certificate_digest,
                &mutual_tls.client_certificate_generation,
            ),
            (
                &mutual_tls.client_private_key,
                &mutual_tls.client_private_key_digest,
                &mutual_tls.client_private_key_generation,
            ),
        ] {
            if file_digest(path)? != *expected
                || crate::utils::stable_file_generation(path).as_ref() != Some(generation)
            {
                return Err(RailError::with_help(
                    "installed distributed TLS identity drifted",
                    "rerun cache setup with the complete distributed identity to repair it",
                ));
            }
        }
    }
    if let Some(existing) = plan
        .receipt_before
        .as_deref()
        .and_then(|bytes| {
            (parse_receipt_version(bytes, &receipt_path).ok() == Some(INSTALLATION_VERSION))
                .then(|| parse_receipt(bytes, &receipt_path))
        })
        .transpose()?
    {
        for profile in crate::cache::profile::load_all(&plan.cargo_home)? {
            let mut selected = existing.clone();
            selected.attach_profile(profile);
            crate::remote_cache::stop_installed_coordinators(&selected);
        }
    }

    ensure_real_directory(&plan.cargo_home)?;
    let owner = plan.cargo_home.join("cargo-rail");
    create_private_directory(&owner)?;
    let install_directory = owner.join(INSTALLATION_DIRECTORY);
    create_private_directory(&install_directory)?;
    let selected_profile = crate::cache::profile::apply_setup(&plan.profile)?;
    plan.receipt.attach_profile(selected_profile);
    for profile in crate::cache::profile::load_all(&plan.cargo_home)? {
        let mut selected = plan.receipt.clone();
        selected.attach_profile(profile);
        crate::remote_cache::stop_installed_coordinators(&selected);
    }
    let _profile_lifecycles = crate::cache::profile::lock_all_exclusive(&plan.cargo_home)?;
    // Global mutations wait for active profile work before taking the
    // installation lock used by compiler children. The remaining order is
    // registry/profile, installation, then session.
    let _installation_lock = lock_installation(&plan.cargo_home)?;
    let _session_lock = lock_session(&plan.receipt)?;
    revalidate_executable_setup(&plan.receipt.wrapper_path, &plan.wrapper, "installed compiler wrapper")?;
    revalidate_executable_setup(&plan.receipt.worker_path, &plan.worker, "installed compiler worker")?;
    for (component, setup) in plan.receipt.compiler_components.iter().zip(&plan.compiler_components) {
        revalidate_executable_setup(&component.path, &setup.file, "installed compiler component")?;
    }
    for (component, expected) in &plan.retired_compiler_components {
        revalidate_compiler_component_removal(component, expected)?;
    }
    if let (Some(setup), Some(distributed)) = (&plan.distributed_worker, &plan.receipt.distributed) {
        revalidate_executable_setup(&distributed.worker_path, setup, "installed distributed compiler worker")?;
    }
    if read_optional_regular(&plan.receipt.session_memo_path()?, MAX_SESSION_MEMO_BYTES)?.is_some() {
        fs::remove_file(plan.receipt.session_memo_path()?)?;
    }
    LocalCas::open_selected(plan.receipt.cache()?)?;
    install_planned_executable(&plan.wrapper, &plan.receipt.wrapper_path)?;
    install_planned_executable(&plan.worker, &plan.receipt.worker_path)?;
    for (component, setup) in plan
        .receipt
        .compiler_components
        .iter_mut()
        .zip(&plan.compiler_components)
    {
        if setup.requires_installation() {
            if setup.executable {
                install_executable_atomic(&setup.file.source, &component.path)?;
            } else {
                install_private_file_atomic(&setup.file.source, &component.path, &component.digest)?;
            }
        }
        if file_digest(&component.path)? != component.digest
            || !compiler_component_mode_matches(&component.path, setup.executable)?
        {
            return Err(RailError::message("installed compiler component failed verification"));
        }
        component.generation = crate::utils::stable_file_generation(&component.path)
            .ok_or_else(|| RailError::message("installed compiler component has no stable local file generation"))?;
    }
    for (component, expected) in &plan.retired_compiler_components {
        revalidate_compiler_component_removal(component, expected)?;
        if expected.is_some() {
            fs::remove_file(&component.path)?;
        }
    }
    if let (Some(setup), Some(distributed)) = (&plan.distributed_worker, &plan.receipt.distributed) {
        install_planned_executable(setup, &distributed.worker_path)?;
    }
    if let (Some(source), Some(mutual_tls)) = (
        plan.source_distributed_identity.as_ref(),
        plan.receipt
            .distributed
            .as_ref()
            .and_then(|distributed| distributed.mutual_tls.as_ref()),
    ) {
        install_private_file_atomic(
            &source.authority,
            &mutual_tls.authority_certificate,
            &source.authority_digest,
        )?;
        install_private_file_atomic(
            &source.client_certificate,
            &mutual_tls.client_certificate,
            &source.client_certificate_digest,
        )?;
        install_private_file_atomic(
            &source.client_private_key,
            &mutual_tls.client_private_key,
            &source.client_private_key_digest,
        )?;
    }
    if file_digest(&plan.receipt.wrapper_path)? != plan.receipt.wrapper_digest {
        return Err(RailError::message(
            "installed compiler wrapper failed content verification",
        ));
    }
    plan.receipt.wrapper_generation = crate::utils::stable_file_generation(&plan.receipt.wrapper_path)
        .ok_or_else(|| RailError::message("installed compiler wrapper has no stable local file generation"))?;
    if file_digest(&plan.receipt.worker_path)? != plan.receipt.worker_digest {
        return Err(RailError::message(
            "installed compiler worker failed content verification",
        ));
    }
    plan.receipt.worker_generation = crate::utils::stable_file_generation(&plan.receipt.worker_path)
        .ok_or_else(|| RailError::message("installed compiler worker has no stable local file generation"))?;
    if let Some(distributed) = plan.receipt.distributed.as_mut() {
        if plan
            .distributed_worker
            .as_ref()
            .map(|setup| setup.source_digest.as_str())
            != Some(distributed.worker_digest.as_str())
            || file_digest(&distributed.worker_path)? != distributed.worker_digest
        {
            return Err(RailError::message(
                "installed distributed compiler worker failed content verification",
            ));
        }
        distributed.worker_generation =
            crate::utils::stable_file_generation(&distributed.worker_path).ok_or_else(|| {
                RailError::message("installed distributed compiler worker has no stable local file generation")
            })?;
        if let Some(mutual_tls) = distributed.mutual_tls.as_mut() {
            for (path, expected) in [
                (&mutual_tls.authority_certificate, &mutual_tls.authority_digest),
                (&mutual_tls.client_certificate, &mutual_tls.client_certificate_digest),
                (&mutual_tls.client_private_key, &mutual_tls.client_private_key_digest),
            ] {
                if file_digest(path)? != *expected {
                    return Err(RailError::message(
                        "installed distributed TLS identity failed content verification",
                    ));
                }
            }
            mutual_tls.authority_generation = crate::utils::stable_file_generation(&mutual_tls.authority_certificate)
                .ok_or_else(|| {
                RailError::message("installed distributed TLS authority has no stable file generation")
            })?;
            mutual_tls.client_certificate_generation =
                crate::utils::stable_file_generation(&mutual_tls.client_certificate).ok_or_else(|| {
                    RailError::message("installed distributed TLS client certificate has no stable file generation")
                })?;
            mutual_tls.client_private_key_generation =
                crate::utils::stable_file_generation(&mutual_tls.client_private_key).ok_or_else(|| {
                    RailError::message("installed distributed TLS client key has no stable file generation")
                })?;
            crate::compiler::distributed::validate_mutual_tls_client_identity(
                &mutual_tls.server_name,
                &mutual_tls.worker_capability_id,
                &mutual_tls.authority_certificate,
                &mutual_tls.client_certificate,
                &mutual_tls.client_private_key,
            )?;
        }
    }
    let encoded_receipt = encode_receipt(&plan.receipt)?;
    if plan.receipt_before.as_deref() != Some(encoded_receipt.as_slice()) {
        if let (Some(stale), Some(quarantine)) = (
            plan.receipt_before
                .as_deref()
                .filter(|_| plan.stale_receipt_quarantine.is_some()),
            plan.stale_receipt_quarantine.as_deref(),
        ) {
            let directory = quarantine
                .parent()
                .ok_or_else(|| RailError::message("installation quarantine receipt has no parent"))?;
            create_private_directory(directory)?;
            match read_optional_regular(quarantine, MAX_RECEIPT_BYTES)? {
                Some(existing) if existing == stale => {}
                Some(_) => {
                    return Err(RailError::message(
                        "installation quarantine receipt conflicts with retained recovery evidence",
                    ));
                }
                None => write_private_atomic(quarantine, stale)?,
            }
            revalidate_optional(&receipt_path, Some(stale), MAX_RECEIPT_BYTES)?;
        }
        write_private_atomic(&receipt_path, &encoded_receipt)?;
    }
    revalidate_optional(&plan.config_path, plan.config_before.as_deref(), 16 * 1024 * 1024)?;
    if plan.config_before.as_deref() != Some(plan.config_after.as_slice()) {
        crate::utils::write_file_atomic(&plan.config_path, &plan.config_after)?;
    }
    Ok(())
}

/// Build a no-write exact removal plan for the selected Cargo home.
pub(crate) fn plan_removal(current_dir: &Path) -> RailResult<RemovalPlan> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let config_path = selected_user_config(&cargo_home);
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let receipt_before = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)?;
    let Some(receipt_bytes) = receipt_before else {
        return Ok(RemovalPlan {
            config_path,
            config_before: None,
            config_after: None,
            receipt_before: None,
            receipt: None,
            wrapper_before_digest: None,
            worker_before_digest: None,
            compiler_component_before_digests: Vec::new(),
            distributed_worker_before_digest: None,
            distributed_identity_before_digests: Vec::new(),
        });
    };
    let receipt = parse_receipt(&receipt_bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != config_path {
        return Err(RailError::message(
            "transparent compiler-cache removal authority does not match the selected Cargo home",
        ));
    }
    let config_before = read_optional_regular(&config_path, 16 * 1024 * 1024)?;
    if config_before.is_none() && !receipt.config_created {
        return Err(RailError::message(
            "Cargo user configuration owned by another authority is missing",
        ));
    }
    let config_after = config_before
        .as_deref()
        .map(|bytes| {
            std::str::from_utf8(bytes)
                .map_err(|_| RailError::message("Cargo user configuration is not valid UTF-8"))
                .and_then(|contents| remove_wrapper_value(contents, &receipt))
        })
        .transpose()?
        .flatten();
    let wrapper_before_digest = match optional_file_digest(&receipt.wrapper_path)? {
        Some(digest) if digest == receipt.wrapper_digest => Some(digest),
        Some(_) => {
            return Err(RailError::message(
                "installed compiler wrapper content changed; removal refused",
            ));
        }
        None => None,
    };
    let worker_before_digest = match optional_file_digest(&receipt.worker_path)? {
        Some(digest) if digest == receipt.worker_digest => Some(digest),
        Some(_) => {
            return Err(RailError::message(
                "installed compiler worker content changed; removal refused",
            ));
        }
        None => None,
    };
    let distributed_worker_before_digest = match receipt.distributed.as_ref() {
        Some(distributed) => match optional_file_digest(&distributed.worker_path)? {
            Some(digest) if digest == distributed.worker_digest => Some(digest),
            Some(_) => {
                return Err(RailError::message(
                    "installed distributed compiler worker content changed; removal refused",
                ));
            }
            None => None,
        },
        None => None,
    };
    let compiler_component_before_digests = receipt
        .compiler_components
        .iter()
        .map(compiler_component_removal_digest)
        .collect::<RailResult<Vec<_>>>()?;
    let distributed_identity_before_digests = receipt
        .distributed
        .as_ref()
        .and_then(|distributed| distributed.mutual_tls.as_ref())
        .map(|mutual_tls| {
            [
                (&mutual_tls.authority_certificate, &mutual_tls.authority_digest),
                (&mutual_tls.client_certificate, &mutual_tls.client_certificate_digest),
                (&mutual_tls.client_private_key, &mutual_tls.client_private_key_digest),
            ]
            .into_iter()
            .map(|(path, expected)| match optional_file_digest(path)? {
                Some(digest) if digest == *expected => Ok(Some(digest)),
                Some(_) => Err(RailError::message(
                    "installed distributed TLS identity changed; removal refused",
                )),
                None => Ok(None),
            })
            .collect::<RailResult<Vec<_>>>()
        })
        .transpose()?
        .unwrap_or_default();
    Ok(RemovalPlan {
        config_path,
        config_before,
        config_after,
        receipt_before: Some(receipt_bytes),
        receipt: Some(receipt),
        wrapper_before_digest,
        worker_before_digest,
        compiler_component_before_digests,
        distributed_worker_before_digest,
        distributed_identity_before_digests,
    })
}

/// Remove only configuration and private installation state bound by one receipt.
pub(crate) fn apply_removal(plan: RemovalPlan) -> RailResult<()> {
    let Some(receipt) = plan.receipt else {
        return Ok(());
    };
    let install_directory = receipt
        .wrapper_path
        .parent()
        .ok_or_else(|| RailError::message("installed compiler wrapper has no parent"))?;
    let receipt_path = install_directory.join(RECEIPT_FILE);
    revalidate_optional(&plan.config_path, plan.config_before.as_deref(), 16 * 1024 * 1024)?;
    revalidate_optional(&receipt_path, plan.receipt_before.as_deref(), MAX_RECEIPT_BYTES)?;
    match (
        &plan.wrapper_before_digest,
        optional_file_digest(&receipt.wrapper_path)?,
    ) {
        (Some(expected), Some(current)) if expected == &current => {}
        (None, None) => {}
        _ => {
            return Err(RailError::with_help(
                "installed compiler wrapper changed after removal planning",
                "rerun the command after inspecting the installation drift",
            ));
        }
    }
    match (&plan.worker_before_digest, optional_file_digest(&receipt.worker_path)?) {
        (Some(expected), Some(current)) if expected == &current => {}
        (None, None) => {}
        _ => {
            return Err(RailError::with_help(
                "installed compiler worker changed after removal planning",
                "rerun the command after inspecting the installation drift",
            ));
        }
    }
    for (component, expected) in receipt
        .compiler_components
        .iter()
        .zip(&plan.compiler_component_before_digests)
    {
        revalidate_compiler_component_removal(component, expected)?;
    }
    if let Some(distributed) = &receipt.distributed {
        match (
            &plan.distributed_worker_before_digest,
            optional_file_digest(&distributed.worker_path)?,
        ) {
            (Some(expected), Some(current)) if expected == &current => {}
            (None, None) => {}
            _ => {
                return Err(RailError::with_help(
                    "installed distributed compiler worker changed after removal planning",
                    "rerun the command after inspecting the installation drift",
                ));
            }
        }
    }
    if let Some(mutual_tls) = receipt
        .distributed
        .as_ref()
        .and_then(|distributed| distributed.mutual_tls.as_ref())
    {
        for ((path, _), expected) in [
            (&mutual_tls.authority_certificate, &mutual_tls.authority_digest),
            (&mutual_tls.client_certificate, &mutual_tls.client_certificate_digest),
            (&mutual_tls.client_private_key, &mutual_tls.client_private_key_digest),
        ]
        .into_iter()
        .zip(&plan.distributed_identity_before_digests)
        {
            match (expected, optional_file_digest(path)?) {
                (Some(expected), Some(current)) if expected == &current => {}
                (None, None) => {}
                _ => {
                    return Err(RailError::with_help(
                        "installed distributed TLS identity changed after removal planning",
                        "rerun the command after inspecting the installation drift",
                    ));
                }
            }
        }
    }
    for profile in crate::cache::profile::load_all(&receipt.cargo_home)? {
        let mut selected = receipt.clone();
        selected.attach_profile(profile);
        crate::remote_cache::stop_installed_coordinators(&selected);
    }
    let _profile_lifecycles = crate::cache::profile::lock_all_exclusive(&receipt.cargo_home)?;
    let _installation_lock = lock_installation(&receipt.cargo_home)?;

    match plan.config_after {
        Some(contents) => crate::utils::write_file_atomic(&plan.config_path, &contents)?,
        None if plan.config_before.is_some() => fs::remove_file(&plan.config_path)?,
        None => {}
    }
    revalidate_optional(&receipt_path, plan.receipt_before.as_deref(), MAX_RECEIPT_BYTES)?;
    let session_lock = lock_session(&receipt)?;
    if plan.wrapper_before_digest.is_some() {
        fs::remove_file(&receipt.wrapper_path)?;
    }
    if plan.worker_before_digest.is_some() {
        fs::remove_file(&receipt.worker_path)?;
    }
    for (component, expected) in receipt
        .compiler_components
        .iter()
        .zip(&plan.compiler_component_before_digests)
    {
        revalidate_compiler_component_removal(component, expected)?;
        if expected.is_some() {
            fs::remove_file(&component.path)?;
        }
    }
    if plan.distributed_worker_before_digest.is_some()
        && let Some(distributed) = &receipt.distributed
    {
        fs::remove_file(&distributed.worker_path)?;
    }
    if let Some(mutual_tls) = receipt
        .distributed
        .as_ref()
        .and_then(|distributed| distributed.mutual_tls.as_ref())
    {
        for (path, present) in [
            &mutual_tls.authority_certificate,
            &mutual_tls.client_certificate,
            &mutual_tls.client_private_key,
        ]
        .into_iter()
        .zip(&plan.distributed_identity_before_digests)
        {
            if present.is_some() {
                fs::remove_file(path)?;
            }
        }
    }
    if read_optional_regular(&receipt.session_memo_path()?, MAX_SESSION_MEMO_BYTES)?.is_some() {
        fs::remove_file(receipt.session_memo_path()?)?;
    }
    if read_optional_regular(&receipt.usage_path()?, MAX_USAGE_BYTES)?.is_some() {
        fs::remove_file(receipt.usage_path()?)?;
    }
    if read_bounded_private_log(&receipt.early_bypass_path()?, MAX_EARLY_BYPASS_BYTES)?.is_some() {
        fs::remove_file(receipt.early_bypass_path()?)?;
    }
    remove_failure_counter_state(&receipt)?;
    remove_distributed_placement_state(&receipt)?;
    fs::remove_file(&receipt_path)?;
    let session_lock_path = receipt.session_lock_path()?;
    drop(session_lock);
    if read_optional_regular(&session_lock_path, 0)?.is_some() {
        fs::remove_file(session_lock_path)?;
    }
    match fs::remove_dir(install_directory) {
        Ok(()) => {}
        Err(error) if error.kind() == std::io::ErrorKind::DirectoryNotEmpty => {}
        Err(error) => return Err(error.into()),
    }
    if let Some(owner) = install_directory.parent() {
        match fs::remove_dir(owner) {
            Ok(()) => {}
            Err(error) if error.kind() == std::io::ErrorKind::DirectoryNotEmpty => {}
            Err(error) => return Err(error.into()),
        }
    }
    Ok(())
}

fn remove_failure_counter_state(receipt: &InstallationReceipt) -> RailResult<()> {
    let lock_path = receipt.failure_counters_lock_path()?;
    let lock = crate::utils::open_cache_lock_file(&lock_path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        lock.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "compiler-cache failure counter lock is not a private regular file",
        ));
    }
    lock.lock()?;
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "compiler-cache failure counter lock changed while it was acquired",
        ));
    }
    let counters = receipt.failure_counters_path()?;
    if read_optional_regular(&counters, MAX_FAILURE_COUNTERS_BYTES)?.is_some() {
        fs::remove_file(counters)?;
    }
    drop(lock);
    if read_optional_regular(&lock_path, 0)?.is_some() {
        fs::remove_file(lock_path)?;
    }
    Ok(())
}

fn remove_distributed_placement_state(receipt: &InstallationReceipt) -> RailResult<()> {
    let lock_path = receipt.distributed_placement_lock_path()?;
    let lock = crate::utils::open_cache_lock_file(&lock_path, true)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;

        lock.set_permissions(fs::Permissions::from_mode(0o600))?;
    }
    if !crate::utils::private_file_matches_path(&lock, &lock_path, 0)? {
        return Err(RailError::message(
            "distributed placement lock is not a private regular file",
        ));
    }
    lock.lock()?;
    let history = receipt.distributed_placement_history_path()?;
    if read_optional_regular(&history, MAX_DISTRIBUTED_PLACEMENT_HISTORY_BYTES)?.is_some() {
        fs::remove_file(history)?;
    }
    drop(lock);
    if read_optional_regular(&lock_path, 0)?.is_some() {
        fs::remove_file(lock_path)?;
    }
    Ok(())
}

/// Load and validate the receipt adjacent to one installed wrapper process.
pub(crate) fn load_for_wrapper(invoked: &Path, workspace_root: &Path) -> RailResult<InstallationReceipt> {
    let invoked = absolute(invoked)?;
    let launcher = std::env::var_os(DIRECT_LAUNCHER_ENV)
        .map(PathBuf::from)
        .ok_or_else(|| RailError::message("installed compiler worker has no launcher capability"))?;
    let launcher = absolute(&launcher)?;
    let directory = invoked
        .parent()
        .ok_or_else(|| RailError::message("installed compiler wrapper has no parent directory"))?;
    let receipt_path = directory.join(RECEIPT_FILE);
    let bytes = read_required_regular(&receipt_path, MAX_RECEIPT_BYTES)?;
    let mut receipt = parse_receipt(&bytes, &receipt_path)?;
    if crate::utils::canonicalize_existing(&receipt.worker_path)? != crate::utils::canonicalize_existing(&invoked)?
        || crate::utils::canonicalize_existing(&receipt.wrapper_path)?
            != crate::utils::canonicalize_existing(&launcher)?
    {
        return Err(RailError::message(
            "installed compiler launcher or worker does not match its setup authority",
        ));
    }
    if crate::utils::stable_file_generation(&invoked).as_ref() != Some(&receipt.worker_generation)
        || crate::utils::stable_file_generation(&launcher).as_ref() != Some(&receipt.wrapper_generation)
        || receipt.compiler_components.iter().any(|component| {
            crate::utils::stable_file_generation(&component.path).as_ref() != Some(&component.generation)
        })
        || receipt.distributed.as_ref().is_some_and(|distributed| {
            crate::utils::stable_file_generation(&distributed.worker_path).as_ref()
                != Some(&distributed.worker_generation)
                || distributed.mutual_tls.as_ref().is_some_and(|mutual_tls| {
                    crate::utils::stable_file_generation(&mutual_tls.authority_certificate).as_ref()
                        != Some(&mutual_tls.authority_generation)
                        || crate::utils::stable_file_generation(&mutual_tls.client_certificate).as_ref()
                            != Some(&mutual_tls.client_certificate_generation)
                        || crate::utils::stable_file_generation(&mutual_tls.client_private_key).as_ref()
                            != Some(&mutual_tls.client_private_key_generation)
                })
        })
    {
        return Err(RailError::message(
            "installed compiler launcher or worker changed after verified setup",
        ));
    }
    let selected = crate::cache::profile::load_locked(&receipt.cargo_home, workspace_root)?;
    let (profile, lock, readiness_local_only) = match selected {
        Some((profile, lock)) => (profile, lock, false),
        None => {
            let (profile, lock) = readiness_profile(&receipt.cargo_home, workspace_root)?.ok_or_else(|| {
                RailError::with_help(
                    "the compiler workspace has no installed cache profile",
                    "run `cargo rail cache setup` from that exact Cargo workspace",
                )
            })?;
            (profile, lock, true)
        }
    };
    receipt.readiness_local_only = readiness_local_only;
    receipt.attach_locked_profile(profile, lock);
    Ok(receipt)
}

// The readiness workspace is deliberately not enrolled. Its capability names
// the enrolled workspace whose local installation it may exercise.
fn readiness_profile(
    cargo_home: &Path,
    source_root: &Path,
) -> RailResult<
    Option<(
        crate::cache::profile::InstalledCacheProfile,
        crate::cache::profile::ProfileLifecycleLock,
    )>,
> {
    let Some(workspace) = std::env::var_os(READINESS_WORKSPACE_ENV).map(PathBuf::from) else {
        return Ok(None);
    };
    let workspace = crate::utils::canonicalize_existing(&workspace)?;
    let source_root = crate::utils::canonicalize_existing(source_root)?;
    let parent = workspace.join("target/cargo-rail/compiler-artifacts-v1");
    let source_parent = source_root.parent();
    let source_name = source_root.file_name().and_then(std::ffi::OsStr::to_str);
    if source_parent != Some(parent.as_path())
        || source_name.is_none_or(|name| !name.starts_with("cargo-rail-cache-readiness-"))
    {
        return Err(RailError::message(
            "cache readiness workspace capability does not authorize this compiler source root",
        ));
    }
    crate::cache::profile::load_locked(cargo_home, &workspace)
}

/// Load and validate the receipt adjacent to a coordinator worker process.
pub(crate) fn load_for_coordinator(invoked: &Path) -> RailResult<InstallationReceipt> {
    let invoked = absolute(invoked)?;
    let directory = invoked
        .parent()
        .ok_or_else(|| RailError::message("installed compiler worker has no parent directory"))?;
    let receipt_path = directory.join(RECEIPT_FILE);
    let bytes = read_required_regular(&receipt_path, MAX_RECEIPT_BYTES)?;
    let mut receipt = parse_receipt(&bytes, &receipt_path)?;
    if crate::utils::canonicalize_existing(&receipt.worker_path)? != crate::utils::canonicalize_existing(&invoked)?
        || crate::utils::stable_file_generation(&invoked).as_ref() != Some(&receipt.worker_generation)
        || file_digest(&invoked)? != receipt.worker_digest
        || receipt.compiler_components.iter().any(|component| {
            crate::utils::stable_file_generation(&component.path).as_ref() != Some(&component.generation)
        })
    {
        return Err(RailError::message(
            "remote coordinator worker does not match its setup authority",
        ));
    }
    let capability = std::env::var(crate::cache::profile::COORDINATOR_PROFILE_ENV)
        .map_err(|_| RailError::message("remote coordinator has no cache profile capability"))?;
    let (profile, lock) = crate::cache::profile::load_locked_coordinator_capability(&receipt.cargo_home, &capability)?;
    receipt.attach_locked_profile(profile, lock);
    Ok(receipt)
}

fn coordinator_state_path(receipt: &InstallationReceipt, identity: &str, suffix: &str) -> RailResult<PathBuf> {
    if identity.len() != 64
        || !identity
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(RailError::message("remote coordinator identity is invalid"));
    }
    Ok(receipt
        .profile_state_directory()?
        .join(format!("{COORDINATOR_STATE_PREFIX}{identity}.{suffix}")))
}

pub(crate) fn coordinator_state_file(receipt: &InstallationReceipt, identity: &str) -> RailResult<PathBuf> {
    coordinator_state_path(receipt, identity, "json")
}

pub(crate) fn coordinator_lock_file(receipt: &InstallationReceipt, identity: &str) -> RailResult<PathBuf> {
    coordinator_state_path(receipt, identity, "lock")
}

pub(crate) fn read_coordinator_state(
    receipt: &InstallationReceipt,
    identity: &str,
    maximum: u64,
) -> RailResult<Option<Vec<u8>>> {
    read_optional_regular(&coordinator_state_file(receipt, identity)?, maximum)
}

pub(crate) fn write_coordinator_state(receipt: &InstallationReceipt, identity: &str, bytes: &[u8]) -> RailResult<()> {
    write_private_atomic(&coordinator_state_file(receipt, identity)?, bytes)
}

pub(crate) fn remove_coordinator_state_if(
    receipt: &InstallationReceipt,
    identity: &str,
    expected: &[u8],
) -> RailResult<()> {
    let path = coordinator_state_file(receipt, identity)?;
    if read_optional_regular(&path, expected.len() as u64)?.as_deref() == Some(expected) {
        fs::remove_file(path)?;
    }
    Ok(())
}

/// Load the optional non-secret remote policy and retain its profile authority.
pub(crate) fn installed_remote(current_dir: &Path) -> RailResult<Option<InstalledRemoteSelection>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let config_path = selected_user_config(&cargo_home);
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(None);
    };
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != config_path {
        return Err(RailError::message(
            "transparent compiler-cache setup authority does not match the selected Cargo home",
        ));
    }
    Ok(
        crate::cache::profile::load_locked(&cargo_home, current_dir)?.map(|(profile, lock)| InstalledRemoteSelection {
            remote: profile.remote().cloned(),
            _profile: lock,
        }),
    )
}

/// Load the local CAS selected for this exact workspace and retain its profile
/// authority for the lifetime of the caller's compiler analysis.
pub(crate) fn installed_local(current_dir: &Path) -> RailResult<Option<InstalledLocalSelection>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let config_path = selected_user_config(&cargo_home);
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(None);
    };
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != config_path {
        return Err(RailError::message(
            "transparent compiler-cache setup authority does not match the selected Cargo home",
        ));
    }
    Ok(
        crate::cache::profile::load_locked(&cargo_home, current_dir)?.map(|(profile, lock)| InstalledLocalSelection {
            cache: profile.cache().clone(),
            remote: profile.remote().cloned(),
            _profile: lock,
        }),
    )
}

/// Return the exact digest only when one selected program is the unchanged
/// compiler wrapper owned by the canonical installation receipt.
pub(crate) fn verified_installed_wrapper_digest(selection: &OsStr, current_dir: &Path) -> RailResult<Option<String>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let config_path = selected_user_config(&cargo_home);
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(None);
    };
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    let selected = crate::executable::resolve_executable_path(selection, current_dir)?;
    if receipt.cargo_home != cargo_home
        || receipt.config_path != config_path
        || crate::utils::canonicalize_existing(&receipt.wrapper_path)? != selected
        || crate::utils::stable_file_generation(&receipt.wrapper_path).as_ref() != Some(&receipt.wrapper_generation)
        || file_digest(&receipt.wrapper_path)? != receipt.wrapper_digest
    {
        return Ok(None);
    }
    Ok(Some(receipt.wrapper_digest))
}

pub(crate) fn stop_current_profile_coordinator(current_dir: &Path) -> RailResult<()> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(());
    };
    let mut receipt = parse_receipt(&bytes, &receipt_path)?;
    if let Some(profile) = crate::cache::profile::load(&cargo_home, current_dir)? {
        receipt.attach_profile(profile);
        crate::remote_cache::stop_installed_coordinators(&receipt);
    }
    Ok(())
}

/// Inspect installation health without creating Cargo or cache state.
pub(crate) fn status(current_dir: &Path) -> RailResult<InstallationStatus> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let config_path = selected_user_config(&cargo_home);
    let storage = installation_storage_status(&cargo_home)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(InstallationStatus {
            state: "not_installed",
            healthy: true,
            installation_integrity: "not_installed",
            component_authentication: "not_installed",
            workspace_enrollment: "not_enrolled",
            observed_reuse: "not_observed",
            receipt_version: None,
            recovery_action: None,
            cargo_home: cargo_home.to_string_lossy().into_owned(),
            config_path: config_path.to_string_lossy().into_owned(),
            wrapper_path: None,
            profile_id: None,
            bound_workspace_root: None,
            trust_domain: None,
            selection_source: "none",
            cache_base: None,
            max_bytes: None,
            root_portability: None,
            distributed: None,
            distributed_policy: None,
            distributed_placement_history: None,
            cargo_l0: "owned_by_cargo_not_observable_when_rustc_is_not_launched",
            owned_bytes: storage.owned_bytes,
            required_bytes: storage.required_bytes,
            reclaimable_bytes: storage.reclaimable_bytes,
            quarantined_receipts: storage.quarantined_receipts,
            usage: empty_usage_status(),
            issues: Vec::new(),
        });
    };
    let version = parse_receipt_version(&bytes, &receipt_path)?;
    if version != INSTALLATION_VERSION {
        let wrapper_path = receipt_path
            .parent()
            .ok_or_else(|| RailError::message("compiler-cache receipt has no parent directory"))?
            .join(WRAPPER_FILE);
        let mut issues = vec![format!(
            "compiler-cache receipt version {version} requires quarantine and reinstallation"
        )];
        match read_optional_regular(&config_path, 16 * 1024 * 1024) {
            Ok(Some(config)) => match std::str::from_utf8(&config)
                .map_err(|_| RailError::message("Cargo user configuration is not valid UTF-8"))
                .and_then(wrapper_value)
            {
                Ok(Some(selected)) if Path::new(&selected) == wrapper_path => {}
                Ok(_) => issues.push("Cargo build.rustc-wrapper does not select the stale installation".to_string()),
                Err(error) => issues.push(error.to_string()),
            },
            Ok(None) => issues.push("Cargo user configuration is missing".to_string()),
            Err(error) => issues.push(error.to_string()),
        }
        return Ok(InstallationStatus {
            state: "stale",
            healthy: false,
            installation_integrity: "stale_receipt",
            component_authentication: "not_authenticated",
            workspace_enrollment: "unknown",
            observed_reuse: "not_observed",
            receipt_version: Some(version),
            recovery_action: Some("cargo rail cache setup"),
            cargo_home: cargo_home.to_string_lossy().into_owned(),
            config_path: config_path.to_string_lossy().into_owned(),
            wrapper_path: Some(wrapper_path.to_string_lossy().into_owned()),
            profile_id: None,
            bound_workspace_root: None,
            trust_domain: None,
            selection_source: "stale_receipt",
            cache_base: None,
            max_bytes: None,
            root_portability: None,
            distributed: None,
            distributed_policy: None,
            distributed_placement_history: None,
            cargo_l0: "owned_by_cargo_not_observable_when_rustc_is_not_launched",
            owned_bytes: storage.owned_bytes,
            required_bytes: storage.required_bytes,
            reclaimable_bytes: storage.reclaimable_bytes,
            quarantined_receipts: storage.quarantined_receipts,
            usage: empty_usage_status(),
            issues,
        });
    }
    let mut receipt = parse_receipt(&bytes, &receipt_path)?;
    let mut issues = Vec::new();
    let mut installation_integrity = "verified";
    let mut component_authentication = "authenticated";
    let profile = match crate::cache::profile::load(&cargo_home, current_dir) {
        Ok(Some(profile)) => {
            receipt.attach_profile(profile.clone());
            Some(profile)
        }
        Ok(None) => None,
        Err(error) => {
            issues.push(format!("workspace cache profile is unavailable: {error}"));
            None
        }
    };
    if receipt.cargo_home != cargo_home || receipt.config_path != config_path {
        installation_integrity = "drifted";
        issues.push("selected Cargo home or active user config changed".to_string());
    }
    match read_optional_regular(&config_path, 16 * 1024 * 1024) {
        Ok(Some(bytes)) => match std::str::from_utf8(&bytes)
            .map_err(|_| RailError::message("Cargo user configuration is not valid UTF-8"))
            .and_then(wrapper_value)
        {
            Ok(Some(value)) if Path::new(&value) == receipt.wrapper_path => {}
            Ok(_) => {
                installation_integrity = "drifted";
                issues.push("Cargo build.rustc-wrapper no longer selects the installed wrapper".to_string());
            }
            Err(error) => {
                installation_integrity = "drifted";
                issues.push(error.to_string());
            }
        },
        Ok(None) => {
            installation_integrity = "drifted";
            issues.push("Cargo user configuration is missing".to_string());
        }
        Err(error) => {
            installation_integrity = "drifted";
            issues.push(error.to_string());
        }
    }
    match file_digest(&receipt.wrapper_path) {
        Ok(digest) if digest == receipt.wrapper_digest => {}
        Ok(_) => {
            component_authentication = "failed";
            issues.push("installed wrapper content changed".to_string());
        }
        Err(error) => {
            component_authentication = "failed";
            issues.push(format!("installed wrapper is unavailable: {error}"));
        }
    }
    match file_digest(&receipt.worker_path) {
        Ok(digest) if digest == receipt.worker_digest => {}
        Ok(_) => {
            component_authentication = "failed";
            issues.push("installed worker content changed".to_string());
        }
        Err(error) => {
            component_authentication = "failed";
            issues.push(format!("installed worker is unavailable: {error}"));
        }
    }
    for component in &receipt.compiler_components {
        match file_digest(&component.path) {
            Ok(digest)
                if digest == component.digest
                    && crate::utils::stable_file_generation(&component.path).as_ref()
                        == Some(&component.generation) => {}
            Ok(_) => {
                component_authentication = "failed";
                issues.push(format!(
                    "installed compiler component '{}' changed",
                    component.path.display()
                ));
            }
            Err(error) => {
                component_authentication = "failed";
                issues.push(format!(
                    "installed compiler component '{}' is unavailable: {error}",
                    component.path.display()
                ));
            }
        }
    }
    if let Some(distributed) = &receipt.distributed {
        match file_digest(&distributed.worker_path) {
            Ok(digest)
                if digest == distributed.worker_digest
                    && crate::utils::stable_file_generation(&distributed.worker_path).as_ref()
                        == Some(&distributed.worker_generation) => {}
            Ok(_) => {
                component_authentication = "failed";
                issues.push("installed distributed worker content changed".to_string());
            }
            Err(error) => {
                component_authentication = "failed";
                issues.push(format!("installed distributed worker is unavailable: {error}"));
            }
        }
        if let Some(mutual_tls) = &distributed.mutual_tls {
            for (role, path, expected, generation) in [
                (
                    "authority",
                    &mutual_tls.authority_certificate,
                    &mutual_tls.authority_digest,
                    &mutual_tls.authority_generation,
                ),
                (
                    "client certificate",
                    &mutual_tls.client_certificate,
                    &mutual_tls.client_certificate_digest,
                    &mutual_tls.client_certificate_generation,
                ),
                (
                    "client key",
                    &mutual_tls.client_private_key,
                    &mutual_tls.client_private_key_digest,
                    &mutual_tls.client_private_key_generation,
                ),
            ] {
                match file_digest(path) {
                    Ok(digest)
                        if digest == *expected
                            && crate::utils::stable_file_generation(path).as_ref() == Some(generation) => {}
                    Ok(_) => {
                        component_authentication = "failed";
                        issues.push(format!("installed distributed TLS {role} changed"));
                    }
                    Err(error) => {
                        component_authentication = "failed";
                        issues.push(format!("installed distributed TLS {role} is unavailable: {error}"));
                    }
                }
            }
            if let Err(error) = crate::compiler::distributed::validate_mutual_tls_client_identity(
                &mutual_tls.server_name,
                &mutual_tls.worker_capability_id,
                &mutual_tls.authority_certificate,
                &mutual_tls.client_certificate,
                &mutual_tls.client_private_key,
            ) {
                component_authentication = "failed";
                issues.push(error.to_string());
            }
        }
    }
    if let Some(profile) = &profile {
        match profile.cache().configured_root() {
            Ok(Some(_)) => {}
            Ok(None) => issues.push("local cache is unavailable; run cargo rail cache setup to repair it".to_string()),
            Err(error) => issues.push(format!("local cache authority is invalid: {error}")),
        }
    }
    if let Err(error) = reject_shadowing_global_wrapper(current_dir, &config_path) {
        installation_integrity = "drifted";
        issues.push(error.to_string());
    }
    if component_authentication == "failed" {
        installation_integrity = "drifted";
    }
    let mut usage = if profile.is_some() {
        match usage_status(&receipt) {
            Ok(usage) => usage,
            Err(error) => {
                issues.push(error.to_string());
                InstallationUsageStatus::default()
            }
        }
    } else {
        empty_usage_status()
    };
    if profile.is_some() {
        populate_failure_reason_status(&receipt, &mut usage);
    }
    let distributed_placement_history = if receipt
        .distributed
        .as_ref()
        .is_some_and(|distributed| distributed.mutual_tls.is_some())
    {
        match crate::compiler::distributed::placement_history_status(&receipt) {
            Ok(status) => Some(status),
            Err(error) => {
                issues.push(format!("distributed placement history is unavailable: {error}"));
                None
            }
        }
    } else {
        None
    };
    Ok(InstallationStatus {
        state: if issues.is_empty() { "installed" } else { "drifted" },
        healthy: issues.is_empty(),
        installation_integrity,
        component_authentication,
        workspace_enrollment: if profile.is_some() { "enrolled" } else { "not_enrolled" },
        observed_reuse: observed_reuse_state(&usage),
        receipt_version: Some(INSTALLATION_VERSION),
        recovery_action: (!issues.is_empty()).then_some("cargo rail cache setup"),
        cargo_home: cargo_home.to_string_lossy().into_owned(),
        config_path: config_path.to_string_lossy().into_owned(),
        wrapper_path: Some(receipt.wrapper_path.to_string_lossy().into_owned()),
        profile_id: profile.as_ref().map(|profile| profile.profile_id().to_string()),
        bound_workspace_root: profile
            .as_ref()
            .map(|profile| profile.selected_root().to_string_lossy().into_owned()),
        trust_domain: profile
            .as_ref()
            .and_then(|profile| profile.cache().trust_domain().map(str::to_string)),
        selection_source: if profile.is_some() { "installed_profile" } else { "none" },
        cache_base: profile
            .as_ref()
            .map(|profile| profile.cache().base().to_string_lossy().into_owned()),
        max_bytes: profile.as_ref().map(|profile| profile.cache().max_bytes()),
        root_portability: profile.as_ref().map(|profile| profile.root_portability().as_str()),
        distributed: receipt
            .distributed
            .as_ref()
            .and_then(|distributed| match distributed.mode.as_str() {
                "local_process_qualification_v1" => Some("local_process_qualification_v1"),
                "mutual_tls_direct_v1" => Some("mutual_tls_direct_v1"),
                _ => None,
            }),
        distributed_policy: receipt
            .distributed
            .as_ref()
            .and_then(|distributed| distributed.placement)
            .map(DistributedPlacementPolicy::as_str),
        distributed_placement_history,
        cargo_l0: "owned_by_cargo_not_observable_when_rustc_is_not_launched",
        owned_bytes: storage.owned_bytes,
        required_bytes: storage.required_bytes,
        reclaimable_bytes: storage.reclaimable_bytes,
        quarantined_receipts: storage.quarantined_receipts,
        usage,
        issues,
    })
}

fn observed_reuse_state(usage: &InstallationUsageStatus) -> &'static str {
    if usage.hits > 0 {
        "verified_hit_observed"
    } else if usage.misses > 0 {
        "misses_only"
    } else if usage.bypasses > 0 {
        "bypasses_only"
    } else if usage.failures > 0 {
        "failures_only"
    } else {
        "not_observed"
    }
}

struct InstallationStorageStatus {
    owned_bytes: u64,
    required_bytes: u64,
    reclaimable_bytes: u64,
    quarantined_receipts: u64,
}

fn installation_storage_status(cargo_home: &Path) -> RailResult<InstallationStorageStatus> {
    let owner = cargo_home.join("cargo-rail");
    let owned_bytes = crate::cache::path_status(&owner)?.map_or(0, |(bytes, _, _)| bytes);
    let quarantine = owner.join(INSTALLATION_QUARANTINE_DIRECTORY);
    let (reclaimable_bytes, quarantined_receipts) =
        crate::cache::path_status(&quarantine)?.map_or((0, 0), |(bytes, files, _)| (bytes, files));
    let required_bytes = owned_bytes
        .checked_sub(reclaimable_bytes)
        .ok_or_else(|| RailError::message("compiler-cache installation storage accounting underflow"))?;
    Ok(InstallationStorageStatus {
        owned_bytes,
        required_bytes,
        reclaimable_bytes,
        quarantined_receipts,
    })
}

/// Inspect the exact local CAS selected by a healthy installation receipt.
pub(crate) fn local_cache_status(current_dir: &Path) -> RailResult<Option<crate::cache::cas::LocalCasStatus>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(None);
    };
    parse_receipt(&bytes, &receipt_path)?;
    let Some((profile, _profile_lock)) = crate::cache::profile::load_locked(&cargo_home, current_dir)? else {
        return Ok(None);
    };
    let Some(root) = profile.cache().configured_root()? else {
        return Ok(None);
    };
    crate::cache::cas::status_at_with_max(&root, profile.cache().max_bytes())
}

/// Preview recovery using only the exact cache authority in the setup receipt.
pub(crate) fn plan_local_cache_recovery(
    current_dir: &Path,
) -> RailResult<Option<crate::cache::cas::LocalCasRecoveryPlan>> {
    let selection = recovery_cache_selection(current_dir)?;
    crate::cache::cas::plan_markerless_recovery(&selection)
}

/// Quarantine one selected markerless CAS and create a fresh owned authority.
pub(crate) fn recover_local_cache(current_dir: &Path) -> RailResult<Option<crate::cache::cas::LocalCasRecoveryPlan>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let bytes = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)?.ok_or_else(|| {
        RailError::with_help(
            "transparent compiler-cache setup receipt is unavailable",
            "run `cargo rail cache setup --check` before attempting local CAS recovery",
        )
    })?;
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != selected_user_config(&cargo_home) {
        return Err(RailError::message(
            "transparent compiler-cache recovery authority does not match the selected Cargo home",
        ));
    }
    let (profile, _profile_lock) = crate::cache::profile::load_exclusive(&cargo_home, current_dir)?
        .ok_or_else(|| RailError::message("the selected workspace has no installed cache profile"))?;
    let selection = profile.cache().clone();
    crate::cache::cas::recover_markerless_selected(&selection)
}

fn recovery_cache_selection(current_dir: &Path) -> RailResult<LocalCacheSelection> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let bytes = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)?.ok_or_else(|| {
        RailError::with_help(
            "transparent compiler-cache setup receipt is unavailable",
            "run `cargo rail cache setup --check` before attempting local CAS recovery",
        )
    })?;
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != selected_user_config(&cargo_home) {
        return Err(RailError::message(
            "transparent compiler-cache recovery authority does not match the selected Cargo home",
        ));
    }
    crate::cache::profile::load(&cargo_home, current_dir)?
        .map(|profile| profile.cache().clone())
        .ok_or_else(|| RailError::message("the selected workspace has no installed cache profile"))
}

/// Remove only the CAS selected by an installation receipt, when one exists.
/// `Some` means the receipt was authoritative even when its CAS was absent.
pub(crate) fn remove_local_cache(current_dir: &Path) -> RailResult<Option<Vec<(PathBuf, u64)>>> {
    let cargo_home = resolve_cargo_home(current_dir)?;
    let receipt_path = cargo_home
        .join("cargo-rail")
        .join(INSTALLATION_DIRECTORY)
        .join(RECEIPT_FILE);
    let Some(bytes) = read_optional_regular(&receipt_path, MAX_RECEIPT_BYTES)? else {
        return Ok(None);
    };
    let receipt = parse_receipt(&bytes, &receipt_path)?;
    if receipt.cargo_home != cargo_home || receipt.config_path != selected_user_config(&cargo_home) {
        return Err(RailError::message(
            "transparent compiler-cache cleanup authority does not match the selected Cargo home",
        ));
    }
    let mut removed = Vec::new();
    let Some((profile, _profile_lock)) = crate::cache::profile::load_exclusive(&cargo_home, current_dir)? else {
        return Ok(Some(removed));
    };
    if let Some(root) = profile.cache().configured_root()?
        && let Some(entry) = crate::cache::cas::remove_owned_root_at(&root)?
    {
        removed.push(entry);
    }
    Ok(Some(removed))
}

fn resolve_cargo_home(current_dir: &Path) -> RailResult<PathBuf> {
    let selected = CargoConfigSnapshot::cargo_home(current_dir)?;
    resolve_requested_path(current_dir, &selected)
}

pub(crate) fn selected_cargo_home(current_dir: &Path) -> RailResult<PathBuf> {
    resolve_cargo_home(current_dir)
}

fn selected_user_config(cargo_home: &Path) -> PathBuf {
    let extensionless = cargo_home.join("config");
    if extensionless.is_file() {
        extensionless
    } else {
        cargo_home.join("config.toml")
    }
}

fn reject_shadowing_global_wrapper(current_dir: &Path, user_config: &Path) -> RailResult<()> {
    for name in ["CARGO_BUILD_RUSTC_WRAPPER", "RUSTC_WRAPPER"] {
        if std::env::var_os(name).is_some() {
            return Err(RailError::with_help(
                format!("{name} shadows Cargo's user rustc-wrapper setting"),
                format!("unset {name} before installing transparent compiler reuse"),
            ));
        }
    }
    let captured = CargoConfigSnapshot::capture(current_dir)?;
    for source in captured.provenance() {
        if source.path() != user_config
            && source
                .settings()
                .get("build")
                .and_then(serde_json::Value::as_object)
                .is_some_and(|build| build.contains_key("rustc-wrapper"))
        {
            return Err(RailError::with_help(
                format!(
                    "Cargo configuration '{}' shadows the user rustc-wrapper setting",
                    source.path().display()
                ),
                "remove the higher-precedence build.rustc-wrapper or keep Cargo-Rail uninstalled for this workspace",
            ));
        }
    }
    Ok(())
}

fn install_wrapper_value(original: &str, wrapper: &Path, owned_wrapper: Option<&Path>) -> RailResult<(String, bool)> {
    let mut document = if original.is_empty() {
        DocumentMut::new()
    } else {
        original
            .parse::<DocumentMut>()
            .map_err(|error| RailError::message(format!("failed to parse Cargo user configuration: {error}")))?
    };
    let selected = wrapper_value_from_document(&document)?;
    if let Some(selected) = selected
        && owned_wrapper.is_none_or(|owned| Path::new(selected) != owned)
    {
        return Err(RailError::with_help(
            format!("Cargo user configuration already selects rustc wrapper '{selected}'"),
            "remove the existing global wrapper before installing Cargo-Rail",
        ));
    }
    let encoded = wrapper
        .to_str()
        .ok_or_else(|| RailError::message("installed compiler wrapper path is not valid UTF-8"))?;
    let build_table_created = document.get("build").is_none();
    match document.get_mut("build") {
        Some(Item::Table(table)) => {
            table.insert("rustc-wrapper", toml_edit::value(encoded));
        }
        Some(Item::Value(Value::InlineTable(table))) => {
            table.insert("rustc-wrapper", Value::from(encoded));
        }
        Some(_) => return Err(RailError::message("Cargo configuration key 'build' is not a table")),
        None => {
            let mut table = Table::new();
            table.insert("rustc-wrapper", toml_edit::value(encoded));
            document.insert("build", Item::Table(table));
        }
    }
    Ok((document.to_string(), build_table_created))
}

fn wrapper_value(contents: &str) -> RailResult<Option<String>> {
    let document = contents
        .parse::<DocumentMut>()
        .map_err(|error| RailError::message(format!("failed to parse Cargo user configuration: {error}")))?;
    wrapper_value_from_document(&document).map(|value| value.map(str::to_string))
}

fn wrapper_value_from_document(document: &DocumentMut) -> RailResult<Option<&str>> {
    let Some(build) = document.get("build") else {
        return Ok(None);
    };
    let value = match build {
        Item::Table(table) => table.get("rustc-wrapper").and_then(Item::as_str),
        Item::Value(Value::InlineTable(table)) => table.get("rustc-wrapper").and_then(Value::as_str),
        _ => return Err(RailError::message("Cargo configuration key 'build' is not a table")),
    };
    if value.is_none()
        && match build {
            Item::Table(table) => table.contains_key("rustc-wrapper"),
            Item::Value(Value::InlineTable(table)) => table.contains_key("rustc-wrapper"),
            _ => false,
        }
    {
        return Err(RailError::message(
            "Cargo configuration key 'build.rustc-wrapper' is not a string",
        ));
    }
    Ok(value)
}

fn remove_wrapper_value(contents: &str, receipt: &InstallationReceipt) -> RailResult<Option<Vec<u8>>> {
    let mut document = contents
        .parse::<DocumentMut>()
        .map_err(|error| RailError::message(format!("failed to parse Cargo user configuration: {error}")))?;
    if let Some(selected) = wrapper_value_from_document(&document)?
        && Path::new(selected) != receipt.wrapper_path
    {
        return Err(RailError::message(
            "Cargo build.rustc-wrapper changed after setup; removal refused",
        ));
    }
    if let Some(build) = document.get_mut("build") {
        match build {
            Item::Table(table) => {
                table.remove("rustc-wrapper");
            }
            Item::Value(Value::InlineTable(table)) => {
                table.remove("rustc-wrapper");
            }
            _ => return Err(RailError::message("Cargo configuration key 'build' is not a table")),
        }
    }
    if receipt.build_table_created
        && document.get("build").is_some_and(|build| match build {
            Item::Table(table) => table.is_empty(),
            Item::Value(Value::InlineTable(table)) => table.is_empty(),
            _ => false,
        })
    {
        document.remove("build");
    }
    let bytes = document.to_string().into_bytes();
    if receipt.config_created && bytes.iter().all(u8::is_ascii_whitespace) {
        Ok(None)
    } else {
        Ok(Some(bytes))
    }
}

pub(super) fn resolve_requested_path(current_dir: &Path, path: &Path) -> RailResult<PathBuf> {
    let path = if path.is_absolute() {
        path.to_path_buf()
    } else {
        current_dir.join(path)
    };
    crate::utils::canonicalize_allow_missing(&path).map_err(Into::into)
}

pub(super) fn create_private_directory(path: &Path) -> RailResult<()> {
    fs::create_dir_all(path)?;
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.is_dir() || crate::utils::is_symlink_or_reparse(&metadata) {
        return Err(RailError::message(format!(
            "private installation path '{}' is not a real directory",
            path.display()
        )));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
    }
    Ok(())
}

fn ensure_real_directory(path: &Path) -> RailResult<()> {
    fs::create_dir_all(path)?;
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.is_dir() || crate::utils::is_symlink_or_reparse(&metadata) {
        return Err(RailError::message(format!(
            "Cargo home '{}' is not a real directory",
            path.display()
        )));
    }
    Ok(())
}

fn require_consistent_component_versions(components: &[(&str, &Path)]) -> RailResult<()> {
    const RECOVERY: &str =
        "reinstall one complete Cargo-Rail component set, then verify it with `cargo rail cache setup --check`";

    let current_executable = std::env::current_exe()?;
    let mut releases = vec![(env!("CARGO_PKG_VERSION").to_string(), vec!["cargo-rail"])];
    for &(name, path) in components {
        let version = if path == current_executable {
            env!("CARGO_PKG_VERSION").to_string()
        } else {
            let output = Command::new(path)
                .arg(crate::compiler::component_version::ARGUMENT)
                .output()
                .map_err(|error| {
                    RailError::with_help(
                        format!("cannot read the Cargo-Rail release version from compiler component '{name}': {error}"),
                        RECOVERY,
                    )
                })?;
            let response = std::str::from_utf8(&output.stdout).ok().and_then(|stdout| {
                let line = stdout.strip_suffix('\n').unwrap_or(stdout);
                let line = line.strip_suffix('\r').unwrap_or(line);
                line.strip_prefix(crate::compiler::component_version::RESPONSE_PREFIX)
                    .and_then(|version| version.strip_prefix('\t'))
                    .filter(|version| {
                        !version.is_empty()
                            && version.len() <= 64
                            && version
                                .bytes()
                                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b'+'))
                    })
            });
            let Some(response) = response.filter(|_| output.status.success() && output.stderr.is_empty()) else {
                return Err(RailError::with_help(
                    format!("compiler component '{name}' did not report a valid Cargo-Rail release version"),
                    RECOVERY,
                ));
            };
            response.to_string()
        };
        if let Some((_, names)) = releases.iter_mut().find(|(release, _)| release == &version) {
            names.push(name);
        } else {
            releases.push((version, vec![name]));
        }
    }
    if releases.len() == 1 {
        return Ok(());
    }
    let versions = releases
        .into_iter()
        .map(|(version, names)| format!("{version} ({})", names.join(", ")))
        .collect::<Vec<_>>()
        .join("; ");
    Err(RailError::with_help(
        format!("compiler components come from inconsistent Cargo-Rail releases: {versions}"),
        RECOVERY,
    ))
}

fn plan_executable_setup(source: PathBuf, destination: &Path, description: &str) -> RailResult<ExecutableSetup> {
    Ok(ExecutableSetup {
        source_digest: installation_source_digest(&source)?,
        source,
        before: observe_installed_executable(destination, description)?,
    })
}

fn compiler_component_mode_matches(path: &Path, executable: bool) -> RailResult<bool> {
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.is_file() || crate::utils::is_symlink_or_reparse(&metadata) {
        return Ok(false);
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        Ok(metadata.permissions().mode() & 0o777 == if executable { 0o700 } else { 0o600 })
    }
    #[cfg(not(unix))]
    {
        let _ = executable;
        Ok(!metadata.permissions().readonly())
    }
}

fn compiler_component_removal_digest(component: &InstalledCompilerComponent) -> RailResult<Option<String>> {
    match optional_file_digest(&component.path)? {
        Some(digest) if digest == component.digest => Ok(Some(digest)),
        Some(_) => Err(RailError::message(
            "installed compiler component changed; removal refused",
        )),
        None => Ok(None),
    }
}

fn revalidate_compiler_component_removal(
    component: &InstalledCompilerComponent,
    expected: &Option<String>,
) -> RailResult<()> {
    if optional_file_digest(&component.path)? != *expected {
        return Err(RailError::with_help(
            "installed compiler component changed after removal planning",
            "rerun the command after inspecting the installation drift",
        ));
    }
    Ok(())
}

fn observe_installed_executable(path: &Path, description: &str) -> RailResult<Option<InstalledExecutable>> {
    match fs::symlink_metadata(path) {
        Ok(_) => {
            let generation_before = crate::utils::stable_file_generation(path)
                .ok_or_else(|| RailError::message(format!("{description} has no stable local file generation")))?;
            let digest = file_digest(path)?;
            let generation = crate::utils::stable_file_generation(path)
                .ok_or_else(|| RailError::message(format!("{description} has no stable local file generation")))?;
            if generation != generation_before {
                return Err(RailError::with_help(
                    format!("{description} changed while setup was planned"),
                    "rerun the command to build a new exact mutation plan",
                ));
            }
            Ok(Some(InstalledExecutable { digest, generation }))
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

fn revalidate_executable_setup(path: &Path, setup: &ExecutableSetup, description: &str) -> RailResult<()> {
    if observe_installed_executable(path, description)?.as_ref() == setup.before.as_ref() {
        Ok(())
    } else {
        Err(RailError::with_help(
            format!("{description} changed after setup planning"),
            "rerun the command to build a new exact mutation plan",
        ))
    }
}

fn install_planned_executable(setup: &ExecutableSetup, destination: &Path) -> RailResult<()> {
    if setup.requires_installation() {
        install_executable_atomic(&setup.source, destination)?;
    }
    Ok(())
}

fn install_executable_atomic(source: &Path, destination: &Path) -> RailResult<()> {
    let parent = destination
        .parent()
        .ok_or_else(|| RailError::message("installed wrapper has no parent"))?;
    let source_metadata = fs::symlink_metadata(source)?;
    if !source_metadata.is_file() || crate::utils::is_symlink_or_reparse(&source_metadata) {
        return Err(RailError::message("compiler wrapper source is not a real regular file"));
    }
    let mut source = File::open(source)?;
    let mut temporary = tempfile::Builder::new()
        .prefix(".cargo-rail-wrapper-")
        .tempfile_in(parent)?;
    std::io::copy(&mut source, &mut temporary)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        temporary.as_file().set_permissions(fs::Permissions::from_mode(0o700))?;
    }
    temporary.as_file().sync_all()?;
    crate::utils::persist_file_atomic(temporary, destination)?;
    Ok(())
}

fn install_private_file_atomic(source: &Path, destination: &Path, expected_digest: &str) -> RailResult<()> {
    let parent = destination
        .parent()
        .ok_or_else(|| RailError::message("installed distributed identity has no parent"))?;
    let metadata = fs::symlink_metadata(source)?;
    if !metadata.is_file() || crate::utils::is_symlink_or_reparse(&metadata) {
        return Err(RailError::message(
            "distributed identity source is not a real regular file",
        ));
    }
    let mut input = File::open(source)?;
    if !crate::utils::opened_file_matches_path(&input, source, metadata.len())? {
        return Err(RailError::message(
            "distributed identity source changed before it was opened",
        ));
    }
    let mut builder = tempfile::Builder::new();
    builder.prefix(".cargo-rail-distributed-identity-");
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        builder.permissions(fs::Permissions::from_mode(0o600));
    }
    let mut temporary = builder.tempfile_in(parent)?;
    std::io::copy(&mut input, &mut temporary)?;
    if !crate::utils::opened_file_matches_path(&input, source, metadata.len())? {
        return Err(RailError::message(
            "distributed identity source changed while it was copied",
        ));
    }
    temporary.as_file().sync_all()?;
    if digest_regular_file(temporary.path(), false)? != expected_digest {
        return Err(RailError::message(
            "distributed identity source changed after setup planning",
        ));
    }
    crate::utils::persist_file_atomic(temporary, destination)?;
    Ok(())
}

pub(super) fn write_private_atomic(path: &Path, bytes: &[u8]) -> RailResult<()> {
    let parent = path
        .parent()
        .ok_or_else(|| RailError::message("private installation file has no parent"))?;
    let mut builder = tempfile::Builder::new();
    builder.prefix(".cargo-rail-install-");
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        builder.permissions(fs::Permissions::from_mode(0o600));
    }
    let mut temporary = builder.tempfile_in(parent)?;
    temporary.write_all(bytes)?;
    temporary.as_file().sync_all()?;
    crate::utils::persist_file_atomic(temporary, path)?;
    Ok(())
}

pub(super) fn read_optional_regular(path: &Path, max_bytes: u64) -> RailResult<Option<Vec<u8>>> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            if !metadata.is_file() || crate::utils::is_symlink_or_reparse(&metadata) || metadata.len() > max_bytes {
                return Err(RailError::message(format!(
                    "installation input '{}' is not a bounded regular file",
                    path.display()
                )));
            }
            let file = File::open(path)?;
            if !crate::utils::private_file_matches_path(&file, path, metadata.len())? {
                return Err(RailError::message(format!(
                    "installation input '{}' changed before it was opened",
                    path.display()
                )));
            }
            let mut bytes = Vec::with_capacity(usize::try_from(metadata.len()).unwrap_or(0));
            file.take(max_bytes.saturating_add(1)).read_to_end(&mut bytes)?;
            if bytes.len() as u64 != metadata.len() {
                return Err(RailError::message(format!(
                    "installation input '{}' changed while it was read",
                    path.display()
                )));
            }
            Ok(Some(bytes))
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

fn read_required_regular(path: &Path, max_bytes: u64) -> RailResult<Vec<u8>> {
    read_optional_regular(path, max_bytes)?
        .ok_or_else(|| RailError::message(format!("required installation file '{}' is missing", path.display())))
}

fn revalidate_optional(path: &Path, expected: Option<&[u8]>, max_bytes: u64) -> RailResult<()> {
    if read_optional_regular(path, max_bytes)?.as_deref() == expected {
        Ok(())
    } else {
        Err(RailError::with_help(
            format!("installation input '{}' changed after planning", path.display()),
            "rerun the command to build a new exact mutation plan",
        ))
    }
}

fn parse_receipt(bytes: &[u8], path: &Path) -> RailResult<InstallationReceipt> {
    let version = parse_receipt_version(bytes, path)?;
    if version != INSTALLATION_VERSION {
        return Err(RailError::with_help(
            format!(
                "unsupported compiler-cache installation receipt version {} at '{}'",
                version,
                path.display()
            ),
            "preserve the installation; use the executable that created it to preview and perform its removal, then run current `cargo rail cache setup`; the originating release version is unknown",
        ));
    }
    let receipt: InstallationReceipt = serde_json::from_slice(bytes)?;
    receipt.validate()?;
    if encode_receipt(&receipt)? != bytes {
        return Err(RailError::message(
            "transparent compiler-cache installation receipt is not canonical",
        ));
    }
    Ok(receipt)
}

fn parse_receipt_version(bytes: &[u8], path: &Path) -> RailResult<u32> {
    #[derive(Deserialize)]
    struct ReceiptVersion {
        version: u32,
    }
    let version: ReceiptVersion = serde_json::from_slice(bytes).map_err(|error| {
        RailError::message(format!(
            "compiler-cache installation receipt at '{}' has no readable version: {error}",
            path.display()
        ))
    })?;
    if version.version == 0 {
        return Err(RailError::message(
            "compiler-cache installation receipt version must be positive",
        ));
    }
    Ok(version.version)
}

fn parse_legacy_receipt(bytes: &[u8], cargo_home: &Path, config_path: &Path) -> RailResult<LegacyInstallationReceipt> {
    let receipt: LegacyInstallationReceipt = serde_json::from_slice(bytes)?;
    let install_directory = cargo_home.join("cargo-rail").join(INSTALLATION_DIRECTORY);
    if receipt.version == INSTALLATION_VERSION
        || !valid_hex_digest(&receipt.authority)
        || receipt.cargo_home != cargo_home
        || receipt.config_path != config_path
        || receipt.wrapper_path != install_directory.join(WRAPPER_FILE)
    {
        return Err(RailError::with_help(
            "stale compiler-cache receipt does not contain safe migration authority",
            "preserve the Cargo-Rail state and inspect the receipt before choosing a manual quarantine path",
        ));
    }
    if let Some(cache) = &receipt.cache {
        LocalCacheSelection::new(
            cache.base().to_path_buf(),
            cache.max_bytes(),
            cache.trust_domain().map(str::to_string),
        )?;
    }
    Ok(receipt)
}

fn encode_receipt(receipt: &InstallationReceipt) -> RailResult<Vec<u8>> {
    let mut bytes = serde_json::to_vec_pretty(receipt)?;
    bytes.push(b'\n');
    if bytes.len() as u64 > MAX_RECEIPT_BYTES {
        return Err(RailError::message(
            "transparent compiler-cache receipt exceeds its bound",
        ));
    }
    Ok(bytes)
}

fn installation_source_digest(path: &Path) -> RailResult<String> {
    digest_regular_file(path, false)
}

fn file_digest(path: &Path) -> RailResult<String> {
    digest_regular_file(path, true)
}

fn digest_regular_file(path: &Path, require_single_link: bool) -> RailResult<String> {
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.is_file() || crate::utils::is_symlink_or_reparse(&metadata) {
        return Err(RailError::message(format!(
            "'{}' is not a real regular file",
            path.display()
        )));
    }
    let mut file = File::open(path)?;
    let matches = if require_single_link {
        crate::utils::private_file_matches_path(&file, path, metadata.len())?
    } else {
        crate::utils::opened_file_matches_path(&file, path, metadata.len())?
    };
    if !matches {
        return Err(RailError::message(format!(
            "'{}' changed before it was opened",
            path.display()
        )));
    }
    let mut hasher = Sha256::new();
    let mut buffer = [0_u8; 64 * 1024];
    let mut read = 0_u64;
    loop {
        let count = file.read(&mut buffer)?;
        if count == 0 {
            break;
        }
        read = read
            .checked_add(count as u64)
            .ok_or_else(|| RailError::message("compiler wrapper size overflow"))?;
        hasher.update(&buffer[..count]);
    }
    let matches = if require_single_link {
        crate::utils::private_file_matches_path(&file, path, metadata.len())?
    } else {
        crate::utils::opened_file_matches_path(&file, path, metadata.len())?
    };
    if read != metadata.len() || !matches {
        return Err(RailError::message(format!(
            "'{}' changed while it was read",
            path.display()
        )));
    }
    Ok(format!(
        "sha256:{}",
        ContentDigest::from_sha256_bytes(hasher.finalize())
    ))
}

fn optional_file_digest(path: &Path) -> RailResult<Option<String>> {
    match fs::symlink_metadata(path) {
        Ok(_) => file_digest(path).map(Some),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

fn absolute(path: &Path) -> RailResult<PathBuf> {
    if path.is_absolute() {
        Ok(path.to_path_buf())
    } else {
        Ok(std::env::current_dir()?.join(path))
    }
}

pub(super) fn random_authority() -> RailResult<String> {
    let mut bytes = [0_u8; 32];
    getrandom::fill(&mut bytes)
        .map_err(|error| RailError::message(format!("failed to acquire installation authority randomness: {error}")))?;
    Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect())
}

fn valid_hex_digest(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

fn valid_sha256(value: &str) -> bool {
    value.strip_prefix("sha256:").is_some_and(valid_hex_digest)
}

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

    #[test]
    fn failure_reason_counters_advance_after_the_usage_ledger_fills() {
        let root = tempfile::tempdir().unwrap();
        let receipt = receipt_with_components(root.path());
        fs::create_dir_all(receipt.profile_state_directory().unwrap()).unwrap();
        let usage = receipt.usage_path().unwrap();
        let full_ledger = vec![b'B'; usize::try_from(MAX_USAGE_BYTES).unwrap()];
        fs::write(&usage, &full_ledger).unwrap();
        let mut expected = std::collections::BTreeMap::from([
            ("complete_action_capture_unavailable".to_string(), 0_u64),
            ("complete_action_identity_unavailable".to_string(), 0_u64),
            ("post_execution_witness_validation_unavailable".to_string(), 0_u64),
        ]);
        for (reason, key) in [
            (
                NativeCacheFailureReason::ActionCapture,
                "complete_action_capture_unavailable",
            ),
            (
                NativeCacheFailureReason::ActionIdentity,
                "complete_action_identity_unavailable",
            ),
            (
                NativeCacheFailureReason::PostExecutionWitness,
                "post_execution_witness_validation_unavailable",
            ),
            (
                NativeCacheFailureReason::ActionCapture,
                "complete_action_capture_unavailable",
            ),
        ] {
            let error = record_usage(&receipt, b'B', Some(reason)).unwrap_err();
            assert!(error.to_string().contains("size bound"), "{error}");
            *expected.get_mut(key).unwrap() += 1;
            assert_eq!(read_failure_counters(&receipt).unwrap().unwrap().status(), expected);
            assert_eq!(fs::read(&usage).unwrap(), full_ledger);
        }
    }

    fn receipt_with_components(cargo_home: &Path) -> InstallationReceipt {
        let directory = cargo_home.join("cargo-rail").join(INSTALLATION_DIRECTORY);
        InstallationReceipt {
            version: INSTALLATION_VERSION,
            authority: "a".repeat(64),
            cargo_home: cargo_home.to_path_buf(),
            config_path: cargo_home.join("config.toml"),
            config_created: true,
            build_table_created: true,
            wrapper_path: directory.join(WRAPPER_FILE),
            wrapper_digest: format!("sha256:{}", "b".repeat(64)),
            wrapper_generation: vec![1],
            worker_path: directory.join(WORKER_FILE),
            worker_digest: format!("sha256:{}", "c".repeat(64)),
            worker_generation: vec![2],
            compiler_components: vec![InstalledCompilerComponent {
                path: directory.join(crate::compiler::driver::compiler_driver_file_name()),
                digest: format!("sha256:{}", "d".repeat(64)),
                generation: vec![3],
            }],
            distributed: None,
            active_profile: None,
            active_profile_lock: None,
            readiness_local_only: false,
        }
    }

    #[test]
    fn compiler_component_receipt_binds_only_named_siblings_and_requires_current_inventory() {
        let root = tempfile::tempdir().expect("receipt fixture");
        let receipt = receipt_with_components(root.path());
        receipt.validate().expect("named compiler component");
        for path in [
            root.path().join("outside"),
            receipt.wrapper_path.clone(),
            receipt.worker_path.clone(),
        ] {
            let mut changed = receipt.clone();
            changed.compiler_components[0].path = path;
            assert_eq!(
                changed.validate().unwrap_err().to_string(),
                "installed compiler component receipt is invalid"
            );
        }
        let mut duplicate = receipt.clone();
        duplicate
            .compiler_components
            .push(duplicate.compiler_components[0].clone());
        duplicate
            .validate()
            .expect_err("duplicated component ownership must fail");

        let mut value = serde_json::to_value(&receipt).expect("receipt encoding");
        value
            .as_object_mut()
            .expect("receipt object")
            .remove("compiler_components");
        let error = parse_receipt(
            &serde_json::to_vec(&value).expect("missing inventory encoding"),
            &root.path().join(RECEIPT_FILE),
        )
        .expect_err("current receipt cannot silently acquire an empty inventory");
        assert!(error.to_string().contains("missing field `compiler_components`"));
        value["version"] = serde_json::json!(4);
        let error = parse_receipt(
            &serde_json::to_vec(&value).expect("previous receipt encoding"),
            &root.path().join(RECEIPT_FILE),
        )
        .expect_err("previous receipt cannot acquire current runtime authority");
        assert!(
            error
                .to_string()
                .contains("unsupported compiler-cache installation receipt version 4")
        );
    }

    #[test]
    fn compiler_component_removal_rejects_changed_and_reappearing_files() {
        let root = tempfile::tempdir().expect("component fixture");
        let path = root
            .path()
            .join(crate::compiler::driver::COMPILER_DRIVER_SOURCE_FILE_NAME);
        fs::write(&path, b"authenticated source bytes").expect("source bytes");
        let component = InstalledCompilerComponent {
            path: path.clone(),
            digest: file_digest(&path).expect("component digest"),
            generation: vec![1],
        };
        let observed = compiler_component_removal_digest(&component).expect("unchanged owned component");
        assert_eq!(observed.as_ref(), Some(&component.digest));
        fs::write(&path, b"changed source bytes").expect("replace source");
        compiler_component_removal_digest(&component).expect_err("changed bytes are not removal authority");
        revalidate_compiler_component_removal(&component, &observed).expect_err("planned bytes changed");
        fs::remove_file(&path).expect("remove component");
        assert_eq!(
            compiler_component_removal_digest(&component).expect("missing component"),
            None
        );
        fs::write(&path, b"authenticated source bytes").expect("reappearing component");
        revalidate_compiler_component_removal(&component, &None).expect_err("a newly present file is outside the plan");
    }

    #[test]
    fn stale_worker_pin_retains_only_installation_ownership_authority() {
        let stale = format!("worker-capability-v3:sha256:{}", "a".repeat(64));
        assert!(valid_installed_worker_capability(&stale));
        assert!(!crate::compiler::distributed::worker_capability_identity_is_valid(
            &stale
        ));
        for invalid in [
            stale.replace("-v3:", "-v03:"),
            stale.replace("-v3:", "-v0:"),
            stale.replace("-v3:", "-v../:"),
            stale.replace("sha256:", "sha512:"),
            stale.to_uppercase(),
        ] {
            assert!(
                !valid_installed_worker_capability(&invalid),
                "accepted malformed pin: {invalid}"
            );
        }
    }

    #[test]
    fn lossless_wrapper_edit_preserves_existing_cargo_configuration() {
        let original = "# retained\n[net]\noffline = true\n\n[build] # retained table\ntarget-dir = 'out'\n";
        let wrapper = Path::new("/tools/cargo-rail-native-rustc-wrapper");
        let (edited, created) = install_wrapper_value(original, wrapper, None).expect("valid edit");
        assert!(!created);
        assert_eq!(edited, format!("{original}rustc-wrapper = \"{}\"\n", wrapper.display()));
    }

    #[test]
    fn setup_refuses_to_adopt_an_unowned_wrapper() {
        let error = install_wrapper_value("[build]\nrustc-wrapper = 'sccache'\n", Path::new("/tools/rail"), None)
            .expect_err("existing wrapper must conflict");
        assert!(error.to_string().contains("already selects rustc wrapper"));
    }

    #[test]
    fn mutation_revalidation_rejects_byte_drift() {
        let directory = tempfile::tempdir().expect("temporary directory");
        let path = directory.path().join("config.toml");
        fs::write(&path, b"before\n").expect("write original");
        fs::write(&path, b"after\n").expect("write drift");
        let error = revalidate_optional(&path, Some(b"before\n"), 1024).expect_err("drift must fail");
        assert!(error.to_string().contains("changed after planning"));
    }

    #[cfg(unix)]
    #[test]
    fn setup_accepts_a_hard_linked_build_artifact_without_weakening_installed_authority() {
        let directory = tempfile::tempdir().expect("temporary directory");
        let source = directory.path().join("source");
        let linked = directory.path().join("linked");
        fs::write(&source, b"compiler wrapper").expect("write source");
        fs::hard_link(&source, &linked).expect("create build-artifact hard link");

        assert_eq!(
            installation_source_digest(&source).expect("hard-linked source digest"),
            installation_source_digest(&linked).expect("hard-linked alias digest")
        );
        assert!(
            file_digest(&source).is_err(),
            "installed authority must stay single-link"
        );
    }
}