saddle-core 0.2.0-rc.20

use std::sync::atomic::{AtomicU64, Ordering};

#[cfg(test)]
use std::ffi::CString;

use ring::signature::{ED25519, UnparsedPublicKey};
#[cfg(test)]
use rustix::fs::{self, AtFlags, Mode, OFlags, StatxFlags};
use serde::Deserialize;
use sha2::{Digest, Sha256};

use crate::deployment_applicability::VerifiedDeploymentDirectoryAuthority;

static BUNDLE_GENERATION: AtomicU64 = AtomicU64::new(1);
static FINAL_DIRECTORY_WHOLE_NONCE: AtomicU64 = AtomicU64::new(1);
#[cfg(test)]
static DIRECTORY_DEPLOYMENT_GENERATION: AtomicU64 = AtomicU64::new(0);

/// Opaque result of checking only the signed deployment subject. Directory
/// bytes and filesystem facts are deliberately not materialized in Core.
///
/// The former raw issuer/pair/parts chain is not in the production export
/// surface.
///
/// ```compile_fail
/// use saddle_core::{
///     ObservabilityDirectoryParts, VerifiedFilesystemServiceIdentityOwner,
///     VerifiedSignedDirectoryIdentityOwner, verify_and_issue_signed_directory_owners,
///     verify_observability_bundle_identity,
/// };
/// ```
pub struct VerifiedSignedDirectorySubjectContinuation {
    pub(crate) authority: VerifiedDeploymentDirectoryAuthority,
}

impl VerifiedSignedDirectorySubjectContinuation {
    pub(crate) fn pair_supervisor_startup_recovering(
        self,
        seed: crate::VerifiedDeploymentSupervisorStartupSeed,
    ) -> Result<
        (Self, crate::VerifiedDeploymentSupervisorStartupOwner),
        (Self, crate::VerifiedDeploymentSupervisorStartupSeed),
    > {
        match self.authority.pair_supervisor_startup_recovering(seed) {
            Ok((authority, supervisor)) => Ok((Self { authority }, supervisor)),
            Err((authority, seed)) => Err((Self { authority }, seed)),
        }
    }

    pub(crate) fn restore_supervisor_startup(
        self,
        supervisor: crate::VerifiedDeploymentSupervisorStartupOwner,
    ) -> (Self, crate::VerifiedDeploymentSupervisorStartupSeed) {
        let (authority, seed) = self.authority.restore_supervisor_startup(supervisor);
        (Self { authority }, seed)
    }

    pub(crate) fn pair_scheduler_observation_recovering(
        self,
        seed: crate::VerifiedDeploymentSchedulerObservationSeed,
    ) -> Result<
        (Self, crate::VerifiedDeploymentSchedulerObservationOwner),
        (Self, crate::VerifiedDeploymentSchedulerObservationSeed),
    > {
        match self.authority.pair_scheduler_observation_recovering(seed) {
            Ok((authority, scheduler)) => Ok((Self { authority }, scheduler)),
            Err((authority, seed)) => Err((Self { authority }, seed)),
        }
    }

    pub(crate) fn restore_scheduler_observation(
        self,
        scheduler: crate::VerifiedDeploymentSchedulerObservationOwner,
    ) -> (Self, crate::VerifiedDeploymentSchedulerObservationSeed) {
        let (authority, seed) = self.authority.restore_scheduler_observation(scheduler);
        (Self { authority }, seed)
    }

    pub(crate) fn pair_gate_build_recovering(
        self,
        seed: crate::VerifiedDeploymentGateBuildAuthoritySeed,
    ) -> Result<
        (Self, crate::VerifiedDeploymentGateBuildOwner),
        (Self, crate::VerifiedDeploymentGateBuildAuthoritySeed),
    > {
        match self.authority.pair_gate_build_recovering(seed) {
            Ok((authority, gate)) => Ok((Self { authority }, gate)),
            Err((authority, seed)) => Err((Self { authority }, seed)),
        }
    }

    pub(crate) fn restore_gate_build(
        self,
        gate: crate::VerifiedDeploymentGateBuildOwner,
    ) -> (Self, crate::VerifiedDeploymentGateBuildAuthoritySeed) {
        let (authority, seed) = self.authority.restore_gate_build(gate);
        (Self { authority }, seed)
    }

    pub(crate) fn pair_startup_time_policy_recovering(
        self,
        seed: crate::VerifiedDeploymentStartupAuthoritySeed,
    ) -> Result<
        (Self, crate::VerifiedStartupTimePolicyOwner),
        (Self, crate::VerifiedDeploymentStartupAuthoritySeed),
    > {
        match self.authority.pair_startup_time_policy_recovering(seed) {
            Ok((authority, time)) => Ok((Self { authority }, time)),
            Err((authority, seed)) => Err((Self { authority }, seed)),
        }
    }

    pub(crate) fn restore_startup_time_policy(
        self,
        time: crate::VerifiedStartupTimePolicyOwner,
    ) -> (Self, crate::VerifiedDeploymentStartupAuthoritySeed) {
        let (authority, seed) = self.authority.restore_startup_time_policy(time);
        (Self { authority }, seed)
    }

    pub(crate) fn deployment_binding_view(&self) -> ([u8; 32], [u8; 32], u64, [u8; 32]) {
        self.authority.deployment_binding_view()
    }
}

/// Opaque same-bundle half paired later with Admission's queue authority.
pub struct ObservabilityBundleIdentityContinuation {
    binding: BundleBinding,
}

/// Verified identity half of one filesystem service observation.
///
/// It deliberately has no public constructor, `Clone`, raw accessor, serde or
/// open adapter trait. The deployment verifier is the only production issuer.
#[cfg(test)]
pub(crate) struct VerifiedFilesystemServiceIdentityOwner {
    target_linux_x86_64: bool,
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    build: [u8; 32],
    issuer_binding: [u8; 32],
}

/// Verified identity and canonical stat payload for one signed directory.
#[cfg(test)]
pub(crate) struct VerifiedSignedDirectoryIdentityOwner<const PATH: usize> {
    bytes: [u8; PATH],
    len: usize,
    provider: [u8; 32],
    deployment: [u8; 32],
    deployment_generation: u64,
    build: [u8; 32],
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    path: [u8; 32],
    device: u64,
    inode: u64,
    mode: u32,
    uid: u32,
    gid: u32,
    mount_id: u64,
    mount_binding: [u8; 32],
    issuer_binding: [u8; 32],
}

/// Linear rollback authority retained until Observability has atomically
/// assembled the concrete production filesystem bundle.
#[cfg(test)]
pub(crate) struct SignedDirectoryIssuerContinuation {
    authority: VerifiedDeploymentDirectoryAuthority,
    issuer_binding: [u8; 32],
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SignedDirectoryIssuerError {
    Encoding,
    Signature,
    Foreign,
    Directory,
    TocTou,
    Replay,
}

#[allow(clippy::result_large_err)]
pub fn verify_signed_directory_subject(
    authority: VerifiedDeploymentDirectoryAuthority,
    evidence: &[u8],
    signature: &[u8],
) -> Result<
    VerifiedSignedDirectorySubjectContinuation,
    (
        SignedDirectoryIssuerError,
        VerifiedDeploymentDirectoryAuthority,
    ),
> {
    let view = authority.directory_view();
    if !canonical_json(evidence)
        || UnparsedPublicKey::new(&ED25519, view.public_key)
            .verify(evidence, signature)
            .is_err()
    {
        return Err((SignedDirectoryIssuerError::Signature, authority));
    }
    let doc: SignedDirectoryDocument = match serde_json::from_slice(evidence) {
        Ok(doc) => doc,
        Err(_) => return Err((SignedDirectoryIssuerError::Encoding, authority)),
    };
    let identity = |value: &str| decode_identity(value);
    if doc.schema != "saddle-f08-signed-directory/1"
        || doc.signature_domain != hex_identity(&sha256(b"saddle/f08/signed-directory/v1"))
        || doc.usage != hex_identity(&sha256(b"observability.log-directory.identity"))
        || identity(&doc.key_identity) != Some(view.key)
        || identity(&doc.bundle_identity) != Some(view.bundle)
        || identity(&doc.provider_identity) != Some(view.provider)
        || identity(&doc.application_identity) != Some(view.application)
        || identity(&doc.binary_identity) != Some(view.binary)
        || identity(&doc.build_identity) != Some(view.build)
        || identity(&doc.source_identity) != Some(view.source)
        || identity(&doc.artifact_identity) != Some(view.artifact)
        || identity(&doc.routes_identity) != Some(view.routes)
        || identity(&doc.deployment_identity) != Some(view.deployment)
        || identity(&doc.environment_identity) != Some(view.environment)
        || identity(&doc.nonce_identity) != Some(view.nonce)
        || identity(&doc.calibration_identity) != Some(view.calibration)
        || doc.deployment_generation != view.generation
    {
        return Err((SignedDirectoryIssuerError::Foreign, authority));
    }
    Ok(VerifiedSignedDirectorySubjectContinuation { authority })
}

#[allow(clippy::too_many_arguments, clippy::result_large_err)]
pub fn issue_observability_bundle_identity(
    subject: VerifiedSignedDirectorySubjectContinuation,
    provider: [u8; 32],
    deployment: [u8; 32],
    deployment_generation: u64,
    build: [u8; 32],
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    path: [u8; 32],
    mount_binding: [u8; 32],
) -> Result<
    (
        VerifiedSignedDirectorySubjectContinuation,
        ObservabilityBundleIdentityContinuation,
        ObservabilityBundleIdentityAuthority,
    ),
    VerifiedSignedDirectorySubjectContinuation,
> {
    if [
        provider,
        deployment,
        build,
        environment,
        filesystem,
        mount,
        service,
        path,
        mount_binding,
    ]
    .contains(&[0; 32])
        || deployment_generation == 0
    {
        return Err(subject);
    }
    let generation =
        match BUNDLE_GENERATION.fetch_update(Ordering::AcqRel, Ordering::Acquire, |value| {
            value.checked_add(1)
        }) {
            Ok(generation) => generation,
            Err(_) => return Err(subject),
        };
    let binding = BundleBinding {
        generation,
        provider,
        deployment,
        deployment_generation,
        build,
        environment,
        filesystem,
        mount,
        service,
        path,
        mount_binding,
    };
    Ok((
        subject,
        ObservabilityBundleIdentityContinuation {
            binding: duplicate_binding(&binding),
        },
        ObservabilityBundleIdentityAuthority { binding },
    ))
}

#[allow(clippy::result_large_err)]
pub fn verify_observability_bundle_binding(
    authority: ObservabilityBundleIdentityAuthority,
    continuation: ObservabilityBundleIdentityContinuation,
) -> Result<
    (),
    (
        ObservabilityBundleIdentityAuthority,
        ObservabilityBundleIdentityContinuation,
    ),
> {
    if authority.binding == continuation.binding {
        Ok(())
    } else {
        Err((authority, continuation))
    }
}

#[allow(clippy::too_many_arguments)]
pub fn verify_signed_directory_subject_deployment_pair(
    subject: VerifiedSignedDirectorySubjectContinuation,
    bundle: [u8; 32],
    provider: [u8; 32],
    application: [u8; 32],
    binary: [u8; 32],
    build: [u8; 32],
    source: [u8; 32],
    artifact: [u8; 32],
    routes: [u8; 32],
    deployment: [u8; 32],
    environment: [u8; 32],
    nonce: [u8; 32],
    calibration: [u8; 32],
) -> Result<VerifiedSignedDirectorySubjectContinuation, VerifiedSignedDirectorySubjectContinuation>
{
    match subject.authority.verify_deployment_pair([
        bundle,
        provider,
        application,
        binary,
        build,
        source,
        artifact,
        routes,
        deployment,
        environment,
        nonce,
        calibration,
    ]) {
        Ok(authority) => Ok(VerifiedSignedDirectorySubjectContinuation { authority }),
        Err(authority) => Err(VerifiedSignedDirectorySubjectContinuation { authority }),
    }
}

pub fn pair_signed_directory_subject_generation(
    subject: VerifiedSignedDirectorySubjectContinuation,
    generated: crate::VerifiedGeneratedComponentGenerationOwner,
) -> Result<
    (
        VerifiedSignedDirectorySubjectContinuation,
        crate::VerifiedComponentGenerationPairingReceipt,
    ),
    (
        VerifiedSignedDirectorySubjectContinuation,
        crate::VerifiedGeneratedComponentGenerationOwner,
    ),
> {
    match subject.authority.pair_generation(generated) {
        Ok((authority, receipt)) => Ok((
            VerifiedSignedDirectorySubjectContinuation { authority },
            receipt,
        )),
        Err((authority, generated)) => Err((
            VerifiedSignedDirectorySubjectContinuation { authority },
            generated,
        )),
    }
}

#[allow(clippy::result_large_err)]
pub fn pair_signed_directory_subject_generation_v2(
    subject: VerifiedSignedDirectorySubjectContinuation,
    generated: crate::VerifiedGeneratedComponentGenerationV2Owner,
) -> Result<
    (
        VerifiedSignedDirectorySubjectContinuation,
        crate::VerifiedGeneratedComponentGenerationV2Owner,
        crate::VerifiedComponentGenerationPairingReceipt,
    ),
    (
        VerifiedSignedDirectorySubjectContinuation,
        crate::VerifiedGeneratedComponentGenerationV2Owner,
    ),
> {
    let transaction = crate::begin_generated_component_generation_v2_final_transaction(generated);
    match transaction.pair_subject_recovering(subject) {
        Ok((transaction, subject, receipt)) => Ok((
            subject,
            crate::restore_generated_component_generation_v2_final_transaction(transaction),
            receipt,
        )),
        Err((transaction, subject)) => Err((
            subject,
            crate::restore_generated_component_generation_v2_final_transaction(transaction),
        )),
    }
}

#[allow(dead_code)]
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SignedDirectoryDocument {
    schema: String,
    signature_domain: String,
    usage: String,
    key_identity: String,
    bundle_identity: String,
    provider_identity: String,
    application_identity: String,
    binary_identity: String,
    build_identity: String,
    source_identity: String,
    artifact_identity: String,
    routes_identity: String,
    deployment_identity: String,
    environment_identity: String,
    nonce_identity: String,
    calibration_identity: String,
    deployment_generation: u64,
    filesystem_identity: String,
    mount_identity: String,
    service_identity: String,
    canonical_unix_bytes_hex: String,
    canonical_unix_bytes_len: usize,
    st_dev: u64,
    st_ino: u64,
    st_mode: u32,
    st_uid: u32,
    st_gid: u32,
    mount_id: u64,
    mount_root_hex: String,
    mount_point_hex: String,
    filesystem_type_hex: String,
}

#[allow(clippy::too_many_arguments, clippy::result_large_err)]
#[cfg(test)]
pub fn verify_and_issue_signed_directory_owners<const PATH: usize>(
    authority: VerifiedDeploymentDirectoryAuthority,
    evidence: &[u8],
    signature: &[u8],
    target_linux_x86_64: bool,
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    build: [u8; 32],
) -> Result<
    (
        SignedDirectoryIssuerContinuation,
        VerifiedFilesystemServiceIdentityOwner,
        VerifiedSignedDirectoryIdentityOwner<PATH>,
    ),
    (
        SignedDirectoryIssuerError,
        VerifiedDeploymentDirectoryAuthority,
    ),
> {
    macro_rules! fail {
        ($error:expr) => {
            return Err(($error, authority))
        };
    }
    let view = authority.directory_view();
    if !canonical_json(evidence)
        || UnparsedPublicKey::new(&ED25519, view.public_key)
            .verify(evidence, signature)
            .is_err()
    {
        fail!(SignedDirectoryIssuerError::Signature);
    }
    let doc: SignedDirectoryDocument = match serde_json::from_slice(evidence) {
        Ok(value) => value,
        Err(_) => fail!(SignedDirectoryIssuerError::Encoding),
    };
    let identity = |value: &str| decode_identity(value);
    if doc.schema != "saddle-f08-signed-directory/1"
        || doc.signature_domain != hex_identity(&sha256(b"saddle/f08/signed-directory/v1"))
        || doc.usage != hex_identity(&sha256(b"observability.log-directory.identity"))
        || identity(&doc.key_identity) != Some(view.key)
        || identity(&doc.bundle_identity) != Some(view.bundle)
        || identity(&doc.provider_identity) != Some(view.provider)
        || identity(&doc.application_identity) != Some(view.application)
        || identity(&doc.binary_identity) != Some(view.binary)
        || identity(&doc.build_identity) != Some(view.build)
        || identity(&doc.source_identity) != Some(view.source)
        || identity(&doc.artifact_identity) != Some(view.artifact)
        || identity(&doc.routes_identity) != Some(view.routes)
        || identity(&doc.deployment_identity) != Some(view.deployment)
        || identity(&doc.environment_identity) != Some(view.environment)
        || identity(&doc.nonce_identity) != Some(view.nonce)
        || identity(&doc.calibration_identity) != Some(view.calibration)
        || doc.deployment_generation != view.generation
        || environment != view.environment
        || build != view.build
        || identity(&doc.filesystem_identity) != Some(filesystem)
        || identity(&doc.mount_identity) != Some(mount)
        || identity(&doc.service_identity) != Some(service)
        || [filesystem, mount, service].contains(&[0; 32])
        || !target_linux_x86_64
    {
        fail!(SignedDirectoryIssuerError::Foreign);
    }
    let path = match decode_hex(&doc.canonical_unix_bytes_hex) {
        Some(value) => value,
        None => fail!(SignedDirectoryIssuerError::Encoding),
    };
    if path.len() != doc.canonical_unix_bytes_len || path.len() >= PATH {
        fail!(SignedDirectoryIssuerError::Directory);
    }
    let mut bytes = [0; PATH];
    bytes[..path.len()].copy_from_slice(&path);
    if !is_canonical_absolute_directory(&bytes, path.len()) {
        fail!(SignedDirectoryIssuerError::Directory);
    }
    if !live_directory_matches(&path, &doc) {
        fail!(SignedDirectoryIssuerError::TocTou);
    }
    if DIRECTORY_DEPLOYMENT_GENERATION
        .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
            (view.generation > current).then_some(view.generation)
        })
        .is_err()
    {
        fail!(SignedDirectoryIssuerError::Replay);
    }
    let mount_material = [
        doc.mount_root_hex.as_bytes(),
        doc.mount_point_hex.as_bytes(),
        doc.filesystem_type_hex.as_bytes(),
        &doc.mount_id.to_le_bytes(),
    ]
    .concat();
    let issuer_binding = sha256(
        &[
            view.bundle.as_slice(),
            view.key.as_slice(),
            filesystem.as_slice(),
            mount.as_slice(),
            service.as_slice(),
            path.as_slice(),
        ]
        .concat(),
    );
    Ok((
        SignedDirectoryIssuerContinuation {
            authority,
            issuer_binding,
        },
        VerifiedFilesystemServiceIdentityOwner {
            target_linux_x86_64,
            environment,
            filesystem,
            mount,
            service,
            build,
            issuer_binding,
        },
        VerifiedSignedDirectoryIdentityOwner {
            bytes,
            len: path.len(),
            provider: view.provider,
            deployment: view.deployment,
            deployment_generation: view.generation,
            build,
            environment,
            filesystem,
            mount,
            service,
            path: sha256(&path),
            device: doc.st_dev,
            inode: doc.st_ino,
            mode: doc.st_mode,
            uid: doc.st_uid,
            gid: doc.st_gid,
            mount_id: doc.mount_id,
            mount_binding: sha256(&mount_material),
            issuer_binding,
        },
    ))
}

#[cfg(test)]
pub(crate) fn signed_directory_owners_match(
    continuation: &SignedDirectoryIssuerContinuation,
    service: &VerifiedFilesystemServiceIdentityOwner,
    directory: &VerifiedSignedDirectoryIdentityOwner<256>,
) -> bool {
    continuation.issuer_binding != [0; 32]
        && continuation.issuer_binding == service.issuer_binding
        && continuation.issuer_binding == directory.issuer_binding
}

#[cfg(test)]
pub(crate) fn consume_signed_directory_issuer_continuation(
    continuation: SignedDirectoryIssuerContinuation,
) -> crate::VerifiedDeploymentBundleApplicabilityOwner {
    continuation.authority.into_applicability()
}

#[cfg(test)]
fn live_directory_matches(path: &[u8], doc: &SignedDirectoryDocument) -> bool {
    let Ok(path) = CString::new(path) else {
        return false;
    };
    let Ok(fd) = fs::openat(
        fs::CWD,
        path.as_c_str(),
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
    ) else {
        return false;
    };
    let Ok(stat) = fs::fstat(&fd) else {
        return false;
    };
    let Ok(statx) = fs::statx(&fd, c"", AtFlags::EMPTY_PATH, StatxFlags::MNT_ID) else {
        return false;
    };
    stat.st_dev == doc.st_dev
        && stat.st_ino == doc.st_ino
        && stat.st_mode == doc.st_mode
        && stat.st_uid == doc.st_uid
        && stat.st_gid == doc.st_gid
        && statx.stx_mnt_id == doc.mount_id
}
fn canonical_json(value: &[u8]) -> bool {
    serde_json::from_slice::<serde_json::Value>(value)
        .ok()
        .and_then(|v| serde_json::to_vec(&v).ok())
        .is_some_and(|v| v == value)
}
fn sha256(value: &[u8]) -> [u8; 32] {
    Sha256::digest(value).into()
}
fn decode_identity(value: &str) -> Option<[u8; 32]> {
    decode_hex(value)?.try_into().ok()
}
fn decode_hex(value: &str) -> Option<Vec<u8>> {
    if value.len() % 2 != 0 {
        return None;
    }
    value
        .as_bytes()
        .chunks_exact(2)
        .map(|p| u8::from_str_radix(std::str::from_utf8(p).ok()?, 16).ok())
        .collect()
}
fn hex_identity(value: &[u8; 32]) -> String {
    value.iter().map(|b| format!("{b:02x}")).collect()
}

/// Physical identity continuation. Only Observability may immediately consume
/// its minimal canonical directory payload after the bundle verifier returns.
#[cfg(test)]
pub(crate) struct ObservabilityDirectoryIdentityContinuation<const PATH: usize> {
    parts: ObservabilityDirectoryParts<PATH>,
    binding: BundleBinding,
}

/// Minimal inline payload required by Observability's open/fstat/statx stage.
#[cfg(test)]
pub(crate) struct ObservabilityDirectoryParts<const PATH: usize> {
    bytes: [u8; PATH],
    len: usize,
    device: u64,
    inode: u64,
    mode: u32,
    uid: u32,
    gid: u32,
    mount_id: u64,
}

/// Single-use same-bundle authority carried opaquely through Admission.
///
/// ```compile_fail
/// use saddle_core::ObservabilityBundleIdentityAuthority;
/// fn needs_clone<T: Clone>() {}
/// needs_clone::<ObservabilityBundleIdentityAuthority>();
/// ```
///
/// ```compile_fail
/// use saddle_core::ObservabilityBundleIdentityAuthority;
/// let _ = ObservabilityBundleIdentityAuthority { binding: panic!() };
/// ```
///
/// Even an external consumer built with every published feature cannot reach a
/// fixture issuer (the crate publishes no such feature or module).
///
/// ```compile_fail
/// let _ = saddle_core::test_support::filesystem_service(
///     [1; 32], [2; 32], [3; 32], [4; 32], [5; 32],
/// );
/// ```
pub struct ObservabilityBundleIdentityAuthority {
    binding: BundleBinding,
}

#[derive(Eq, PartialEq)]
struct BundleBinding {
    generation: u64,
    provider: [u8; 32],
    deployment: [u8; 32],
    deployment_generation: u64,
    build: [u8; 32],
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    path: [u8; 32],
    mount_binding: [u8; 32],
}

#[derive(Eq, PartialEq)]
pub(crate) struct FinalDirectoryBinding {
    generation: u64,
    deployment_generation: u64,
    build: [u8; 32],
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    path: [u8; 32],
}

#[cfg(test)]
impl FinalDirectoryBinding {
    pub(crate) fn is_fully_bound(&self) -> bool {
        self.generation != 0
            && self.deployment_generation != 0
            && self.build != [0; 32]
            && self.environment != [0; 32]
            && self.filesystem != [0; 32]
            && self.mount != [0; 32]
            && self.service != [0; 32]
            && self.path != [0; 32]
    }
}

fn final_directory_binding(binding: &BundleBinding) -> FinalDirectoryBinding {
    FinalDirectoryBinding {
        generation: binding.generation,
        deployment_generation: binding.deployment_generation,
        build: binding.build,
        environment: binding.environment,
        filesystem: binding.filesystem,
        mount: binding.mount,
        service: binding.service,
        path: binding.path,
    }
}

/// Atomically proves that the subject and physical continuation came from the
/// same verified Observability whole. The original bundle halves are returned
/// on both success and failure; only opaque, non-clone receipts are added.
#[doc(hidden)]
#[allow(clippy::result_large_err)]
pub fn issue_final_directory_receipts(
    subject: VerifiedSignedDirectorySubjectContinuation,
    authority: ObservabilityBundleIdentityAuthority,
    continuation: ObservabilityBundleIdentityContinuation,
) -> Result<
    (
        ObservabilityBundleIdentityAuthority,
        ObservabilityBundleIdentityContinuation,
        crate::VerifiedFinalDirectoryReceiptPair,
    ),
    (
        VerifiedSignedDirectorySubjectContinuation,
        ObservabilityBundleIdentityAuthority,
        ObservabilityBundleIdentityContinuation,
    ),
> {
    let deployment = subject.deployment_binding_view();
    if authority.binding != continuation.binding
        || authority.binding.deployment_generation != deployment.2
        || authority.binding.build != deployment.3
    {
        return Err((subject, authority, continuation));
    }
    let whole_nonce = match FINAL_DIRECTORY_WHOLE_NONCE.fetch_update(
        Ordering::AcqRel,
        Ordering::Acquire,
        |nonce| nonce.checked_add(1),
    ) {
        Ok(nonce) if nonce != 0 => nonce,
        _ => return Err((subject, authority, continuation)),
    };
    let physical_binding = final_directory_binding(&authority.binding);
    let subject_binding = final_directory_binding(&authority.binding);
    let body_binding = final_directory_binding(&authority.binding);
    Ok((
        authority,
        continuation,
        crate::VerifiedFinalDirectoryReceiptPair {
            directory: crate::VerifiedFinalDirectoryReceipt {
                whole_nonce,
                binding: physical_binding,
                pair_nonce: 0,
            },
            subject: crate::VerifiedFinalSubjectReceipt {
                subject,
                whole_nonce,
                binding: subject_binding,
                pair_nonce: 0,
            },
            body: crate::deployment_applicability::VerifiedFinalDirectoryBodyBinding {
                whole_nonce,
                binding: body_binding,
            },
        },
    ))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg(test)]
pub(crate) enum ObservabilityIdentityError {
    InvalidIdentity,
    InvalidDirectory,
    ForeignBundle,
    Overflow,
}

/// Consuming comparison against an independently verified Observability work
/// proof. It reveals no field from the identity owner.
#[allow(clippy::too_many_arguments)]
#[allow(clippy::result_large_err)]
#[cfg(test)]
pub fn verify_filesystem_service_identity(
    owner: VerifiedFilesystemServiceIdentityOwner,
    target_linux_x86_64: bool,
    environment: [u8; 32],
    filesystem: [u8; 32],
    mount: [u8; 32],
    service: [u8; 32],
    build: [u8; 32],
) -> Result<VerifiedFilesystemServiceIdentityOwner, VerifiedFilesystemServiceIdentityOwner> {
    if owner.target_linux_x86_64 == target_linux_x86_64
        && owner.environment == environment
        && owner.filesystem == filesystem
        && owner.mount == mount
        && owner.service == service
        && owner.build == build
    {
        Ok(owner)
    } else {
        Err(owner)
    }
}

#[allow(clippy::result_large_err)]
#[cfg(test)]
pub fn verify_observability_bundle_identity<const PATH: usize>(
    service: VerifiedFilesystemServiceIdentityOwner,
    directory: VerifiedSignedDirectoryIdentityOwner<PATH>,
) -> Result<
    (
        ObservabilityDirectoryIdentityContinuation<PATH>,
        ObservabilityBundleIdentityAuthority,
    ),
    (
        ObservabilityIdentityError,
        VerifiedFilesystemServiceIdentityOwner,
        VerifiedSignedDirectoryIdentityOwner<PATH>,
    ),
> {
    let invalid = service.target_linux_x86_64
        && service.environment != [0; 32]
        && service.filesystem != [0; 32]
        && service.mount != [0; 32]
        && service.service != [0; 32]
        && service.build != [0; 32]
        && directory.provider != [0; 32]
        && directory.deployment != [0; 32]
        && directory.deployment_generation != 0
        && directory.path != [0; 32]
        && directory.mount_binding != [0; 32]
        && directory.environment == service.environment
        && directory.filesystem == service.filesystem
        && directory.mount == service.mount
        && directory.service == service.service
        && directory.build == service.build
        && service.issuer_binding != [0; 32]
        && service.issuer_binding == directory.issuer_binding;
    if !invalid {
        return Err((
            ObservabilityIdentityError::InvalidIdentity,
            service,
            directory,
        ));
    }
    if !is_canonical_absolute_directory(&directory.bytes, directory.len)
        || directory.len >= PATH
        || directory.device == 0
        || directory.inode == 0
        || directory.mode == 0
        || directory.mount_id == 0
    {
        return Err((
            ObservabilityIdentityError::InvalidDirectory,
            service,
            directory,
        ));
    }
    let generation =
        match BUNDLE_GENERATION.fetch_update(Ordering::AcqRel, Ordering::Acquire, |value| {
            value.checked_add(1)
        }) {
            Ok(generation) => generation,
            Err(_) => return Err((ObservabilityIdentityError::Overflow, service, directory)),
        };
    let binding = BundleBinding {
        generation,
        provider: directory.provider,
        deployment: directory.deployment,
        deployment_generation: directory.deployment_generation,
        build: directory.build,
        environment: directory.environment,
        filesystem: directory.filesystem,
        mount: directory.mount,
        service: directory.service,
        path: directory.path,
        mount_binding: directory.mount_binding,
    };
    let continuation = ObservabilityDirectoryIdentityContinuation {
        parts: ObservabilityDirectoryParts {
            bytes: directory.bytes,
            len: directory.len,
            device: directory.device,
            inode: directory.inode,
            mode: directory.mode,
            uid: directory.uid,
            gid: directory.gid,
            mount_id: directory.mount_id,
        },
        binding: duplicate_binding(&binding),
    };
    Ok((
        continuation,
        ObservabilityBundleIdentityAuthority { binding },
    ))
}

#[cfg(test)]
fn is_canonical_absolute_directory<const PATH: usize>(bytes: &[u8; PATH], len: usize) -> bool {
    if len == 0 || len > PATH || bytes[0] != b'/' || bytes[..len].contains(&0) {
        return false;
    }
    if len == 1 {
        return true;
    }
    if bytes[1] == b'/' || bytes[len - 1] == b'/' {
        return false;
    }
    !bytes[..len]
        .split(|byte| *byte == b'/')
        .skip(1)
        .any(|component| component == b"." || component == b".." || component.is_empty())
}

#[allow(clippy::result_large_err)]
#[cfg(test)]
pub fn verify_observability_bundle_pair<const PATH: usize>(
    authority: ObservabilityBundleIdentityAuthority,
    continuation: ObservabilityDirectoryIdentityContinuation<PATH>,
) -> Result<
    ObservabilityDirectoryParts<PATH>,
    (
        ObservabilityIdentityError,
        ObservabilityBundleIdentityAuthority,
        ObservabilityDirectoryIdentityContinuation<PATH>,
    ),
> {
    if authority.binding != continuation.binding {
        return Err((
            ObservabilityIdentityError::ForeignBundle,
            authority,
            continuation,
        ));
    }
    Ok(continuation.parts)
}

#[cfg(test)]
impl<const PATH: usize> ObservabilityDirectoryParts<PATH> {
    /// A1 consuming transfer; no borrowed/raw identity accessor exists.
    #[doc(hidden)]
    pub fn into_observability_parts(self) -> ([u8; PATH], usize, u64, u64, u32, u32, u32, u64) {
        (
            self.bytes,
            self.len,
            self.device,
            self.inode,
            self.mode,
            self.uid,
            self.gid,
            self.mount_id,
        )
    }
}

fn duplicate_binding(binding: &BundleBinding) -> BundleBinding {
    BundleBinding {
        generation: binding.generation,
        provider: binding.provider,
        deployment: binding.deployment,
        deployment_generation: binding.deployment_generation,
        build: binding.build,
        environment: binding.environment,
        filesystem: binding.filesystem,
        mount: binding.mount,
        service: binding.service,
        path: binding.path,
        mount_binding: binding.mount_binding,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ring::{
        rand::SystemRandom,
        signature::{Ed25519KeyPair, KeyPair},
    };
    use std::os::unix::fs::MetadataExt;

    fn service(marker: u8) -> VerifiedFilesystemServiceIdentityOwner {
        VerifiedFilesystemServiceIdentityOwner {
            target_linux_x86_64: true,
            environment: [marker; 32],
            filesystem: [marker.wrapping_add(1); 32],
            mount: [marker.wrapping_add(2); 32],
            service: [marker.wrapping_add(3); 32],
            build: [marker.wrapping_add(4); 32],
            issuer_binding: [marker.wrapping_add(9); 32],
        }
    }

    fn directory(marker: u8) -> VerifiedSignedDirectoryIdentityOwner<64> {
        let mut bytes = [0_u8; 64];
        bytes[..15].copy_from_slice(b"/var/log/saddle");
        VerifiedSignedDirectoryIdentityOwner {
            bytes,
            len: 15,
            provider: [marker.wrapping_add(5); 32],
            deployment: [marker.wrapping_add(6); 32],
            deployment_generation: u64::from(marker),
            build: [marker.wrapping_add(4); 32],
            environment: [marker; 32],
            filesystem: [marker.wrapping_add(1); 32],
            mount: [marker.wrapping_add(2); 32],
            service: [marker.wrapping_add(3); 32],
            path: [marker.wrapping_add(7); 32],
            device: u64::from(marker),
            inode: u64::from(marker) + 1,
            mode: 0o040700,
            uid: 1,
            gid: 1,
            mount_id: u64::from(marker) + 2,
            mount_binding: [marker.wrapping_add(8); 32],
            issuer_binding: [marker.wrapping_add(9); 32],
        }
    }

    fn issuer_failure<const PATH: usize>(
        result: Result<
            (
                SignedDirectoryIssuerContinuation,
                VerifiedFilesystemServiceIdentityOwner,
                VerifiedSignedDirectoryIdentityOwner<PATH>,
            ),
            (
                SignedDirectoryIssuerError,
                VerifiedDeploymentDirectoryAuthority,
            ),
        >,
    ) -> (
        SignedDirectoryIssuerError,
        VerifiedDeploymentDirectoryAuthority,
    ) {
        match result {
            Err(value) => value,
            Ok(_) => panic!("issuer must fail"),
        }
    }

    fn directory_authority(
        applicability: crate::VerifiedDeploymentBundleApplicabilityOwner,
    ) -> VerifiedDeploymentDirectoryAuthority {
        let (subject, _resource, permit) =
            match crate::consume_deployment_binding_split(applicability) {
                Ok(parts) => parts,
                Err(_) => panic!("fixture applicability must split"),
            };
        match crate::pair_deployment_directory_permit(subject, permit) {
            Ok(authority) => authority,
            Err(_) => panic!("same split directory permit must pair"),
        }
    }

    fn resource_and_subject(
        marker: u8,
        generation: u64,
    ) -> (
        crate::VerifiedRuntimeResourceParts,
        VerifiedSignedDirectorySubjectContinuation,
        crate::VerifiedDeploymentStartupAuthoritySeed,
        crate::VerifiedDeploymentGateBuildAuthoritySeed,
        crate::VerifiedDeploymentSchedulerObservationSeed,
        crate::VerifiedDeploymentSupervisorStartupSeed,
    ) {
        let applicability = crate::deployment_applicability::fixture_directory_applicability(
            [marker; 32],
            generation,
        );
        let (subject, resource, directory) = crate::consume_deployment_binding_split(applicability)
            .unwrap_or_else(|_| panic!("fixture must split"));
        let parts = crate::deployment_applicability::fixture_runtime_resource_parts(resource);
        let authority = crate::pair_deployment_directory_permit(subject, directory)
            .unwrap_or_else(|_| panic!("same split must pair"));
        (
            parts,
            VerifiedSignedDirectorySubjectContinuation { authority },
            crate::deployment_applicability::fixture_startup_time_seed(generation),
            crate::deployment_applicability::fixture_gate_build_seed(generation),
            crate::deployment_applicability::fixture_scheduler_observation_seed(generation),
            crate::deployment_applicability::fixture_supervisor_startup_seed(generation),
        )
    }

    fn final_inputs(
        marker: u8,
        generation: u64,
    ) -> (
        crate::VerifiedRuntimeResourceParts,
        crate::VerifiedFinalDirectoryReceiptPair,
        crate::VerifiedGeneratedComponentGenerationV2FinalTransaction,
        crate::VerifiedDeploymentStartupAuthoritySeed,
        crate::VerifiedDeploymentGateBuildAuthoritySeed,
        crate::VerifiedDeploymentSchedulerObservationSeed,
        crate::VerifiedDeploymentSupervisorStartupSeed,
    ) {
        let (parts, subject, seed, gate_seed, scheduler_seed, supervisor_seed) =
            resource_and_subject(marker, generation);
        let deployment = subject.deployment_binding_view();
        let binding = BundleBinding {
            generation,
            provider: [marker.wrapping_add(1); 32],
            deployment: [marker.wrapping_add(2); 32],
            deployment_generation: deployment.2,
            build: deployment.3,
            environment: [marker.wrapping_add(3); 32],
            filesystem: [marker.wrapping_add(4); 32],
            mount: [marker.wrapping_add(5); 32],
            service: [marker.wrapping_add(6); 32],
            path: [marker.wrapping_add(7); 32],
            mount_binding: [marker.wrapping_add(8); 32],
        };
        let authority = ObservabilityBundleIdentityAuthority {
            binding: duplicate_binding(&binding),
        };
        let continuation = ObservabilityBundleIdentityContinuation { binding };
        let (_authority, _continuation, directory_pair) =
            issue_final_directory_receipts(subject, authority, continuation)
                .unwrap_or_else(|_| panic!("same whole must issue receipts"));
        (
            parts,
            directory_pair,
            crate::generation_pair::fixture_upgrade_generated_generation_v2(
                crate::generation_pair::fixture_generated_generation_half(generation),
            ),
            seed,
            gate_seed,
            scheduler_seed,
            supervisor_seed,
        )
    }

    #[test]
    fn signed_directory_v2_cross_rejects_both_directions_and_originals_retry() {
        let (_, subject_a, _, _, _, _) = resource_and_subject(41, 41);
        let (_, subject_b, _, _, _, _) = resource_and_subject(42, 42);
        let generated_a = crate::generation_pair::fixture_generated_generation_v2_owner(41);
        let generated_b = crate::generation_pair::fixture_generated_generation_v2_owner(42);

        let (subject_a, generated_b) =
            match pair_signed_directory_subject_generation_v2(subject_a, generated_b) {
                Err(owners) => owners,
                Ok(_) => panic!("A subject must reject B generation"),
            };
        let (subject_b, generated_a) =
            match pair_signed_directory_subject_generation_v2(subject_b, generated_a) {
                Err(owners) => owners,
                Ok(_) => panic!("B subject must reject A generation"),
            };
        assert!(pair_signed_directory_subject_generation_v2(subject_a, generated_a).is_ok());
        assert!(pair_signed_directory_subject_generation_v2(subject_b, generated_b).is_ok());
    }

    pub(super) fn final_receipts(
        marker: u8,
        generation: u64,
    ) -> crate::VerifiedFinalPairingReceipts {
        let (parts, directory_pair, generated, seed, gate_seed, scheduler_seed, supervisor_seed) =
            final_inputs(marker, generation);
        crate::pair_final_startup_receipts(
            parts,
            directory_pair,
            generated,
            seed,
            gate_seed,
            scheduler_seed,
            supervisor_seed,
        )
        .unwrap_or_else(|_| panic!("same deployment receipts must pair"))
    }

    #[test]
    fn final_pair_rejects_crossed_time_seed_then_original_outers_retry() {
        let (
            first_parts,
            first_directory,
            first_generated,
            first_seed,
            first_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        ) = final_inputs(0xe1, 111);
        let (
            second_parts,
            second_directory,
            second_generated,
            second_seed,
            second_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        ) = final_inputs(0xe2, 112);
        let (
            first_parts,
            first_directory,
            first_generated,
            second_seed,
            first_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        ) = crate::pair_final_startup_receipts(
            first_parts,
            first_directory,
            first_generated,
            second_seed,
            first_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        )
        .err()
        .unwrap_or_else(|| panic!("A outer plus B seed must reject"));
        let (
            second_parts,
            second_directory,
            second_generated,
            first_seed,
            second_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        ) = crate::pair_final_startup_receipts(
            second_parts,
            second_directory,
            second_generated,
            first_seed,
            second_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        )
        .err()
        .unwrap_or_else(|| panic!("B outer plus A seed must reject"));
        assert!(
            crate::pair_final_startup_receipts(
                first_parts,
                first_directory,
                first_generated,
                first_seed,
                first_gate_seed,
                first_scheduler_seed,
                first_supervisor_seed,
            )
            .is_ok()
        );
        assert!(
            crate::pair_final_startup_receipts(
                second_parts,
                second_directory,
                second_generated,
                second_seed,
                second_gate_seed,
                second_scheduler_seed,
                second_supervisor_seed,
            )
            .is_ok()
        );
    }

    #[test]
    fn final_pair_rejects_crossed_gate_seed_then_original_outers_retry() {
        let (
            first_parts,
            first_directory,
            first_generated,
            first_seed,
            first_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        ) = final_inputs(0xe3, 113);
        let (
            second_parts,
            second_directory,
            second_generated,
            second_seed,
            second_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        ) = final_inputs(0xe4, 114);
        let (
            first_parts,
            first_directory,
            first_generated,
            first_seed,
            second_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        ) = crate::pair_final_startup_receipts(
            first_parts,
            first_directory,
            first_generated,
            first_seed,
            second_gate_seed,
            first_scheduler_seed,
            first_supervisor_seed,
        )
        .err()
        .unwrap_or_else(|| panic!("A outer plus B GateBuild seed must reject"));
        let (
            second_parts,
            second_directory,
            second_generated,
            second_seed,
            first_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        ) = crate::pair_final_startup_receipts(
            second_parts,
            second_directory,
            second_generated,
            second_seed,
            first_gate_seed,
            second_scheduler_seed,
            second_supervisor_seed,
        )
        .err()
        .unwrap_or_else(|| panic!("B outer plus A GateBuild seed must reject"));
        assert!(
            crate::pair_final_startup_receipts(
                first_parts,
                first_directory,
                first_generated,
                first_seed,
                first_gate_seed,
                first_scheduler_seed,
                first_supervisor_seed,
            )
            .is_ok()
        );
        assert!(
            crate::pair_final_startup_receipts(
                second_parts,
                second_directory,
                second_generated,
                second_seed,
                second_gate_seed,
                second_scheduler_seed,
                second_supervisor_seed,
            )
            .is_ok()
        );
    }

    #[test]
    fn final_pair_rejects_crossed_scheduler_seed_then_original_outers_retry() {
        let (a_parts, a_directory, a_generated, a_time, a_gate, a_scheduler, a_supervisor) =
            final_inputs(0xe5, 115);
        let (b_parts, b_directory, b_generated, b_time, b_gate, b_scheduler, b_supervisor) =
            final_inputs(0xe6, 116);
        let (a_parts, a_directory, a_generated, a_time, a_gate, b_scheduler, a_supervisor) =
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                a_generated,
                a_time,
                a_gate,
                b_scheduler,
                a_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("A outer plus B scheduler seed must reject"));
        let (b_parts, b_directory, b_generated, b_time, b_gate, a_scheduler, b_supervisor) =
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                b_generated,
                b_time,
                b_gate,
                a_scheduler,
                b_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("B outer plus A scheduler seed must reject"));
        assert!(
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                a_generated,
                a_time,
                a_gate,
                a_scheduler,
                a_supervisor,
            )
            .is_ok()
        );
        assert!(
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                b_generated,
                b_time,
                b_gate,
                b_scheduler,
                b_supervisor,
            )
            .is_ok()
        );
    }

    #[test]
    fn final_pair_rejects_crossed_supervisor_seed_then_seven_outers_retry() {
        let (a_parts, a_directory, a_generated, a_time, a_gate, a_scheduler, a_supervisor) =
            final_inputs(0xe7, 117);
        let (b_parts, b_directory, b_generated, b_time, b_gate, b_scheduler, b_supervisor) =
            final_inputs(0xe8, 118);
        let (a_parts, a_directory, a_generated, a_time, a_gate, a_scheduler, b_supervisor) =
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                a_generated,
                a_time,
                a_gate,
                a_scheduler,
                b_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("A outer plus B supervisor seed must reject"));
        let (b_parts, b_directory, b_generated, b_time, b_gate, b_scheduler, a_supervisor) =
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                b_generated,
                b_time,
                b_gate,
                b_scheduler,
                a_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("B outer plus A supervisor seed must reject"));
        assert!(
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                a_generated,
                a_time,
                a_gate,
                a_scheduler,
                a_supervisor,
            )
            .is_ok()
        );
        assert!(
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                b_generated,
                b_time,
                b_gate,
                b_scheduler,
                b_supervisor,
            )
            .is_ok()
        );
    }

    #[test]
    fn generation_failure_restores_supervisor_then_all_seven_outers_retry() {
        let (a_parts, a_directory, a_generated, a_time, a_gate, a_scheduler, a_supervisor) =
            final_inputs(0xe9, 119);
        let (b_parts, b_directory, b_generated, b_time, b_gate, b_scheduler, b_supervisor) =
            final_inputs(0xea, 120);
        let (a_parts, a_directory, b_generated, a_time, a_gate, a_scheduler, a_supervisor) =
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                b_generated,
                a_time,
                a_gate,
                a_scheduler,
                a_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("generation mismatch after supervisor must reject"));
        let (b_parts, b_directory, a_generated, b_time, b_gate, b_scheduler, b_supervisor) =
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                a_generated,
                b_time,
                b_gate,
                b_scheduler,
                b_supervisor,
            )
            .err()
            .unwrap_or_else(|| panic!("reverse generation mismatch must reject"));
        assert!(
            crate::pair_final_startup_receipts(
                a_parts,
                a_directory,
                a_generated,
                a_time,
                a_gate,
                a_scheduler,
                a_supervisor,
            )
            .is_ok()
        );
        assert!(
            crate::pair_final_startup_receipts(
                b_parts,
                b_directory,
                b_generated,
                b_time,
                b_gate,
                b_scheduler,
                b_supervisor,
            )
            .is_ok()
        );
    }

    #[test]
    fn four_final_receipts_reject_cross_then_accept_original_regrouping() {
        let first = final_receipts(0xa1, 91);
        let second = final_receipts(0xb2, 92);
        let crate::VerifiedFinalPairingReceipts {
            resource: first_resource,
            directory_pair: first_pair,
            generated: first_generated,
            time: first_time,
            gate: first_gate,
            scheduler: first_scheduler,
            supervisor: first_supervisor,
        } = first;
        let crate::VerifiedFinalPairingReceipts {
            resource: second_resource,
            directory_pair: second_pair,
            generated: second_generated,
            time: second_time,
            gate: second_gate,
            scheduler: second_scheduler,
            supervisor: second_supervisor,
        } = second;
        let crate::VerifiedFinalDirectoryReceiptPair {
            directory: first_directory,
            subject: first_subject,
            body: first_body,
        } = first_pair;
        let crate::VerifiedFinalDirectoryReceiptPair {
            directory: second_directory,
            subject: second_subject,
            body: second_body,
        } = second_pair;
        let crossed_first =
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: first_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: second_directory,
                    subject: first_subject,
                    body: first_body,
                },
                generated: first_generated,
                time: first_time,
                gate: first_gate,
                scheduler: first_scheduler,
                supervisor: first_supervisor,
            })
            .err()
            .unwrap_or_else(|| panic!("crossed receipt must reject"));
        let crossed_second =
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: second_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: first_directory,
                    subject: second_subject,
                    body: second_body,
                },
                generated: second_generated,
                time: second_time,
                gate: second_gate,
                scheduler: second_scheduler,
                supervisor: second_supervisor,
            })
            .err()
            .unwrap_or_else(|| panic!("reverse crossed receipt must reject"));
        let crate::VerifiedFinalPairingReceipts {
            resource: first_resource,
            directory_pair: crossed_first_pair,
            generated: first_generated,
            time: first_time,
            gate: first_gate,
            scheduler: first_scheduler,
            supervisor: first_supervisor,
        } = crossed_first;
        let crate::VerifiedFinalPairingReceipts {
            resource: second_resource,
            directory_pair: crossed_second_pair,
            generated: second_generated,
            time: second_time,
            gate: second_gate,
            scheduler: second_scheduler,
            supervisor: second_supervisor,
        } = crossed_second;
        assert!(
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: first_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: crossed_second_pair.directory,
                    subject: crossed_first_pair.subject,
                    body: crossed_first_pair.body,
                },
                generated: first_generated,
                time: first_time,
                gate: first_gate,
                scheduler: first_scheduler,
                supervisor: first_supervisor,
            })
            .is_ok()
        );
        assert!(
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: second_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: crossed_first_pair.directory,
                    subject: crossed_second_pair.subject,
                    body: crossed_second_pair.body,
                },
                generated: second_generated,
                time: second_time,
                gate: second_gate,
                scheduler: second_scheduler,
                supervisor: second_supervisor,
            })
            .is_ok()
        );
    }

    #[test]
    fn final_directory_receipts_reject_crossed_body_then_original_groups_retry() {
        let first = final_receipts(0xc1, 101);
        let second = final_receipts(0xd2, 102);
        let crate::VerifiedFinalPairingReceipts {
            resource: first_resource,
            directory_pair: first_pair,
            generated: first_generated,
            time: first_time,
            gate: first_gate,
            scheduler: first_scheduler,
            supervisor: first_supervisor,
        } = first;
        let crate::VerifiedFinalPairingReceipts {
            resource: second_resource,
            directory_pair: second_pair,
            generated: second_generated,
            time: second_time,
            gate: second_gate,
            scheduler: second_scheduler,
            supervisor: second_supervisor,
        } = second;
        let crate::VerifiedFinalDirectoryReceiptPair {
            directory: first_directory,
            subject: first_subject,
            body: first_body,
        } = first_pair;
        let crate::VerifiedFinalDirectoryReceiptPair {
            directory: second_directory,
            subject: second_subject,
            body: second_body,
        } = second_pair;
        let first = crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
            resource: first_resource,
            directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                directory: first_directory,
                subject: first_subject,
                body: second_body,
            },
            generated: first_generated,
            time: first_time,
            gate: first_gate,
            scheduler: first_scheduler,
            supervisor: first_supervisor,
        })
        .err()
        .unwrap_or_else(|| panic!("A receipts plus B body must reject"));
        let second = crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
            resource: second_resource,
            directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                directory: second_directory,
                subject: second_subject,
                body: first_body,
            },
            generated: second_generated,
            time: second_time,
            gate: second_gate,
            scheduler: second_scheduler,
            supervisor: second_supervisor,
        })
        .err()
        .unwrap_or_else(|| panic!("B receipts plus A body must reject"));
        let crate::VerifiedFinalPairingReceipts {
            resource: first_resource,
            directory_pair: first_pair,
            generated: first_generated,
            time: first_time,
            gate: first_gate,
            scheduler: first_scheduler,
            supervisor: first_supervisor,
        } = first;
        let crate::VerifiedFinalPairingReceipts {
            resource: second_resource,
            directory_pair: second_pair,
            generated: second_generated,
            time: second_time,
            gate: second_gate,
            scheduler: second_scheduler,
            supervisor: second_supervisor,
        } = second;
        assert!(
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: first_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: first_pair.directory,
                    subject: first_pair.subject,
                    body: second_pair.body,
                },
                generated: first_generated,
                time: first_time,
                gate: first_gate,
                scheduler: first_scheduler,
                supervisor: first_supervisor,
            })
            .is_ok()
        );
        assert!(
            crate::consume_final_pairing_receipts(crate::VerifiedFinalPairingReceipts {
                resource: second_resource,
                directory_pair: crate::VerifiedFinalDirectoryReceiptPair {
                    directory: second_pair.directory,
                    subject: second_pair.subject,
                    body: first_pair.body,
                },
                generated: second_generated,
                time: second_time,
                gate: second_gate,
                scheduler: second_scheduler,
                supervisor: second_supervisor,
            })
            .is_ok()
        );
    }

    #[test]
    fn finished_directory_seal_retains_creator_binding_after_atomic_consume() {
        let receipts = final_receipts(0xe3, 103);
        let parts = match crate::consume_final_pairing_receipts(receipts) {
            Ok(parts) => parts,
            Err(_) => panic!("same-whole receipts must consume"),
        };
        assert!(parts.filesystem_creator_binding_is_retained());
    }

    #[test]
    fn two_valid_bundles_cannot_cross_pair_and_failures_return_both_owners() {
        let (first_continuation, first_authority) =
            match verify_observability_bundle_identity(service(11), directory(11)) {
                Ok(value) => value,
                Err(_) => panic!("first bundle is valid"),
            };
        let (second_continuation, second_authority) =
            match verify_observability_bundle_identity(service(21), directory(21)) {
                Ok(value) => value,
                Err(_) => panic!("second bundle is valid"),
            };

        assert_ne!(first_authority.binding.generation, 0);
        assert_ne!(second_authority.binding.generation, 0);
        assert_ne!(first_continuation.binding.path, [0; 32]);
        assert_ne!(second_continuation.binding.path, [0; 32]);

        let (_, first_authority, second_continuation) =
            match verify_observability_bundle_pair(first_authority, second_continuation) {
                Ok(_) => panic!("crossed bundle must fail"),
                Err(value) => value,
            };
        let (_, second_authority, first_continuation) =
            match verify_observability_bundle_pair(second_authority, first_continuation) {
                Ok(_) => panic!("crossed bundle must fail"),
                Err(value) => value,
            };

        let first = match verify_observability_bundle_pair(first_authority, first_continuation) {
            Ok(value) => value,
            Err(_) => panic!("restored first owners must pair"),
        };
        let second = match verify_observability_bundle_pair(second_authority, second_continuation) {
            Ok(value) => value,
            Err(_) => panic!("restored second owners must pair"),
        };
        assert_ne!(
            first.into_observability_parts().2,
            second.into_observability_parts().2
        );
    }

    #[test]
    fn service_or_directory_identity_drift_returns_original_linear_inputs() {
        let service = service(11);
        let directory = directory(12);
        let (error, service, directory) =
            match verify_observability_bundle_identity(service, directory) {
                Ok(_) => panic!("drift must fail"),
                Err(value) => value,
            };
        assert_eq!(error, ObservabilityIdentityError::InvalidIdentity);
        let service = match verify_filesystem_service_identity(
            service, true, [11; 32], [12; 32], [13; 32], [14; 32], [15; 32],
        ) {
            Ok(value) => value,
            Err(_) => panic!("returned service owner is unchanged"),
        };
        assert!(verify_observability_bundle_identity(service, directory).is_err());
    }

    #[test]
    fn equal_filesystem_facts_from_different_issuers_cannot_cross_and_can_retry() {
        let first_service = service(51);
        let first_directory = directory(51);
        let mut second_service = service(51);
        let mut second_directory = directory(51);
        second_service.issuer_binding = [91; 32];
        second_directory.issuer_binding = [91; 32];

        let (_, first_service, second_directory) =
            match verify_observability_bundle_identity(first_service, second_directory) {
                Ok(_) => panic!("cross issuer service/directory must fail"),
                Err(value) => value,
            };
        let (_, second_service, first_directory) =
            match verify_observability_bundle_identity(second_service, first_directory) {
                Ok(_) => panic!("reverse cross issuer service/directory must fail"),
                Err(value) => value,
            };

        assert!(verify_observability_bundle_identity(first_service, first_directory).is_ok());
        assert!(verify_observability_bundle_identity(second_service, second_directory).is_ok());
    }

    #[test]
    fn directory_identity_requires_canonical_absolute_bytes() {
        for invalid in [
            b"/var/../log".as_slice(),
            b"/var/./log".as_slice(),
            b"/var/log/".as_slice(),
            b"//var/log".as_slice(),
        ] {
            let mut candidate = directory(31);
            candidate.bytes = [0; 64];
            candidate.bytes[..invalid.len()].copy_from_slice(invalid);
            candidate.len = invalid.len();
            let (error, service, candidate) =
                match verify_observability_bundle_identity(service(31), candidate) {
                    Ok(_) => panic!("non-canonical directory must fail"),
                    Err(value) => value,
                };
            assert_eq!(error, ObservabilityIdentityError::InvalidDirectory);
            assert!(verify_observability_bundle_identity(service, candidate).is_err());
        }

        assert!(verify_observability_bundle_identity(service(41), directory(41)).is_ok());
    }

    #[test]
    fn signed_directory_issuer_rejects_tamper_foreign_toc_tou_and_replay() {
        let rng = SystemRandom::new();
        let pkcs8 = Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
        let key = Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();
        let public: [u8; 32] = key.public_key().as_ref().try_into().unwrap();
        let generation = 30_001;
        let path =
            std::env::temp_dir().join(format!("saddle-f08-directory-{}", std::process::id()));
        let moved = path.with_extension("moved");
        let _ = std::fs::remove_dir(&path);
        let _ = std::fs::remove_dir(&moved);
        std::fs::create_dir(&path).unwrap();
        let metadata = std::fs::metadata(&path).unwrap();
        let cpath = CString::new(path.as_os_str().as_encoded_bytes()).unwrap();
        let fd = fs::openat(
            fs::CWD,
            cpath.as_c_str(),
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .unwrap();
        let statx = fs::statx(&fd, c"", AtFlags::EMPTY_PATH, StatxFlags::MNT_ID).unwrap();
        let evidence=serde_json::to_vec(&serde_json::json!({
            "schema":"saddle-f08-signed-directory/1","signature_domain":hex_identity(&sha256(b"saddle/f08/signed-directory/v1")),
            "usage":hex_identity(&sha256(b"observability.log-directory.identity")),"key_identity":hex_identity(&sha256(&public)),
            "bundle_identity":hex_identity(&[13;32]),"provider_identity":hex_identity(&[1;32]),"application_identity":hex_identity(&[2;32]),
            "binary_identity":hex_identity(&[3;32]),"build_identity":hex_identity(&[4;32]),"source_identity":hex_identity(&[5;32]),
            "artifact_identity":hex_identity(&[6;32]),"routes_identity":hex_identity(&[7;32]),"deployment_identity":hex_identity(&[8;32]),
            "environment_identity":hex_identity(&[9;32]),"nonce_identity":hex_identity(&[14;32]),"calibration_identity":hex_identity(&[15;32]),
            "deployment_generation":generation,"filesystem_identity":hex_identity(&[31;32]),"mount_identity":hex_identity(&[32;32]),
            "service_identity":hex_identity(&[33;32]),"canonical_unix_bytes_hex":path.as_os_str().as_encoded_bytes().iter().map(|b|format!("{b:02x}")).collect::<String>(),
            "canonical_unix_bytes_len":path.as_os_str().as_encoded_bytes().len(),"st_dev":metadata.dev(),"st_ino":metadata.ino(),
            "st_mode":metadata.mode(),"st_uid":metadata.uid(),"st_gid":metadata.gid(),"mount_id":statx.stx_mnt_id,
            "mount_root_hex":"2f","mount_point_hex":"2f","filesystem_type_hex":"74657374"})).unwrap();
        let signature = key.sign(&evidence);
        let authority = directory_authority(
            crate::deployment_applicability::fixture_directory_applicability(public, generation),
        );
        let (_, authority) = issuer_failure(verify_and_issue_signed_directory_owners::<256>(
            authority, &evidence, &[0; 64], true, [9; 32], [31; 32], [32; 32], [33; 32], [4; 32],
        ));
        let (_, authority) = issuer_failure(verify_and_issue_signed_directory_owners::<256>(
            authority,
            &evidence,
            signature.as_ref(),
            true,
            [9; 32],
            [31; 32],
            [32; 32],
            [99; 32],
            [4; 32],
        ));
        std::fs::rename(&path, &moved).unwrap();
        let (_, authority) = issuer_failure(verify_and_issue_signed_directory_owners::<256>(
            authority,
            &evidence,
            signature.as_ref(),
            true,
            [9; 32],
            [31; 32],
            [32; 32],
            [33; 32],
            [4; 32],
        ));
        std::fs::rename(&moved, &path).unwrap();
        let (continuation, service, directory) = match verify_and_issue_signed_directory_owners::<256>(
            authority,
            &evidence,
            signature.as_ref(),
            true,
            [9; 32],
            [31; 32],
            [32; 32],
            [33; 32],
            [4; 32],
        ) {
            Ok(value) => value,
            Err(_) => panic!("issuer must succeed"),
        };
        assert!(signed_directory_owners_match(
            &continuation,
            &service,
            &directory
        ));
        assert_ne!(continuation.issuer_binding, [0; 32]);
        assert_ne!(service.issuer_binding, [0; 32]);
        assert_ne!(directory.path, [0; 32]);
        assert_ne!(directory.device, 0);
        assert_ne!(directory.inode, 0);
        assert_ne!(directory.mount_id, 0);
        consume_signed_directory_issuer_continuation(continuation);
        let replay = directory_authority(
            crate::deployment_applicability::fixture_directory_applicability(public, generation),
        );
        assert_eq!(
            issuer_failure(verify_and_issue_signed_directory_owners::<256>(
                replay,
                &evidence,
                signature.as_ref(),
                true,
                [9; 32],
                [31; 32],
                [32; 32],
                [33; 32],
                [4; 32]
            ))
            .0,
            SignedDirectoryIssuerError::Replay
        );
        std::fs::remove_dir(&path).unwrap();
    }
}