u280_evidence_oracle/

lib.rs

//! Independent cryptographic promotion gate for sealed U280 Phase-B evidence.
//!
//! This crate consumes, but never modifies, the archive emitted by the C++ XRT
//! transport runner. It recomputes every SHA-256-profile WOTS signature with
//! the software reference implementation pinned by this workspace, verifies
//! the complete transport envelope, and writes a separately sealed result.
//! It also verifies and binds the producer's canonical coordination chain and
//! native-XRT card identity chain. Coordination is a structurally authenticated
//! procedural ledger assertion, not cryptographic ledger proof or global
//! nonrollback evidence. A successful result is still not a hardware-completion
//! claim: route, measured performance, ownership, replay uniqueness, and final
//! promotion belong to a later combiner.
//! A distinct entry point in this package consumes only the deliberately
//! non-hardware mock-HBM source-simulation format and cannot accept this
//! transport archive in its place.
#![forbid(unsafe_code)]

mod archive;
mod model;
mod sealed_result;
mod source;

pub use sealed_result::{
    VerifiedClockObservation, VerifiedHardwareCardIdentity, VerifiedHardwareXclbinIdentity,
    VerifiedOracleReceiptCapture, VerifiedPerformanceObservations, VerifiedPerformanceSample,
    VerifiedRequestArtifactBinding, VerifiedResultArtifactBinding, VerifiedResultRequest,
    VerifiedRoutedClockEvidenceBinding, VerifiedSealedResult, VerifiedXclbinBinding,
    VerifiedXrtBuildIdentity, completed_result_receipt_line, open_verified_oracle_receipt_capture,
    verify_typed_sealed_result,
};
pub use source::{SourceOracleConfig, SourceOracleFailure, SourceOracleOutcome, execute_source};

use anyhow::{Context, Result, ensure};
use archive::{
    ArchiveSnapshot, CheckedFile, FileRecord, ResultDirectory, canonical_transport_root,
    create_exclusive_file, finish_writer, hex_lower, is_lower_hex, open_verified_file,
    prepare_result_directory, read_verified_small, seal_result_directory, sha256_bytes,
    verify_sealed_manifest, verify_transport_archive, verify_transport_archive_again,
    write_exclusive,
};
use hashsigs_reference::{
    FUSED_OUTPUT_BYTES, HashSigsSha256GenericV1, MessageDigest, PrivateSeed, SIGNATURE_BYTES,
    Sha256GenericWots,
};
use model::{
    ArtifactBinding, BatchResult, CaptureCoreAcceptance, ClockBinding, CoordinationBurn,
    CoordinationBurnGrant, CoordinationGrant, CoordinationIntent, CoordinationReceipt, Diagnostics,
    DigestBinding, FinalHardwareIdentity, HardwarePlatformIdentity, OracleRequest, Percentiles,
    PerformanceAcceptance, PerformancePolicyBasis, PreRunHardwareIdentity, ProgrammedImageIdentity,
    RatePercentiles, ReservationReceipt, ResultsEnvelope, RootArtifactBinding, RunMetadata,
    SummaryDecoded, TimingRecord, TransportContract, ValidationRecord, WholeKernelAcceptance,
    Word32, Word64, XrtBuildIdentity, XrtPlatformReport, parse_canonical_json, parse_strict_json,
    parse_xrt_report_json,
};
use serde::Serialize;
use sha2::{Digest as _, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::thread;

const PROFILE: &str = "HASHSIGS_SHA256_GENERIC_V1";
const REQUEST_KIND: &str = "hashsigs_phase_b_rust_oracle_request";
const RESULT_KIND: &str = "hashsigs_phase_b_rust_oracle_result";
/// Schema emitted by a completed Phase-B hardware-oracle comparison.
pub const PHASE_B_TYPED_RESULT_SCHEMA: u64 = 3;
/// Closed result contract consumed by [`verify_typed_sealed_result`].
pub const PHASE_B_TYPED_RESULT_CONTRACT: &str = "hashsigs-u280-phase-b-typed-result-v1";
/// CLI receipt schema that explicitly binds the typed result contract.
pub const PHASE_B_ORACLE_RECEIPT_SCHEMA: u64 = 2;
const INPUT_BYTES_PER_JOB: u64 = 64;
const INPUT_BYTES_PER_JOB_USIZE: usize = 64;
const OUTPUT_SLOT_BYTES: u64 = 4_096;
const OUTPUT_SLOT_BYTES_USIZE: usize = 4_096;
const OUTPUT_VALID_BYTES: u64 = 2_208;
const OUTPUT_VALID_BYTES_USIZE: usize = 2_208;
const OUTPUT_PADDING_BYTES: u64 = 1_888;
const SUMMARY_BYTES: u64 = 64;
const MAXIMUM_BATCH: u32 = 65_535;
const MAXIMUM_JSON_BYTES: u64 = 16 * 1024 * 1024;
const ABI_WORD: u32 = 0x4853_0001;
const LAST_STATUS_SUCCESS: u32 = 0x0f00_0000;
const DEFAULT_KERNEL: &str = "hashsigs_sha256_hbm_kernel:{hashsigs_sha256_hbm_kernel_1}";
const TIMED_REGION: &str = "xrt::kernel submission through run.wait completion";
const OUTPUT_POISON: &str = "a5 xor ((offset*131+5b) mod 256)";
const SUMMARY_POISON: &str = "c3 xor ((offset*29) mod 256)";
const RATE_POLICY: &str = "requested clock is metadata; routed rates remain non-promotable until the external Rust oracle validates every output";
const PERFORMANCE_BLOCKING_REASON: &str = "Phase-B transport performance is non-promotable until the independent Rust oracle validates every output and binds the sealed archive";
const REQUIRED_PERFORMANCE_BATCH: u32 = 16_384;
const REQUIRED_PERFORMANCE_SAMPLES: usize = 30;
const PERFORMANCE_CLOCK_BOUND_HZ: u64 = 250_000_000;
const MAXIMUM_P95_CAPTURE_CYCLES_EXACT: u64 = 500;
const MAXIMUM_P95_CAPTURE_CYCLES: f64 = 500.0;
const MINIMUM_SIGNATURES_PER_SECOND_EXACT: u64 = 500_000;
const MINIMUM_SIGNATURES_PER_SECOND: f64 = 500_000.0;
const EXPECTED_DEVICE_NAME: &str = "xilinx_u280_gen3x16_xdma_base_1";
const EXPECTED_DEPLOYMENT_PLATFORM: &str = "xilinx_u280_gen3x16_xdma_1_202211_1";
const EXPECTED_FPGA_PART: &str = "xcu280-fsvh2892-2L-e";
const EXPECTED_INTERFACE_UUID: &str = "fb2b2c5a-19ed-6359-3fea-95f51fbc8eb9";
const EXPECTED_LOGIC_UUID: &str = "283bab8f-654d-8674-968f-4da57f7fa5d7";
const PRE_RUN_IDENTITY_CONTRACT: &str = "hashsigs-u280-phase-b-hardware-identity-pre-run-v1";
const FINAL_IDENTITY_CONTRACT: &str = "hashsigs-u280-phase-b-hardware-identity-v1";
const PRE_RUN_IDENTITY_FILE: &str = "hardware-identity-pre-run.json";
const FINAL_IDENTITY_FILE: &str = "hardware-identity.json";
const PLATFORM_PRE_LOAD_FILE: &str = "xrt-platform-pre-load.json";
const PLATFORM_POST_RUN_FILE: &str = "xrt-platform-post-run.json";
const XRT_HOST_FILE: &str = "xrt-host.json";
const MAXIMUM_PLATFORM_REPORT_BYTES: u64 = 4 * 1024 * 1024;
const MAXIMUM_XRT_HOST_REPORT_BYTES: u64 = 64 * 1024;
const MAXIMUM_COORDINATION_RECORD_BYTES: u64 = 64 * 1024;
const COORDINATION_INTENT_FILE: &str = "coordination-intent.json";
const COORDINATION_BURN_FILE: &str = "coordination-burn.json";
const COORDINATION_RECEIPT_FILE: &str = "coordination-receipt.json";
const COORDINATION_INTENT_CONTRACT: &str = "hashsigs-u280-phase-b-coordination-intent-v1";
const COORDINATION_BURN_CONTRACT: &str = "hashsigs-u280-phase-b-coordination-burn-v1";
const COORDINATION_RECEIPT_CONTRACT: &str = "hashsigs-u280-phase-b-coordination-receipt-v2";
const COORDINATION_OWNER: &str = "hashsigs-rhdl";
const COORDINATION_COMMAND_CLASS: &str = "phase-b-xrt-card";
const COORDINATION_LOCK_CONTEXT: &str = "descriptor-v1";
const COORDINATION_LOCK_ACQUISITION: &str = "nonblocking-exclusive-flock";
const COORDINATION_LOCK_SUFFIX: &str = "/.cache/rhdl-fpga/u280.lock";
const COORDINATION_LEDGER_IDS: [&str; 2] = [
    "hashsigs-project-coordination",
    "ed25519-agent-coordination",
];
const RESULT_EVIDENCE_BOUNDARY: &str = "cryptographic payload, immutable transport binding, a structurally authenticated procedural ledger assertion, and unpromoted native-XRT card identity only; this is not cryptographic ledger proof or global nonrollback evidence, and route, performance, ownership, replay uniqueness, and final hardware completion require a later combiner";

const SAMPLE_FILES: [&str; 11] = [
    "diagnostics.json",
    "input.bin",
    "oracle-request.json",
    "oracle-request.sha256",
    "output.bin",
    "reservation.json",
    "summary-decoded.json",
    "summary.bin",
    "timing.json",
    "transport.json",
    "validation.json",
];

/// Configuration for one immutable archive validation.
#[derive(Clone, Debug)]
pub struct OracleConfig {
    /// Directory containing the transport runner's top-level `SHA256SUMS`.
    pub transport_root: PathBuf,
    /// Brand-new directory in which the oracle will write separate evidence.
    pub result_directory: PathBuf,
    /// Maximum number of scoped software-oracle workers.
    pub threads: usize,
}

/// Completed result returned after the separate directory has been sealed.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OracleOutcome {
    /// `true` only when every requested job and every padding region matched.
    pub cryptographic_oracle_valid: bool,
    /// Number of oracle-request samples processed.
    pub samples: usize,
    /// Number of WOTS jobs recomputed.
    pub jobs: u64,
    /// SHA-256 of the result directory's own `SHA256SUMS`.
    pub result_manifest_sha256: String,
    /// Exact byte count of the result directory's own `SHA256SUMS`.
    pub result_manifest_bytes: u64,
    /// Canonical path of the separately created result directory.
    pub result_directory: PathBuf,
}

/// A structural validation failure whose separate evidence directory was
/// successfully sealed and can therefore be anchored by the CLI.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OracleFailure {
    /// Canonical path of the separately created result directory.
    pub result_directory: PathBuf,
    /// SHA-256 of the result directory's own `SHA256SUMS`.
    pub result_manifest_sha256: String,
    /// Exact byte count of the result directory's own `SHA256SUMS`.
    pub result_manifest_bytes: u64,
    /// Original structural validation error, rendered with its context chain.
    pub failure: String,
}

impl std::fmt::Display for OracleFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "{}; failure evidence sealed in {}",
            self.failure,
            self.result_directory.display()
        )
    }
}

impl std::error::Error for OracleFailure {}

/// Revalidates every entry in a sealed result and requires the external
/// manifest byte-count/digest receipt captured when that result was created.
///
/// # Errors
///
/// Returns an error for a malformed receipt, any manifest/file mismatch, or a
/// result directory that no longer matches the externally captured receipt.
pub fn verify_sealed_result(
    result_directory: &Path,
    result_manifest_bytes: u64,
    result_manifest_sha256: &str,
) -> Result<()> {
    ensure!(
        is_lower_hex(result_manifest_sha256, 64),
        "result manifest SHA-256 is not canonical"
    );
    verify_sealed_manifest(
        result_directory,
        &FileRecord {
            bytes: result_manifest_bytes,
            sha256: result_manifest_sha256.to_owned(),
        },
    )
}

#[derive(Clone, Debug)]
struct SampleSpec {
    relative: String,
    batch: u32,
    phase: &'static str,
    index: usize,
}

#[derive(Clone, Debug)]
struct JobComputation {
    transcript: JobTranscript,
    seed_sha256: [u8; 32],
    padding_match: bool,
}

#[derive(Clone, Debug, Serialize)]
struct JobTranscript {
    schema: u64,
    sample: String,
    job_index: u32,
    input_sha256: String,
    expected_sha256: String,
    observed_sha256: String,
    r#match: bool,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
struct ManifestBinding {
    file: String,
    bytes: u64,
    sha256: String,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
struct ResultXclbinBinding {
    path: String,
    bytes: u64,
    sha256: String,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
struct ResultRequestArtifactBinding {
    file: String,
    logical_bytes: u64,
    stored_bytes: u64,
    sha256: String,
}

#[derive(Clone, Debug, Serialize)]
struct RequestResult {
    sample: String,
    phase: &'static str,
    sample_index: usize,
    request_sha256: String,
    run_id: String,
    coordination_intent: ManifestBinding,
    coordination_burn: ManifestBinding,
    coordination_receipt: ManifestBinding,
    batch: u32,
    nonce_hex: String,
    reservation_record_sha256: String,
    xclbin: ResultXclbinBinding,
    hardware_identity_pre_run: ManifestBinding,
    input: ResultRequestArtifactBinding,
    output: ResultRequestArtifactBinding,
    jobs: u32,
    matched_jobs: u32,
    padding_valid: bool,
    cryptographic_oracle_valid: bool,
}

#[derive(Clone, Debug, Serialize)]
// Flat booleans preserve the published success-evidence JSON schema.
#[allow(clippy::struct_excessive_bools)]
struct SuccessSummary {
    schema: u64,
    kind: &'static str,
    verifier_contract: &'static str,
    profile: &'static str,
    producer: &'static str,
    pinned_hashsigs_rs_commit: &'static str,
    transport_manifest: ManifestBinding,
    run_id: String,
    coordination_intent: ManifestBinding,
    coordination_burn: ManifestBinding,
    coordination_receipt: ManifestBinding,
    final_hardware_identity: ManifestBinding,
    worker_threads: usize,
    sample_count: usize,
    total_jobs: u64,
    matched_jobs: u64,
    mismatched_jobs: u64,
    all_padding_valid: bool,
    requests: Vec<RequestResult>,
    cryptographic_oracle_valid: bool,
    card_identity_promotable: bool,
    performance_promotable: bool,
    hardware_completion_promotable: bool,
    evidence_boundary: &'static str,
}

#[derive(Clone, Debug, Serialize)]
struct FailureSummary {
    schema: u64,
    kind: &'static str,
    profile: &'static str,
    transport_manifest: Option<ManifestBinding>,
    cryptographic_oracle_valid: Option<bool>,
    card_identity_promotable: bool,
    performance_promotable: bool,
    hardware_completion_promotable: bool,
    failure: String,
}

#[derive(Clone, Debug)]
struct ValidationOutput {
    summary: SuccessSummary,
    jobs_record: FileRecord,
}

#[derive(Clone, Copy, Debug)]
struct SampleMetrics {
    batch: u32,
    kernel_nanoseconds: u64,
    memory_span: u64,
    capture_span: u64,
}

#[derive(Clone, Debug)]
struct ValidatedHardwareIdentity {
    pre_run_record: FileRecord,
    final_record: FileRecord,
    pre_run: PreRunHardwareIdentity,
}

#[derive(Clone, Debug)]
struct ValidatedCoordination {
    intent: CoordinationIntent,
    receipt: CoordinationReceipt,
    intent_record: FileRecord,
    burn_record: FileRecord,
    receipt_record: FileRecord,
}

#[derive(Serialize)]
struct CoordinationBurnHashBody<'a> {
    schema: u64,
    contract: &'a str,
    sequence: u64,
    run_id: &'a str,
    intent: &'a DigestBinding,
    grants: &'a [CoordinationBurnGrant],
    previous_sha256: &'a str,
}

#[derive(Debug)]
struct ProcessedSamples {
    jobs_record: FileRecord,
    request_results: Vec<RequestResult>,
    measured_metrics: Vec<SampleMetrics>,
    total_jobs: u64,
    matched_jobs: u64,
    all_padding_valid: bool,
}

#[derive(Clone, Copy, Debug)]
struct SampleComputationSummary {
    matched_jobs: u32,
    padding_valid: bool,
}

/// Validates a sealed Phase-B archive and creates a distinct sealed result.
///
/// # Errors
///
/// Returns an error for malformed archives, filesystem races, transport
/// contract failures, or result-write failures. When the result directory can
/// be created safely, structural failures leave `failure.json` and a result
/// `SHA256SUMS` behind for audit and return an [`OracleFailure`] that carries
/// the external manifest receipt. Cryptographic mismatches produce a complete
/// summary with `cryptographic_oracle_valid:false` and return `Ok` so callers
/// can distinguish a completed negative comparison from malformed evidence.
pub fn execute(config: &OracleConfig) -> Result<OracleOutcome> {
    ensure!(
        (1..=256).contains(&config.threads),
        "worker thread count must be between 1 and 256"
    );
    let transport_root = canonical_transport_root(&config.transport_root)?;
    let result_directory = prepare_result_directory(&transport_root, &config.result_directory)?;
    let mut transport_manifest = None;
    let operation: Result<OracleOutcome> = (|| {
        let snapshot = verify_transport_archive(&transport_root)?;
        transport_manifest = Some(manifest_binding(&snapshot.manifest));
        let output = validate_archive(&snapshot, &result_directory, config.threads)?;
        let valid = output.summary.cryptographic_oracle_valid;
        let samples = output.summary.sample_count;
        let jobs = output.summary.total_jobs;
        let summary = canonical_json_line(&output.summary)?;
        let summary_record = write_exclusive(&result_directory, "summary.json", &summary)?;
        let expected = BTreeMap::from([
            ("jobs.jsonl".to_owned(), output.jobs_record),
            ("summary.json".to_owned(), summary_record),
        ]);
        let result_manifest = seal_result_directory(&result_directory, &expected)?;
        Ok(OracleOutcome {
            cryptographic_oracle_valid: valid,
            samples,
            jobs,
            result_manifest_sha256: result_manifest.sha256,
            result_manifest_bytes: result_manifest.bytes,
            result_directory: result_directory.path.clone(),
        })
    })();

    match operation {
        Ok(outcome) => Ok(outcome),
        Err(error) => {
            for name in ["SHA256SUMS", "jobs.jsonl", "summary.json"] {
                if let Err(cleanup_error) = result_directory.remove_if_exists(name) {
                    return Err(error.context(format!(
                        "also failed to remove partial result {name}: {cleanup_error:#}"
                    )));
                }
            }
            let failure = FailureSummary {
                schema: 1,
                kind: "hashsigs_phase_b_rust_oracle_failure",
                profile: PROFILE,
                transport_manifest,
                cryptographic_oracle_valid: None,
                card_identity_promotable: false,
                performance_promotable: false,
                hardware_completion_promotable: false,
                failure: format!("{error:#}"),
            };
            let write_result = canonical_json_line(&failure).and_then(|bytes| {
                let record = write_exclusive(&result_directory, "failure.json", &bytes)?;
                let expected = BTreeMap::from([("failure.json".to_owned(), record)]);
                seal_result_directory(&result_directory, &expected)
            });
            let result_manifest = match write_result {
                Ok(record) => record,
                Err(evidence_error) => {
                    return Err(error.context(format!(
                        "also failed to seal failure evidence: {evidence_error:#}"
                    )));
                }
            };
            Err(OracleFailure {
                result_directory: result_directory.path.clone(),
                result_manifest_sha256: result_manifest.sha256,
                result_manifest_bytes: result_manifest.bytes,
                failure: format!("{error:#}"),
            }
            .into())
        }
    }
}

fn validate_archive(
    snapshot: &ArchiveSnapshot,
    result_directory: &ResultDirectory,
    threads: usize,
) -> Result<ValidationOutput> {
    let coordination = validate_coordination_records(snapshot)?;
    let metadata: RunMetadata = parse_archive_json(snapshot, "run-metadata.json")?;
    validate_run_metadata(&metadata, &coordination)?;
    let hardware_identity = validate_hardware_identity(snapshot, &metadata)?;
    validate_coordination_hardware(&coordination, &hardware_identity)?;
    let results: ResultsEnvelope = parse_archive_json(snapshot, "results.json")?;
    validate_results(
        &results,
        &metadata,
        &hardware_identity.final_record,
        &coordination,
    )?;
    let samples = expected_layout(snapshot, &metadata)?;
    let processed = process_samples(
        snapshot,
        result_directory,
        threads,
        &metadata,
        &samples,
        &hardware_identity.pre_run_record,
        &coordination,
    )?;
    validate_results_statistics(&results, &metadata, &processed.measured_metrics)?;

    let after = verify_transport_archive_again(snapshot)?;
    ensure!(
        &after == snapshot,
        "transport archive changed during oracle execution"
    );
    let cryptographic_oracle_valid =
        processed.matched_jobs == processed.total_jobs && processed.all_padding_valid;
    let mismatched_jobs = processed
        .total_jobs
        .checked_sub(processed.matched_jobs)
        .context("mismatch counter underflow")?;
    Ok(ValidationOutput {
        jobs_record: processed.jobs_record,
        summary: SuccessSummary {
            schema: PHASE_B_TYPED_RESULT_SCHEMA,
            kind: RESULT_KIND,
            verifier_contract: PHASE_B_TYPED_RESULT_CONTRACT,
            profile: PROFILE,
            producer: "independent Rust Sha256GenericWots oracle",
            pinned_hashsigs_rs_commit: "2d315dd4168804b7cbc51c51a1bf7ca27bf74140",
            transport_manifest: manifest_binding(&snapshot.manifest),
            run_id: coordination.intent.run_id.clone(),
            coordination_intent: named_binding(
                COORDINATION_INTENT_FILE,
                &coordination.intent_record,
            ),
            coordination_burn: named_binding(COORDINATION_BURN_FILE, &coordination.burn_record),
            coordination_receipt: named_binding(
                COORDINATION_RECEIPT_FILE,
                &coordination.receipt_record,
            ),
            final_hardware_identity: named_binding(
                FINAL_IDENTITY_FILE,
                &hardware_identity.final_record,
            ),
            worker_threads: threads,
            sample_count: samples.len(),
            total_jobs: processed.total_jobs,
            matched_jobs: processed.matched_jobs,
            mismatched_jobs,
            all_padding_valid: processed.all_padding_valid,
            requests: processed.request_results,
            cryptographic_oracle_valid,
            card_identity_promotable: false,
            performance_promotable: false,
            hardware_completion_promotable: false,
            evidence_boundary: RESULT_EVIDENCE_BOUNDARY,
        },
    })
}

fn process_samples(
    snapshot: &ArchiveSnapshot,
    result_directory: &ResultDirectory,
    threads: usize,
    metadata: &RunMetadata,
    samples: &[SampleSpec],
    pre_run_identity_record: &FileRecord,
    coordination: &ValidatedCoordination,
) -> Result<ProcessedSamples> {
    let mut transcript = create_exclusive_file(result_directory, "jobs.jsonl")?;
    let mut request_results = Vec::with_capacity(samples.len());
    let mut validation_state = SampleValidationState {
        metadata,
        pre_run_identity_record,
        coordination,
        reservation_ids: HashSet::new(),
        nonces: HashSet::new(),
        reservation_chain: None,
    };
    let mut private_seed_digests = HashSet::new();
    let mut measured_metrics = Vec::new();
    let mut total_jobs = 0_u64;
    let mut matched_jobs = 0_u64;
    let mut all_padding_valid = true;

    for sample in samples {
        let validated = validate_sample_contract(snapshot, sample, &mut validation_state)?;
        let computations = compute_sample_jobs(snapshot, sample, threads)?;
        let computation =
            write_sample_computations(&mut transcript, computations, &mut private_seed_digests)?;
        transcript.flush().context("cannot flush job transcript")?;
        transcript
            .get_ref()
            .sync_all()
            .context("cannot fsync job transcript")?;

        if sample.phase == "measured" {
            measured_metrics.push(validated.metrics);
        }
        total_jobs = total_jobs
            .checked_add(u64::from(sample.batch))
            .context("total job counter overflow")?;
        matched_jobs = matched_jobs
            .checked_add(u64::from(computation.matched_jobs))
            .context("matched job counter overflow")?;
        all_padding_valid &= computation.padding_valid;
        request_results.push(sample_request_result(sample, validated, computation));
    }

    let jobs_record = finish_writer(transcript)?;
    result_directory.sync()?;
    Ok(ProcessedSamples {
        jobs_record,
        request_results,
        measured_metrics,
        total_jobs,
        matched_jobs,
        all_padding_valid,
    })
}

fn write_sample_computations(
    transcript: &mut impl Write,
    computations: Vec<JobComputation>,
    private_seed_digests: &mut HashSet<[u8; 32]>,
) -> Result<SampleComputationSummary> {
    let mut matched_jobs = 0_u32;
    let mut padding_valid = true;
    for computation in computations {
        ensure!(
            private_seed_digests.insert(computation.seed_sha256),
            "transport archive reuses a private WOTS seed"
        );
        if computation.transcript.r#match {
            matched_jobs = matched_jobs
                .checked_add(1)
                .context("sample match counter overflow")?;
        }
        padding_valid &= computation.padding_match;
        serde_json::to_writer(&mut *transcript, &computation.transcript)
            .context("cannot serialize job transcript")?;
        transcript
            .write_all(b"\n")
            .context("cannot write job transcript")?;
    }
    Ok(SampleComputationSummary {
        matched_jobs,
        padding_valid,
    })
}

fn sample_request_result(
    sample: &SampleSpec,
    validated: ValidatedSample,
    computation: SampleComputationSummary,
) -> RequestResult {
    RequestResult {
        sample: sample.relative.clone(),
        phase: sample.phase,
        sample_index: sample.index,
        request_sha256: validated.request_sha256,
        run_id: validated.request.run_id.clone(),
        coordination_intent: root_binding(&validated.request.coordination_intent),
        coordination_burn: root_binding(&validated.request.coordination_burn),
        coordination_receipt: root_binding(&validated.request.coordination_receipt),
        batch: sample.batch,
        nonce_hex: validated.request.nonce_hex.clone(),
        reservation_record_sha256: validated.request.reservation_record_sha256.clone(),
        xclbin: ResultXclbinBinding {
            path: validated.request.xclbin.path.clone(),
            bytes: validated.request.xclbin.bytes,
            sha256: validated.request.xclbin.sha256.clone(),
        },
        hardware_identity_pre_run: root_binding(&validated.request.hardware_identity_pre_run),
        input: request_artifact_binding(&validated.request.artifacts.input),
        output: request_artifact_binding(&validated.request.artifacts.full_output_allocation),
        jobs: sample.batch,
        matched_jobs: computation.matched_jobs,
        padding_valid: computation.padding_valid,
        cryptographic_oracle_valid: computation.matched_jobs == sample.batch
            && computation.padding_valid,
    }
}

fn manifest_binding(record: &FileRecord) -> ManifestBinding {
    named_binding("SHA256SUMS", record)
}

fn named_binding(file: &str, record: &FileRecord) -> ManifestBinding {
    ManifestBinding {
        file: file.to_owned(),
        bytes: record.bytes,
        sha256: record.sha256.clone(),
    }
}

fn root_binding(binding: &RootArtifactBinding) -> ManifestBinding {
    ManifestBinding {
        file: binding.archive_root_file.clone(),
        bytes: binding.bytes,
        sha256: binding.sha256.clone(),
    }
}

fn request_artifact_binding(binding: &ArtifactBinding) -> ResultRequestArtifactBinding {
    ResultRequestArtifactBinding {
        file: binding.file.clone(),
        logical_bytes: binding.logical_bytes,
        stored_bytes: binding.stored_bytes,
        sha256: binding.sha256.clone(),
    }
}

fn canonical_json_line<T: Serialize>(value: &T) -> Result<Vec<u8>> {
    let mut bytes = serde_json::to_vec(value).context("cannot serialize canonical JSON")?;
    bytes.push(b'\n');
    Ok(bytes)
}

fn parse_archive_json<T>(snapshot: &ArchiveSnapshot, relative: &str) -> Result<T>
where
    T: serde::de::DeserializeOwned,
{
    let bytes = read_verified_small(snapshot, relative, MAXIMUM_JSON_BYTES)?;
    parse_strict_json(&bytes, relative)
}

fn parse_coordination_json<T>(snapshot: &ArchiveSnapshot, relative: &str) -> Result<T>
where
    T: serde::de::DeserializeOwned + Serialize,
{
    let bytes = read_verified_small(snapshot, relative, MAXIMUM_COORDINATION_RECORD_BYTES)?;
    parse_canonical_json(&bytes, relative)
}

fn validate_coordination_records(snapshot: &ArchiveSnapshot) -> Result<ValidatedCoordination> {
    let intent_record = archive_file_record(snapshot, COORDINATION_INTENT_FILE)?;
    let burn_record = archive_file_record(snapshot, COORDINATION_BURN_FILE)?;
    let receipt_record = archive_file_record(snapshot, COORDINATION_RECEIPT_FILE)?;
    let intent: CoordinationIntent = parse_coordination_json(snapshot, COORDINATION_INTENT_FILE)?;
    validate_coordination_intent(&intent)?;
    let burn: CoordinationBurn = parse_coordination_json(snapshot, COORDINATION_BURN_FILE)?;
    validate_coordination_burn(&burn, &intent, &intent_record)?;
    let receipt: CoordinationReceipt =
        parse_coordination_json(snapshot, COORDINATION_RECEIPT_FILE)?;
    validate_coordination_receipt(
        snapshot,
        &receipt,
        &intent,
        &burn,
        &intent_record,
        &burn_record,
    )?;
    Ok(ValidatedCoordination {
        intent,
        receipt,
        intent_record,
        burn_record,
        receipt_record,
    })
}

fn validate_coordination_intent(intent: &CoordinationIntent) -> Result<()> {
    ensure!(intent.schema == 1, "coordination intent schema must be 1");
    ensure!(
        intent.contract == COORDINATION_INTENT_CONTRACT,
        "coordination intent contract changed"
    );
    ensure!(
        is_nonzero_lower_hex(&intent.run_id, 32),
        "coordination run id is not canonical"
    );
    ensure!(
        intent.owner == COORDINATION_OWNER,
        "coordination intent owner changed"
    );
    ensure!(
        intent.command_class == COORDINATION_COMMAND_CLASS,
        "coordination intent command class changed"
    );
    ensure!(
        valid_utc(&intent.created_utc),
        "coordination intent creation time is not canonical UTC"
    );
    ensure!(
        intent.lock.context == COORDINATION_LOCK_CONTEXT
            && intent.lock.acquisition == COORDINATION_LOCK_ACQUISITION
            && canonical_coordination_lock_path(&intent.lock.path),
        "coordination intent lock binding is noncanonical"
    );
    ensure!(
        intent.grants.len() == COORDINATION_LEDGER_IDS.len(),
        "coordination intent must contain exactly two ordered grants"
    );
    for (index, grant) in intent.grants.iter().enumerate() {
        validate_coordination_grant(grant, COORDINATION_LEDGER_IDS[index], &intent.run_id)?;
        ensure!(
            grant.granted_utc.as_str() <= intent.created_utc.as_str(),
            "coordination grant time is later than intent creation"
        );
    }
    ensure!(
        intent.grants[0].entry_id != intent.grants[1].entry_id,
        "coordination intent grant entry ids must differ"
    );
    Ok(())
}

fn validate_coordination_grant(
    grant: &CoordinationGrant,
    expected_ledger: &str,
    run_id: &str,
) -> Result<()> {
    ensure!(
        grant.ledger_id == expected_ledger
            && canonical_coordination_entry_id(&grant.entry_id)
            && valid_utc(&grant.granted_utc)
            && grant.run_id == run_id
            && grant.entry_bytes > 0
            && is_nonzero_lower_hex(&grant.entry_sha256, 64),
        "coordination grant violates its canonical ledger binding"
    );
    Ok(())
}

fn validate_coordination_burn(
    burn: &CoordinationBurn,
    intent: &CoordinationIntent,
    intent_record: &FileRecord,
) -> Result<()> {
    ensure!(burn.schema == 1, "coordination burn schema must be 1");
    ensure!(
        burn.contract == COORDINATION_BURN_CONTRACT,
        "coordination burn contract changed"
    );
    ensure!(
        burn.run_id == intent.run_id,
        "coordination burn run id differs from intent"
    );
    validate_digest(&burn.intent, "coordination burn intent")?;
    ensure!(
        burn.intent.bytes == intent_record.bytes && burn.intent.sha256 == intent_record.sha256,
        "coordination burn does not bind the archived intent"
    );
    ensure!(
        burn.grants.len() == intent.grants.len(),
        "coordination burn must contain exactly two ordered grants"
    );
    for (index, (burn_grant, intent_grant)) in burn.grants.iter().zip(&intent.grants).enumerate() {
        ensure!(
            burn_grant.ledger_id == COORDINATION_LEDGER_IDS[index]
                && burn_grant.ledger_id == intent_grant.ledger_id
                && burn_grant.entry_id == intent_grant.entry_id
                && burn_grant.entry_bytes == intent_grant.entry_bytes
                && burn_grant.entry_sha256 == intent_grant.entry_sha256,
            "coordination burn grant differs from intent"
        );
    }
    ensure!(
        is_lower_hex(&burn.previous_sha256, 64),
        "coordination burn previous SHA-256 is not canonical"
    );
    let zero_hash = "0".repeat(64);
    if burn.sequence == 0 {
        ensure!(
            burn.previous_sha256.as_str() == zero_hash.as_str(),
            "initial coordination burn previous hash must be zero"
        );
    } else {
        ensure!(
            burn.previous_sha256.as_str() != zero_hash.as_str(),
            "noninitial coordination burn previous hash must be nonzero"
        );
    }
    ensure!(
        is_lower_hex(&burn.record_sha256, 64),
        "coordination burn record SHA-256 is not canonical"
    );
    let body = CoordinationBurnHashBody {
        schema: burn.schema,
        contract: &burn.contract,
        sequence: burn.sequence,
        run_id: &burn.run_id,
        intent: &burn.intent,
        grants: &burn.grants,
        previous_sha256: &burn.previous_sha256,
    };
    let body_bytes =
        serde_json::to_vec(&body).context("cannot serialize coordination burn hash body")?;
    ensure!(
        sha256_bytes(&body_bytes) == burn.record_sha256,
        "coordination burn record SHA-256 differs from its canonical body"
    );
    Ok(())
}

fn validate_coordination_receipt(
    snapshot: &ArchiveSnapshot,
    receipt: &CoordinationReceipt,
    intent: &CoordinationIntent,
    burn: &CoordinationBurn,
    intent_record: &FileRecord,
    burn_record: &FileRecord,
) -> Result<()> {
    ensure!(receipt.schema == 2, "coordination receipt schema must be 2");
    ensure!(
        receipt.contract == COORDINATION_RECEIPT_CONTRACT,
        "coordination receipt contract changed"
    );
    ensure!(
        receipt.run_id == intent.run_id && receipt.run_id == burn.run_id,
        "coordination receipt run id differs from intent or burn"
    );
    validate_root_binding_against_record(
        &receipt.intent,
        COORDINATION_INTENT_FILE,
        intent_record,
        "coordination receipt intent",
    )?;
    validate_root_binding_against_record(
        &receipt.coordination_burn,
        COORDINATION_BURN_FILE,
        burn_record,
        "coordination receipt burn",
    )?;
    ensure!(
        receipt.grants == intent.grants,
        "coordination receipt grants differ from intent"
    );
    ensure!(
        receipt.lock.path == intent.lock.path
            && receipt.lock.context == intent.lock.context
            && receipt.lock.acquisition == intent.lock.acquisition
            && receipt.lock.device > 0
            && receipt.lock.inode > 0
            && u32::try_from(receipt.lock.euid).is_ok(),
        "coordination receipt lock differs from intent or lacks descriptor identity"
    );
    ensure!(
        u32::try_from(receipt.hardware.device_index).is_ok()
            && canonical_bdf(&receipt.hardware.bdf)
            && canonical_serial(&receipt.hardware.xmc_serial_number)
            && receipt.hardware.shell_vbnv == EXPECTED_DEVICE_NAME
            && receipt.hardware.logic_uuid == EXPECTED_LOGIC_UUID
            && receipt.hardware.interface_uuid == EXPECTED_INTERFACE_UUID,
        "coordination receipt hardware identity is noncanonical"
    );
    ensure!(
        receipt.xclbin.bytes > 0
            && is_lower_hex(&receipt.xclbin.sha256, 64)
            && canonical_uuid(&receipt.xclbin.uuid)
            && receipt.xclbin.target == "hw",
        "coordination receipt xclbin binding is noncanonical"
    );
    validate_root_artifact_binding(
        snapshot,
        &receipt.hardware_identity_pre_run,
        PRE_RUN_IDENTITY_FILE,
        "coordination receipt pre-run hardware identity",
    )?;
    ensure!(
        !receipt.card_identity_promotable
            && !receipt.performance_promotable
            && !receipt.hardware_completion_promotable,
        "coordination receipt makes a premature promotion claim"
    );
    Ok(())
}

fn validate_coordination_hardware(
    coordination: &ValidatedCoordination,
    hardware_identity: &ValidatedHardwareIdentity,
) -> Result<()> {
    let receipt = &coordination.receipt;
    let pre_run = &hardware_identity.pre_run;
    ensure!(
        receipt.hardware_identity_pre_run.bytes == hardware_identity.pre_run_record.bytes
            && receipt.hardware_identity_pre_run.sha256 == hardware_identity.pre_run_record.sha256,
        "coordination receipt pre-run binding differs from parsed identity"
    );
    ensure!(
        receipt.hardware.device_index == pre_run.device.index
            && receipt.hardware.bdf == pre_run.device.bdf
            && receipt.hardware.xmc_serial_number == pre_run.platform.xmc_serial_number
            && receipt.hardware.shell_vbnv == pre_run.platform.shell_vbnv
            && receipt.hardware.logic_uuid == pre_run.platform.logic_uuid
            && receipt.hardware.interface_uuid == pre_run.device.interface_uuid
            && receipt.hardware.interface_uuid == pre_run.pinned_target.interface_uuid,
        "coordination receipt hardware fields differ from native pre-run identity"
    );
    ensure!(
        receipt.xclbin.bytes == pre_run.xclbin.digest.bytes
            && receipt.xclbin.sha256 == pre_run.xclbin.digest.sha256
            && receipt.xclbin.uuid == pre_run.xclbin.uuid
            && receipt.xclbin.target == pre_run.xclbin.target,
        "coordination receipt xclbin differs from native pre-run identity"
    );
    Ok(())
}

fn validate_coordination_consumer_binding(
    run_id: &str,
    intent: &RootArtifactBinding,
    burn: &RootArtifactBinding,
    receipt: &RootArtifactBinding,
    coordination: &ValidatedCoordination,
    label: &str,
) -> Result<()> {
    ensure!(
        run_id == coordination.intent.run_id,
        "{label} run id differs from coordination intent"
    );
    validate_root_binding_against_record(
        intent,
        COORDINATION_INTENT_FILE,
        &coordination.intent_record,
        &format!("{label} coordination intent"),
    )?;
    validate_root_binding_against_record(
        burn,
        COORDINATION_BURN_FILE,
        &coordination.burn_record,
        &format!("{label} coordination burn"),
    )?;
    validate_root_binding_against_record(
        receipt,
        COORDINATION_RECEIPT_FILE,
        &coordination.receipt_record,
        &format!("{label} coordination receipt"),
    )?;
    Ok(())
}

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

fn canonical_coordination_lock_path(value: &str) -> bool {
    value.is_ascii()
        && value.starts_with('/')
        && value.len() > COORDINATION_LOCK_SUFFIX.len()
        && value.ends_with(COORDINATION_LOCK_SUFFIX)
        && !value.contains("//")
        && value
            .split('/')
            .skip(1)
            .all(|component| !component.is_empty() && component != "." && component != "..")
}

fn is_nonzero_lower_hex(value: &str, width: usize) -> bool {
    is_lower_hex(value, width) && value.bytes().any(|byte| byte != b'0')
}

fn validate_run_metadata(
    metadata: &RunMetadata,
    coordination: &ValidatedCoordination,
) -> Result<()> {
    ensure!(metadata.schema == 2, "run metadata schema must be 2");
    ensure!(
        metadata.runner == "phase_b_hbm",
        "archive is not a Phase-B HBM run"
    );
    validate_coordination_consumer_binding(
        &metadata.run_id,
        &metadata.coordination_intent,
        &metadata.coordination_burn,
        &metadata.coordination_receipt,
        coordination,
        "run metadata",
    )?;
    ensure!(
        valid_utc(&metadata.created_utc),
        "created_utc is not canonical UTC"
    );
    ensure!(
        metadata.created_utc.as_str() >= coordination.intent.created_utc.as_str(),
        "run metadata predates coordination intent"
    );
    ensure!(
        u32::try_from(metadata.device_index).is_ok(),
        "run metadata device index does not fit u32"
    );
    ensure!(
        metadata.kernel_name == DEFAULT_KERNEL,
        "unexpected HBM kernel identity"
    );
    validate_digest(&metadata.xclbin.digest, "metadata xclbin")?;
    ensure!(!metadata.xclbin.path.is_empty(), "xclbin path is empty");
    ensure!(
        metadata.measured_runs_per_batch > 0,
        "measured run count must be positive"
    );
    ensure!(!metadata.batches.is_empty(), "run metadata has no batches");
    ensure!(metadata.timeout_ms > 0, "run timeout must be positive");
    validate_clock(&metadata.clock)?;
    ensure!(
        metadata.cryptographic_control == "Rust/RHDL kernel only",
        "metadata cryptographic-control boundary changed"
    );
    ensure!(
        metadata.output_oracle == "external Rust oracle required",
        "metadata does not require the external Rust oracle"
    );
    let mut batches = HashSet::new();
    for &batch in &metadata.batches {
        ensure!(
            batch <= MAXIMUM_BATCH,
            "metadata batch exceeds the shell ABI"
        );
        ensure!(batches.insert(batch), "metadata repeats a batch");
    }
    Ok(())
}

fn valid_utc(text: &str) -> bool {
    let bytes = text.as_bytes();
    bytes.len() == 20 && bytes[19] == b'Z' && valid_calendar_timestamp(&bytes[..19], b'T')
}

fn validate_hardware_identity(
    snapshot: &ArchiveSnapshot,
    metadata: &RunMetadata,
) -> Result<ValidatedHardwareIdentity> {
    let pre_run_record = archive_file_record(snapshot, PRE_RUN_IDENTITY_FILE)?;
    let final_record = archive_file_record(snapshot, FINAL_IDENTITY_FILE)?;
    let pre_run: PreRunHardwareIdentity = parse_archive_json(snapshot, PRE_RUN_IDENTITY_FILE)?;
    let final_identity: FinalHardwareIdentity = parse_archive_json(snapshot, FINAL_IDENTITY_FILE)?;

    validate_pre_run_identity(&pre_run, metadata)?;
    validate_final_identity(&final_identity, metadata)?;
    validate_identity_continuity(&pre_run, &final_identity)?;
    validate_identity_report_bindings(snapshot, &pre_run, &final_identity, &pre_run_record)?;
    validate_raw_identity_reports(snapshot, &pre_run, &final_identity)?;

    Ok(ValidatedHardwareIdentity {
        pre_run_record,
        final_record,
        pre_run,
    })
}

fn validate_identity_continuity(
    pre_run: &PreRunHardwareIdentity,
    final_identity: &FinalHardwareIdentity,
) -> Result<()> {
    ensure!(
        pre_run.capture == final_identity.capture
            && pre_run.pinned_target == final_identity.pinned_target
            && pre_run.device == final_identity.device
            && pre_run.platform == final_identity.platform
            && pre_run.xrt_build == final_identity.xrt_build
            && pre_run.xclbin == final_identity.xclbin,
        "pre-run and final hardware identity fields differ"
    );
    ensure!(
        pre_run.programmed_image.load_return_uuid
            == final_identity.programmed_image.load_return_uuid
            && pre_run.programmed_image.post_load_uuid
                == final_identity.programmed_image.post_load_uuid,
        "programmed-image UUID changed between pre-run and final identity"
    );
    Ok(())
}

fn validate_identity_report_bindings(
    snapshot: &ArchiveSnapshot,
    pre_run: &PreRunHardwareIdentity,
    final_identity: &FinalHardwareIdentity,
    pre_run_record: &FileRecord,
) -> Result<()> {
    validate_root_artifact_binding(
        snapshot,
        &pre_run.raw_reports.platform_pre_load,
        PLATFORM_PRE_LOAD_FILE,
        "pre-run platform report",
    )?;
    ensure!(
        pre_run.raw_reports.platform_post_run.is_none(),
        "pre-run identity unexpectedly binds a post-run platform report"
    );
    validate_root_artifact_binding(
        snapshot,
        &pre_run.raw_reports.xrt_host,
        XRT_HOST_FILE,
        "pre-run XRT host report",
    )?;
    validate_root_artifact_binding(
        snapshot,
        &final_identity.raw_reports.platform_pre_load,
        PLATFORM_PRE_LOAD_FILE,
        "final pre-load platform report",
    )?;
    let final_post_run = final_identity
        .raw_reports
        .platform_post_run
        .as_ref()
        .context("final identity omits its post-run platform report")?;
    validate_root_artifact_binding(
        snapshot,
        final_post_run,
        PLATFORM_POST_RUN_FILE,
        "final post-run platform report",
    )?;
    validate_root_artifact_binding(
        snapshot,
        &final_identity.raw_reports.xrt_host,
        XRT_HOST_FILE,
        "final XRT host report",
    )?;
    validate_root_artifact_binding(
        snapshot,
        &final_identity.pre_run_record,
        PRE_RUN_IDENTITY_FILE,
        "final pre-run identity record",
    )?;
    ensure!(
        final_identity.pre_run_record.bytes == pre_run_record.bytes
            && final_identity.pre_run_record.sha256 == pre_run_record.sha256,
        "final identity does not bind the parsed pre-run identity"
    );
    ensure!(
        pre_run.raw_reports.platform_pre_load == final_identity.raw_reports.platform_pre_load
            && pre_run.raw_reports.xrt_host == final_identity.raw_reports.xrt_host,
        "pre-run and final identity raw-report bindings differ"
    );
    Ok(())
}

fn validate_raw_identity_reports(
    snapshot: &ArchiveSnapshot,
    pre_run: &PreRunHardwareIdentity,
    final_identity: &FinalHardwareIdentity,
) -> Result<()> {
    let platform_pre = parse_xrt_platform_report(snapshot, PLATFORM_PRE_LOAD_FILE)?;
    let platform_post = parse_xrt_platform_report(snapshot, PLATFORM_POST_RUN_FILE)?;
    ensure!(
        platform_pre == pre_run.platform,
        "pre-load XRT platform report differs from pre-run identity"
    );
    ensure!(
        platform_post == final_identity.platform,
        "post-run XRT platform report differs from final identity"
    );
    ensure!(
        platform_pre == platform_post,
        "XRT platform identity changed between pre-load and post-run reports"
    );

    let host_bytes = read_verified_small(snapshot, XRT_HOST_FILE, MAXIMUM_XRT_HOST_REPORT_BYTES)?;
    let raw_xrt_build: XrtBuildIdentity = parse_xrt_report_json(&host_bytes, XRT_HOST_FILE)?;
    validate_xrt_build(&raw_xrt_build, false)?;
    ensure!(
        raw_xrt_build.version == pre_run.xrt_build.version
            && raw_xrt_build.branch == pre_run.xrt_build.branch
            && raw_xrt_build.hash.to_ascii_lowercase() == pre_run.xrt_build.hash
            && raw_xrt_build.build_date == pre_run.xrt_build.build_date,
        "raw XRT host report differs from hardware identity"
    );
    Ok(())
}

fn archive_file_record(snapshot: &ArchiveSnapshot, file: &str) -> Result<FileRecord> {
    snapshot
        .files
        .get(file)
        .cloned()
        .with_context(|| format!("archive manifest omits {file}"))
}

fn validate_root_artifact_binding(
    snapshot: &ArchiveSnapshot,
    binding: &RootArtifactBinding,
    expected_file: &str,
    label: &str,
) -> Result<()> {
    let record = archive_file_record(snapshot, expected_file)?;
    validate_root_binding_against_record(binding, expected_file, &record, label)
}

fn validate_root_binding_against_record(
    binding: &RootArtifactBinding,
    expected_file: &str,
    record: &FileRecord,
    label: &str,
) -> Result<()> {
    ensure!(
        binding.archive_root_file == expected_file,
        "{label} names the wrong archive-root file"
    );
    ensure!(binding.bytes > 0, "{label} byte count must be positive");
    ensure!(
        is_lower_hex(&binding.sha256, 64),
        "{label} SHA-256 is not canonical"
    );
    ensure!(
        binding.bytes == record.bytes && binding.sha256 == record.sha256,
        "{label} differs from the top-level manifest"
    );
    Ok(())
}

fn validate_pre_run_identity(
    identity: &PreRunHardwareIdentity,
    metadata: &RunMetadata,
) -> Result<()> {
    validate_identity_fields(
        identity.schema,
        &identity.contract,
        PRE_RUN_IDENTITY_CONTRACT,
        &identity.capture,
        &identity.pinned_target,
        &identity.device,
        &identity.platform,
        &identity.xrt_build,
        &identity.xclbin,
        identity.hardware_completion_promotable,
        metadata,
    )?;
    validate_programmed_image(&identity.programmed_image, &identity.xclbin.uuid, false)
}

fn validate_final_identity(identity: &FinalHardwareIdentity, metadata: &RunMetadata) -> Result<()> {
    validate_identity_fields(
        identity.schema,
        &identity.contract,
        FINAL_IDENTITY_CONTRACT,
        &identity.capture,
        &identity.pinned_target,
        &identity.device,
        &identity.platform,
        &identity.xrt_build,
        &identity.xclbin,
        identity.hardware_completion_promotable,
        metadata,
    )?;
    validate_programmed_image(&identity.programmed_image, &identity.xclbin.uuid, true)
}

#[allow(clippy::too_many_arguments)]
fn validate_identity_fields(
    schema: u64,
    contract: &str,
    expected_contract: &str,
    capture: &model::HardwareIdentityCapture,
    pinned: &model::PinnedHardwareTarget,
    device: &model::HardwareDeviceIdentity,
    platform: &HardwarePlatformIdentity,
    xrt_build: &XrtBuildIdentity,
    xclbin: &model::HardwareXclbinIdentity,
    hardware_completion_promotable: bool,
    metadata: &RunMetadata,
) -> Result<()> {
    ensure!(schema == 1, "hardware identity schema must be 1");
    ensure!(
        contract == expected_contract,
        "hardware identity contract kind mismatch"
    );
    ensure!(
        capture.in_process_xrt_native_cpp && !capture.subprocess_or_sysfs_fallback_accepted,
        "hardware identity was not captured by the required in-process XRT API"
    );
    ensure!(
        pinned.device_name == EXPECTED_DEVICE_NAME
            && pinned.deployment_platform == EXPECTED_DEPLOYMENT_PLATFORM
            && pinned.fpga_part == EXPECTED_FPGA_PART
            && pinned.interface_uuid == EXPECTED_INTERFACE_UUID
            && pinned.logic_uuid == EXPECTED_LOGIC_UUID,
        "hardware identity pinned-target constants differ from the U280 contract"
    );
    ensure!(
        device.index == metadata.device_index,
        "hardware identity device index differs from run metadata"
    );
    ensure!(
        canonical_bdf(&device.bdf),
        "hardware identity BDF is not canonical"
    );
    ensure!(
        device.name == EXPECTED_DEVICE_NAME && device.interface_uuid == EXPECTED_INTERFACE_UUID,
        "typed hardware device identity differs from the pinned U280"
    );
    validate_platform_identity(platform)?;
    validate_xrt_build(xrt_build, true)?;
    validate_digest(&xclbin.digest, "hardware identity xclbin")?;
    ensure!(
        xclbin.path == metadata.xclbin.path && xclbin.digest == metadata.xclbin.digest,
        "hardware identity xclbin binding differs from run metadata"
    );
    ensure!(canonical_uuid(&xclbin.uuid), "xclbin UUID is not canonical");
    ensure!(
        xclbin.interface_uuid == EXPECTED_INTERFACE_UUID
            && xclbin.xsa_name == EXPECTED_DEPLOYMENT_PLATFORM
            && xclbin.fpga_part == EXPECTED_FPGA_PART
            && xclbin.target == "hw",
        "hardware identity xclbin target differs from the pinned U280 hardware image"
    );
    ensure!(
        !hardware_completion_promotable,
        "hardware identity makes a premature promotion claim"
    );
    Ok(())
}

fn validate_programmed_image(
    image: &ProgrammedImageIdentity,
    xclbin_uuid: &str,
    final_record: bool,
) -> Result<()> {
    ensure!(
        canonical_uuid(&image.load_return_uuid)
            && canonical_uuid(&image.post_load_uuid)
            && image.load_return_uuid == xclbin_uuid
            && image.post_load_uuid == xclbin_uuid
            && image.load_post_load_equal,
        "load-return/post-load UUID continuity is invalid"
    );
    if final_record {
        ensure!(
            image.post_run_uuid.as_deref() == Some(xclbin_uuid)
                && image.load_post_load_post_run_equal == Some(true),
            "final programmed-image UUID continuity is invalid"
        );
    } else {
        ensure!(
            image.post_run_uuid.is_none() && image.load_post_load_post_run_equal.is_none(),
            "pre-run identity contains a post-run UUID decision"
        );
    }
    Ok(())
}

fn parse_xrt_platform_report(
    snapshot: &ArchiveSnapshot,
    file: &str,
) -> Result<HardwarePlatformIdentity> {
    let bytes = read_verified_small(snapshot, file, MAXIMUM_PLATFORM_REPORT_BYTES)?;
    let report: XrtPlatformReport = parse_xrt_report_json(&bytes, file)?;
    ensure!(
        report.platforms.len() == 1,
        "XRT platform report must describe exactly one platform"
    );
    let platform = &report.platforms[0];
    let identity = HardwarePlatformIdentity {
        shell_vbnv: platform.static_region.vbnv.clone(),
        logic_uuid: platform.static_region.logic_uuid.to_ascii_lowercase(),
        fpga_part: platform.static_region.fpga_name.clone(),
        xmc_serial_number: platform
            .controller
            .card_mgmt_controller
            .serial_number
            .clone(),
    };
    validate_platform_identity(&identity)?;
    Ok(identity)
}

fn validate_platform_identity(platform: &HardwarePlatformIdentity) -> Result<()> {
    ensure!(
        platform.shell_vbnv == EXPECTED_DEVICE_NAME
            && platform.logic_uuid == EXPECTED_LOGIC_UUID
            && platform.fpga_part == EXPECTED_FPGA_PART,
        "hardware platform identity differs from the pinned U280"
    );
    ensure!(
        canonical_serial(&platform.xmc_serial_number),
        "XMC serial number is blank, fallback, or noncanonical"
    );
    Ok(())
}

fn validate_xrt_build(build: &XrtBuildIdentity, require_lower_hash: bool) -> Result<()> {
    let mut version = build.version.split('.');
    let version_valid = version.next() == Some("2")
        && version.next() == Some("16")
        && version.next().is_some_and(|patch| {
            !patch.is_empty() && patch.bytes().all(|byte| byte.is_ascii_digit())
        })
        && version.next().is_none();
    let hash_valid = (7..=64).contains(&build.hash.len())
        && build.hash.bytes().all(|byte| byte.is_ascii_hexdigit())
        && (!require_lower_hash
            || build
                .hash
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)));
    ensure!(
        version_valid
            && build.branch == "2023.2"
            && hash_valid
            && valid_xrt_build_date(&build.build_date),
        "XRT build identity is incomplete or is not the pinned 2023.2 ABI"
    );
    Ok(())
}

fn valid_xrt_build_date(value: &str) -> bool {
    valid_calendar_timestamp(value.as_bytes(), b' ')
}

fn valid_calendar_timestamp(bytes: &[u8], date_time_separator: u8) -> bool {
    if bytes.len() != 19
        || bytes[4] != b'-'
        || bytes[7] != b'-'
        || bytes[10] != date_time_separator
        || bytes[13] != b':'
        || bytes[16] != b':'
        || !bytes
            .iter()
            .enumerate()
            .all(|(index, byte)| matches!(index, 4 | 7 | 10 | 13 | 16) || byte.is_ascii_digit())
    {
        return false;
    }
    let decimal = |digits: &[u8]| {
        digits.iter().try_fold(0_u32, |value, byte| {
            value.checked_mul(10)?.checked_add(u32::from(*byte - b'0'))
        })
    };
    let (Some(year), Some(month), Some(day), Some(hour), Some(minute), Some(second)) = (
        decimal(&bytes[0..4]),
        decimal(&bytes[5..7]),
        decimal(&bytes[8..10]),
        decimal(&bytes[11..13]),
        decimal(&bytes[14..16]),
        decimal(&bytes[17..19]),
    ) else {
        return false;
    };
    let leap_year =
        year.is_multiple_of(4) && (!year.is_multiple_of(100) || year.is_multiple_of(400));
    let maximum_day = match month {
        1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
        4 | 6 | 9 | 11 => 30,
        2 if leap_year => 29,
        2 => 28,
        _ => return false,
    };
    (2000..=2100).contains(&year)
        && (1..=maximum_day).contains(&day)
        && hour <= 23
        && minute <= 59
        && second <= 59
}

fn canonical_uuid(value: &str) -> bool {
    let bytes = value.as_bytes();
    bytes.len() == 36
        && bytes.iter().enumerate().all(|(index, byte)| {
            if matches!(index, 8 | 13 | 18 | 23) {
                *byte == b'-'
            } else {
                byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)
            }
        })
        && bytes
            .iter()
            .enumerate()
            .any(|(index, byte)| !matches!(index, 8 | 13 | 18 | 23) && *byte != b'0')
}

fn canonical_bdf(value: &str) -> bool {
    let bytes = value.as_bytes();
    if bytes.len() != 12
        || bytes[4] != b':'
        || bytes[7] != b':'
        || bytes[10] != b'.'
        || !bytes[11].is_ascii_digit()
        || bytes[11] > b'7'
        || !bytes.iter().enumerate().all(|(index, byte)| {
            matches!(index, 4 | 7 | 10) || byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)
        })
    {
        return false;
    }
    let hex_nibble = |byte: u8| match byte {
        b'0'..=b'9' => byte - b'0',
        b'a'..=b'f' => byte - b'a' + 10,
        _ => unreachable!("shape check admitted a non-hex BDF digit"),
    };
    let device = u16::from(hex_nibble(bytes[8])) * 16 + u16::from(hex_nibble(bytes[9]));
    device <= 0x1f
}

fn canonical_serial(value: &str) -> bool {
    let lower = value.to_ascii_lowercase();
    !value.is_empty()
        && value.len() <= 128
        && !matches!(lower.as_str(), "n/a" | "na" | "unknown" | "none")
        && value.bytes().any(|byte| byte.is_ascii_alphanumeric())
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
}

fn validate_results(
    results: &ResultsEnvelope,
    metadata: &RunMetadata,
    final_identity_record: &FileRecord,
    coordination: &ValidatedCoordination,
) -> Result<()> {
    ensure!(results.schema == 2, "results schema must be 2");
    ensure!(
        results.runner == "phase_b_hbm",
        "results runner kind mismatch"
    );
    validate_coordination_consumer_binding(
        &results.run_id,
        &results.coordination_intent,
        &results.coordination_burn,
        &results.coordination_receipt,
        coordination,
        "results",
    )?;
    ensure!(
        results.transport_valid,
        "host transport validation is false"
    );
    ensure!(
        results.cryptographic_oracle_valid.is_none(),
        "transport results pre-populated the Rust-oracle decision"
    );
    ensure!(
        !results.hardware_signature_evidence_complete,
        "transport results make a premature hardware-signature claim"
    );
    ensure!(
        !results.hardware_completion_promotable,
        "transport results make a premature hardware-completion claim"
    );
    validate_root_binding_against_record(
        &results.hardware_identity,
        FINAL_IDENTITY_FILE,
        final_identity_record,
        "results final hardware identity",
    )?;
    ensure!(
        results.clock == metadata.clock,
        "results clock binding differs from metadata"
    );
    ensure!(
        results.rate_policy == RATE_POLICY,
        "results rate policy changed"
    );
    ensure!(
        results
            .performance_acceptance
            .cryptographic_oracle_valid
            .is_none(),
        "performance acceptance pre-populated the oracle decision"
    );
    ensure!(
        !results.performance_acceptance.accepted && !results.performance_acceptance.promotable,
        "performance acceptance is prematurely promotable"
    );
    ensure!(
        results.batches.len() == metadata.batches.len(),
        "results batch list length differs from metadata"
    );
    for (value, expected_batch) in results.batches.iter().zip(&metadata.batches) {
        ensure!(
            value.batch == *expected_batch,
            "results batch order differs from metadata"
        );
        ensure!(
            value.measured_samples == metadata.measured_runs_per_batch,
            "results measured sample count differs from metadata"
        );
    }
    Ok(())
}

fn nearest_rank(mut values: Vec<f64>, percentile_numerator: usize) -> Option<f64> {
    if values.is_empty() || !(1..=100).contains(&percentile_numerator) {
        return None;
    }
    values.sort_by(f64::total_cmp);
    let rank = values
        .len()
        .checked_mul(percentile_numerator)?
        .div_ceil(100);
    values.get(rank.checked_sub(1)?).copied()
}

fn percentiles(values: &[f64]) -> Option<Percentiles> {
    Some(Percentiles {
        p50: nearest_rank(values.to_vec(), 50)?,
        p95: nearest_rank(values.to_vec(), 95)?,
        p99: nearest_rank(values.to_vec(), 99)?,
    })
}

fn rate_percentiles(values: &[f64]) -> Option<RatePercentiles> {
    Some(RatePercentiles {
        p5_conservative: nearest_rank(values.to_vec(), 5)?,
        p50: nearest_rank(values.to_vec(), 50)?,
        p95_fast_tail: nearest_rank(values.to_vec(), 95)?,
    })
}

// These values are serialized as human-readable floating-point rate reports;
// exact pass/fail promotion arithmetic belongs to the later hardware promoter.
#[allow(clippy::cast_precision_loss)]
fn u64_to_f64_for_reporting(value: u64) -> f64 {
    value as f64
}

fn expected_batch_result(
    batch: u32,
    metrics: &[SampleMetrics],
    clock: &ClockBinding,
) -> Result<BatchResult> {
    let samples = metrics
        .iter()
        .copied()
        .filter(|sample| sample.batch == batch)
        .collect::<Vec<_>>();
    let mut whole_kernel_rates = Vec::new();
    let mut whole_kernel_nanoseconds = Vec::new();
    let mut memory_rates = Vec::new();
    let mut capture_rates = Vec::new();
    let mut memory_cycles = Vec::new();
    let mut capture_cycles = Vec::new();
    for sample in &samples {
        if batch > 0 {
            let kernel_nanoseconds = u64_to_f64_for_reporting(sample.kernel_nanoseconds);
            whole_kernel_rates.push(f64::from(batch) * 1.0e9 / kernel_nanoseconds);
            whole_kernel_nanoseconds.push(kernel_nanoseconds / f64::from(batch));
        }
        if batch >= 2 {
            let intervals = f64::from(batch - 1);
            let memory_span = u64_to_f64_for_reporting(sample.memory_span);
            let capture_span = u64_to_f64_for_reporting(sample.capture_span);
            memory_cycles.push(memory_span / intervals);
            capture_cycles.push(capture_span / intervals);
            if let Some(clock_hz) = clock.routed_clock_hz {
                ensure!(
                    sample.memory_span > 0 && sample.capture_span > 0,
                    "routed rate span is zero"
                );
                let clock_hz = u64_to_f64_for_reporting(clock_hz);
                memory_rates.push(intervals * clock_hz / memory_span);
                capture_rates.push(intervals * clock_hz / capture_span);
            }
        }
    }
    let memory_percentiles = percentiles(&memory_cycles);
    let capture_percentiles = percentiles(&capture_cycles);
    let routed_memory_rate_at_p95_cycles = match (&memory_percentiles, clock.routed_clock_hz) {
        (Some(value), Some(clock_hz)) => Some(u64_to_f64_for_reporting(clock_hz) / value.p95),
        _ => None,
    };
    let routed_capture_rate_at_p95_cycles = match (&capture_percentiles, clock.routed_clock_hz) {
        (Some(value), Some(clock_hz)) => Some(u64_to_f64_for_reporting(clock_hz) / value.p95),
        _ => None,
    };
    Ok(BatchResult {
        batch,
        measured_samples: samples.len(),
        whole_kernel_nanoseconds_per_result: percentiles(&whole_kernel_nanoseconds),
        whole_kernel_signatures_per_second: rate_percentiles(&whole_kernel_rates),
        memory_retirement_cycles_per_result: memory_percentiles,
        capture_core_cycles_per_result: capture_percentiles,
        routed_memory_retirement_signatures_per_second: rate_percentiles(&memory_rates),
        routed_memory_rate_at_p95_cycles,
        routed_capture_signatures_per_second: rate_percentiles(&capture_rates),
        routed_capture_rate_at_p95_cycles,
    })
}

fn expected_performance_acceptance(
    metadata: &RunMetadata,
    metrics: &[SampleMetrics],
) -> Result<PerformanceAcceptance> {
    let required_batch_requested = metadata.batches.contains(&REQUIRED_PERFORMANCE_BATCH);
    let statistics = expected_batch_result(REQUIRED_PERFORMANCE_BATCH, metrics, &metadata.clock)?;
    let capture_p95 = statistics
        .capture_core_cycles_per_result
        .as_ref()
        .map(|value| value.p95);
    let capture_rate_at_bound =
        capture_p95.map(|cycles| u64_to_f64_for_reporting(PERFORMANCE_CLOCK_BOUND_HZ) / cycles);
    let whole_kernel_p95 = statistics
        .whole_kernel_nanoseconds_per_result
        .as_ref()
        .map(|value| value.p95);
    let whole_kernel_rate_at_p95 = whole_kernel_p95.map(|nanoseconds| 1.0e9 / nanoseconds);
    let capture_cycles_threshold_met =
        capture_p95.map(|cycles| cycles <= MAXIMUM_P95_CAPTURE_CYCLES);
    let capture_rate_threshold_met =
        capture_rate_at_bound.map(|rate| rate >= MINIMUM_SIGNATURES_PER_SECOND);
    let whole_kernel_rate_threshold_met =
        whole_kernel_rate_at_p95.map(|rate| rate >= MINIMUM_SIGNATURES_PER_SECOND);
    let sample_count_requirement_met = statistics.measured_samples >= REQUIRED_PERFORMANCE_SAMPLES;
    let threshold_observations_complete = required_batch_requested
        && sample_count_requirement_met
        && capture_p95.is_some()
        && capture_rate_at_bound.is_some()
        && whole_kernel_p95.is_some()
        && whole_kernel_rate_at_p95.is_some()
        && statistics.measured_samples
            == metrics
                .iter()
                .filter(|sample| sample.batch == REQUIRED_PERFORMANCE_BATCH)
                .count();
    let performance_thresholds_met = threshold_observations_complete
        && capture_cycles_threshold_met == Some(true)
        && capture_rate_threshold_met == Some(true)
        && whole_kernel_rate_threshold_met == Some(true);
    Ok(PerformanceAcceptance {
        required_positive_large_batch: REQUIRED_PERFORMANCE_BATCH,
        required_batch_requested,
        required_measured_samples: REQUIRED_PERFORMANCE_SAMPLES,
        observed_measured_samples: statistics.measured_samples,
        sample_count_requirement_met,
        policy_basis: PerformancePolicyBasis {
            clock_bound_hz: PERFORMANCE_CLOCK_BOUND_HZ,
            maximum_p95_capture_core_cycles_per_result: MAXIMUM_P95_CAPTURE_CYCLES,
            minimum_signatures_per_second: MINIMUM_SIGNATURES_PER_SECOND,
            clock_bound_is_arithmetic_normalization_not_routed_clock_evidence: true,
        },
        capture_core: CaptureCoreAcceptance {
            p95_cycles_per_result: capture_p95,
            rate_at_p95_cycles: capture_rate_at_bound,
            cycles_threshold_met: capture_cycles_threshold_met,
            rate_threshold_met: capture_rate_threshold_met,
        },
        whole_kernel: WholeKernelAcceptance {
            p95_nanoseconds_per_result: whole_kernel_p95,
            rate_at_p95_elapsed: whole_kernel_rate_at_p95,
            rate_threshold_met: whole_kernel_rate_threshold_met,
        },
        threshold_observations_complete,
        performance_thresholds_met,
        cryptographic_oracle_valid: None,
        accepted: false,
        promotable: false,
        blocking_reason: PERFORMANCE_BLOCKING_REASON.to_owned(),
    })
}

fn validate_results_statistics(
    results: &ResultsEnvelope,
    metadata: &RunMetadata,
    metrics: &[SampleMetrics],
) -> Result<()> {
    let expected_batches = metadata
        .batches
        .iter()
        .map(|&batch| expected_batch_result(batch, metrics, &metadata.clock))
        .collect::<Result<Vec<_>>>()?;
    ensure!(
        results.batches == expected_batches,
        "results batch statistics differ from sealed sample timing and summaries"
    );
    let expected_acceptance = expected_performance_acceptance(metadata, metrics)?;
    ensure!(
        results.performance_acceptance == expected_acceptance,
        "performance acceptance differs from the exact Phase-B policy recomputation"
    );
    Ok(())
}

fn validate_digest(digest: &DigestBinding, label: &str) -> Result<()> {
    ensure!(digest.bytes > 0, "{label} byte count must be positive");
    ensure!(
        is_lower_hex(&digest.sha256, 64),
        "{label} SHA-256 is not canonical"
    );
    Ok(())
}

fn validate_clock(clock: &ClockBinding) -> Result<()> {
    if let Some(requested) = clock.requested_clock_hz_metadata {
        ensure!(requested > 0, "requested clock metadata must be positive");
    }
    ensure!(
        clock.routed_clock_hz.is_some() == clock.routed_clock_evidence.is_some(),
        "routed clock and timing evidence must appear together"
    );
    if let Some(routed) = clock.routed_clock_hz {
        ensure!(routed > 0, "routed clock must be positive");
    }
    if let Some(evidence) = &clock.routed_clock_evidence {
        ensure!(
            !evidence.path.is_empty(),
            "routed-clock evidence path is empty"
        );
        validate_digest(&evidence.digest, "routed-clock evidence")?;
    }
    ensure!(
        clock.supports_hardware_rate_claim == clock.routed_clock_hz.is_some(),
        "clock rate-claim flag does not match its evidence"
    );
    Ok(())
}

fn expected_layout(snapshot: &ArchiveSnapshot, metadata: &RunMetadata) -> Result<Vec<SampleSpec>> {
    let mut expected_files = BTreeSet::from([
        COORDINATION_BURN_FILE.to_owned(),
        COORDINATION_INTENT_FILE.to_owned(),
        COORDINATION_RECEIPT_FILE.to_owned(),
        FINAL_IDENTITY_FILE.to_owned(),
        PRE_RUN_IDENTITY_FILE.to_owned(),
        "results.json".to_owned(),
        "run-metadata.json".to_owned(),
        PLATFORM_POST_RUN_FILE.to_owned(),
        PLATFORM_PRE_LOAD_FILE.to_owned(),
        XRT_HOST_FILE.to_owned(),
    ]);
    let mut expected_directories = BTreeSet::from(["samples".to_owned()]);
    let mut samples = Vec::new();
    for &batch in &metadata.batches {
        let batch_directory = format!("samples/batch-{batch:05}");
        expected_directories.insert(batch_directory.clone());
        for (phase, count) in [
            ("warmup", metadata.warmups_per_batch),
            ("measured", metadata.measured_runs_per_batch),
        ] {
            for index in 0..count {
                let relative = format!("{batch_directory}/{phase}-{index:03}");
                expected_directories.insert(relative.clone());
                for file in SAMPLE_FILES {
                    expected_files.insert(format!("{relative}/{file}"));
                }
                samples.push(SampleSpec {
                    relative,
                    batch,
                    phase,
                    index,
                });
            }
        }
    }
    ensure!(
        snapshot.files.keys().cloned().collect::<BTreeSet<_>>() == expected_files,
        "successful transport archive has missing or extra files"
    );
    ensure!(
        snapshot.directories == expected_directories,
        "successful transport archive has missing, extra, or empty directories"
    );
    ensure!(
        !samples.is_empty(),
        "transport archive has no oracle requests"
    );
    Ok(samples)
}

struct ValidatedSample {
    request: OracleRequest,
    request_sha256: String,
    metrics: SampleMetrics,
}

struct SampleValidationState<'a> {
    metadata: &'a RunMetadata,
    pre_run_identity_record: &'a FileRecord,
    coordination: &'a ValidatedCoordination,
    reservation_ids: HashSet<String>,
    nonces: HashSet<String>,
    reservation_chain: Option<(u64, String)>,
}

fn validate_sample_contract(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    state: &mut SampleValidationState<'_>,
) -> Result<ValidatedSample> {
    let (request, request_sha256) = validate_sample_request(
        snapshot,
        sample,
        state.metadata,
        state.pre_run_identity_record,
        state.coordination,
    )?;
    validate_request_artifacts(snapshot, sample, &request)?;
    validate_sample_reservation(
        snapshot,
        sample,
        &request,
        &mut state.reservation_ids,
        &mut state.nonces,
        &mut state.reservation_chain,
    )?;
    let metrics = validate_transport_artifacts(snapshot, sample, &request)?;
    Ok(ValidatedSample {
        request,
        request_sha256,
        metrics,
    })
}

fn validate_sample_request(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    metadata: &RunMetadata,
    pre_run_identity_record: &FileRecord,
    coordination: &ValidatedCoordination,
) -> Result<(OracleRequest, String)> {
    let request_path = sample_file(sample, "oracle-request.json");
    let request_bytes = read_verified_small(snapshot, &request_path, MAXIMUM_JSON_BYTES)?;
    let request_sha256 = sha256_bytes(&request_bytes);
    let request: OracleRequest = parse_strict_json(&request_bytes, &request_path)?;
    let sidecar_path = sample_file(sample, "oracle-request.sha256");
    let sidecar = read_verified_small(snapshot, &sidecar_path, 256)?;
    let expected_sidecar = format!("{request_sha256}  oracle-request.json\n");
    ensure!(
        sidecar == expected_sidecar.as_bytes(),
        "oracle-request sidecar mismatch"
    );

    ensure!(request.schema == 2, "oracle request schema must be 2");
    ensure!(request.kind == REQUEST_KIND, "oracle request kind mismatch");
    validate_coordination_consumer_binding(
        &request.run_id,
        &request.coordination_intent,
        &request.coordination_burn,
        &request.coordination_receipt,
        coordination,
        &format!("oracle request {request_path}"),
    )?;
    ensure!(
        request.profile == PROFILE,
        "oracle request profile mismatch"
    );
    ensure!(
        request.batch == sample.batch,
        "request batch does not match sample path"
    );
    ensure!(
        is_lower_hex(&request.nonce_hex, 16),
        "request nonce is not canonical 64-bit hex"
    );
    ensure!(
        u64::from_str_radix(&request.nonce_hex, 16)? == request.nonce_u64,
        "request nonce hex and integer differ"
    );
    ensure!(
        is_lower_hex(&request.reservation_record_sha256, 64),
        "request reservation digest is not canonical"
    );
    ensure!(
        request.xclbin.path == metadata.xclbin.path
            && request.xclbin.bytes == metadata.xclbin.digest.bytes
            && request.xclbin.sha256 == metadata.xclbin.digest.sha256,
        "request xclbin binding differs from run metadata"
    );
    ensure!(
        request.clock == metadata.clock,
        "request clock binding differs from metadata"
    );
    validate_root_binding_against_record(
        &request.hardware_identity_pre_run,
        PRE_RUN_IDENTITY_FILE,
        pre_run_identity_record,
        "oracle request pre-run hardware identity",
    )?;
    ensure!(
        request.cryptographic_oracle_valid.is_none(),
        "request pre-populated the Rust-oracle decision"
    );
    ensure!(
        !request.hardware_completion_promotable,
        "request is prematurely hardware-promotable"
    );
    let must_bind = request
        .required_oracle_result
        .must_bind
        .iter()
        .map(String::as_str)
        .collect::<Vec<_>>();
    ensure!(
        request.required_oracle_result.producer == "independent Rust oracle"
            && must_bind.len() == 7
            && must_bind[0] == "oracle-request.json sha256"
            && must_bind[1] == "coordination-intent.json sha256"
            && must_bind[2] == "coordination-burn.json sha256"
            && must_bind[3] == "coordination-receipt.json sha256"
            && must_bind[4] == "hardware-identity-pre-run.json sha256"
            && must_bind[5] == "top-level SHA256SUMS sha256"
            && must_bind[6] == "every input/output job"
            && request
                .required_oracle_result
                .must_not_mutate_transport_archive,
        "request's required-oracle contract changed"
    );
    Ok((request, request_sha256))
}

fn validate_request_artifacts(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    request: &OracleRequest,
) -> Result<()> {
    let logical_input = u64::from(sample.batch)
        .checked_mul(INPUT_BYTES_PER_JOB)
        .context("input byte count overflow")?;
    let logical_output = u64::from(sample.batch)
        .checked_mul(OUTPUT_SLOT_BYTES)
        .context("output byte count overflow")?;
    let output_allocation = if sample.batch == 0 {
        OUTPUT_SLOT_BYTES
    } else {
        logical_output
    };
    validate_artifact(
        snapshot,
        sample,
        &request.artifacts.input,
        "input.bin",
        logical_input,
        logical_input,
    )?;
    validate_artifact(
        snapshot,
        sample,
        &request.artifacts.full_output_allocation,
        "output.bin",
        output_allocation,
        output_allocation,
    )?;
    validate_artifact(
        snapshot,
        sample,
        &request.artifacts.summary,
        "summary.bin",
        SUMMARY_BYTES,
        SUMMARY_BYTES,
    )?;
    for (artifact, name) in [
        (&request.artifacts.diagnostic_registers, "diagnostics.json"),
        (
            &request.artifacts.untimed_transport_validation,
            "validation.json",
        ),
        (&request.artifacts.timing, "timing.json"),
        (&request.artifacts.transport_contract, "transport.json"),
    ] {
        let bytes = snapshot
            .files
            .get(&sample_file(sample, name))
            .with_context(|| format!("manifest omits {name}"))?
            .bytes;
        validate_artifact(snapshot, sample, artifact, name, bytes, bytes)?;
    }
    ensure!(
        request.output_contract.slot_bytes == OUTPUT_SLOT_BYTES
            && request.output_contract.valid_bytes_per_job == OUTPUT_VALID_BYTES
            && request.output_contract.padding_bytes_per_job == OUTPUT_PADDING_BYTES
            && request.output_contract.requested_output_slot_bytes == logical_output
            && request.output_contract.full_output_allocation_bytes == output_allocation,
        "request output contract has the wrong geometry"
    );
    Ok(())
}

fn validate_sample_reservation(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    request: &OracleRequest,
    reservation_ids: &mut HashSet<String>,
    nonces: &mut HashSet<String>,
    reservation_chain: &mut Option<(u64, String)>,
) -> Result<()> {
    let receipt_path = sample_file(sample, "reservation.json");
    let receipt_bytes = read_verified_small(snapshot, &receipt_path, MAXIMUM_JSON_BYTES)?;
    let receipt: ReservationReceipt = parse_strict_json(&receipt_bytes, &receipt_path)?;
    validate_reservation(&receipt, &receipt_bytes, request)?;
    advance_reservation_chain(reservation_chain, &receipt)?;
    ensure!(
        reservation_ids.insert(receipt.reservation_id),
        "transport archive repeats a reservation identity"
    );
    ensure!(
        nonces.insert(receipt.nonce),
        "transport archive repeats a Phase-B nonce"
    );
    Ok(())
}

fn validate_artifact(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    artifact: &ArtifactBinding,
    expected_file: &str,
    expected_logical: u64,
    expected_stored: u64,
) -> Result<()> {
    ensure!(
        artifact.file == expected_file,
        "oracle artifact filename mismatch"
    );
    ensure!(
        artifact.logical_bytes == expected_logical && artifact.stored_bytes == expected_stored,
        "oracle artifact byte geometry mismatch for {expected_file}"
    );
    ensure!(
        is_lower_hex(&artifact.sha256, 64),
        "oracle artifact SHA-256 is not canonical"
    );
    let sealed = snapshot
        .files
        .get(&sample_file(sample, expected_file))
        .with_context(|| format!("manifest omits {expected_file}"))?;
    ensure!(
        sealed.bytes == artifact.stored_bytes && sealed.sha256 == artifact.sha256,
        "oracle artifact binding differs from top manifest for {expected_file}"
    );
    Ok(())
}

fn validate_reservation(
    receipt: &ReservationReceipt,
    raw: &[u8],
    request: &OracleRequest,
) -> Result<()> {
    ensure!(receipt.schema == 1, "reservation schema must be 1");
    ensure!(receipt.phase == "phase_b", "reservation is not Phase B");
    ensure!(
        is_lower_hex(&receipt.reservation_id, 32),
        "reservation ID is malformed"
    );
    ensure!(
        receipt.nonce == request.nonce_hex,
        "reservation nonce differs from request"
    );
    ensure!(
        receipt.base.is_none(),
        "Phase-B reservation unexpectedly has a range base"
    );
    ensure!(
        receipt.batch == request.batch,
        "reservation batch differs from request"
    );
    ensure!(
        receipt.input_bytes == request.artifacts.input.stored_bytes,
        "reservation input length differs from request"
    );
    ensure!(
        receipt.input_sha256.as_deref() == Some(request.artifacts.input.sha256.as_str()),
        "reservation input digest differs from request"
    );
    ensure!(
        receipt.seed_count == request.batch,
        "reservation seed count differs from batch"
    );
    ensure!(
        is_lower_hex(&receipt.previous_sha256, 64),
        "reservation prior hash is malformed"
    );
    ensure!(
        receipt.record_sha256 == request.reservation_record_sha256,
        "reservation record digest differs from request"
    );
    let body = format!(
        "{{\"schema\":1,\"sequence\":{},\"phase\":\"phase_b\",\"reservation_id\":\"{}\",\"nonce\":\"{}\",\"base\":null,\"batch\":{},\"input_bytes\":{},\"input_sha256\":\"{}\",\"seed_count\":{},\"previous_sha256\":\"{}\"}}",
        receipt.sequence,
        receipt.reservation_id,
        receipt.nonce,
        receipt.batch,
        receipt.input_bytes,
        receipt.input_sha256.as_deref().expect("validated Some"),
        receipt.seed_count,
        receipt.previous_sha256,
    );
    ensure!(
        sha256_bytes(body.as_bytes()) == receipt.record_sha256,
        "reservation chain hash mismatch"
    );
    let mut complete = body;
    complete.pop();
    complete.push_str(",\"record_sha256\":\"");
    complete.push_str(&receipt.record_sha256);
    complete.push_str("\"}\n");
    ensure!(
        raw == complete.as_bytes(),
        "reservation receipt is not canonical JSONL"
    );
    Ok(())
}

fn advance_reservation_chain(
    chain: &mut Option<(u64, String)>,
    receipt: &ReservationReceipt,
) -> Result<()> {
    if let Some((previous_sequence, previous_digest)) = chain.as_ref() {
        ensure!(
            receipt.sequence
                == previous_sequence
                    .checked_add(1)
                    .context("reservation sequence overflow")?,
            "Phase-B reservation sequence is not contiguous in execution order"
        );
        ensure!(
            &receipt.previous_sha256 == previous_digest,
            "Phase-B reservation previous digest does not bind the prior sample"
        );
    }
    *chain = Some((receipt.sequence, receipt.record_sha256.clone()));
    Ok(())
}

fn validate_transport_artifacts(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    request: &OracleRequest,
) -> Result<SampleMetrics> {
    let validation: ValidationRecord =
        parse_archive_json(snapshot, &sample_file(sample, "validation.json"))?;
    ensure!(
        validation.schema == 1
            && validation.transport_valid
            && validation.validated_after_timing
            && validation.cryptographic_oracle_valid.is_none()
            && validation.external_rust_oracle_required,
        "untimed transport validation is not a strict successful handoff"
    );

    let transport: TransportContract =
        parse_archive_json(snapshot, &sample_file(sample, "transport.json"))?;
    validate_transport_contract(&transport, request)?;
    let diagnostics: Diagnostics =
        parse_archive_json(snapshot, &sample_file(sample, "diagnostics.json"))?;
    validate_diagnostics(&diagnostics, request, &transport)?;
    let timing: TimingRecord = parse_archive_json(snapshot, &sample_file(sample, "timing.json"))?;
    validate_timing(&timing)?;
    let summary_bytes =
        read_verified_small(snapshot, &sample_file(sample, "summary.bin"), SUMMARY_BYTES)?;
    ensure!(
        summary_bytes.len() == usize::try_from(SUMMARY_BYTES)?,
        "summary size mismatch"
    );
    let summary = decode_summary(&summary_bytes);
    let decoded: SummaryDecoded =
        parse_archive_json(snapshot, &sample_file(sample, "summary-decoded.json"))?;
    ensure!(
        summary == decoded,
        "decoded summary JSON differs from summary.bin"
    );
    validate_summary(&summary, &diagnostics, request)?;
    Ok(SampleMetrics {
        batch: request.batch,
        kernel_nanoseconds: timing.kernel_nanoseconds,
        memory_span: diagnostics.memory_span.decimal,
        capture_span: summary.capture_span_cycles,
    })
}

fn validate_transport_contract(
    transport: &TransportContract,
    request: &OracleRequest,
) -> Result<()> {
    let logical_input = u64::from(request.batch) * INPUT_BYTES_PER_JOB;
    let logical_output = u64::from(request.batch) * OUTPUT_SLOT_BYTES;
    let input_allocation = if request.batch == 0 {
        INPUT_BYTES_PER_JOB
    } else {
        logical_input
    };
    let output_allocation = if request.batch == 0 {
        OUTPUT_SLOT_BYTES
    } else {
        logical_output
    };
    ensure!(
        transport.schema == 1 && transport.batch == request.batch,
        "transport schema/batch mismatch"
    );
    ensure!(
        transport.abi_word == "0x48530001",
        "transport ABI word mismatch"
    );
    let argument_order = transport
        .kernel_argument_order
        .iter()
        .map(String::as_str)
        .collect::<Vec<_>>();
    ensure!(
        argument_order.len() == 6
            && argument_order[0] == "inputs"
            && argument_order[1] == "outputs"
            && argument_order[2] == "summary"
            && argument_order[3] == "batch"
            && argument_order[4] == "abi_word"
            && argument_order[5] == "nonce",
        "kernel argument order mismatch"
    );
    ensure!(
        transport.group_ids.inputs >= 0
            && transport.group_ids.outputs >= 0
            && transport.group_ids.summary >= 0
            && transport.group_ids.inputs != transport.group_ids.outputs
            && transport.group_ids.outputs == transport.group_ids.summary,
        "HBM group contract mismatch"
    );
    ensure!(
        transport.logical_bytes.inputs == logical_input
            && transport.logical_bytes.outputs == logical_output
            && transport.logical_bytes.summary == SUMMARY_BYTES
            && transport.bo_allocation_bytes.inputs == input_allocation
            && transport.bo_allocation_bytes.outputs == output_allocation
            && transport.bo_allocation_bytes.summary == SUMMARY_BYTES,
        "transport byte geometry mismatch"
    );
    ensure!(
        transport.zero_batch_sentinel == (request.batch == 0),
        "zero-batch sentinel flag mismatch"
    );
    ensure!(
        transport.reservation_record_sha256 == request.reservation_record_sha256,
        "transport reservation binding differs from request"
    );
    ensure!(
        transport.poison.output == OUTPUT_POISON && transport.poison.summary == SUMMARY_POISON,
        "transport poison contract mismatch"
    );
    Ok(())
}

fn validate_timing(timing: &TimingRecord) -> Result<()> {
    ensure!(timing.schema == 1, "timing schema must be 1");
    ensure!(
        timing.timed_region == TIMED_REGION,
        "timed-region label mismatch"
    );
    // Strict deserialization requires these telemetry fields to be present,
    // but schema 1 does not require their out-of-kernel durations to be nonzero.
    #[allow(clippy::no_effect_underscore_binding)]
    let _outside_timed_region = (
        timing.host_to_device_nanoseconds,
        timing.device_to_host_nanoseconds,
        timing.diagnostic_read_nanoseconds,
    );
    ensure!(
        timing.kernel_nanoseconds > 0,
        "kernel timing must be positive"
    );
    ensure!(
        timing.completion_state == 4,
        "XRT completion state is not COMPLETED"
    );
    Ok(())
}

fn validate_diagnostics(
    diagnostics: &Diagnostics,
    request: &OracleRequest,
    transport: &TransportContract,
) -> Result<()> {
    ensure!(diagnostics.schema == 1, "diagnostics schema must be 1");
    for word in [
        &diagnostics.ap_ctrl,
        &diagnostics.gie,
        &diagnostics.ier,
        &diagnostics.isr,
        &diagnostics.batch,
        &diagnostics.abi_word,
        &diagnostics.last_status,
        &diagnostics.progress0,
        &diagnostics.progress1,
        &diagnostics.queue_status,
        &diagnostics.first_error_job,
        &diagnostics.first_error_detail,
    ] {
        validate_word32(word)?;
    }
    for word in [
        &diagnostics.input_base,
        &diagnostics.output_base,
        &diagnostics.summary_base,
        &diagnostics.nonce,
        &diagnostics.output_stall,
        &diagnostics.memory_first,
        &diagnostics.memory_last,
        &diagnostics.memory_span,
    ] {
        validate_word64(word)?;
    }
    let batch = request.batch;
    ensure!(
        diagnostics.last_status.decimal == LAST_STATUS_SUCCESS
            && diagnostics.ap_ctrl.decimal & 0x0c == 0x0c
            && diagnostics.batch.decimal == batch
            && diagnostics.abi_word.decimal == ABI_WORD
            && diagnostics.nonce.decimal == request.nonce_u64
            && diagnostics.progress0.decimal & 0xffff == batch
            && diagnostics.progress0.decimal >> 16 == batch
            && diagnostics.progress1.decimal & 0xffff == batch
            && diagnostics.progress1.decimal >> 16 == 0
            && diagnostics.queue_status.decimal == 0
            && diagnostics.first_error_job.decimal == u32::MAX
            && diagnostics.first_error_detail.decimal == 0,
        "terminal diagnostic contract mismatch"
    );
    let output_end = diagnostics
        .output_base
        .decimal
        .checked_add(transport.bo_allocation_bytes.outputs)
        .context("output BO address overflow")?;
    let summary_end = diagnostics
        .summary_base
        .decimal
        .checked_add(SUMMARY_BYTES)
        .context("summary BO address overflow")?;
    ensure!(
        output_end <= diagnostics.summary_base.decimal
            || summary_end <= diagnostics.output_base.decimal,
        "output and summary BO ranges overlap"
    );
    Ok(())
}

fn validate_word32(word: &Word32) -> Result<()> {
    ensure!(
        word.hex == format!("0x{:08x}", word.decimal),
        "u32 diagnostic hex mismatch"
    );
    Ok(())
}

fn validate_word64(word: &Word64) -> Result<()> {
    ensure!(
        word.hex == format!("0x{:016x}", word.decimal),
        "u64 diagnostic hex mismatch"
    );
    Ok(())
}

fn decode_summary(bytes: &[u8]) -> SummaryDecoded {
    SummaryDecoded {
        schema: 1,
        magic: format!("0x{:08x}", read_le_u32(bytes, 0)),
        revision: read_le_u16(bytes, 4),
        profile: bytes[6],
        terminal: bytes[7],
        nonce: read_le_u64(bytes, 8),
        payload_interval_cycles: read_le_u64(bytes, 16),
        core_latency_cycles: read_le_u64(bytes, 24),
        capture_span_cycles: read_le_u64(bytes, 32),
        batch: read_le_u16(bytes, 40),
        input_ar: read_le_u16(bytes, 42),
        input_r: read_le_u16(bytes, 44),
        admissions: read_le_u16(bytes, 46),
        frames: read_le_u16(bytes, 48),
        signer_errors: read_le_u16(bytes, 50),
        payload_aw: read_le_u16(bytes, 52),
        payload_b: read_le_u16(bytes, 54),
        payload_w: read_le_u32(bytes, 56),
        memory_completed: read_le_u16(bytes, 60),
        error_flags: read_le_u16(bytes, 62),
    }
}

fn read_le_u16(bytes: &[u8], offset: usize) -> u16 {
    u16::from_le_bytes(
        bytes[offset..offset + 2]
            .try_into()
            .expect("fixed summary span"),
    )
}

fn read_le_u32(bytes: &[u8], offset: usize) -> u32 {
    u32::from_le_bytes(
        bytes[offset..offset + 4]
            .try_into()
            .expect("fixed summary span"),
    )
}

fn read_le_u64(bytes: &[u8], offset: usize) -> u64 {
    u64::from_le_bytes(
        bytes[offset..offset + 8]
            .try_into()
            .expect("fixed summary span"),
    )
}

fn validate_summary(
    summary: &SummaryDecoded,
    diagnostics: &Diagnostics,
    request: &OracleRequest,
) -> Result<()> {
    let batch16 = u16::try_from(request.batch).context("batch exceeds summary width")?;
    ensure!(
        summary.schema == 1
            && summary.magic == "0x31525348"
            && summary.revision == 1
            && summary.profile == 1
            && summary.terminal == 0
            && summary.nonce == request.nonce_u64
            && summary.batch == batch16
            && summary.input_ar == batch16
            && summary.input_r == batch16
            && summary.admissions == batch16
            && summary.frames == batch16
            && summary.signer_errors == 0
            && summary.payload_aw == batch16
            && summary.payload_b == batch16
            && summary.payload_w == 35 * request.batch
            && summary.memory_completed == batch16
            && summary.error_flags == 0,
        "HSR1 summary/counter contract mismatch"
    );
    match request.batch {
        0 => ensure!(
            summary.payload_interval_cycles == 1
                && summary.core_latency_cycles == 0
                && summary.capture_span_cycles == 0
                && diagnostics.memory_first.decimal == 0
                && diagnostics.memory_last.decimal == 0
                && diagnostics.memory_span.decimal == 0,
            "zero-batch timing contract mismatch"
        ),
        1 => ensure!(
            summary.payload_interval_cycles > 0
                && summary.core_latency_cycles > 0
                && summary.capture_span_cycles == 0
                && diagnostics.memory_span.decimal == 0
                && diagnostics.memory_first.decimal == diagnostics.memory_last.decimal,
            "singleton timing contract mismatch"
        ),
        _ => ensure!(
            summary.payload_interval_cycles > 0
                && summary.core_latency_cycles > 0
                && summary.capture_span_cycles > 0
                && diagnostics.memory_span.decimal > 0
                && diagnostics.memory_last.decimal >= diagnostics.memory_first.decimal
                && diagnostics.memory_span.decimal
                    == diagnostics.memory_last.decimal - diagnostics.memory_first.decimal,
            "sustained timing contract mismatch"
        ),
    }
    Ok(())
}

fn compute_sample_jobs(
    snapshot: &ArchiveSnapshot,
    sample: &SampleSpec,
    threads: usize,
) -> Result<Vec<JobComputation>> {
    if sample.batch == 0 {
        let output = open_verified_file(snapshot, &sample_file(sample, "output.bin"))?;
        let mut sentinel = [0_u8; OUTPUT_SLOT_BYTES_USIZE];
        output.read_exact_at(&mut sentinel, 0)?;
        output.finish()?;
        ensure!(
            sentinel.iter().enumerate().all(|(offset, byte)| {
                let offset = u64::try_from(offset).expect("output slot offset fits in u64");
                *byte == output_poison(offset)
            }),
            "zero-batch output sentinel changed"
        );
        return Ok(Vec::new());
    }
    let jobs = usize::try_from(sample.batch)?;
    let workers = threads.min(jobs);
    let chunk = jobs.div_ceil(workers);
    let input_relative = sample_file(sample, "input.bin");
    let output_relative = sample_file(sample, "output.bin");
    let sample_relative = sample.relative.clone();
    // Each large artifact is fully manifest-bound exactly once before any
    // worker can read it.  FileExt positional reads safely share these retained
    // descriptors without reopening an attacker-substitutable pathname.
    let input_file = open_verified_file(snapshot, &input_relative)?;
    let output_file = open_verified_file(snapshot, &output_relative)?;
    let mut ranges = Vec::new();
    let mut start = 0_usize;
    while start < jobs {
        let end = (start + chunk).min(jobs);
        ranges.push((start, end));
        start = end;
    }

    let computation = thread::scope(|scope| -> Result<Vec<JobComputation>> {
        let mut handles = Vec::with_capacity(ranges.len());
        for (start, end) in ranges {
            let sample_relative = sample_relative.clone();
            let input_file = &input_file;
            let output_file = &output_file;
            handles.push(scope.spawn(move || {
                compute_job_range(&sample_relative, input_file, output_file, start, end)
            }));
        }
        let mut computations = Vec::with_capacity(jobs);
        for handle in handles {
            let range = handle
                .join()
                .map_err(|_| anyhow::anyhow!("software-oracle worker panicked"))??;
            computations.extend(range);
        }
        ensure!(
            computations.len() == jobs,
            "software-oracle worker lost a job"
        );
        for (index, computation) in computations.iter().enumerate() {
            ensure!(
                usize::try_from(computation.transcript.job_index)? == index,
                "software-oracle results are not in deterministic job order"
            );
        }
        Ok(computations)
    });
    let input_finish = input_file.finish();
    let output_finish = output_file.finish();
    input_finish?;
    output_finish?;
    computation
}

fn compute_job_range(
    sample_relative: &str,
    input_file: &CheckedFile,
    output_file: &CheckedFile,
    start: usize,
    end: usize,
) -> Result<Vec<JobComputation>> {
    let oracle = Sha256GenericWots::new();
    let mut computations = Vec::with_capacity(end - start);
    for job in start..end {
        let job_u64 = u64::try_from(job)?;
        let mut input = [0_u8; INPUT_BYTES_PER_JOB_USIZE];
        let mut output = [0_u8; OUTPUT_SLOT_BYTES_USIZE];
        input_file.read_exact_at(&mut input, job_u64 * INPUT_BYTES_PER_JOB)?;
        output_file.read_exact_at(&mut output, job_u64 * OUTPUT_SLOT_BYTES)?;
        let private_seed: [u8; 32] = input[..32].try_into().expect("fixed input seed span");
        let message: [u8; 32] = input[32..].try_into().expect("fixed input message span");
        let seed_sha256: [u8; 32] = Sha256::digest(private_seed).into();
        let fused = oracle.sign_and_public_key_from_private_seed(
            PrivateSeed::<HashSigsSha256GenericV1>::new(private_seed),
            &MessageDigest::new(message),
        );
        let signature = fused.signature.to_bytes();
        let public_key = fused.public_key.to_bytes();
        let mut expected = [0_u8; FUSED_OUTPUT_BYTES];
        expected[..SIGNATURE_BYTES].copy_from_slice(&signature);
        expected[SIGNATURE_BYTES..].copy_from_slice(&public_key);
        let observed = &output[..OUTPUT_VALID_BYTES_USIZE];
        let payload_match = observed == expected.as_slice();
        let padding_match =
            output[OUTPUT_VALID_BYTES_USIZE..]
                .iter()
                .enumerate()
                .all(|(index, byte)| {
                    let index = u64::try_from(index).expect("output padding index fits in u64");
                    let global = job_u64 * OUTPUT_SLOT_BYTES + OUTPUT_VALID_BYTES + index;
                    *byte == output_poison(global)
                });
        computations.push(JobComputation {
            transcript: JobTranscript {
                schema: 1,
                sample: sample_relative.to_owned(),
                job_index: u32::try_from(job)?,
                input_sha256: hex_lower(&Sha256::digest(input)),
                expected_sha256: hex_lower(&Sha256::digest(expected)),
                observed_sha256: hex_lower(&Sha256::digest(observed)),
                r#match: payload_match,
            },
            seed_sha256,
            padding_match,
        });
        input.fill(0);
        output.fill(0);
    }
    Ok(computations)
}

fn output_poison(offset: u64) -> u8 {
    let mixed = offset.wrapping_mul(131).wrapping_add(0x5b) & 0xff;
    0xa5 ^ u8::try_from(mixed).expect("masked to one byte")
}

fn sample_file(sample: &SampleSpec, file: &str) -> String {
    format!("{}/{file}", sample.relative)
}

#[cfg(test)]
mod tests {
    use super::{
        TIMED_REGION, advance_reservation_chain, canonical_bdf, canonical_serial, output_poison,
        valid_utc, valid_xrt_build_date, validate_timing,
    };
    use crate::model::{ReservationReceipt, TimingRecord};

    #[test]
    fn poison_matches_cpp_boundaries() {
        assert_eq!(output_poison(0), 0xfe);
        assert_eq!(output_poison(2_207), 0x1d);
        assert_eq!(output_poison(2_208), 0x9e);
        assert_eq!(output_poison(4_095), 0x7d);
        assert_eq!(output_poison(4_096), 0xfe);
    }

    #[test]
    fn ancillary_timing_can_be_zero_but_kernel_timing_must_be_positive() {
        let timing = TimingRecord {
            schema: 1,
            timed_region: TIMED_REGION.to_owned(),
            host_to_device_nanoseconds: 0,
            kernel_nanoseconds: 1,
            device_to_host_nanoseconds: 0,
            diagnostic_read_nanoseconds: 0,
            completion_state: 4,
        };
        assert!(validate_timing(&timing).is_ok());

        let no_kernel_time = TimingRecord {
            kernel_nanoseconds: 0,
            ..timing
        };
        assert!(validate_timing(&no_kernel_time).is_err());
    }

    fn receipt(sequence: u64, previous: &str, record: &str) -> ReservationReceipt {
        ReservationReceipt {
            schema: 1,
            sequence,
            phase: "phase_b".to_owned(),
            reservation_id: "00".repeat(16),
            nonce: "00".repeat(8),
            base: None,
            batch: 1,
            input_bytes: 64,
            input_sha256: Some("11".repeat(32)),
            seed_count: 1,
            previous_sha256: previous.to_owned(),
            record_sha256: record.to_owned(),
        }
    }

    #[test]
    fn reservation_chain_requires_contiguous_sequence_and_digest() {
        let first_digest = "22".repeat(32);
        let mut chain = None;
        advance_reservation_chain(&mut chain, &receipt(41, &"00".repeat(32), &first_digest))
            .unwrap();
        advance_reservation_chain(&mut chain, &receipt(42, &first_digest, &"33".repeat(32)))
            .unwrap();

        let mut bad_sequence = Some((41, first_digest.clone()));
        assert!(
            advance_reservation_chain(
                &mut bad_sequence,
                &receipt(43, &first_digest, &"44".repeat(32)),
            )
            .is_err()
        );
        let mut bad_digest = Some((41, first_digest));
        assert!(
            advance_reservation_chain(
                &mut bad_digest,
                &receipt(42, &"55".repeat(32), &"66".repeat(32)),
            )
            .is_err()
        );
    }

    #[test]
    fn hardware_identity_text_fields_are_semantically_canonical() {
        assert!(canonical_bdf("0000:65:00.1"));
        assert!(canonical_bdf("ffff:ff:1f.7"));
        assert!(!canonical_bdf("0000:65:20.0"));
        assert!(!canonical_bdf("0000:65:ff.0"));
        assert!(canonical_serial("XMC-SERIAL_001.2"));
        assert!(!canonical_serial("---"));
        assert!(!canonical_serial("N/A"));
        assert!(valid_xrt_build_date("2024-02-29 23:59:59"));
        assert!(!valid_xrt_build_date("2023-02-29 12:00:00"));
        assert!(!valid_xrt_build_date("2023-13-01 12:00:00"));
        assert!(!valid_xrt_build_date("2023-12-01 24:00:00"));
    }

    #[test]
    fn utc_timestamp_requires_a_real_calendar_instant() {
        assert!(valid_utc("2024-02-29T23:59:59Z"));
        for impossible in [
            "2023-02-29T12:00:00Z",
            "2100-02-29T12:00:00Z",
            "2026-04-31T12:00:00Z",
            "2026-00-15T12:00:00Z",
            "2026-13-01T12:00:00Z",
            "2026-07-15T24:00:00Z",
            "2026-07-15T12:60:00Z",
            "2026-07-15T12:00:60Z",
        ] {
            assert!(!valid_utc(impossible), "accepted {impossible}");
        }
    }
}