ahri_tre_secrets/

managed.rs

use crate::{ManagedSecretReference, SecretMaterial};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
use std::num::NonZeroU64;
#[cfg(target_os = "linux")]
use std::os::fd::AsRawFd;
#[cfg(unix)]
use std::os::unix::fs::{DirBuilderExt, MetadataExt, OpenOptionsExt};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use thiserror::Error;
use uuid::Uuid;

#[cfg(any(test, feature = "test-fixtures"))]
mod absence_fixture;

/// Fixed production root for the encrypted Managed-secret store.
pub const MANAGED_SECRET_ROOT: &str = "/var/lib/ahri-tre/secrets";
/// Fixed production destination for an offline root-identity rotation.
pub const MANAGED_SECRET_ROTATION_ROOT: &str = "/var/lib/ahri-tre/secrets-rotation";
/// Fixed production projection for the deployment X25519 identity.
pub const ROOT_IDENTITY_PATH: &str = "/run/secrets/ahri-tre/root-identity/value";
/// Fixed production projection for the next deployment X25519 identity.
pub const NEXT_ROOT_IDENTITY_PATH: &str = "/run/secrets/ahri-tre/root-identity-next/value";

const STORE_FORMAT_VERSION: u64 = 1;
const DATASTORE_REFERENCE_INVENTORY_VERSION: u64 = 1;
const ENVELOPE_FORMAT_VERSION: u8 = 1;
const ENVELOPE_MAGIC: &[u8; 16] = b"AHRI-TRE-SECRET\0";
const HEADER_FILE: &str = "header.json";
const LOCK_FILE: &str = "store.lock";
const MANIFEST_FILE: &str = "manifest.age";
const OBJECTS_DIRECTORY: &str = "objects";
const MAX_HEADER_BYTES: u64 = 16 * 1024;
const MAX_MANIFEST_BYTES: u64 = 64 * 1024 * 1024;
const MAX_CIPHERTEXT_BYTES: u64 = 256 * 1024;
const ZERO_HASH: [u8; 32] = [0; 32];
const LOCK_CONTENT: &[u8] = b"ahri-tre-managed-store-lock-v1\n";

/// Aggregate, disclosure-safe evidence from complete offline verification.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ManagedSecretVerification {
    deployment_id: Uuid,
    recipient_fingerprint: String,
    active_secret_count: usize,
    audit_record_count: usize,
}

impl ManagedSecretVerification {
    pub fn deployment_id(&self) -> Uuid {
        self.deployment_id
    }

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

    pub fn active_secret_count(&self) -> usize {
        self.active_secret_count
    }

    pub fn audit_record_count(&self) -> usize {
        self.audit_record_count
    }
}

/// Aggregate, disclosure-safe evidence from a staged offline root rotation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RootRotationVerification {
    deployment_id: Uuid,
    current_recipient_fingerprint: String,
    next_recipient_fingerprint: String,
    active_secret_count: usize,
    audit_record_count: usize,
}

impl RootRotationVerification {
    pub fn deployment_id(&self) -> Uuid {
        self.deployment_id
    }

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

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

    pub fn active_secret_count(&self) -> usize {
        self.active_secret_count
    }

    pub fn audit_record_count(&self) -> usize {
        self.audit_record_count
    }
}

/// A monotonically increasing Managed-secret version.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct ManagedSecretVersion(NonZeroU64);

impl ManagedSecretVersion {
    pub fn get(self) -> u64 {
        self.0.get()
    }

    fn first() -> Self {
        Self(NonZeroU64::MIN)
    }

    fn next(self) -> Result<Self, ManagedSecretError> {
        self.get()
            .checked_add(1)
            .and_then(NonZeroU64::new)
            .map(Self)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict))
    }
}

/// Safe metadata returned after a Managed-secret mutation or resolution.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ManagedSecretMetadata {
    reference: ManagedSecretReference,
    version: ManagedSecretVersion,
    created_at: DateTime<Utc>,
}

impl ManagedSecretMetadata {
    pub fn reference(&self) -> &ManagedSecretReference {
        &self.reference
    }

    pub fn version(&self) -> ManagedSecretVersion {
        self.version
    }

    pub fn created_at(&self) -> DateTime<Utc> {
        self.created_at
    }
}

/// Protected Managed-secret material with its safe resolution metadata.
pub struct ResolvedManagedSecret {
    metadata: ManagedSecretMetadata,
    material: SecretMaterial,
}

impl ResolvedManagedSecret {
    pub fn metadata(&self) -> &ManagedSecretMetadata {
        &self.metadata
    }

    pub fn material(&self) -> &SecretMaterial {
        &self.material
    }
}

impl fmt::Debug for ResolvedManagedSecret {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ResolvedManagedSecret")
            .field("metadata", &self.metadata)
            .field("material", &self.material)
            .finish()
    }
}

/// A durable, owner-specific replacement that is not active until its owning
/// authority has applied and verified the staged material.
pub struct StagedManagedSecret {
    stage_id: Uuid,
    metadata: ManagedSecretMetadata,
    material: SecretMaterial,
}

impl StagedManagedSecret {
    pub fn metadata(&self) -> &ManagedSecretMetadata {
        &self.metadata
    }

    pub fn material(&self) -> &SecretMaterial {
        &self.material
    }
}

impl fmt::Debug for StagedManagedSecret {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("StagedManagedSecret")
            .field("stage_id", &self.stage_id)
            .field("metadata", &self.metadata)
            .field("material", &self.material)
            .finish()
    }
}
/// Narrow backend contract consumed by trusted bootstrap and workflows.
pub trait ManagedSecretBackend: Send + Sync {
    fn create(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError>;

    fn resolve(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ResolvedManagedSecret, ManagedSecretError>;

    /// Authenticates the store manifest and looks up safe active metadata
    /// without opening the selected Secret envelope.
    fn lookup_active_metadata(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError>;
}

mod owner_backend_sealed {
    pub trait Sealed {}
}

/// Internal mutation primitives available only to explicit credential-owner
/// workflows. There is intentionally no generic operator overwrite command.
pub trait ManagedSecretOwnerBackend: ManagedSecretBackend + owner_backend_sealed::Sealed {
    fn stage_owner_replacement(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
        expected_active_version: ManagedSecretVersion,
    ) -> Result<StagedManagedSecret, ManagedSecretError>;

    fn resume_owner_replacement(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Option<StagedManagedSecret>, ManagedSecretError>;

    fn activate_owner_replacement(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError>;

    fn activate_owner_replacements(
        &self,
        staged: &[&StagedManagedSecret],
    ) -> Result<Vec<ManagedSecretMetadata>, ManagedSecretError>;

    fn owner_replacement_is_active(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<bool, ManagedSecretError>;

    fn discard_owner_replacement(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<(), ManagedSecretError>;
}

mod trusted_backend_sealed {
    pub trait Sealed {}
}

/// Repository-internal capability for Trusted-runtime reference retention and removal.
#[doc(hidden)]
pub trait ManagedSecretTrustedBackend:
    ManagedSecretOwnerBackend + trusted_backend_sealed::Sealed
{
    fn replace_trusted_reference_holder(
        &self,
        holder: &str,
        references: &[ManagedSecretReference],
    ) -> Result<(), ManagedSecretError>;

    fn remove_trusted_reference_holder(&self, holder: &str) -> Result<(), ManagedSecretError>;

    fn trusted_held_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretError>;

    fn trusted_reference_holders(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Vec<String>, ManagedSecretError>;

    fn datastore_reference_inventory_is_complete(&self) -> Result<bool, ManagedSecretError>;

    fn remove_with_repository_evidence(
        &self,
        reference: &ManagedSecretReference,
        proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError>;
}

/// Parsed deployment root identity. Its private key has no formatting surface.
pub struct ManagedSecretIdentity {
    identity: age::x25519::Identity,
    recipient: age::x25519::Recipient,
}

impl ManagedSecretIdentity {
    /// Parses a protected age X25519 identity without exposing it through an
    /// ordinary string-returning interface.
    pub fn from_material(material: &SecretMaterial) -> Result<Self, ManagedSecretError> {
        material.expose(|bytes| {
            let encoded = std::str::from_utf8(bytes)
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?
                .trim();
            let identity = age::x25519::Identity::from_str(encoded)
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?;
            let recipient = identity.to_public();
            Ok(Self {
                identity,
                recipient,
            })
        })
    }

    /// Loads the identity from its fixed production projection.
    pub fn load() -> Result<Self, ManagedSecretError> {
        let material = crate::load_root_identity_material()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?;
        Self::from_material(&material)
    }

    /// Loads the next identity from its fixed production projection.
    pub fn load_next() -> Result<Self, ManagedSecretError> {
        let material = crate::load_next_root_identity_material()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?;
        Self::from_material(&material)
    }

    /// Internal acceptance-harness seam. Production callers must use
    /// [`Self::load`], which retains the fixed injected-secret contract.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn load_from_injected_root_for_test(root: &Path) -> Result<Self, ManagedSecretError> {
        let material = crate::load_root_identity_material_from_root(root)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?;
        Self::from_material(&material)
    }

    /// Internal acceptance-harness seam for the fixed next-identity projection.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn load_next_from_injected_root_for_test(root: &Path) -> Result<Self, ManagedSecretError> {
        let material = crate::load_next_root_identity_material_from_root(root)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::InvalidIdentity))?;
        Self::from_material(&material)
    }

    fn recipient_string(&self) -> String {
        self.recipient.to_string()
    }
}

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

/// Version-1 encrypted local Managed-secret backend.
pub struct LocalManagedSecretStore {
    root: PathBuf,
    anchor: Arc<StoreAnchor>,
    deployment_id: Uuid,
    identity: Arc<ManagedSecretIdentity>,
    faults: Arc<dyn MutationFaultInjector>,
}

/// Read-only capability that authenticates only explicitly selected Managed
/// Secret envelopes and exposes safe metadata for them.
pub struct SelectedManagedSecretReader {
    store: LocalManagedSecretStore,
}

impl SelectedManagedSecretReader {
    /// Resolves only the selected reference under the read-only store anchor.
    pub fn resolve(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ResolvedManagedSecret, ManagedSecretError> {
        self.store.resolve(reference)
    }

    pub fn inspect(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.store.inspect(reference)
    }
}

impl fmt::Debug for SelectedManagedSecretReader {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SelectedManagedSecretReader")
            .field("store", &self.store)
            .finish()
    }
}

impl fmt::Debug for LocalManagedSecretStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("LocalManagedSecretStore")
            .field("root", &ahri_tre_security::REDACTED_RESTRICTED_REFERENCE)
            .field("deployment_id", &self.deployment_id)
            .field("recipient", &self.identity.recipient_string())
            .finish_non_exhaustive()
    }
}

struct StoreAnchor {
    root: File,
    objects: File,
    root_identity: DirectoryIdentity,
    objects_identity: DirectoryIdentity,
    require_write: bool,
}

impl StoreAnchor {
    fn open(root_path: &Path, require_write: bool) -> Result<Self, ManagedSecretError> {
        let root = open_directory(root_path)?;
        let root_metadata = root
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&root_metadata, require_write)?;
        let objects = open_directory_at(&root, Path::new(OBJECTS_DIRECTORY))?;
        let objects_metadata = objects
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&objects_metadata, require_write)?;
        Ok(Self {
            root_identity: directory_identity(&root_metadata),
            objects_identity: directory_identity(&objects_metadata),
            root,
            objects,
            require_write,
        })
    }

    fn validate_current(&self, root_path: &Path) -> Result<(), ManagedSecretError> {
        let metadata = fs::symlink_metadata(root_path)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&metadata, self.require_write)?;
        if directory_identity(&metadata) != self.root_identity {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
        let current_objects = open_directory_at(&self.root, Path::new(OBJECTS_DIRECTORY))?;
        let metadata = current_objects
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&metadata, self.require_write)?;
        if directory_identity(&metadata) != self.objects_identity {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Copy, PartialEq, Eq)]
struct DirectoryIdentity {
    device: u64,
    inode: u64,
}

#[cfg(unix)]
fn directory_identity(metadata: &fs::Metadata) -> DirectoryIdentity {
    DirectoryIdentity {
        device: metadata.dev(),
        inode: metadata.ino(),
    }
}

#[cfg(not(unix))]
fn directory_identity(_metadata: &fs::Metadata) -> DirectoryIdentity {
    DirectoryIdentity {
        device: 0,
        inode: 0,
    }
}

impl LocalManagedSecretStore {
    pub fn initialize(
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::initialize_at(Path::new(MANAGED_SECRET_ROOT), deployment_id, identity)
    }

    pub fn open(
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::open_at(Path::new(MANAGED_SECRET_ROOT), deployment_id, identity)
    }

    /// Opens a completely verified store without recovery or another mutation.
    pub fn open_verified(
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::open_verified_at(Path::new(MANAGED_SECRET_ROOT), deployment_id, identity)
    }

    /// Opens the fixed store for selected, read-only resolution without
    /// recovering state or authenticating unrelated Secret envelopes.
    pub fn open_selected_read_only(
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<SelectedManagedSecretReader, ManagedSecretError> {
        Self::open_selected_read_only_at(Path::new(MANAGED_SECRET_ROOT), deployment_id, identity)
    }

    /// Completely verifies the fixed active store without repair or mutation.
    pub fn verify_all(
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<ManagedSecretVerification, ManagedSecretError> {
        Self::verify_all_at(Path::new(MANAGED_SECRET_ROOT), deployment_id, identity)
    }

    /// Stages a complete root-identity rotation at the fixed production paths.
    /// The active store and both identity projections are never replaced.
    pub fn rotate_root(deployment_id: Uuid) -> Result<RootRotationVerification, RootRotationError> {
        Self::rotate_root_at_with_identity_loader(
            Path::new(MANAGED_SECRET_ROOT),
            Path::new(MANAGED_SECRET_ROTATION_ROOT),
            deployment_id,
            true,
            &NoRotationFaults,
            || {
                Ok((
                    ManagedSecretIdentity::load()?,
                    ManagedSecretIdentity::load_next()?,
                ))
            },
        )
    }

    /// Internal acceptance-harness seam. Production verification uses the
    /// fixed active store root.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn verify_all_at_for_test(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<ManagedSecretVerification, ManagedSecretError> {
        Self::verify_all_at(root, deployment_id, identity)
    }

    fn verify_all_at(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<ManagedSecretVerification, ManagedSecretError> {
        ensure_supported_platform()?;
        validate_directory(root, false)?;
        let anchor = StoreAnchor::open(root, false)?;
        let _lock = StoreLock::acquire(&anchor.root, LockMode::NonBlockingShared)?;
        let verified = verify_anchored_store(&anchor, root, deployment_id, &identity)?;
        Ok(verified.report())
    }

    /// Internal acceptance-harness seam. Production rotation also proves that
    /// the active store is mounted read-only.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn rotate_root_at_for_test(
        active_root: &Path,
        staging_root: &Path,
        deployment_id: Uuid,
        current_identity: ManagedSecretIdentity,
        next_identity: ManagedSecretIdentity,
    ) -> Result<RootRotationVerification, RootRotationError> {
        Self::rotate_root_at(
            active_root,
            staging_root,
            deployment_id,
            current_identity,
            next_identity,
            false,
        )
    }

    #[cfg(any(test, feature = "test-fixtures"))]
    fn rotate_root_at(
        active_root: &Path,
        staging_root: &Path,
        deployment_id: Uuid,
        current_identity: ManagedSecretIdentity,
        next_identity: ManagedSecretIdentity,
        require_read_only_active_mount: bool,
    ) -> Result<RootRotationVerification, RootRotationError> {
        Self::rotate_root_at_with_faults(
            active_root,
            staging_root,
            deployment_id,
            current_identity,
            next_identity,
            require_read_only_active_mount,
            &NoRotationFaults,
        )
    }

    #[cfg(any(test, feature = "test-fixtures"))]
    fn rotate_root_at_with_faults(
        active_root: &Path,
        staging_root: &Path,
        deployment_id: Uuid,
        current_identity: ManagedSecretIdentity,
        next_identity: ManagedSecretIdentity,
        require_read_only_active_mount: bool,
        faults: &dyn RotationFaultInjector,
    ) -> Result<RootRotationVerification, RootRotationError> {
        Self::rotate_root_at_with_identity_loader(
            active_root,
            staging_root,
            deployment_id,
            require_read_only_active_mount,
            faults,
            move || Ok((current_identity, next_identity)),
        )
    }

    fn rotate_root_at_with_identity_loader(
        active_root: &Path,
        staging_root: &Path,
        deployment_id: Uuid,
        require_read_only_active_mount: bool,
        faults: &dyn RotationFaultInjector,
        load_identities: impl FnOnce() -> Result<
            (ManagedSecretIdentity, ManagedSecretIdentity),
            ManagedSecretError,
        >,
    ) -> Result<RootRotationVerification, RootRotationError> {
        ensure_supported_platform()
            .and_then(|()| validate_directory(active_root, false))
            .map_err(|error| RootRotationError::at(RootRotationPhase::Prerequisites, error))?;
        let active = StoreAnchor::open(active_root, false)
            .map_err(|error| RootRotationError::at(RootRotationPhase::Prerequisites, error))?;
        faults
            .check(RotationCheckpoint::Lock)
            .map_err(|error| RootRotationError::at(RootRotationPhase::Lock, error))?;
        let _lock = StoreLock::acquire(&active.root, LockMode::NonBlockingExclusive)
            .map_err(|error| RootRotationError::at(RootRotationPhase::Lock, error))?;

        let (current_identity, next_identity) = load_identities()
            .map_err(|error| RootRotationError::at(RootRotationPhase::Prerequisites, error))?;

        if require_read_only_active_mount
            && !directory_mount_is_read_only(&active.root)
                .map_err(|error| RootRotationError::at(RootRotationPhase::Prerequisites, error))?
        {
            return Err(RootRotationError::at(
                RootRotationPhase::Prerequisites,
                ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem),
            ));
        }
        let staging = validate_empty_staging_directory(staging_root, &active)
            .map_err(|error| RootRotationError::at(RootRotationPhase::Prerequisites, error))?;
        if current_identity.recipient_string() == next_identity.recipient_string() {
            return Err(RootRotationError::at(
                RootRotationPhase::Prerequisites,
                ManagedSecretError::new(ManagedSecretErrorKind::RecipientMismatch),
            ));
        }
        faults
            .check(RotationCheckpoint::SourceDecrypt)
            .and_then(|()| faults.check(RotationCheckpoint::SourceAudit))
            .map_err(|error| RootRotationError::at(RootRotationPhase::SourceVerification, error))?;
        let verified =
            verify_anchored_store(&active, active_root, deployment_id, &current_identity).map_err(
                |error| RootRotationError::at(RootRotationPhase::SourceVerification, error),
            )?;

        let staged_anchor = populate_rotation_staging(&staging, &verified, &next_identity, faults)
            .map_err(|error| RootRotationError::at(RootRotationPhase::StagingWrite, error))?;
        faults
            .check(RotationCheckpoint::DestinationVerification)
            .map_err(|error| {
                RootRotationError::at(RootRotationPhase::DestinationVerification, error)
            })?;
        let _staged_lock = StoreLock::acquire(&staged_anchor.root, LockMode::NonBlockingShared)
            .map_err(|error| {
                RootRotationError::at(RootRotationPhase::DestinationVerification, error)
            })?;
        let staged =
            verify_anchored_store(&staged_anchor, &staging.path, deployment_id, &next_identity)
                .map_err(|error| {
                    RootRotationError::at(RootRotationPhase::DestinationVerification, error)
                })?
                .report();
        if staged.active_secret_count != verified.manifest.entries.len()
            || staged.audit_record_count != verified.manifest.audit.len()
        {
            return Err(RootRotationError::at(
                RootRotationPhase::DestinationVerification,
                ManagedSecretError::new(ManagedSecretErrorKind::Integrity),
            ));
        }
        Ok(RootRotationVerification {
            deployment_id,
            current_recipient_fingerprint: recipient_fingerprint(&verified.header.recipient),
            next_recipient_fingerprint: staged.recipient_fingerprint,
            active_secret_count: staged.active_secret_count,
            audit_record_count: staged.audit_record_count,
        })
    }

    /// Returns safe metadata for every active Managed secret in canonical
    /// reference order. No ciphertext or Secret material crosses this seam.
    pub fn list(&self) -> Result<Vec<ManagedSecretMetadata>, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        validate_store_tree(&self.anchor, &manifest)?;
        self.anchor.validate_current(&self.root)?;
        Ok(manifest
            .entries
            .into_iter()
            .map(|(reference, entry)| ManagedSecretMetadata {
                reference: ManagedSecretReference::from_str(&reference)
                    .expect("validated manifest contains canonical references"),
                version: entry.version,
                created_at: entry.created_at,
            })
            .collect())
    }

    /// Authenticates one active envelope and returns only its safe metadata.
    pub fn inspect(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        Ok(self.resolve(reference)?.metadata().clone())
    }

    fn lookup_active_metadata(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        validate_store_tree(&self.anchor, &manifest)?;
        let entry = manifest
            .entries
            .get(reference.as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        self.anchor.validate_current(&self.root)?;
        Ok(ManagedSecretMetadata {
            reference: reference.clone(),
            version: entry.version,
            created_at: entry.created_at,
        })
    }

    /// Persists a new encrypted version without making it active. Only the
    /// credential's owning authority may carry this capability through apply
    /// and verification to activation.
    fn stage_replacement(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
        expected_active_version: ManagedSecretVersion,
    ) -> Result<StagedManagedSecret, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        recover_uncommitted_files(&self.anchor, &manifest)?;
        let active = manifest
            .entries
            .get(reference.as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        if active.version != expected_active_version {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        if manifest.staged.contains_key(reference.as_str()) {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::AlreadyExists,
            ));
        }

        let version = expected_active_version.next()?;
        let created_at = Utc::now();
        let metadata = ManagedSecretMetadata {
            reference: reference.clone(),
            version,
            created_at,
        };
        let plaintext = encode_secret_envelope(
            self.deployment_id,
            &self.identity.recipient_string(),
            &metadata,
            material,
        )?;
        let ciphertext = encrypt(&self.identity.recipient, &plaintext)?;
        let object_name = format!("object-{}.age", Uuid::new_v4());
        atomic_write(
            &self.anchor.objects,
            &object_name,
            &ciphertext,
            self.faults.as_ref(),
            MutationPhase::Object,
        )
        .map_err(AtomicWriteFailure::into_error)?;
        if let Err(error) = verify_persisted_object(
            &self.anchor.objects,
            &object_name,
            self.deployment_id,
            &self.identity,
            &metadata,
            material,
        ) {
            remove_file_if_present(&self.anchor.objects, &object_name)?;
            sync_directory(&self.anchor.objects)?;
            return Err(error);
        }

        let stage_id = Uuid::new_v4();
        manifest.staged.insert(
            reference.as_str().to_string(),
            StagedManifestEntry {
                stage_id,
                expected_active_version,
                version,
                object: object_name.clone(),
                created_at,
            },
        );
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            if failure.renamed {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::CommitUncertain,
                ));
            }
            remove_file_if_present(&self.anchor.objects, &object_name)?;
            sync_directory(&self.anchor.objects)?;
            return Err(failure.into_error());
        }
        let material = material
            .expose(|bytes| SecretMaterial::try_from(bytes.to_vec()))
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::SizeLimit))?;
        Ok(StagedManagedSecret {
            stage_id,
            metadata,
            material,
        })
    }

    /// Reopens a durable pending replacement after process restart.
    fn pending_replacement(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Option<StagedManagedSecret>, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        validate_store_tree(&self.anchor, &manifest)?;
        let Some(staged) = manifest.staged.get(reference.as_str()) else {
            return Ok(None);
        };
        let entry = ManifestEntry {
            version: staged.version,
            object: staged.object.clone(),
            created_at: staged.created_at,
        };
        let ciphertext =
            read_store_file(&self.anchor.objects, &entry.object, MAX_CIPHERTEXT_BYTES)?;
        let plaintext = decrypt(&self.identity.identity, &ciphertext)?;
        let resolved = decode_secret_envelope(
            &plaintext,
            self.deployment_id,
            &self.identity.recipient_string(),
            reference,
            &entry,
        )?;
        Ok(Some(StagedManagedSecret {
            stage_id: staged.stage_id,
            metadata: resolved.metadata,
            material: resolved.material,
        }))
    }

    /// Atomically makes an owner-verified staged version active and then
    /// destroys the previous active ciphertext.
    fn activate_replacement(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        let pending = manifest
            .staged
            .get(staged.metadata.reference().as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        if pending.stage_id != staged.stage_id
            || pending.version != staged.metadata.version()
            || pending.created_at != staged.metadata.created_at()
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        verify_persisted_object(
            &self.anchor.objects,
            &pending.object,
            self.deployment_id,
            &self.identity,
            &staged.metadata,
            &staged.material,
        )?;
        let active = manifest
            .entries
            .get(staged.metadata.reference().as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        if active.version != pending.expected_active_version {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        let prior_object = active.object.clone();
        let pending = manifest
            .staged
            .remove(staged.metadata.reference().as_str())
            .expect("validated staged replacement remains present");
        manifest.entries.insert(
            staged.metadata.reference().as_str().to_string(),
            ManifestEntry {
                version: pending.version,
                object: pending.object,
                created_at: pending.created_at,
            },
        );
        manifest.generation = manifest
            .generation
            .checked_add(1)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict))?;
        manifest.audit.push(AuditRecord::mutation(
            manifest.audit.last(),
            AuditOperation::Replace,
            staged.metadata.reference(),
            staged.metadata.version(),
            staged.metadata.created_at(),
            self.deployment_id,
        )?);
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            return Err(if failure.renamed {
                ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain)
            } else {
                failure.into_error()
            });
        }
        remove_file_if_present(&self.anchor.objects, &prior_object)
            .and_then(|()| sync_directory(&self.anchor.objects))
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        self.anchor
            .validate_current(&self.root)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        Ok(staged.metadata.clone())
    }

    /// Atomically activates a complete owner transaction in one manifest commit.
    fn activate_replacements(
        &self,
        staged: &[&StagedManagedSecret],
    ) -> Result<Vec<ManagedSecretMetadata>, ManagedSecretError> {
        if staged.is_empty() {
            return Ok(Vec::new());
        }
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        let mut references = BTreeSet::new();
        let mut prior_objects = Vec::with_capacity(staged.len());
        for candidate in staged {
            let reference = candidate.metadata.reference();
            if !references.insert(reference.as_str()) {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::VersionConflict,
                ));
            }
            let pending = manifest
                .staged
                .get(reference.as_str())
                .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
            if pending.stage_id != candidate.stage_id
                || pending.version != candidate.metadata.version()
                || pending.created_at != candidate.metadata.created_at()
            {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::VersionConflict,
                ));
            }
            verify_persisted_object(
                &self.anchor.objects,
                &pending.object,
                self.deployment_id,
                &self.identity,
                &candidate.metadata,
                &candidate.material,
            )?;
            let active = manifest
                .entries
                .get(reference.as_str())
                .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
            if active.version != pending.expected_active_version {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::VersionConflict,
                ));
            }
            prior_objects.push(active.object.clone());
        }
        for candidate in staged {
            let reference = candidate.metadata.reference();
            let pending = manifest
                .staged
                .remove(reference.as_str())
                .expect("validated staged replacement remains present");
            manifest.entries.insert(
                reference.as_str().to_string(),
                ManifestEntry {
                    version: pending.version,
                    object: pending.object,
                    created_at: pending.created_at,
                },
            );
            manifest.audit.push(AuditRecord::mutation(
                manifest.audit.last(),
                AuditOperation::Replace,
                reference,
                candidate.metadata.version(),
                candidate.metadata.created_at(),
                self.deployment_id,
            )?);
        }
        manifest.generation =
            manifest
                .generation
                .checked_add(u64::try_from(staged.len()).map_err(|_| {
                    ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict)
                })?)
                .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict))?;
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            return Err(if failure.renamed {
                ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain)
            } else {
                failure.into_error()
            });
        }
        for object in prior_objects {
            remove_file_if_present(&self.anchor.objects, &object)
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        }
        sync_directory(&self.anchor.objects)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        self.anchor
            .validate_current(&self.root)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        Ok(staged
            .iter()
            .map(|candidate| candidate.metadata.clone())
            .collect())
    }

    /// Authenticates the manifest and reports whether an uncertain activation committed.
    fn replacement_is_active(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<bool, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        Ok(!manifest
            .staged
            .contains_key(staged.metadata.reference().as_str())
            && manifest
                .entries
                .get(staged.metadata.reference().as_str())
                .is_some_and(|entry| {
                    entry.version == staged.metadata.version()
                        && entry.created_at == staged.metadata.created_at()
                }))
    }

    /// Destroys a rejected staged value without changing the active version.
    fn discard_replacement(&self, staged: &StagedManagedSecret) -> Result<(), ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        let pending = manifest
            .staged
            .get(staged.metadata.reference().as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        if pending.stage_id != staged.stage_id
            || pending.version != staged.metadata.version()
            || pending.created_at != staged.metadata.created_at()
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        let object = pending.object.clone();
        manifest
            .staged
            .remove(staged.metadata.reference().as_str())
            .expect("validated staged replacement remains present");
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) && !failure.renamed
        {
            return Err(failure.into_error());
        }
        remove_file_if_present(&self.anchor.objects, &object)?;
        sync_directory(&self.anchor.objects)?;
        self.anchor.validate_current(&self.root)
    }

    /// Durably replaces the safe Managed-reference snapshot retained by one Session.
    fn replace_reference_holder(
        &self,
        holder: &str,
        references: &[ManagedSecretReference],
    ) -> Result<(), ManagedSecretError> {
        if !valid_reference_holder(holder) || references.is_empty() {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        recover_uncommitted_files(&self.anchor, &manifest)?;
        let references = references
            .iter()
            .map(|reference| {
                if manifest.entries.contains_key(reference.as_str()) {
                    Ok(reference.as_str().to_string())
                } else {
                    Err(ManagedSecretError::new(ManagedSecretErrorKind::Missing))
                }
            })
            .collect::<Result<BTreeSet<_>, _>>()?;
        if manifest.reference_holders.get(holder) == Some(&references) {
            return Ok(());
        }
        manifest
            .reference_holders
            .insert(holder.to_string(), references);
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            return Err(if failure.renamed {
                ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain)
            } else {
                failure.into_error()
            });
        }
        self.anchor.validate_current(&self.root)
    }

    /// Durably releases the reference snapshot for one closed Session.
    fn remove_reference_holder(&self, holder: &str) -> Result<(), ManagedSecretError> {
        if !valid_reference_holder(holder) {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        if manifest.reference_holders.remove(holder).is_none() {
            return Ok(());
        }
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            return Err(if failure.renamed {
                ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain)
            } else {
                failure.into_error()
            });
        }
        self.anchor.validate_current(&self.root)
    }

    /// Reads the authenticated durable Session-reference index.
    fn held_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        Ok(
            if manifest
                .reference_holders
                .values()
                .any(|references| references.contains(reference.as_str()))
            {
                ManagedSecretReferenceStatus::Referenced
            } else {
                ManagedSecretReferenceStatus::Unreferenced
            },
        )
    }

    fn reference_holders(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Vec<String>, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        Ok(manifest
            .reference_holders
            .iter()
            .filter(|(_, references)| references.contains(reference.as_str()))
            .map(|(holder, _)| holder.clone())
            .collect())
    }

    fn datastore_reference_inventory_is_complete(&self) -> Result<bool, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        Ok(manifest.datastore_reference_inventory_version == DATASTORE_REFERENCE_INVENTORY_VERSION)
    }

    /// Removes a Managed Secret only when the caller supplies proof issued by
    /// every authoritative reference owner. The audit tombstone is durable and
    /// prevents the canonical reference from being silently recreated.
    fn remove_proven_unreferenced(
        &self,
        reference: &ManagedSecretReference,
        proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        if proof.reference != *reference {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        recover_uncommitted_files(&self.anchor, &manifest)?;
        if manifest
            .reference_holders
            .values()
            .any(|references| references.contains(reference.as_str()))
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        if manifest.staged.contains_key(reference.as_str()) {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::VersionConflict,
            ));
        }
        let active = manifest
            .entries
            .remove(reference.as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        let metadata = ManagedSecretMetadata {
            reference: reference.clone(),
            version: active.version,
            created_at: active.created_at,
        };
        manifest.generation = manifest
            .generation
            .checked_add(1)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict))?;
        manifest.audit.push(AuditRecord::mutation(
            manifest.audit.last(),
            AuditOperation::Remove,
            reference,
            active.version,
            Utc::now(),
            self.deployment_id,
        )?);
        if let Err(failure) = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        ) {
            return Err(if failure.renamed {
                ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain)
            } else {
                failure.into_error()
            });
        }
        remove_file_if_present(&self.anchor.objects, &active.object)
            .and_then(|()| sync_directory(&self.anchor.objects))
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))?;
        self.anchor.validate_current(&self.root)?;
        Ok(metadata)
    }

    /// Internal acceptance-harness seam. Production callers must use
    /// [`Self::initialize`], which retains the fixed store root.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn initialize_at_for_test(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::initialize_at(root, deployment_id, identity)
    }

    fn initialize_at(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::initialize_at_with_faults(root, deployment_id, identity, Arc::new(NoMutationFaults))
    }

    fn initialize_at_with_faults(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
        faults: Arc<dyn MutationFaultInjector>,
    ) -> Result<Self, ManagedSecretError> {
        ensure_supported_platform()?;
        if deployment_id.is_nil() {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::DeploymentMismatch,
            ));
        }
        if fs::symlink_metadata(root).is_ok() {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::AlreadyInitialized,
            ));
        }

        let identity = Arc::new(identity);
        let created_at = Utc::now();
        let header = StoreHeader {
            format_version: STORE_FORMAT_VERSION,
            deployment_id,
            recipient: identity.recipient_string(),
            created_at,
        };
        let audit = AuditRecord::bootstrap(deployment_id, created_at)?;
        let manifest = StoreManifest {
            format_version: STORE_FORMAT_VERSION,
            deployment_id,
            recipient: header.recipient.clone(),
            generation: 0,
            entries: BTreeMap::new(),
            staged: BTreeMap::new(),
            reference_holders: BTreeMap::new(),
            datastore_reference_inventory_version: DATASTORE_REFERENCE_INVENTORY_VERSION,
            audit: vec![audit],
        };

        let parent = root
            .parent()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let file_name = root
            .file_name()
            .and_then(|value| value.to_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let staging = parent.join(format!(".{file_name}.init-{}", Uuid::new_v4()));
        if let Err(error) = create_initial_store(&staging, &header, &manifest, &identity) {
            let _ = fs::remove_dir_all(&staging);
            return Err(error);
        }

        if let Err(error) = rename_directory_create_only(&staging, root) {
            let _ = fs::remove_dir_all(&staging);
            return Err(error);
        }
        if faults
            .check(MutationCheckpoint::DirectorySync(
                MutationPhase::Initialization,
            ))
            .and_then(|()| sync_directory_path(parent))
            .is_err()
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::CommitUncertain,
            ));
        }

        Self::open_at_with_faults(root, deployment_id, identity, faults)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::CommitUncertain))
    }

    /// Internal acceptance-harness seam. Production callers must use
    /// [`Self::open`], which retains the fixed store root.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn open_at_for_test(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::open_at(root, deployment_id, identity)
    }

    /// Internal bootstrap-harness seam for non-mutating verified open.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn open_verified_at_for_test(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::open_verified_at(root, deployment_id, identity)
    }

    /// Internal acceptance-harness seam for selected, read-only resolution.
    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn open_selected_read_only_at_for_test(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<SelectedManagedSecretReader, ManagedSecretError> {
        Self::open_selected_read_only_at(root, deployment_id, identity)
    }

    fn open_verified_at(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        ensure_supported_platform()?;
        validate_directory(root, false)?;
        let anchor = Arc::new(StoreAnchor::open(root, false)?);
        let identity = Arc::new(identity);
        let lock = StoreLock::acquire(&anchor.root, LockMode::Shared)?;
        verify_anchored_store(&anchor, root, deployment_id, &identity)?;
        drop(lock);
        Ok(Self {
            root: root.to_path_buf(),
            anchor,
            deployment_id,
            identity,
            faults: Arc::new(NoMutationFaults),
        })
    }

    fn open_selected_read_only_at(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<SelectedManagedSecretReader, ManagedSecretError> {
        ensure_supported_platform()?;
        validate_directory(root, false)?;
        let anchor = Arc::new(StoreAnchor::open(root, false)?);
        let identity = Arc::new(identity);
        let _lock = StoreLock::acquire(&anchor.root, LockMode::Shared)?;
        anchor.validate_current(root)?;
        let header = read_header(&anchor.root)?;
        validate_header(&header, deployment_id, &identity)?;
        if read_store_file(&anchor.root, LOCK_FILE, MAX_HEADER_BYTES)? != LOCK_CONTENT {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        let manifest = read_manifest(&anchor.root, &identity)?;
        validate_manifest(&manifest, &header)?;
        validate_store_control_plane(&anchor)?;
        anchor.validate_current(root)?;
        Ok(SelectedManagedSecretReader {
            store: Self {
                root: root.to_path_buf(),
                anchor,
                deployment_id,
                identity,
                faults: Arc::new(NoMutationFaults),
            },
        })
    }

    fn open_at(
        root: &Path,
        deployment_id: Uuid,
        identity: ManagedSecretIdentity,
    ) -> Result<Self, ManagedSecretError> {
        Self::open_at_with_faults(
            root,
            deployment_id,
            Arc::new(identity),
            Arc::new(NoMutationFaults),
        )
    }

    fn open_at_with_faults(
        root: &Path,
        deployment_id: Uuid,
        identity: Arc<ManagedSecretIdentity>,
        faults: Arc<dyn MutationFaultInjector>,
    ) -> Result<Self, ManagedSecretError> {
        ensure_supported_platform()?;
        validate_directory(root, true)?;
        let anchor = Arc::new(StoreAnchor::open(root, true)?);
        let lock = StoreLock::acquire(&anchor.root, LockMode::Exclusive)?;
        let header = read_header(&anchor.root)?;
        validate_header(&header, deployment_id, &identity)?;
        let manifest = read_manifest(&anchor.root, &identity)?;
        validate_manifest(&manifest, &header)?;
        recover_uncommitted_files(&anchor, &manifest)?;
        validate_store_tree(&anchor, &manifest)?;
        anchor.validate_current(root)?;
        drop(lock);
        Ok(Self {
            root: root.to_path_buf(),
            anchor,
            deployment_id,
            identity,
            faults,
        })
    }

    fn write_version(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
        expected_version: Option<ManagedSecretVersion>,
        operation: AuditOperation,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        if self.faults.check(MutationCheckpoint::Lock).is_err() {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::Coordination,
            ));
        }
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Exclusive)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        if self.faults.check(MutationCheckpoint::Decrypt).is_err() {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        let mut manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        recover_uncommitted_files(&self.anchor, &manifest)?;

        let current = manifest.entries.get(reference.as_str());
        let version = match (current, expected_version) {
            (None, None)
                if !manifest
                    .audit
                    .iter()
                    .any(|record| record.reference.as_deref() == Some(reference.as_str())) =>
            {
                ManagedSecretVersion::first()
            }
            (Some(_), None) => {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::AlreadyExists,
                ));
            }
            (Some(entry), Some(expected)) if entry.version == expected => expected.next()?,
            _ => {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::VersionConflict,
                ));
            }
        };
        let created_at = Utc::now();
        let metadata = ManagedSecretMetadata {
            reference: reference.clone(),
            version,
            created_at,
        };
        self.faults.check(MutationCheckpoint::Audit)?;
        let plaintext = encode_secret_envelope(
            self.deployment_id,
            &self.identity.recipient_string(),
            &metadata,
            material,
        )?;
        let ciphertext = encrypt(&self.identity.recipient, &plaintext)?;
        let object_name = format!("object-{}.age", Uuid::new_v4());
        atomic_write(
            &self.anchor.objects,
            &object_name,
            &ciphertext,
            self.faults.as_ref(),
            MutationPhase::Object,
        )
        .map_err(AtomicWriteFailure::into_error)?;
        let verification = self
            .faults
            .check(MutationCheckpoint::Verification)
            .and_then(|()| {
                verify_persisted_object(
                    &self.anchor.objects,
                    &object_name,
                    self.deployment_id,
                    &self.identity,
                    &metadata,
                    material,
                )
            });
        if let Err(error) = verification {
            remove_file_if_present(&self.anchor.objects, &object_name)?;
            sync_directory(&self.anchor.objects)?;
            return Err(error);
        }

        let prior_object = current.map(|entry| entry.object.clone());
        manifest.generation = manifest
            .generation
            .checked_add(1)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::VersionConflict))?;
        manifest.entries.insert(
            reference.as_str().to_string(),
            ManifestEntry {
                version,
                object: object_name.clone(),
                created_at,
            },
        );
        let audit = AuditRecord::mutation(
            manifest.audit.last(),
            operation,
            reference,
            version,
            created_at,
            self.deployment_id,
        )?;
        manifest.audit.push(audit);

        let manifest_result = write_manifest(
            &self.anchor.root,
            &manifest,
            &self.identity,
            self.faults.as_ref(),
        );
        if let Err(failure) = manifest_result {
            if failure.renamed {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::CommitUncertain,
                ));
            }
            remove_file_if_present(&self.anchor.objects, &object_name)?;
            sync_directory(&self.anchor.objects)?;
            return Err(failure.into_error());
        }

        if let Some(prior_object) = prior_object
            && self
                .faults
                .check(MutationCheckpoint::CleanupRemove)
                .and_then(|()| remove_file_if_present(&self.anchor.objects, &prior_object))
                .and_then(|()| self.faults.check(MutationCheckpoint::CleanupSync))
                .and_then(|()| sync_directory(&self.anchor.objects))
                .is_err()
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::CommitUncertain,
            ));
        }
        self.anchor.validate_current(&self.root)?;
        Ok(metadata)
    }

    #[cfg(any(test, feature = "test-fixtures"))]
    #[doc(hidden)]
    pub fn replace_for_test(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
        expected_version: ManagedSecretVersion,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.write_version(
            reference,
            material,
            Some(expected_version),
            AuditOperation::Replace,
        )
    }
}

impl ManagedSecretBackend for LocalManagedSecretStore {
    fn create(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.write_version(reference, material, None, AuditOperation::Create)
    }

    fn resolve(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ResolvedManagedSecret, ManagedSecretError> {
        self.anchor.validate_current(&self.root)?;
        let _lock = StoreLock::acquire(&self.anchor.root, LockMode::Shared)?;
        self.anchor.validate_current(&self.root)?;
        let header = read_header(&self.anchor.root)?;
        validate_header(&header, self.deployment_id, &self.identity)?;
        let manifest = read_manifest(&self.anchor.root, &self.identity)?;
        validate_manifest(&manifest, &header)?;
        let entry = manifest
            .entries
            .get(reference.as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Missing))?;
        let ciphertext =
            read_store_file(&self.anchor.objects, &entry.object, MAX_CIPHERTEXT_BYTES)?;
        let plaintext = decrypt(&self.identity.identity, &ciphertext)?;
        let resolved = decode_secret_envelope(
            &plaintext,
            self.deployment_id,
            &self.identity.recipient_string(),
            reference,
            entry,
        )?;
        self.anchor.validate_current(&self.root)?;
        Ok(resolved)
    }

    fn lookup_active_metadata(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        LocalManagedSecretStore::lookup_active_metadata(self, reference)
    }
}

impl owner_backend_sealed::Sealed for LocalManagedSecretStore {}
impl trusted_backend_sealed::Sealed for LocalManagedSecretStore {}

impl ManagedSecretOwnerBackend for LocalManagedSecretStore {
    fn stage_owner_replacement(
        &self,
        reference: &ManagedSecretReference,
        material: &SecretMaterial,
        expected_active_version: ManagedSecretVersion,
    ) -> Result<StagedManagedSecret, ManagedSecretError> {
        self.stage_replacement(reference, material, expected_active_version)
    }

    fn resume_owner_replacement(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Option<StagedManagedSecret>, ManagedSecretError> {
        self.pending_replacement(reference)
    }

    fn activate_owner_replacement(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.activate_replacement(staged)
    }

    fn activate_owner_replacements(
        &self,
        staged: &[&StagedManagedSecret],
    ) -> Result<Vec<ManagedSecretMetadata>, ManagedSecretError> {
        self.activate_replacements(staged)
    }

    fn owner_replacement_is_active(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<bool, ManagedSecretError> {
        self.replacement_is_active(staged)
    }

    fn discard_owner_replacement(
        &self,
        staged: &StagedManagedSecret,
    ) -> Result<(), ManagedSecretError> {
        self.discard_replacement(staged)
    }
}

impl ManagedSecretTrustedBackend for LocalManagedSecretStore {
    fn replace_trusted_reference_holder(
        &self,
        holder: &str,
        references: &[ManagedSecretReference],
    ) -> Result<(), ManagedSecretError> {
        self.replace_reference_holder(holder, references)
    }

    fn remove_trusted_reference_holder(&self, holder: &str) -> Result<(), ManagedSecretError> {
        self.remove_reference_holder(holder)
    }

    fn trusted_held_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretError> {
        self.held_reference_status(reference)
    }

    fn trusted_reference_holders(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Vec<String>, ManagedSecretError> {
        self.reference_holders(reference)
    }

    fn datastore_reference_inventory_is_complete(&self) -> Result<bool, ManagedSecretError> {
        LocalManagedSecretStore::datastore_reference_inventory_is_complete(self)
    }

    fn remove_with_repository_evidence(
        &self,
        reference: &ManagedSecretReference,
        proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, ManagedSecretError> {
        self.remove_proven_unreferenced(reference, proof)
    }
}

/// Stable, disclosure-safe Managed-secret failure categories.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ManagedSecretErrorKind {
    AlreadyInitialized,
    NotInitialized,
    AlreadyExists,
    Missing,
    InvalidIdentity,
    DeploymentMismatch,
    RecipientMismatch,
    UnsafeFilesystem,
    Coordination,
    Integrity,
    VersionConflict,
    SizeLimit,
    CommitUncertain,
    Mutation,
    UnsupportedPlatform,
}

/// A disclosure-safe Managed-secret backend error.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
#[error("Managed-secret operation failed: {kind:?}")]
pub struct ManagedSecretError {
    kind: ManagedSecretErrorKind,
}

/// Disclosure-safe phase of an offline root-identity rotation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootRotationPhase {
    Lock,
    Prerequisites,
    SourceVerification,
    StagingWrite,
    DestinationVerification,
}

/// A disclosure-safe root-rotation failure with aggregate phase context.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
#[error("Managed-secret root rotation failed during {phase:?}: {source}")]
pub struct RootRotationError {
    phase: RootRotationPhase,
    source: ManagedSecretError,
}

impl RootRotationError {
    fn at(phase: RootRotationPhase, source: ManagedSecretError) -> Self {
        Self { phase, source }
    }

    pub fn phase(&self) -> RootRotationPhase {
        self.phase
    }

    pub fn kind(&self) -> ManagedSecretErrorKind {
        self.source.kind()
    }
}

impl ManagedSecretError {
    fn new(kind: ManagedSecretErrorKind) -> Self {
        Self { kind }
    }

    pub fn kind(&self) -> ManagedSecretErrorKind {
        self.kind
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct StoreHeader {
    format_version: u64,
    deployment_id: Uuid,
    recipient: String,
    created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct StoreManifest {
    format_version: u64,
    deployment_id: Uuid,
    recipient: String,
    generation: u64,
    entries: BTreeMap<String, ManifestEntry>,
    #[serde(default)]
    staged: BTreeMap<String, StagedManifestEntry>,
    #[serde(default)]
    reference_holders: BTreeMap<String, BTreeSet<String>>,
    #[serde(default)]
    datastore_reference_inventory_version: u64,
    audit: Vec<AuditRecord>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ManifestEntry {
    version: ManagedSecretVersion,
    object: String,
    created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct StagedManifestEntry {
    stage_id: Uuid,
    expected_active_version: ManagedSecretVersion,
    version: ManagedSecretVersion,
    object: String,
    created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum AuditOperation {
    Bootstrap,
    Create,
    Replace,
    Remove,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct AuditActor {
    effective_uid: u32,
    effective_gid: u32,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct AuditRecord {
    sequence: u64,
    operation: AuditOperation,
    reference: Option<String>,
    version: Option<ManagedSecretVersion>,
    occurred_at: DateTime<Utc>,
    deployment_id: Uuid,
    actor: AuditActor,
    previous_hash: [u8; 32],
    record_hash: [u8; 32],
}

#[derive(Serialize)]
struct AuditHashInput<'a> {
    sequence: u64,
    operation: AuditOperation,
    reference: &'a Option<String>,
    version: Option<ManagedSecretVersion>,
    occurred_at: DateTime<Utc>,
    deployment_id: Uuid,
    actor: &'a AuditActor,
    previous_hash: [u8; 32],
}

impl AuditRecord {
    fn bootstrap(
        deployment_id: Uuid,
        occurred_at: DateTime<Utc>,
    ) -> Result<Self, ManagedSecretError> {
        Self::new(
            1,
            AuditOperation::Bootstrap,
            None,
            None,
            occurred_at,
            deployment_id,
            ZERO_HASH,
        )
    }

    fn mutation(
        previous: Option<&Self>,
        operation: AuditOperation,
        reference: &ManagedSecretReference,
        version: ManagedSecretVersion,
        occurred_at: DateTime<Utc>,
        deployment_id: Uuid,
    ) -> Result<Self, ManagedSecretError> {
        let previous =
            previous.ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Self::new(
            previous
                .sequence
                .checked_add(1)
                .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?,
            operation,
            Some(reference.as_str().to_string()),
            Some(version),
            occurred_at,
            deployment_id,
            previous.record_hash,
        )
    }

    fn new(
        sequence: u64,
        operation: AuditOperation,
        reference: Option<String>,
        version: Option<ManagedSecretVersion>,
        occurred_at: DateTime<Utc>,
        deployment_id: Uuid,
        previous_hash: [u8; 32],
    ) -> Result<Self, ManagedSecretError> {
        let actor = current_actor();
        let mut record = Self {
            sequence,
            operation,
            reference,
            version,
            occurred_at,
            deployment_id,
            actor,
            previous_hash,
            record_hash: ZERO_HASH,
        };
        record.record_hash = record.calculate_hash()?;
        Ok(record)
    }

    fn calculate_hash(&self) -> Result<[u8; 32], ManagedSecretError> {
        let input = AuditHashInput {
            sequence: self.sequence,
            operation: self.operation,
            reference: &self.reference,
            version: self.version,
            occurred_at: self.occurred_at,
            deployment_id: self.deployment_id,
            actor: &self.actor,
            previous_hash: self.previous_hash,
        };
        let encoded = serde_json::to_vec(&input)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Ok(Sha256::digest(encoded).into())
    }
}

fn current_actor() -> AuditActor {
    #[cfg(unix)]
    {
        // SAFETY: These libc calls read process credentials and have no preconditions.
        let effective_uid = unsafe { libc::geteuid() };
        // SAFETY: These libc calls read process credentials and have no preconditions.
        let effective_gid = unsafe { libc::getegid() };
        AuditActor {
            effective_uid,
            effective_gid,
        }
    }
    #[cfg(not(unix))]
    {
        AuditActor {
            effective_uid: 0,
            effective_gid: 0,
        }
    }
}

fn create_initial_store(
    staging: &Path,
    header: &StoreHeader,
    manifest: &StoreManifest,
    identity: &ManagedSecretIdentity,
) -> Result<(), ManagedSecretError> {
    create_private_directory(staging)?;
    let objects = staging.join(OBJECTS_DIRECTORY);
    create_private_directory(&objects)?;
    write_new_file(&staging.join(LOCK_FILE), LOCK_CONTENT)?;
    let header_bytes = serde_json::to_vec(header)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    write_new_file(&staging.join(HEADER_FILE), &header_bytes)?;
    let manifest_bytes = serialize_and_encrypt_manifest(manifest, identity)?;
    write_new_file(&staging.join(MANIFEST_FILE), &manifest_bytes)?;
    sync_directory_path(&objects)?;
    sync_directory_path(staging)
}

fn write_new_file(path: &Path, bytes: &[u8]) -> Result<(), ManagedSecretError> {
    let mut options = OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    options
        .mode(0o600)
        .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW);
    let mut file = options
        .open(path)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    file.write_all(bytes)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    file.sync_all()
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
}

fn read_header(root: &File) -> Result<StoreHeader, ManagedSecretError> {
    let bytes = read_store_file(root, HEADER_FILE, MAX_HEADER_BYTES)?;
    serde_json::from_slice(&bytes)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))
}

fn validate_header(
    header: &StoreHeader,
    deployment_id: Uuid,
    identity: &ManagedSecretIdentity,
) -> Result<(), ManagedSecretError> {
    if header.format_version != STORE_FORMAT_VERSION {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    if header.deployment_id != deployment_id {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::DeploymentMismatch,
        ));
    }
    if header.recipient != identity.recipient_string() {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::RecipientMismatch,
        ));
    }
    age::x25519::Recipient::from_str(&header.recipient)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    Ok(())
}

fn read_manifest(
    root: &File,
    identity: &ManagedSecretIdentity,
) -> Result<StoreManifest, ManagedSecretError> {
    let ciphertext = read_store_file(root, MANIFEST_FILE, MAX_MANIFEST_BYTES)?;
    let plaintext = decrypt(&identity.identity, &ciphertext)?;
    serde_json::from_slice(&plaintext)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))
}

fn write_manifest(
    root: &File,
    manifest: &StoreManifest,
    identity: &ManagedSecretIdentity,
    faults: &dyn MutationFaultInjector,
) -> Result<(), AtomicWriteFailure> {
    let ciphertext =
        serialize_and_encrypt_manifest(manifest, identity).map_err(|error| AtomicWriteFailure {
            error,
            renamed: false,
        })?;
    if ciphertext.len() as u64 > MAX_MANIFEST_BYTES {
        return Err(AtomicWriteFailure {
            error: ManagedSecretError::new(ManagedSecretErrorKind::SizeLimit),
            renamed: false,
        });
    }
    atomic_write(
        root,
        MANIFEST_FILE,
        &ciphertext,
        faults,
        MutationPhase::Manifest,
    )
}

fn serialize_and_encrypt_manifest(
    manifest: &StoreManifest,
    identity: &ManagedSecretIdentity,
) -> Result<Vec<u8>, ManagedSecretError> {
    let plaintext = serde_json::to_vec(manifest)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    encrypt(&identity.recipient, &plaintext)
}

fn valid_reference_holder(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 128
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b':'))
}

fn validate_manifest(
    manifest: &StoreManifest,
    header: &StoreHeader,
) -> Result<(), ManagedSecretError> {
    if manifest.format_version != STORE_FORMAT_VERSION
        || manifest.deployment_id != header.deployment_id
        || manifest.recipient != header.recipient
        || manifest.datastore_reference_inventory_version > DATASTORE_REFERENCE_INVENTORY_VERSION
        || manifest.audit.is_empty()
    {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }

    for (holder, references) in &manifest.reference_holders {
        if !valid_reference_holder(holder) || references.is_empty() {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        for reference in references {
            let parsed = ManagedSecretReference::from_str(reference)
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
            if parsed.as_str() != reference || !manifest.entries.contains_key(reference) {
                return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
            }
        }
    }

    let mut previous_hash = ZERO_HASH;
    let mut previous_time = None;
    for (index, record) in manifest.audit.iter().enumerate() {
        let expected_sequence = u64::try_from(index)
            .ok()
            .and_then(|value| value.checked_add(1))
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        if record.sequence != expected_sequence
            || record.deployment_id != header.deployment_id
            || record.previous_hash != previous_hash
            || record.calculate_hash()? != record.record_hash
            || previous_time.is_some_and(|previous| record.occurred_at < previous)
        {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        if index == 0
            && (!matches!(record.operation, AuditOperation::Bootstrap)
                || record.reference.is_some()
                || record.version.is_some()
                || record.occurred_at != header.created_at)
        {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        previous_hash = record.record_hash;
        previous_time = Some(record.occurred_at);
    }

    let mutation_count = manifest
        .audit
        .len()
        .checked_sub(1)
        .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    if manifest.generation != mutation_count as u64 {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }

    let mut audited_versions =
        BTreeMap::<String, (ManagedSecretVersion, DateTime<Utc>, bool)>::new();
    for record in manifest.audit.iter().skip(1) {
        let reference = record
            .reference
            .as_deref()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let parsed = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let version = record
            .version
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        if parsed.as_str() != reference {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        let next_state = match (record.operation, audited_versions.get(reference).copied()) {
            (AuditOperation::Create, None) if version == ManagedSecretVersion::first() => {
                (version, record.occurred_at, true)
            }
            (AuditOperation::Replace, Some((previous, _, true))) if previous.next()? == version => {
                (version, record.occurred_at, true)
            }
            (AuditOperation::Remove, Some((current, activated_at, true))) if current == version => {
                (current, activated_at, false)
            }
            _ => {
                return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
            }
        };
        audited_versions.insert(reference.to_string(), next_state);
    }

    for (reference, entry) in &manifest.entries {
        let parsed = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        if parsed.as_str() != reference
            || audited_versions.get(reference).copied()
                != Some((entry.version, entry.created_at, true))
            || !valid_object_name(&entry.object)
        {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
    }
    let mut object_names = manifest
        .entries
        .values()
        .map(|entry| entry.object.as_str())
        .collect::<BTreeSet<_>>();
    for (reference, staged) in &manifest.staged {
        let parsed = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let active = manifest
            .entries
            .get(reference)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        if parsed.as_str() != reference
            || active.version != staged.expected_active_version
            || staged.version != staged.expected_active_version.next()?
            || !valid_object_name(&staged.object)
            || !object_names.insert(&staged.object)
        {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
    }
    if audited_versions
        .values()
        .filter(|(_, _, active)| *active)
        .count()
        != manifest.entries.len()
    {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    Ok(())
}

// `material` is only read by the Linux rotation path; verification still
// decrypts it everywhere so the store is proven readable before reporting.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
struct VerifiedStore {
    header: StoreHeader,
    manifest: StoreManifest,
    material: BTreeMap<String, SecretMaterial>,
}

impl VerifiedStore {
    fn report(&self) -> ManagedSecretVerification {
        ManagedSecretVerification {
            deployment_id: self.header.deployment_id,
            recipient_fingerprint: recipient_fingerprint(&self.header.recipient),
            active_secret_count: self.manifest.entries.len(),
            audit_record_count: self.manifest.audit.len(),
        }
    }
}

fn verify_anchored_store(
    anchor: &StoreAnchor,
    root: &Path,
    deployment_id: Uuid,
    identity: &ManagedSecretIdentity,
) -> Result<VerifiedStore, ManagedSecretError> {
    anchor.validate_current(root)?;
    let header = read_header(&anchor.root)?;
    validate_header(&header, deployment_id, identity)?;
    if read_store_file(&anchor.root, LOCK_FILE, MAX_HEADER_BYTES)? != LOCK_CONTENT {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    let manifest = read_manifest(&anchor.root, identity)?;
    validate_manifest(&manifest, &header)?;
    validate_store_tree(anchor, &manifest)?;

    let mut material = BTreeMap::new();
    for (reference, entry) in &manifest.entries {
        let reference = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let ciphertext = read_store_file(&anchor.objects, &entry.object, MAX_CIPHERTEXT_BYTES)?;
        let plaintext = decrypt(&identity.identity, &ciphertext)?;
        let resolved = decode_secret_envelope(
            &plaintext,
            deployment_id,
            &identity.recipient_string(),
            &reference,
            entry,
        )?;
        material.insert(reference.as_str().to_string(), resolved.material);
    }
    for (reference, staged) in &manifest.staged {
        let reference = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let entry = ManifestEntry {
            version: staged.version,
            object: staged.object.clone(),
            created_at: staged.created_at,
        };
        let ciphertext = read_store_file(&anchor.objects, &entry.object, MAX_CIPHERTEXT_BYTES)?;
        let plaintext = decrypt(&identity.identity, &ciphertext)?;
        decode_secret_envelope(
            &plaintext,
            deployment_id,
            &identity.recipient_string(),
            &reference,
            &entry,
        )?;
    }
    anchor.validate_current(root)?;
    Ok(VerifiedStore {
        header,
        manifest,
        material,
    })
}

fn recipient_fingerprint(recipient: &str) -> String {
    let digest = Sha256::digest(recipient.as_bytes());
    let mut encoded = String::with_capacity(digest.len() * 2);
    const HEX: &[u8; 16] = b"0123456789abcdef";
    for byte in digest {
        encoded.push(char::from(HEX[usize::from(byte >> 4)]));
        encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    encoded
}

// Only the Linux rotation path populates staging, so the descriptor and
// identity are unused elsewhere.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
struct EmptyRotationStaging {
    path: PathBuf,
    root: File,
    identity: DirectoryIdentity,
}

#[cfg(target_os = "linux")]
impl EmptyRotationStaging {
    fn validate_current(&self) -> Result<(), ManagedSecretError> {
        let metadata = fs::symlink_metadata(&self.path)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&metadata, true)?;
        if directory_identity(&metadata) != self.identity {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
        Ok(())
    }

    fn populated_anchor(&self, objects: File) -> Result<StoreAnchor, ManagedSecretError> {
        self.validate_current()?;
        let root = self
            .root
            .try_clone()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let objects_metadata = objects
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        validate_directory_metadata(&objects_metadata, true)?;
        Ok(StoreAnchor {
            root,
            objects,
            root_identity: self.identity,
            objects_identity: directory_identity(&objects_metadata),
            require_write: true,
        })
    }
}

fn validate_empty_staging_directory(
    staging_root: &Path,
    active: &StoreAnchor,
) -> Result<EmptyRotationStaging, ManagedSecretError> {
    validate_directory(staging_root, true)?;
    let staging = open_directory(staging_root)?;
    let metadata = staging
        .metadata()
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    validate_directory_metadata(&metadata, true)?;
    let identity = directory_identity(&metadata);
    if identity == active.root_identity
        || directory_mount_is_read_only(&staging)?
        || read_directory(&staging)?.next().is_some()
    {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsafeFilesystem,
        ));
    }
    Ok(EmptyRotationStaging {
        path: staging_root.to_path_buf(),
        root: staging,
        identity,
    })
}

#[cfg(target_os = "linux")]
fn populate_rotation_staging(
    staging: &EmptyRotationStaging,
    source: &VerifiedStore,
    next_identity: &ManagedSecretIdentity,
    faults: &dyn RotationFaultInjector,
) -> Result<StoreAnchor, ManagedSecretError> {
    staging.validate_current()?;
    faults.check(RotationCheckpoint::CreateObjectsDirectory)?;
    rustix::fs::mkdirat(
        &staging.root,
        OBJECTS_DIRECTORY,
        rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR | rustix::fs::Mode::XUSR,
    )
    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    let objects = open_directory_at(&staging.root, Path::new(OBJECTS_DIRECTORY))?;
    for (reference, entry) in &source.manifest.entries {
        let reference = ManagedSecretReference::from_str(reference)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let metadata = ManagedSecretMetadata {
            reference,
            version: entry.version,
            created_at: entry.created_at,
        };
        let material = source
            .material
            .get(metadata.reference().as_str())
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        let plaintext = encode_secret_envelope(
            source.header.deployment_id,
            &next_identity.recipient_string(),
            &metadata,
            material,
        )?;
        let ciphertext = encrypt(&next_identity.recipient, &plaintext)?;
        write_rotation_file(
            &objects,
            &entry.object,
            &ciphertext,
            RotationFile::Object,
            faults,
        )?;
    }

    let mut header = source.header.clone();
    header.recipient = next_identity.recipient_string();
    let mut manifest = source.manifest.clone();
    manifest.recipient.clone_from(&header.recipient);
    write_rotation_file(
        &staging.root,
        LOCK_FILE,
        LOCK_CONTENT,
        RotationFile::Lock,
        faults,
    )?;
    let header_bytes = serde_json::to_vec(&header)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    write_rotation_file(
        &staging.root,
        HEADER_FILE,
        &header_bytes,
        RotationFile::Header,
        faults,
    )?;
    let manifest_bytes = serialize_and_encrypt_manifest(&manifest, next_identity)?;
    write_rotation_file(
        &staging.root,
        MANIFEST_FILE,
        &manifest_bytes,
        RotationFile::Manifest,
        faults,
    )?;
    faults.check(RotationCheckpoint::SyncDirectory(
        RotationDirectory::Objects,
    ))?;
    sync_directory(&objects)?;
    faults.check(RotationCheckpoint::SyncDirectory(RotationDirectory::Root))?;
    sync_directory(&staging.root)?;
    staging.populated_anchor(objects)
}

#[cfg(not(target_os = "linux"))]
fn populate_rotation_staging(
    _staging: &EmptyRotationStaging,
    _source: &VerifiedStore,
    _next_identity: &ManagedSecretIdentity,
    _faults: &dyn RotationFaultInjector,
) -> Result<StoreAnchor, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

#[cfg(target_os = "linux")]
fn write_rotation_file(
    directory: &File,
    name: &str,
    bytes: &[u8],
    file_kind: RotationFile,
    faults: &dyn RotationFaultInjector,
) -> Result<(), ManagedSecretError> {
    faults.check(RotationCheckpoint::Write(file_kind))?;
    let descriptor = rustix::fs::openat2(
        directory,
        name,
        rustix::fs::OFlags::WRONLY
            | rustix::fs::OFlags::CREATE
            | rustix::fs::OFlags::EXCL
            | rustix::fs::OFlags::CLOEXEC
            | rustix::fs::OFlags::NOFOLLOW,
        rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR,
        rustix::fs::ResolveFlags::BENEATH
            | rustix::fs::ResolveFlags::NO_MAGICLINKS
            | rustix::fs::ResolveFlags::NO_SYMLINKS,
    )
    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    let mut file = File::from(descriptor);
    file.write_all(bytes)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    faults.check(RotationCheckpoint::SyncFile(file_kind))?;
    file.sync_all()
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
}

#[cfg(target_os = "linux")]
fn directory_mount_is_read_only(directory: &File) -> Result<bool, ManagedSecretError> {
    let status = rustix::fs::fstatvfs(directory)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    Ok(status
        .f_flag
        .contains(rustix::fs::StatVfsMountFlags::RDONLY))
}

#[cfg(not(target_os = "linux"))]
fn directory_mount_is_read_only(_directory: &File) -> Result<bool, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

fn encrypt(
    recipient: &age::x25519::Recipient,
    plaintext: &[u8],
) -> Result<Vec<u8>, ManagedSecretError> {
    age::encrypt(recipient, plaintext)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
}

fn decrypt(
    identity: &age::x25519::Identity,
    ciphertext: &[u8],
) -> Result<Vec<u8>, ManagedSecretError> {
    age::decrypt(identity, ciphertext)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))
}

fn encode_secret_envelope(
    deployment_id: Uuid,
    recipient: &str,
    metadata: &ManagedSecretMetadata,
    material: &SecretMaterial,
) -> Result<Vec<u8>, ManagedSecretError> {
    let reference = metadata.reference.as_str().as_bytes();
    let recipient = recipient.as_bytes();
    let reference_len = u16::try_from(reference.len())
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    let recipient_len = u16::try_from(recipient.len())
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    material.expose(|secret| {
        let material_len = u32::try_from(secret.len())
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        let mut encoded = Vec::with_capacity(
            ENVELOPE_MAGIC.len() + 55 + recipient.len() + reference.len() + secret.len(),
        );
        encoded.extend_from_slice(ENVELOPE_MAGIC);
        encoded.push(ENVELOPE_FORMAT_VERSION);
        encoded.extend_from_slice(deployment_id.as_bytes());
        encoded.extend_from_slice(&metadata.version.get().to_be_bytes());
        encoded.extend_from_slice(&metadata.created_at.timestamp().to_be_bytes());
        encoded.extend_from_slice(&metadata.created_at.timestamp_subsec_nanos().to_be_bytes());
        encoded.extend_from_slice(&recipient_len.to_be_bytes());
        encoded.extend_from_slice(recipient);
        encoded.extend_from_slice(&reference_len.to_be_bytes());
        encoded.extend_from_slice(reference);
        encoded.extend_from_slice(&material_len.to_be_bytes());
        encoded.extend_from_slice(secret);
        Ok(encoded)
    })
}

fn decode_secret_envelope(
    plaintext: &[u8],
    deployment_id: Uuid,
    recipient: &str,
    reference: &ManagedSecretReference,
    entry: &ManifestEntry,
) -> Result<ResolvedManagedSecret, ManagedSecretError> {
    let mut input = EnvelopeReader::new(plaintext);
    if input.take(ENVELOPE_MAGIC.len())? != ENVELOPE_MAGIC
        || input.u8()? != ENVELOPE_FORMAT_VERSION
        || input.take(16)? != deployment_id.as_bytes()
    {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    let version = ManagedSecretVersion(
        NonZeroU64::new(input.u64()?)
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?,
    );
    let seconds = input.i64()?;
    let nanos = input.u32()?;
    let created_at = DateTime::<Utc>::from_timestamp(seconds, nanos)
        .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    let envelope_recipient = input.length_prefixed_u16_string()?;
    let envelope_reference = input.length_prefixed_u16_string()?;
    let material_len = input.u32()? as usize;
    if material_len > crate::MAX_SECRET_MATERIAL_BYTES {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    let material_bytes = input.take(material_len)?.to_vec();
    if !input.is_empty()
        || envelope_recipient != recipient
        || envelope_reference != reference.as_str()
        || version != entry.version
        || created_at != entry.created_at
    {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    let material = SecretMaterial::try_from(material_bytes)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    Ok(ResolvedManagedSecret {
        metadata: ManagedSecretMetadata {
            reference: reference.clone(),
            version,
            created_at,
        },
        material,
    })
}

struct EnvelopeReader<'a> {
    remaining: &'a [u8],
}

impl<'a> EnvelopeReader<'a> {
    fn new(input: &'a [u8]) -> Self {
        Self { remaining: input }
    }

    fn take(&mut self, length: usize) -> Result<&'a [u8], ManagedSecretError> {
        if self.remaining.len() < length {
            return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
        }
        let (value, remaining) = self.remaining.split_at(length);
        self.remaining = remaining;
        Ok(value)
    }

    fn u8(&mut self) -> Result<u8, ManagedSecretError> {
        Ok(self.take(1)?[0])
    }

    fn u16(&mut self) -> Result<u16, ManagedSecretError> {
        let bytes: [u8; 2] = self
            .take(2)?
            .try_into()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Ok(u16::from_be_bytes(bytes))
    }

    fn u32(&mut self) -> Result<u32, ManagedSecretError> {
        let bytes: [u8; 4] = self
            .take(4)?
            .try_into()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Ok(u32::from_be_bytes(bytes))
    }

    fn u64(&mut self) -> Result<u64, ManagedSecretError> {
        let bytes: [u8; 8] = self
            .take(8)?
            .try_into()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Ok(u64::from_be_bytes(bytes))
    }

    fn i64(&mut self) -> Result<i64, ManagedSecretError> {
        let bytes: [u8; 8] = self
            .take(8)?
            .try_into()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
        Ok(i64::from_be_bytes(bytes))
    }

    fn length_prefixed_u16_string(&mut self) -> Result<&'a str, ManagedSecretError> {
        let length = self.u16()? as usize;
        std::str::from_utf8(self.take(length)?)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))
    }

    fn is_empty(&self) -> bool {
        self.remaining.is_empty()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MutationPhase {
    Initialization,
    Object,
    Manifest,
}

// Fault-injection checkpoints are only reached by the Linux mutation path.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MutationCheckpoint {
    Lock,
    Decrypt,
    Audit,
    Verification,
    Write(MutationPhase),
    Sync(MutationPhase),
    Rename(MutationPhase),
    DirectorySync(MutationPhase),
    CleanupRemove,
    CleanupSync,
}

trait MutationFaultInjector: Send + Sync {
    fn check(&self, checkpoint: MutationCheckpoint) -> Result<(), ManagedSecretError>;
}

struct NoMutationFaults;

impl MutationFaultInjector for NoMutationFaults {
    fn check(&self, _checkpoint: MutationCheckpoint) -> Result<(), ManagedSecretError> {
        Ok(())
    }
}

// Fault-injection checkpoints are only reached by the Linux rotation path.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RotationCheckpoint {
    Lock,
    SourceDecrypt,
    SourceAudit,
    CreateObjectsDirectory,
    Write(RotationFile),
    SyncFile(RotationFile),
    SyncDirectory(RotationDirectory),
    DestinationVerification,
}

#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RotationFile {
    Object,
    Lock,
    Header,
    Manifest,
}

#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RotationDirectory {
    Objects,
    Root,
}

trait RotationFaultInjector {
    fn check(&self, checkpoint: RotationCheckpoint) -> Result<(), ManagedSecretError>;
}

struct NoRotationFaults;

impl RotationFaultInjector for NoRotationFaults {
    fn check(&self, _checkpoint: RotationCheckpoint) -> Result<(), ManagedSecretError> {
        Ok(())
    }
}

#[cfg(target_os = "linux")]
fn atomic_write(
    directory: &File,
    destination: &str,
    bytes: &[u8],
    faults: &dyn MutationFaultInjector,
    phase: MutationPhase,
) -> Result<(), AtomicWriteFailure> {
    let temp = format!(".mutation-{}.tmp", Uuid::new_v4());
    let mut renamed = false;
    let result = (|| {
        faults.check(MutationCheckpoint::Write(phase))?;
        let descriptor = rustix::fs::openat2(
            directory,
            temp.as_str(),
            rustix::fs::OFlags::WRONLY
                | rustix::fs::OFlags::CREATE
                | rustix::fs::OFlags::EXCL
                | rustix::fs::OFlags::CLOEXEC
                | rustix::fs::OFlags::NOFOLLOW,
            rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR,
            rustix::fs::ResolveFlags::BENEATH
                | rustix::fs::ResolveFlags::NO_MAGICLINKS
                | rustix::fs::ResolveFlags::NO_SYMLINKS,
        )
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        let mut file = File::from(descriptor);
        file.write_all(bytes)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        faults.check(MutationCheckpoint::Sync(phase))?;
        file.sync_all()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        drop(file);
        faults.check(MutationCheckpoint::Rename(phase))?;
        rustix::fs::renameat(directory, temp.as_str(), directory, destination)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        renamed = true;
        faults.check(MutationCheckpoint::DirectorySync(phase))?;
        sync_directory(directory)
    })();
    if result.is_err() {
        let _ = rustix::fs::unlinkat(directory, temp.as_str(), rustix::fs::AtFlags::empty());
    }
    result.map_err(|error| AtomicWriteFailure { error, renamed })
}

#[cfg(not(target_os = "linux"))]
fn atomic_write(
    _directory: &File,
    _destination: &str,
    _bytes: &[u8],
    _faults: &dyn MutationFaultInjector,
    _phase: MutationPhase,
) -> Result<(), AtomicWriteFailure> {
    Err(AtomicWriteFailure {
        error: ManagedSecretError::new(ManagedSecretErrorKind::UnsupportedPlatform),
        renamed: false,
    })
}

#[derive(Debug)]
struct AtomicWriteFailure {
    error: ManagedSecretError,
    renamed: bool,
}

impl AtomicWriteFailure {
    fn into_error(self) -> ManagedSecretError {
        self.error
    }
}

#[cfg(target_os = "linux")]
fn recover_uncommitted_files(
    anchor: &StoreAnchor,
    manifest: &StoreManifest,
) -> Result<(), ManagedSecretError> {
    let referenced: BTreeSet<&str> = manifest
        .entries
        .values()
        .map(|entry| entry.object.as_str())
        .chain(manifest.staged.values().map(|entry| entry.object.as_str()))
        .collect();
    for entry in read_directory(&anchor.objects)? {
        let entry =
            entry.map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let name = entry.file_name();
        let name = name
            .to_str()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        if !referenced.contains(name) {
            let metadata = entry
                .metadata()
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
            validate_regular_metadata(
                &metadata,
                OwnershipPolicy::EffectiveUser,
                FileModePolicy::Store,
            )?;
            if valid_object_name(name) || valid_mutation_temp_name(name) {
                rustix::fs::unlinkat(&anchor.objects, name, rustix::fs::AtFlags::empty())
                    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
            } else {
                return Err(ManagedSecretError::new(
                    ManagedSecretErrorKind::UnsafeFilesystem,
                ));
            }
        }
    }
    for entry in read_directory(&anchor.root)? {
        let entry =
            entry.map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let name = entry.file_name();
        let name = name
            .to_str()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        if valid_mutation_temp_name(name) {
            let metadata = entry
                .metadata()
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
            validate_regular_metadata(
                &metadata,
                OwnershipPolicy::EffectiveUser,
                FileModePolicy::Store,
            )?;
            rustix::fs::unlinkat(&anchor.root, name, rustix::fs::AtFlags::empty())
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
        }
    }
    sync_directory(&anchor.objects)?;
    sync_directory(&anchor.root)
}

#[cfg(not(target_os = "linux"))]
fn recover_uncommitted_files(
    _anchor: &StoreAnchor,
    _manifest: &StoreManifest,
) -> Result<(), ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

fn validate_store_tree(
    anchor: &StoreAnchor,
    manifest: &StoreManifest,
) -> Result<(), ManagedSecretError> {
    validate_store_control_plane(anchor)?;

    let allowed_objects: BTreeSet<&str> = manifest
        .entries
        .values()
        .map(|entry| entry.object.as_str())
        .chain(manifest.staged.values().map(|entry| entry.object.as_str()))
        .collect();
    for entry in read_directory(&anchor.objects)? {
        let entry =
            entry.map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let name = entry.file_name();
        let name = name
            .to_str()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        if !allowed_objects.contains(name) {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
        let _ = open_regular_file(
            &anchor.objects,
            name,
            OwnershipPolicy::EffectiveUser,
            FileModePolicy::Store,
            MAX_CIPHERTEXT_BYTES,
        )?;
    }
    Ok(())
}

fn validate_store_control_plane(anchor: &StoreAnchor) -> Result<(), ManagedSecretError> {
    validate_directory_metadata(
        &anchor
            .root
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?,
        anchor.require_write,
    )?;
    validate_directory_metadata(
        &anchor
            .objects
            .metadata()
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?,
        anchor.require_write,
    )?;
    for name in [HEADER_FILE, LOCK_FILE, MANIFEST_FILE] {
        let _ = open_regular_file(
            &anchor.root,
            name,
            OwnershipPolicy::EffectiveUser,
            FileModePolicy::Store,
            if name == MANIFEST_FILE {
                MAX_MANIFEST_BYTES
            } else {
                MAX_HEADER_BYTES
            },
        )?;
    }
    let allowed_root: BTreeSet<&str> = [HEADER_FILE, LOCK_FILE, MANIFEST_FILE, OBJECTS_DIRECTORY]
        .into_iter()
        .collect();
    for entry in read_directory(&anchor.root)? {
        let entry =
            entry.map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        let name = entry.file_name();
        let name = name
            .to_str()
            .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
        if !allowed_root.contains(name) {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
    }
    Ok(())
}

fn valid_object_name(name: &str) -> bool {
    name.strip_prefix("object-")
        .and_then(|value| value.strip_suffix(".age"))
        .and_then(|value| Uuid::parse_str(value).ok())
        .is_some()
}

#[cfg(target_os = "linux")]
fn valid_mutation_temp_name(name: &str) -> bool {
    name.strip_prefix(".mutation-")
        .and_then(|value| value.strip_suffix(".tmp"))
        .and_then(|value| Uuid::parse_str(value).ok())
        .is_some()
}

fn read_store_file(
    directory: &File,
    name: &str,
    maximum: u64,
) -> Result<Vec<u8>, ManagedSecretError> {
    let file = open_regular_file(
        directory,
        name,
        OwnershipPolicy::EffectiveUser,
        FileModePolicy::Store,
        maximum,
    )?;
    read_bounded(file, maximum)
}

fn verify_persisted_object(
    objects: &File,
    object_name: &str,
    deployment_id: Uuid,
    identity: &ManagedSecretIdentity,
    metadata: &ManagedSecretMetadata,
    expected_material: &SecretMaterial,
) -> Result<(), ManagedSecretError> {
    let ciphertext = read_store_file(objects, object_name, MAX_CIPHERTEXT_BYTES)?;
    let plaintext = decrypt(&identity.identity, &ciphertext)?;
    let entry = ManifestEntry {
        version: metadata.version,
        object: object_name.to_string(),
        created_at: metadata.created_at,
    };
    let resolved = decode_secret_envelope(
        &plaintext,
        deployment_id,
        &identity.recipient_string(),
        &metadata.reference,
        &entry,
    )?;
    let material_matches = resolved
        .material
        .expose(|actual| expected_material.expose(|expected| actual == expected));
    if resolved.metadata != *metadata || !material_matches {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    Ok(())
}

fn read_bounded(file: File, maximum: u64) -> Result<Vec<u8>, ManagedSecretError> {
    let limit = maximum
        .checked_add(1)
        .ok_or_else(|| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    let mut bytes = Vec::new();
    file.take(limit)
        .read_to_end(&mut bytes)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Integrity))?;
    if bytes.len() as u64 > maximum {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    Ok(bytes)
}

#[derive(Clone, Copy)]
enum OwnershipPolicy {
    EffectiveUser,
}

#[derive(Clone, Copy)]
enum FileModePolicy {
    Store,
}

#[cfg(target_os = "linux")]
fn open_regular_file(
    directory: &File,
    name: &str,
    ownership: OwnershipPolicy,
    mode: FileModePolicy,
    maximum: u64,
) -> Result<File, ManagedSecretError> {
    let descriptor = rustix::fs::openat2(
        directory,
        name,
        rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::CLOEXEC | rustix::fs::OFlags::NOFOLLOW,
        rustix::fs::Mode::empty(),
        rustix::fs::ResolveFlags::BENEATH
            | rustix::fs::ResolveFlags::NO_MAGICLINKS
            | rustix::fs::ResolveFlags::NO_SYMLINKS,
    )
    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    let file = File::from(descriptor);
    let metadata = file
        .metadata()
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    validate_regular_metadata(&metadata, ownership, mode)?;
    if metadata.len() > maximum {
        return Err(ManagedSecretError::new(ManagedSecretErrorKind::Integrity));
    }
    Ok(file)
}

#[cfg(not(target_os = "linux"))]
fn open_regular_file(
    _directory: &File,
    _name: &str,
    _ownership: OwnershipPolicy,
    _mode: FileModePolicy,
    _maximum: u64,
) -> Result<File, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

#[cfg(target_os = "linux")]
fn validate_regular_metadata(
    metadata: &fs::Metadata,
    ownership: OwnershipPolicy,
    mode_policy: FileModePolicy,
) -> Result<(), ManagedSecretError> {
    if !metadata.is_file() {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsafeFilesystem,
        ));
    }
    validate_regular_unix_attributes(
        metadata.uid(),
        metadata.gid(),
        metadata.nlink(),
        metadata.mode() & 0o777,
        ownership,
        mode_policy,
    )
}

#[cfg(any(target_os = "linux", test))]
fn validate_regular_unix_attributes(
    uid: u32,
    gid: u32,
    link_count: u64,
    mode: u32,
    ownership: OwnershipPolicy,
    mode_policy: FileModePolicy,
) -> Result<(), ManagedSecretError> {
    let owner_matches = match ownership {
        OwnershipPolicy::EffectiveUser => uid == effective_uid() && gid == effective_gid(),
    };
    let unsafe_bits = match mode_policy {
        FileModePolicy::Store => mode & !0o640,
    };
    let readable = match mode_policy {
        FileModePolicy::Store => mode & 0o400 != 0,
    };
    if link_count != 1 || !owner_matches || unsafe_bits != 0 || !readable {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsafeFilesystem,
        ));
    }
    Ok(())
}

fn validate_directory(path: &Path, require_write: bool) -> Result<(), ManagedSecretError> {
    let metadata = fs::symlink_metadata(path)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::NotInitialized))?;
    validate_directory_metadata(&metadata, require_write)
}

fn validate_directory_metadata(
    metadata: &fs::Metadata,
    require_write: bool,
) -> Result<(), ManagedSecretError> {
    if !metadata.is_dir() {
        return Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsafeFilesystem,
        ));
    }
    #[cfg(unix)]
    {
        let mode = metadata.mode() & 0o777;
        if metadata.uid() != effective_uid()
            || metadata.gid() != effective_gid()
            || mode & !0o750 != 0
            || mode & 0o500 != 0o500
            || (require_write && mode & 0o200 == 0)
        {
            return Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsafeFilesystem,
            ));
        }
    }
    Ok(())
}

#[cfg(target_os = "linux")]
fn open_directory(path: &Path) -> Result<File, ManagedSecretError> {
    let descriptor = rustix::fs::open(
        path,
        rustix::fs::OFlags::RDONLY
            | rustix::fs::OFlags::DIRECTORY
            | rustix::fs::OFlags::CLOEXEC
            | rustix::fs::OFlags::NOFOLLOW,
        rustix::fs::Mode::empty(),
    )
    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    Ok(File::from(descriptor))
}

#[cfg(not(target_os = "linux"))]
fn open_directory(_path: &Path) -> Result<File, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

#[cfg(target_os = "linux")]
fn open_directory_at(directory: &File, path: &Path) -> Result<File, ManagedSecretError> {
    let descriptor = rustix::fs::openat2(
        directory,
        path,
        rustix::fs::OFlags::RDONLY
            | rustix::fs::OFlags::DIRECTORY
            | rustix::fs::OFlags::CLOEXEC
            | rustix::fs::OFlags::NOFOLLOW,
        rustix::fs::Mode::empty(),
        rustix::fs::ResolveFlags::BENEATH
            | rustix::fs::ResolveFlags::NO_MAGICLINKS
            | rustix::fs::ResolveFlags::NO_SYMLINKS,
    )
    .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))?;
    Ok(File::from(descriptor))
}

#[cfg(not(target_os = "linux"))]
fn open_directory_at(_directory: &File, _path: &Path) -> Result<File, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

#[cfg(target_os = "linux")]
fn read_directory(directory: &File) -> Result<fs::ReadDir, ManagedSecretError> {
    fs::read_dir(format!("/proc/self/fd/{}", directory.as_raw_fd()))
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::UnsafeFilesystem))
}

#[cfg(not(target_os = "linux"))]
fn read_directory(_directory: &File) -> Result<fs::ReadDir, ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

fn create_private_directory(path: &Path) -> Result<(), ManagedSecretError> {
    #[cfg(unix)]
    {
        let mut builder = fs::DirBuilder::new();
        builder.mode(0o700);
        builder
            .create(path)
            .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
    }
    #[cfg(not(unix))]
    {
        let _ = path;
        Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsupportedPlatform,
        ))
    }
}

fn effective_uid() -> u32 {
    #[cfg(unix)]
    {
        // SAFETY: This libc call reads the process credential and has no preconditions.
        unsafe { libc::geteuid() }
    }
    #[cfg(not(unix))]
    {
        0
    }
}

fn effective_gid() -> u32 {
    #[cfg(unix)]
    {
        // SAFETY: This libc call reads the process credential and has no preconditions.
        unsafe { libc::getegid() }
    }
    #[cfg(not(unix))]
    {
        0
    }
}

struct StoreLock {
    // Held only so `Drop` can release the `flock` on Linux.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    file: File,
}

enum LockMode {
    Shared,
    Exclusive,
    NonBlockingShared,
    NonBlockingExclusive,
}

impl StoreLock {
    fn acquire(root: &File, mode: LockMode) -> Result<Self, ManagedSecretError> {
        let file = open_regular_file(
            root,
            LOCK_FILE,
            OwnershipPolicy::EffectiveUser,
            FileModePolicy::Store,
            MAX_HEADER_BYTES,
        )?;
        #[cfg(target_os = "linux")]
        {
            let operation = match mode {
                LockMode::Shared => rustix::fs::FlockOperation::LockShared,
                LockMode::Exclusive => rustix::fs::FlockOperation::LockExclusive,
                LockMode::NonBlockingShared => rustix::fs::FlockOperation::NonBlockingLockShared,
                LockMode::NonBlockingExclusive => {
                    rustix::fs::FlockOperation::NonBlockingLockExclusive
                }
            };
            rustix::fs::flock(&file, operation)
                .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Coordination))?;
            Ok(Self { file })
        }
        #[cfg(not(target_os = "linux"))]
        {
            let _ = (file, mode);
            Err(ManagedSecretError::new(
                ManagedSecretErrorKind::UnsupportedPlatform,
            ))
        }
    }
}

impl Drop for StoreLock {
    fn drop(&mut self) {
        #[cfg(target_os = "linux")]
        {
            let _ = rustix::fs::flock(&self.file, rustix::fs::FlockOperation::Unlock);
        }
    }
}

fn sync_directory(directory: &File) -> Result<(), ManagedSecretError> {
    directory
        .sync_all()
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
}

fn sync_directory_path(path: &Path) -> Result<(), ManagedSecretError> {
    let directory = open_directory(path)
        .map_err(|_| ManagedSecretError::new(ManagedSecretErrorKind::Mutation))?;
    sync_directory(&directory)
}

#[cfg(target_os = "linux")]
fn remove_file_if_present(directory: &File, name: &str) -> Result<(), ManagedSecretError> {
    match rustix::fs::unlinkat(directory, name, rustix::fs::AtFlags::empty()) {
        Ok(()) => Ok(()),
        Err(error) if error == rustix::io::Errno::NOENT => Ok(()),
        Err(_) => Err(ManagedSecretError::new(ManagedSecretErrorKind::Mutation)),
    }
}

#[cfg(not(target_os = "linux"))]
fn remove_file_if_present(_directory: &File, _name: &str) -> Result<(), ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

#[cfg(target_os = "linux")]
fn rename_directory_create_only(
    source: &Path,
    destination: &Path,
) -> Result<(), ManagedSecretError> {
    rustix::fs::renameat_with(
        rustix::fs::CWD,
        source,
        rustix::fs::CWD,
        destination,
        rustix::fs::RenameFlags::NOREPLACE,
    )
    .map_err(|error| {
        if error == rustix::io::Errno::EXIST {
            ManagedSecretError::new(ManagedSecretErrorKind::AlreadyInitialized)
        } else {
            ManagedSecretError::new(ManagedSecretErrorKind::Mutation)
        }
    })
}

#[cfg(not(target_os = "linux"))]
fn rename_directory_create_only(
    _source: &Path,
    _destination: &Path,
) -> Result<(), ManagedSecretError> {
    Err(ManagedSecretError::new(
        ManagedSecretErrorKind::UnsupportedPlatform,
    ))
}

fn ensure_supported_platform() -> Result<(), ManagedSecretError> {
    if cfg!(target_os = "linux") {
        Ok(())
    } else {
        Err(ManagedSecretError::new(
            ManagedSecretErrorKind::UnsupportedPlatform,
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::SecretMaterial;
    use age::secrecy::ExposeSecret;
    use std::os::unix::fs::PermissionsExt;
    use std::str::FromStr;
    use std::sync::{Barrier, Mutex};
    use std::thread;

    struct FixedReferenceAuthority(Option<ManagedSecretReferenceStatus>);

    impl ManagedSecretReferenceAuthority for FixedReferenceAuthority {
        fn reference_status(
            &self,
            _reference: &ManagedSecretReference,
        ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
            self.0
                .ok_or_else(ManagedSecretReferenceInspectionError::unavailable)
        }
    }

    #[test]
    fn backend_creates_and_resolves_authenticated_material() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = ManagedSecretReference::from_str("managed://postgres/admin").unwrap();
        let material = SecretMaterial::try_from(b"credential-canary".to_vec()).unwrap();

        let metadata = store.create(&reference, &material).unwrap();
        assert_eq!(metadata.reference(), &reference);
        assert_eq!(metadata.version().get(), 1);

        let resolved = store.resolve(&reference).unwrap();
        assert_eq!(resolved.metadata(), &metadata);
        resolved
            .material()
            .expose(|value| assert_eq!(value, b"credential-canary"));
    }

    #[test]
    fn owner_replacement_is_durable_but_inactive_until_atomic_activation() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded_identity) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://datastore/owner/credential");
        let prior = store
            .create(&reference, &secret(b"prior-credential"))
            .unwrap();

        let staged = store
            .stage_replacement(&reference, &secret(b"staged-credential"), prior.version())
            .unwrap();
        assert_eq!(staged.metadata().version().get(), 2);
        store
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"prior-credential"));
        drop(store);

        let reopened = LocalManagedSecretStore::open_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded_identity),
        )
        .unwrap();
        let resumed = reopened.pending_replacement(&reference).unwrap().unwrap();
        resumed
            .material()
            .expose(|value| assert_eq!(value, b"staged-credential"));
        let activated = reopened.activate_replacement(&resumed).unwrap();
        assert_eq!(activated.version().get(), 2);
        reopened
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"staged-credential"));
        assert_eq!(read_directory(&reopened.anchor.objects).unwrap().count(), 1);
    }
    #[test]
    fn owner_replacement_batch_activates_all_artifacts_in_one_generation() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded_identity) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let oauth = managed_reference("managed://authentication/oauth/client-secret");
        let runtime = managed_reference("managed://authentication/runtime/credential");
        let oauth_prior = store.create(&oauth, &secret(b"oauth-v1")).unwrap();
        let runtime_prior = store.create(&runtime, &secret(b"runtime-v1")).unwrap();
        let oauth_staged = store
            .stage_replacement(&oauth, &secret(b"oauth-v2"), oauth_prior.version())
            .unwrap();
        let runtime_staged = store
            .stage_replacement(&runtime, &secret(b"runtime-v2"), runtime_prior.version())
            .unwrap();
        let prior_generation = read_manifest(&store.anchor.root, &store.identity)
            .unwrap()
            .generation;

        let activated = store
            .activate_replacements(&[&oauth_staged, &runtime_staged])
            .unwrap();
        assert_eq!(activated.len(), 2);
        drop(store);

        let reopened = LocalManagedSecretStore::open_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded_identity),
        )
        .unwrap();
        reopened
            .resolve(&oauth)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"oauth-v2"));
        reopened
            .resolve(&runtime)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"runtime-v2"));
        let manifest = read_manifest(&reopened.anchor.root, &reopened.identity).unwrap();
        assert_eq!(manifest.generation, prior_generation + 2);
        assert!(manifest.staged.is_empty());
    }

    #[test]
    fn failed_owner_workflow_discards_stage_and_preserves_prior_active_credential() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://authentication/owner/artifact");
        let prior = store
            .create(&reference, &secret(b"working-credential"))
            .unwrap();
        let staged = store
            .stage_replacement(&reference, &secret(b"rejected-credential"), prior.version())
            .unwrap();

        store.discard_replacement(&staged).unwrap();

        assert!(store.pending_replacement(&reference).unwrap().is_none());
        store
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"working-credential"));
        assert_eq!(read_directory(&store.anchor.objects).unwrap().count(), 1);
    }

    #[test]
    fn removal_requires_absence_proof_from_every_reference_authority() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded_identity) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://datastore/owner/credential");
        let metadata = store
            .create(&reference, &secret(b"removal-canary"))
            .unwrap();
        let clear = FixedReferenceAuthority(Some(ManagedSecretReferenceStatus::Unreferenced));
        let referenced = FixedReferenceAuthority(Some(ManagedSecretReferenceStatus::Referenced));
        let unavailable = FixedReferenceAuthority(None);

        let blocked = ManagedSecretReferenceAuthorities::new(&clear, &referenced, &clear, &clear)
            .prove_unreferenced(&reference)
            .unwrap_err();
        assert_eq!(
            blocked.authority(),
            ManagedSecretReferenceAuthorityKind::DatastoreBindings
        );
        assert_eq!(
            blocked.kind(),
            ManagedSecretRemovalProofErrorKind::Referenced
        );
        assert_eq!(store.inspect(&reference).unwrap(), metadata);

        let unproven = ManagedSecretReferenceAuthorities::new(&clear, &clear, &unavailable, &clear)
            .prove_unreferenced(&reference)
            .unwrap_err();
        assert_eq!(
            unproven.authority(),
            ManagedSecretReferenceAuthorityKind::SessionRecords
        );
        assert_eq!(
            unproven.kind(),
            ManagedSecretRemovalProofErrorKind::InspectionUnavailable
        );
        assert_eq!(store.inspect(&reference).unwrap(), metadata);

        let proof = ManagedSecretReferenceAuthorities::new(&clear, &clear, &clear, &clear)
            .prove_unreferenced(&reference)
            .unwrap();
        let removed = store.remove_proven_unreferenced(&reference, proof).unwrap();
        assert_eq!(removed, metadata);
        assert_eq!(
            store.resolve(&reference).unwrap_err().kind(),
            ManagedSecretErrorKind::Missing
        );
        drop(store);

        let verification = LocalManagedSecretStore::verify_all_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded_identity),
        )
        .unwrap();
        assert_eq!(verification.active_secret_count(), 0);
        assert_eq!(verification.audit_record_count(), 3);
    }
    #[test]
    fn datastore_reference_inventory_epoch_is_authenticated_and_legacy_stores_fail_closed() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        assert!(store.datastore_reference_inventory_is_complete().unwrap());

        let mut manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        manifest.datastore_reference_inventory_version = 0;
        write_manifest(
            &store.anchor.root,
            &manifest,
            &store.identity,
            store.faults.as_ref(),
        )
        .unwrap();
        assert!(!store.datastore_reference_inventory_is_complete().unwrap());

        manifest.datastore_reference_inventory_version = DATASTORE_REFERENCE_INVENTORY_VERSION + 1;
        write_manifest(
            &store.anchor.root,
            &manifest,
            &store.identity,
            store.faults.as_ref(),
        )
        .unwrap();
        assert_eq!(
            store
                .datastore_reference_inventory_is_complete()
                .unwrap_err()
                .kind(),
            ManagedSecretErrorKind::Integrity
        );
    }

    #[test]
    fn durable_reference_holder_survives_restart_and_invalidates_stale_absence_proof() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded_identity) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://datastore/owner/credential");
        store.create(&reference, &secret(b"credential-v1")).unwrap();
        let clear = FixedReferenceAuthority(Some(ManagedSecretReferenceStatus::Unreferenced));
        let stale_proof = ManagedSecretReferenceAuthorities::new(&clear, &clear, &clear, &clear)
            .prove_unreferenced(&reference)
            .unwrap();
        store
            .replace_reference_holder("session:1a", std::slice::from_ref(&reference))
            .unwrap();
        drop(store);

        let reopened = LocalManagedSecretStore::open_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded_identity),
        )
        .unwrap();
        assert_eq!(
            reopened.held_reference_status(&reference).unwrap(),
            ManagedSecretReferenceStatus::Referenced
        );
        assert_eq!(
            reopened
                .remove_proven_unreferenced(&reference, stale_proof)
                .unwrap_err()
                .kind(),
            ManagedSecretErrorKind::VersionConflict
        );

        reopened.remove_reference_holder("session:1a").unwrap();
        let fresh_proof = ManagedSecretReferenceAuthorities::new(&clear, &clear, &clear, &clear)
            .prove_unreferenced(&reference)
            .unwrap();
        reopened
            .remove_proven_unreferenced(&reference, fresh_proof)
            .unwrap();
    }

    #[test]
    fn list_and_inspect_return_only_safe_active_metadata() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let first = managed_reference("managed://oauth/client-token");
        let second = managed_reference("managed://postgres/admin");
        store.create(&second, &secret(b"postgres-canary")).unwrap();
        store.create(&first, &secret(b"oauth-canary")).unwrap();

        let listed = store.list().unwrap();
        assert_eq!(listed.len(), 2);
        assert_eq!(listed[0].reference(), &first);
        assert_eq!(listed[1].reference(), &second);
        assert_eq!(listed[0].version().get(), 1);

        let inspected = store.inspect(&first).unwrap();
        assert_eq!(inspected, listed[0]);
        let rendered = serde_json::to_string(&(listed, inspected)).unwrap();
        assert!(!rendered.contains("oauth-canary"));
        assert!(!rendered.contains("postgres-canary"));
        assert!(!rendered.contains(fixture.root().to_string_lossy().as_ref()));
    }

    #[test]
    fn initialization_is_create_only_and_identity_debug_is_redacted() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let debug = format!("{identity:?}");
        assert!(!debug.contains(&encoded));
        LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity).unwrap();

        let (different_identity, _) = test_identity_pair();
        let error = LocalManagedSecretStore::initialize_at(
            fixture.root(),
            deployment_id,
            different_identity,
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::AlreadyInitialized);
    }

    #[test]
    fn initialization_reports_commit_uncertain_after_the_store_rename() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let error = LocalManagedSecretStore::initialize_at_with_faults(
            fixture.root(),
            deployment_id,
            identity,
            Arc::new(FailAt(MutationCheckpoint::DirectorySync(
                MutationPhase::Initialization,
            ))),
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::CommitUncertain);

        let reopened = LocalManagedSecretStore::open_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap();
        assert_eq!(
            read_manifest(&reopened.anchor.root, &reopened.identity)
                .unwrap()
                .generation,
            0
        );
        drop(reopened);

        let (different_identity, _) = test_identity_pair();
        let retry = LocalManagedSecretStore::initialize_at(
            fixture.root(),
            deployment_id,
            different_identity,
        )
        .unwrap_err();
        assert_eq!(retry.kind(), ManagedSecretErrorKind::AlreadyInitialized);
    }

    #[test]
    fn backend_round_trips_the_maximum_secret_material_size() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://bounds/maximum");
        let material = secret(&vec![b'x'; crate::MAX_SECRET_MATERIAL_BYTES]);

        store.create(&reference, &material).unwrap();
        store
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value.len(), crate::MAX_SECRET_MATERIAL_BYTES));
        assert!(SecretMaterial::try_from(vec![0; crate::MAX_SECRET_MATERIAL_BYTES + 1]).is_err());
    }

    #[test]
    fn active_versions_increase_monotonically_and_create_never_overwrites() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://oauth/token");
        let first = secret(b"first-version");
        let second = secret(b"second-version");
        let third = secret(b"third-version");

        let first_metadata = store.create(&reference, &first).unwrap();
        let duplicate = store.create(&reference, &second).unwrap_err();
        assert_eq!(duplicate.kind(), ManagedSecretErrorKind::AlreadyExists);

        let second_metadata = store
            .replace_for_test(&reference, &second, first_metadata.version())
            .unwrap();
        let stale = store
            .replace_for_test(&reference, &third, first_metadata.version())
            .unwrap_err();
        assert_eq!(stale.kind(), ManagedSecretErrorKind::VersionConflict);
        let third_metadata = store
            .replace_for_test(&reference, &third, second_metadata.version())
            .unwrap();
        assert_eq!(second_metadata.version().get(), 2);
        assert_eq!(third_metadata.version().get(), 3);

        let resolved = store.resolve(&reference).unwrap();
        assert_eq!(resolved.metadata().version().get(), 3);
        resolved
            .material()
            .expose(|value| assert_eq!(value, b"third-version"));
        assert_eq!(
            fs::read_dir(fixture.root().join(OBJECTS_DIRECTORY))
                .unwrap()
                .count(),
            1
        );
    }

    #[test]
    fn opening_rejects_deployment_and_recipient_mismatch() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity).unwrap();

        let wrong_deployment = LocalManagedSecretStore::open_at(
            fixture.root(),
            Uuid::new_v4(),
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(
            wrong_deployment.kind(),
            ManagedSecretErrorKind::DeploymentMismatch
        );

        let (wrong_identity, _) = test_identity_pair();
        let wrong_recipient =
            LocalManagedSecretStore::open_at(fixture.root(), deployment_id, wrong_identity)
                .unwrap_err();
        assert_eq!(
            wrong_recipient.kind(),
            ManagedSecretErrorKind::RecipientMismatch
        );
    }

    #[test]
    fn ciphertext_relocation_and_corruption_fail_authentication() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let first_reference = managed_reference("managed://datastore/first");
        let second_reference = managed_reference("managed://datastore/second");
        store
            .create(&first_reference, &secret(b"first-secret"))
            .unwrap();
        store
            .create(&second_reference, &secret(b"second-secret"))
            .unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let first_path = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[first_reference.as_str()].object);
        let second_path = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[second_reference.as_str()].object);
        let swap_path = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(format!(".mutation-{}.tmp", Uuid::new_v4()));
        fs::rename(&first_path, &swap_path).unwrap();
        fs::rename(&second_path, &first_path).unwrap();
        fs::rename(&swap_path, &second_path).unwrap();

        let relocated = store.resolve(&first_reference).unwrap_err();
        assert_eq!(relocated.kind(), ManagedSecretErrorKind::Integrity);

        let mut ciphertext = fs::read(&first_path).unwrap();
        let middle = ciphertext.len() / 2;
        ciphertext[middle] ^= 0x80;
        fs::write(&first_path, ciphertext).unwrap();
        let corrupted = store.resolve(&first_reference).unwrap_err();
        assert_eq!(corrupted.kind(), ManagedSecretErrorKind::Integrity);
    }

    #[test]
    fn unsafe_store_permissions_fail_closed() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://runtime/key");
        store.create(&reference, &secret(b"runtime-key")).unwrap();

        fs::set_permissions(fixture.root(), fs::Permissions::from_mode(0o770)).unwrap();
        let error = store.resolve(&reference).unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
    }

    #[test]
    fn unsafe_modes_on_every_store_file_fail_closed() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://filesystem/file-modes");
        store.create(&reference, &secret(b"mode-canary")).unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let object = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[reference.as_str()].object);

        for path in [
            fixture.root().join(HEADER_FILE),
            fixture.root().join(LOCK_FILE),
            fixture.root().join(MANIFEST_FILE),
            object,
        ] {
            fs::set_permissions(&path, fs::Permissions::from_mode(0o644)).unwrap();
            let error = store.resolve(&reference).unwrap_err();
            assert_eq!(error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
            fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap();
            store.resolve(&reference).unwrap();
        }
    }

    #[test]
    fn store_file_security_attributes_require_effective_uid_gid_and_one_link() {
        let uid = effective_uid();
        let gid = effective_gid();
        assert!(
            validate_regular_unix_attributes(
                uid,
                gid,
                1,
                0o600,
                OwnershipPolicy::EffectiveUser,
                FileModePolicy::Store,
            )
            .is_ok()
        );
        for (candidate_uid, candidate_gid, links, mode) in [
            (uid.wrapping_add(1), gid, 1, 0o600),
            (uid, gid.wrapping_add(1), 1, 0o600),
            (uid, gid, 2, 0o600),
            (uid, gid, 1, 0o644),
        ] {
            let error = validate_regular_unix_attributes(
                candidate_uid,
                candidate_gid,
                links,
                mode,
                OwnershipPolicy::EffectiveUser,
                FileModePolicy::Store,
            )
            .unwrap_err();
            assert_eq!(error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
        }
    }

    #[test]
    fn replaced_objects_directory_cannot_redirect_reads_or_writes() {
        use std::os::unix::fs::symlink;

        let fixture = StoreFixture::new();
        let outside = StoreFixture::new();
        fs::create_dir(outside.root()).unwrap();
        fs::set_permissions(outside.root(), fs::Permissions::from_mode(0o700)).unwrap();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://filesystem/anchored");
        store
            .create(&reference, &secret(b"anchored-value"))
            .unwrap();

        let objects = fixture.root().join(OBJECTS_DIRECTORY);
        fs::rename(&objects, fixture.root().join("objects-original")).unwrap();
        symlink(outside.root(), &objects).unwrap();

        let read_error = store.resolve(&reference).unwrap_err();
        assert_eq!(read_error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
        let write_error = store
            .create(
                &managed_reference("managed://filesystem/redirected"),
                &secret(b"must-not-escape"),
            )
            .unwrap_err();
        assert_eq!(write_error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
        assert_eq!(fs::read_dir(outside.root()).unwrap().count(), 0);
    }

    #[test]
    fn concurrent_writers_preserve_every_committed_reference() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = Arc::new(
            LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                .unwrap(),
        );
        let barrier = Arc::new(Barrier::new(9));
        let mut writers = Vec::new();
        for index in 0..8 {
            let store = Arc::clone(&store);
            let barrier = Arc::clone(&barrier);
            writers.push(thread::spawn(move || {
                let reference = managed_reference(&format!("managed://concurrent/item-{index}"));
                let value = secret(format!("value-{index}").as_bytes());
                barrier.wait();
                store.create(&reference, &value).unwrap();
            }));
        }
        barrier.wait();
        for writer in writers {
            writer.join().unwrap();
        }
        for index in 0..8 {
            let reference = managed_reference(&format!("managed://concurrent/item-{index}"));
            let expected = format!("value-{index}");
            store
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, expected.as_bytes()));
        }
    }

    #[test]
    fn concurrent_readers_observe_only_complete_active_versions() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = Arc::new(
            LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                .unwrap(),
        );
        let reference = managed_reference("managed://concurrent/active");
        let first = store.create(&reference, &secret(b"version-one")).unwrap();
        let barrier = Arc::new(Barrier::new(9));
        let mut readers = Vec::new();
        for _ in 0..8 {
            let store = Arc::clone(&store);
            let reference = reference.clone();
            let barrier = Arc::clone(&barrier);
            readers.push(thread::spawn(move || {
                barrier.wait();
                for _ in 0..16 {
                    let resolved = store.resolve(&reference).unwrap();
                    let version = resolved.metadata().version().get();
                    resolved.material().expose(|value| match version {
                        1 => assert_eq!(value, b"version-one"),
                        2 => assert_eq!(value, b"version-two"),
                        _ => panic!("unexpected active version"),
                    });
                }
            }));
        }
        barrier.wait();
        let second = store
            .replace_for_test(&reference, &secret(b"version-two"), first.version())
            .unwrap();
        assert_eq!(second.version().get(), 2);
        for reader in readers {
            reader.join().unwrap();
        }
    }

    #[test]
    fn injected_mutation_failures_preserve_the_prior_active_version() {
        let checkpoints = [
            MutationCheckpoint::Write(MutationPhase::Object),
            MutationCheckpoint::Sync(MutationPhase::Object),
            MutationCheckpoint::Rename(MutationPhase::Object),
            MutationCheckpoint::DirectorySync(MutationPhase::Object),
            MutationCheckpoint::Write(MutationPhase::Manifest),
            MutationCheckpoint::Sync(MutationPhase::Manifest),
            MutationCheckpoint::Rename(MutationPhase::Manifest),
        ];
        for checkpoint in checkpoints {
            let fixture = StoreFixture::new();
            let deployment_id = Uuid::new_v4();
            let (identity, encoded) = test_identity_pair();
            let initial =
                LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                    .unwrap();
            let reference = managed_reference("managed://failure/active");
            let first = initial
                .create(&reference, &secret(b"stable-value"))
                .unwrap();
            drop(initial);

            let store = LocalManagedSecretStore::open_at_with_faults(
                fixture.root(),
                deployment_id,
                Arc::new(identity_from_text(&encoded)),
                Arc::new(FailAt(checkpoint)),
            )
            .unwrap();
            let failure = store
                .replace_for_test(&reference, &secret(b"failed-value"), first.version())
                .unwrap_err();
            assert_eq!(failure.kind(), ManagedSecretErrorKind::Mutation);
            store
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, b"stable-value"));
            drop(store);

            let reopened = LocalManagedSecretStore::open_at(
                fixture.root(),
                deployment_id,
                identity_from_text(&encoded),
            )
            .unwrap();
            reopened
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, b"stable-value"));
        }
    }

    #[test]
    fn injected_coordination_decrypt_audit_and_verification_fail_closed() {
        for (checkpoint, expected_kind) in [
            (
                MutationCheckpoint::Lock,
                ManagedSecretErrorKind::Coordination,
            ),
            (
                MutationCheckpoint::Decrypt,
                ManagedSecretErrorKind::Integrity,
            ),
            (MutationCheckpoint::Audit, ManagedSecretErrorKind::Mutation),
            (
                MutationCheckpoint::Verification,
                ManagedSecretErrorKind::Mutation,
            ),
        ] {
            let fixture = StoreFixture::new();
            let deployment_id = Uuid::new_v4();
            let (identity, encoded) = test_identity_pair();
            let initial =
                LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                    .unwrap();
            let reference = managed_reference("managed://failure/non-write-phase");
            let first = initial
                .create(&reference, &secret(b"stable-value"))
                .unwrap();
            drop(initial);

            let store = LocalManagedSecretStore::open_at_with_faults(
                fixture.root(),
                deployment_id,
                Arc::new(identity_from_text(&encoded)),
                Arc::new(FailAt(checkpoint)),
            )
            .unwrap();
            let error = store
                .replace_for_test(&reference, &secret(b"failed-value"), first.version())
                .unwrap_err();
            assert_eq!(error.kind(), expected_kind);
            store
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, b"stable-value"));
            assert_eq!(
                fs::read_dir(fixture.root().join(OBJECTS_DIRECTORY))
                    .unwrap()
                    .count(),
                1
            );
        }
    }

    #[test]
    fn post_activation_sync_failure_preserves_a_recoverable_committed_state() {
        for checkpoint in [
            MutationCheckpoint::DirectorySync(MutationPhase::Manifest),
            MutationCheckpoint::CleanupRemove,
            MutationCheckpoint::CleanupSync,
        ] {
            let fixture = StoreFixture::new();
            let deployment_id = Uuid::new_v4();
            let (identity, encoded) = test_identity_pair();
            let initial =
                LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                    .unwrap();
            let reference = managed_reference("managed://failure/commit-uncertain");
            let first = initial.create(&reference, &secret(b"first-value")).unwrap();
            drop(initial);

            let store = LocalManagedSecretStore::open_at_with_faults(
                fixture.root(),
                deployment_id,
                Arc::new(identity_from_text(&encoded)),
                Arc::new(FailAt(checkpoint)),
            )
            .unwrap();
            let error = store
                .replace_for_test(&reference, &secret(b"second-value"), first.version())
                .unwrap_err();
            assert_eq!(error.kind(), ManagedSecretErrorKind::CommitUncertain);
            store
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, b"second-value"));
            drop(store);

            let reopened = LocalManagedSecretStore::open_at(
                fixture.root(),
                deployment_id,
                identity_from_text(&encoded),
            )
            .unwrap();
            reopened
                .resolve(&reference)
                .unwrap()
                .material()
                .expose(|value| assert_eq!(value, b"second-value"));
            assert_eq!(
                fs::read_dir(fixture.root().join(OBJECTS_DIRECTORY))
                    .unwrap()
                    .count(),
                1
            );
        }
    }

    #[test]
    fn audit_history_is_secret_free_at_rest_and_chain_corruption_is_detected() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://audit/canary");
        let canary = b"audit-must-not-contain-this-secret-canary";
        store.create(&reference, &secret(canary)).unwrap();

        let mut manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let plaintext_manifest = serde_json::to_vec(&manifest).unwrap();
        assert!(!contains_bytes(&plaintext_manifest, canary));
        assert_eq!(manifest.audit.len(), 2);
        assert_eq!(manifest.audit[1].actor.effective_uid, effective_uid());
        assert_eq!(manifest.audit[1].deployment_id, deployment_id);
        assert_eq!(
            manifest.audit[1].reference.as_deref(),
            Some(reference.as_str())
        );

        for path in [
            fixture.root().join(HEADER_FILE),
            fixture.root().join(MANIFEST_FILE),
        ] {
            assert!(!contains_bytes(&fs::read(path).unwrap(), canary));
        }
        let object = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[reference.as_str()].object);
        assert!(!contains_bytes(&fs::read(object).unwrap(), canary));

        manifest.audit[1].previous_hash[0] ^= 1;
        let ciphertext = serialize_and_encrypt_manifest(&manifest, &store.identity).unwrap();
        atomic_write(
            &store.anchor.root,
            MANIFEST_FILE,
            &ciphertext,
            &NoMutationFaults,
            MutationPhase::Manifest,
        )
        .unwrap();
        let error = store.resolve(&reference).unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::Integrity);
    }

    #[test]
    fn uncommitted_ciphertext_is_recovered_without_changing_active_state() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://recovery/active");
        store.create(&reference, &secret(b"active-value")).unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let active = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[reference.as_str()].object);
        let orphan = fixture
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(format!("object-{}.age", Uuid::new_v4()));
        fs::copy(active, &orphan).unwrap();
        fs::set_permissions(&orphan, fs::Permissions::from_mode(0o600)).unwrap();
        drop(store);

        let reopened = LocalManagedSecretStore::open_at(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap();
        assert!(!orphan.exists());
        reopened
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"active-value"));
    }

    #[test]
    fn complete_verification_authenticates_every_active_secret_without_repair() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        store
            .create(
                &managed_reference("managed://recovery/first"),
                &secret(b"first-recovery-canary"),
            )
            .unwrap();
        store
            .create(
                &managed_reference("managed://recovery/second"),
                &secret(b"second-recovery-canary"),
            )
            .unwrap();
        drop(store);
        let before = StoreSnapshot::capture(fixture.root());

        let report = LocalManagedSecretStore::verify_all_at_for_test(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap();

        assert_eq!(report.deployment_id(), deployment_id);
        assert_eq!(report.active_secret_count(), 2);
        assert_eq!(report.audit_record_count(), 3);
        assert_eq!(report.recipient_fingerprint().len(), 64);
        assert_eq!(StoreSnapshot::capture(fixture.root()), before);
    }

    #[test]
    fn complete_verification_rejects_exact_recovery_mismatches_and_extra_state() {
        let fixture = StoreFixture::new();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        store
            .create(
                &managed_reference("managed://recovery/exact"),
                &secret(b"exact-recovery-canary"),
            )
            .unwrap();
        drop(store);

        let deployment_error = LocalManagedSecretStore::verify_all_at_for_test(
            fixture.root(),
            Uuid::new_v4(),
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(
            deployment_error.kind(),
            ManagedSecretErrorKind::DeploymentMismatch
        );
        let (other_identity, _) = test_identity_pair();
        let identity_error = LocalManagedSecretStore::verify_all_at_for_test(
            fixture.root(),
            deployment_id,
            other_identity,
        )
        .unwrap_err();
        assert_eq!(
            identity_error.kind(),
            ManagedSecretErrorKind::RecipientMismatch
        );

        let extra = fixture.root().join(OBJECTS_DIRECTORY).join("unexpected");
        fs::write(&extra, b"extra").unwrap();
        fs::set_permissions(&extra, fs::Permissions::from_mode(0o600)).unwrap();
        let extra_error = LocalManagedSecretStore::verify_all_at_for_test(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(extra_error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
    }

    #[test]
    fn complete_verification_rejects_missing_corrupt_relocated_and_broken_history() {
        let deployment_id = Uuid::new_v4();

        let missing = StoreFixture::new();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(missing.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://recovery/missing");
        store
            .create(&reference, &secret(b"missing-canary"))
            .unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        fs::remove_file(
            missing
                .root()
                .join(OBJECTS_DIRECTORY)
                .join(&manifest.entries[reference.as_str()].object),
        )
        .unwrap();
        drop(store);
        let error = LocalManagedSecretStore::verify_all_at_for_test(
            missing.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);

        let corrupt = StoreFixture::new();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(corrupt.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://recovery/corrupt");
        store
            .create(&reference, &secret(b"corrupt-canary"))
            .unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let path = corrupt
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[reference.as_str()].object);
        let mut ciphertext = fs::read(&path).unwrap();
        let middle = ciphertext.len() / 2;
        ciphertext[middle] ^= 1;
        fs::write(path, ciphertext).unwrap();
        drop(store);
        let error = LocalManagedSecretStore::verify_all_at_for_test(
            corrupt.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::Integrity);

        let relocated = StoreFixture::new();
        let (identity, encoded) = test_identity_pair();
        let store =
            LocalManagedSecretStore::initialize_at(relocated.root(), deployment_id, identity)
                .unwrap();
        let first = managed_reference("managed://recovery/relocated-first");
        let second = managed_reference("managed://recovery/relocated-second");
        store.create(&first, &secret(b"first-canary")).unwrap();
        store.create(&second, &secret(b"second-canary")).unwrap();
        let manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        let first_path = relocated
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[first.as_str()].object);
        let second_path = relocated
            .root()
            .join(OBJECTS_DIRECTORY)
            .join(&manifest.entries[second.as_str()].object);
        let swap = relocated.root().join(OBJECTS_DIRECTORY).join("swap");
        fs::rename(&first_path, &swap).unwrap();
        fs::rename(&second_path, &first_path).unwrap();
        fs::rename(&swap, &second_path).unwrap();
        drop(store);
        let error = LocalManagedSecretStore::verify_all_at_for_test(
            relocated.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::Integrity);

        let history = StoreFixture::new();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(history.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://recovery/history");
        store
            .create(&reference, &secret(b"history-canary"))
            .unwrap();
        let mut manifest = read_manifest(&store.anchor.root, &store.identity).unwrap();
        manifest.audit[1].previous_hash[0] ^= 1;
        let ciphertext = serialize_and_encrypt_manifest(&manifest, &store.identity).unwrap();
        atomic_write(
            &store.anchor.root,
            MANIFEST_FILE,
            &ciphertext,
            &NoMutationFaults,
            MutationPhase::Manifest,
        )
        .unwrap();
        drop(store);
        let error = LocalManagedSecretStore::verify_all_at_for_test(
            history.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(error.kind(), ManagedSecretErrorKind::Integrity);
    }

    #[test]
    fn offline_root_rotation_stages_a_fully_verified_store_without_mutating_active_state() {
        let fixture = StoreFixture::new();
        let staging = StoreFixture::new();
        fs::create_dir(staging.root()).unwrap();
        fs::set_permissions(staging.root(), fs::Permissions::from_mode(0o700)).unwrap();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let (next_identity, next_encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let reference = managed_reference("managed://rotation/canary");
        store
            .create(&reference, &secret(b"root-rotation-canary"))
            .unwrap();
        drop(store);
        let active_before = StoreSnapshot::capture(fixture.root());

        let report = LocalManagedSecretStore::rotate_root_at_for_test(
            fixture.root(),
            staging.root(),
            deployment_id,
            identity_from_text(&encoded),
            next_identity,
        )
        .unwrap();

        assert_eq!(report.deployment_id(), deployment_id);
        assert_eq!(report.active_secret_count(), 1);
        assert_eq!(report.audit_record_count(), 2);
        assert_ne!(
            report.current_recipient_fingerprint(),
            report.next_recipient_fingerprint()
        );
        assert_eq!(StoreSnapshot::capture(fixture.root()), active_before);
        LocalManagedSecretStore::verify_all_at_for_test(
            fixture.root(),
            deployment_id,
            identity_from_text(&encoded),
        )
        .unwrap();
        LocalManagedSecretStore::verify_all_at_for_test(
            staging.root(),
            deployment_id,
            identity_from_text(&next_encoded),
        )
        .unwrap();
        let staged = LocalManagedSecretStore::open_at(
            staging.root(),
            deployment_id,
            identity_from_text(&next_encoded),
        )
        .unwrap();
        staged
            .resolve(&reference)
            .unwrap()
            .material()
            .expose(|value| assert_eq!(value, b"root-rotation-canary"));
    }

    #[test]
    fn offline_root_rotation_refuses_prerequisites_before_staging_writes() {
        let fixture = StoreFixture::new();
        let staging = StoreFixture::new();
        fs::create_dir(staging.root()).unwrap();
        fs::set_permissions(staging.root(), fs::Permissions::from_mode(0o700)).unwrap();
        let deployment_id = Uuid::new_v4();
        let (identity, encoded) = test_identity_pair();
        let store = LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
            .unwrap();
        let active = StoreAnchor::open(fixture.root(), true).unwrap();
        let held_lock = StoreLock::acquire(&active.root, LockMode::Exclusive).unwrap();
        let (next_identity, _) = test_identity_pair();

        let lock_error = LocalManagedSecretStore::rotate_root_at_for_test(
            fixture.root(),
            staging.root(),
            deployment_id,
            identity_from_text(&encoded),
            next_identity,
        )
        .unwrap_err();
        assert_eq!(lock_error.kind(), ManagedSecretErrorKind::Coordination);
        assert_eq!(fs::read_dir(staging.root()).unwrap().count(), 0);
        drop(held_lock);
        drop(active);
        drop(store);

        let same_identity_error = LocalManagedSecretStore::rotate_root_at_for_test(
            fixture.root(),
            staging.root(),
            deployment_id,
            identity_from_text(&encoded),
            identity_from_text(&encoded),
        )
        .unwrap_err();
        assert_eq!(
            same_identity_error.kind(),
            ManagedSecretErrorKind::RecipientMismatch
        );
        assert_eq!(fs::read_dir(staging.root()).unwrap().count(), 0);

        fs::write(staging.root().join("operator-evidence"), b"retain").unwrap();
        fs::set_permissions(
            staging.root().join("operator-evidence"),
            fs::Permissions::from_mode(0o600),
        )
        .unwrap();
        let (next_identity, _) = test_identity_pair();
        let nonempty_error = LocalManagedSecretStore::rotate_root_at_for_test(
            fixture.root(),
            staging.root(),
            deployment_id,
            identity_from_text(&encoded),
            next_identity,
        )
        .unwrap_err();
        assert_eq!(
            nonempty_error.kind(),
            ManagedSecretErrorKind::UnsafeFilesystem
        );
        assert_eq!(
            fs::read(staging.root().join("operator-evidence")).unwrap(),
            b"retain"
        );
    }

    #[test]
    fn failed_root_rotation_preserves_active_state_and_staging_evidence() {
        let checkpoints = [
            RotationCheckpoint::Lock,
            RotationCheckpoint::SourceDecrypt,
            RotationCheckpoint::SourceAudit,
            RotationCheckpoint::CreateObjectsDirectory,
            RotationCheckpoint::Write(RotationFile::Object),
            RotationCheckpoint::SyncFile(RotationFile::Object),
            RotationCheckpoint::Write(RotationFile::Lock),
            RotationCheckpoint::SyncFile(RotationFile::Lock),
            RotationCheckpoint::Write(RotationFile::Header),
            RotationCheckpoint::SyncFile(RotationFile::Header),
            RotationCheckpoint::Write(RotationFile::Manifest),
            RotationCheckpoint::SyncFile(RotationFile::Manifest),
            RotationCheckpoint::SyncDirectory(RotationDirectory::Objects),
            RotationCheckpoint::SyncDirectory(RotationDirectory::Root),
            RotationCheckpoint::DestinationVerification,
        ];
        for checkpoint in checkpoints {
            let fixture = StoreFixture::new();
            let staging = StoreFixture::new();
            fs::create_dir(staging.root()).unwrap();
            fs::set_permissions(staging.root(), fs::Permissions::from_mode(0o700)).unwrap();
            let deployment_id = Uuid::new_v4();
            let (identity, encoded) = test_identity_pair();
            let (next_identity, _) = test_identity_pair();
            let store =
                LocalManagedSecretStore::initialize_at(fixture.root(), deployment_id, identity)
                    .unwrap();
            store
                .create(
                    &managed_reference("managed://rotation/failure-evidence"),
                    &secret(b"rotation-failure-canary"),
                )
                .unwrap();
            drop(store);
            let active_before = StoreSnapshot::capture(fixture.root());

            let injected_failure = RecordingRotationFailure::new(checkpoint, staging.root());
            let error = LocalManagedSecretStore::rotate_root_at_with_faults(
                fixture.root(),
                staging.root(),
                deployment_id,
                identity_from_text(&encoded),
                next_identity,
                false,
                &injected_failure,
            )
            .unwrap_err();

            assert_eq!(error.kind(), ManagedSecretErrorKind::Mutation);
            let expected_phase = match checkpoint {
                RotationCheckpoint::Lock => RootRotationPhase::Lock,
                RotationCheckpoint::SourceDecrypt | RotationCheckpoint::SourceAudit => {
                    RootRotationPhase::SourceVerification
                }
                RotationCheckpoint::DestinationVerification => {
                    RootRotationPhase::DestinationVerification
                }
                _ => RootRotationPhase::StagingWrite,
            };
            assert_eq!(error.phase(), expected_phase);
            assert_eq!(StoreSnapshot::capture(fixture.root()), active_before);
            assert_eq!(
                StoreSnapshot::capture(staging.root()),
                injected_failure.snapshot_at_failure()
            );
        }
    }

    #[test]
    fn rotation_staging_anchor_rejects_path_substitution_without_redirecting_writes() {
        use std::os::unix::fs::symlink;

        let active_fixture = StoreFixture::new();
        let staging_fixture = StoreFixture::new();
        let outside = StoreFixture::new();
        fs::create_dir(staging_fixture.root()).unwrap();
        fs::create_dir(outside.root()).unwrap();
        fs::set_permissions(staging_fixture.root(), fs::Permissions::from_mode(0o700)).unwrap();
        fs::set_permissions(outside.root(), fs::Permissions::from_mode(0o700)).unwrap();
        let deployment_id = Uuid::new_v4();
        let (identity, _) = test_identity_pair();
        let store =
            LocalManagedSecretStore::initialize_at(active_fixture.root(), deployment_id, identity)
                .unwrap();
        let active = StoreAnchor::open(active_fixture.root(), true).unwrap();
        let staging = validate_empty_staging_directory(staging_fixture.root(), &active).unwrap();
        let moved = staging_fixture.root().with_extension("moved");
        fs::rename(staging_fixture.root(), &moved).unwrap();
        symlink(outside.root(), staging_fixture.root()).unwrap();
        let (_, next_encoded) = test_identity_pair();
        let verified = verify_anchored_store(
            &store.anchor,
            active_fixture.root(),
            deployment_id,
            &store.identity,
        )
        .unwrap();

        let error = match populate_rotation_staging(
            &staging,
            &verified,
            &identity_from_text(&next_encoded),
            &NoRotationFaults,
        ) {
            Ok(_) => panic!("substituted staging path must fail"),
            Err(error) => error,
        };

        assert_eq!(error.kind(), ManagedSecretErrorKind::UnsafeFilesystem);
        assert_eq!(fs::read_dir(outside.root()).unwrap().count(), 0);
        assert_eq!(fs::read_dir(&moved).unwrap().count(), 0);
        fs::remove_file(staging_fixture.root()).unwrap();
        fs::rename(moved, staging_fixture.root()).unwrap();
    }

    #[derive(Debug, PartialEq, Eq)]
    struct StoreSnapshot(Vec<(PathBuf, Vec<u8>, u32)>);

    impl StoreSnapshot {
        fn capture(root: &Path) -> Self {
            fn visit(base: &Path, current: &Path, entries: &mut Vec<(PathBuf, Vec<u8>, u32)>) {
                let mut children = fs::read_dir(current)
                    .unwrap()
                    .map(|entry| entry.unwrap())
                    .collect::<Vec<_>>();
                children.sort_by_key(|entry| entry.file_name());
                for child in children {
                    let path = child.path();
                    let metadata = child.metadata().unwrap();
                    let relative = path.strip_prefix(base).unwrap().to_path_buf();
                    if metadata.is_dir() {
                        entries.push((
                            relative.clone(),
                            Vec::new(),
                            metadata.permissions().mode() & 0o777,
                        ));
                        visit(base, &path, entries);
                    } else {
                        entries.push((
                            relative,
                            fs::read(path).unwrap(),
                            metadata.permissions().mode() & 0o777,
                        ));
                    }
                }
            }
            let mut entries = Vec::new();
            visit(root, root, &mut entries);
            Self(entries)
        }
    }

    struct FailAt(MutationCheckpoint);

    impl MutationFaultInjector for FailAt {
        fn check(&self, checkpoint: MutationCheckpoint) -> Result<(), ManagedSecretError> {
            if checkpoint == self.0 {
                Err(ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
            } else {
                Ok(())
            }
        }
    }

    struct RecordingRotationFailure {
        checkpoint: RotationCheckpoint,
        staging_root: PathBuf,
        snapshot: Mutex<Option<StoreSnapshot>>,
    }

    impl RecordingRotationFailure {
        fn new(checkpoint: RotationCheckpoint, staging_root: &Path) -> Self {
            Self {
                checkpoint,
                staging_root: staging_root.to_path_buf(),
                snapshot: Mutex::new(None),
            }
        }

        fn snapshot_at_failure(&self) -> StoreSnapshot {
            self.snapshot
                .lock()
                .unwrap()
                .take()
                .expect("injected checkpoint must be reached")
        }
    }

    impl RotationFaultInjector for RecordingRotationFailure {
        fn check(&self, checkpoint: RotationCheckpoint) -> Result<(), ManagedSecretError> {
            if checkpoint == self.checkpoint {
                *self.snapshot.lock().unwrap() = Some(StoreSnapshot::capture(&self.staging_root));
                Err(ManagedSecretError::new(ManagedSecretErrorKind::Mutation))
            } else {
                Ok(())
            }
        }
    }

    fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
        haystack
            .windows(needle.len())
            .any(|window| window == needle)
    }

    fn managed_reference(value: &str) -> ManagedSecretReference {
        ManagedSecretReference::from_str(value).unwrap()
    }

    fn secret(value: &[u8]) -> SecretMaterial {
        SecretMaterial::try_from(value.to_vec()).unwrap()
    }

    fn test_identity_pair() -> (ManagedSecretIdentity, String) {
        let identity = age::x25519::Identity::generate();
        let encoded = identity.to_string().expose_secret().to_string();
        let recipient = identity.to_public();
        (
            ManagedSecretIdentity {
                identity,
                recipient,
            },
            encoded,
        )
    }

    fn identity_from_text(encoded: &str) -> ManagedSecretIdentity {
        ManagedSecretIdentity::from_material(&secret(encoded.as_bytes())).unwrap()
    }

    struct StoreFixture {
        root: PathBuf,
    }

    impl StoreFixture {
        fn new() -> Self {
            let root = std::env::temp_dir()
                .join(format!("ahri-tre-managed-secret-test-{}", Uuid::new_v4()));
            Self { root }
        }

        fn root(&self) -> &Path {
            &self.root
        }
    }

    impl Drop for StoreFixture {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.root);
        }
    }
}
/// Whether an owning subsystem currently retains a logical reference to a
/// Managed Secret. No Secret material or storage path is exposed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ManagedSecretReferenceStatus {
    Referenced,
    Unreferenced,
}

/// Safe failure returned when a reference-owning subsystem cannot be inspected.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
#[error("Managed-secret reference inspection is unavailable")]
pub struct ManagedSecretReferenceInspectionError;

impl ManagedSecretReferenceInspectionError {
    pub fn unavailable() -> Self {
        Self
    }
}

/// Narrow capability implemented by each authoritative reference owner.
pub trait ManagedSecretReferenceAuthority: Send + Sync {
    fn reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError>;
}

/// The authoritative subsystem that blocked a Managed-secret removal proof.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ManagedSecretReferenceAuthorityKind {
    ApplicationConfiguration,
    DatastoreBindings,
    SessionRecords,
    AuthenticationArtifacts,
}

/// Disclosure-safe reason that removal proof could not be issued.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ManagedSecretRemovalProofErrorKind {
    Referenced,
    InspectionUnavailable,
}

/// A fail-closed result from the fixed Managed-secret reference checks.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
#[error("Managed-secret removal proof failed for {authority:?}: {kind:?}")]
pub struct ManagedSecretRemovalProofError {
    authority: ManagedSecretReferenceAuthorityKind,
    kind: ManagedSecretRemovalProofErrorKind,
}

impl ManagedSecretRemovalProofError {
    pub fn authority(&self) -> ManagedSecretReferenceAuthorityKind {
        self.authority
    }

    pub fn kind(&self) -> ManagedSecretRemovalProofErrorKind {
        self.kind
    }
}

/// Fixed set of reference owners that must all prove absence before removal.
pub struct ManagedSecretReferenceAuthorities<'a> {
    application_configuration: &'a dyn ManagedSecretReferenceAuthority,
    datastore_bindings: &'a dyn ManagedSecretReferenceAuthority,
    session_records: &'a dyn ManagedSecretReferenceAuthority,
    authentication_artifacts: &'a dyn ManagedSecretReferenceAuthority,
}

impl<'a> ManagedSecretReferenceAuthorities<'a> {
    pub fn new(
        application_configuration: &'a dyn ManagedSecretReferenceAuthority,
        datastore_bindings: &'a dyn ManagedSecretReferenceAuthority,
        session_records: &'a dyn ManagedSecretReferenceAuthority,
        authentication_artifacts: &'a dyn ManagedSecretReferenceAuthority,
    ) -> Self {
        Self {
            application_configuration,
            datastore_bindings,
            session_records,
            authentication_artifacts,
        }
    }

    pub fn prove_unreferenced(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretRemovalProof, ManagedSecretRemovalProofError> {
        let authorities = [
            (
                ManagedSecretReferenceAuthorityKind::ApplicationConfiguration,
                self.application_configuration,
            ),
            (
                ManagedSecretReferenceAuthorityKind::DatastoreBindings,
                self.datastore_bindings,
            ),
            (
                ManagedSecretReferenceAuthorityKind::SessionRecords,
                self.session_records,
            ),
            (
                ManagedSecretReferenceAuthorityKind::AuthenticationArtifacts,
                self.authentication_artifacts,
            ),
        ];
        for (authority, inspector) in authorities {
            let status = inspector.reference_status(reference).map_err(|_| {
                ManagedSecretRemovalProofError {
                    authority,
                    kind: ManagedSecretRemovalProofErrorKind::InspectionUnavailable,
                }
            })?;
            if status == ManagedSecretReferenceStatus::Referenced {
                return Err(ManagedSecretRemovalProofError {
                    authority,
                    kind: ManagedSecretRemovalProofErrorKind::Referenced,
                });
            }
        }
        Ok(ManagedSecretRemovalProof {
            reference: reference.clone(),
        })
    }
}

/// Unforgeable evidence that every required reference authority reported
/// absence for one canonical Managed-secret reference.
#[derive(Debug)]
pub struct ManagedSecretRemovalProof {
    reference: ManagedSecretReference,
}