test(detection-core): validate end-to-end readiness paths

2026-11-13 01:48:17 +00:00 · 2026-11-04 10:53:09 +01:00
parent 586543d799
commit 62b1d90734
12 changed files with 1012 additions and 1 deletions
@@ -410,7 +410,7 @@
  - _Depends: 10.6_

 - [ ] 11. Validate readiness, failure visibility, and implementation boundaries
- [ ] 11.1 Run both reference families through the complete synthetic readiness path
+- [x] 11.1 Run both reference families through the complete synthetic readiness path
  - Start from a verified nonzero input and run recognition, transitions, security object/constraints, pair, development, reviewed synthetic admission, guarded spend, and consumer verdict.
  - Produce violation, bounded satisfaction, and bounded silence for both unlike families.
  - Require current prerequisite readiness identities at every later stage.
@@ -0,0 +1,430 @@
+"""Test-owned composition of the production detection-core readiness contracts."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, replace
+from pathlib import Path
+from typing import cast
+
+from detection_core_fixtures import InMemoryContentResolver
+from detection_core_source_authority import SourceAuthority, bind_source_inputs, bind_source_record, load_source_authority
+from test_core_admission_context import _resign_manifest
+from test_core_admission_metrics import (
+    _case_result,
+    _MetricSetup,
+    _rename_named_inputs,
+    _resign_census,
+    _resolver,
+    _setup,
+)
+from test_core_reference_families import _contract, _record
+from test_core_validation_v2 import _catalog, _named_inputs, _with_fact_set
+from test_detection_core_admission_adapter import _custodian, _reviewer, _Store
+
+from benchmarks.valid import readiness_contract
+from benchmarks.valid.admission_run import V2AdmissionRunInput, run_v2_admission
+from benchmarks.valid.population_custody import AdmissionSpendAuthority
+from openultrasast.core import admission_context, canonical, consumer, evidence, family, validation
+from openultrasast.core._admission_funnel import derive_case_funnel
+from openultrasast.core._admission_replay import AdmissionReplay
+from openultrasast.core.admission_case import derive_admission_case_result_digest, derive_case_coverage_digest
+from openultrasast.core.admission_context import AdmissionValidationContext
+from openultrasast.core.admission_metrics import AdmissionCaseResult, AdmissionCaseRole
+from openultrasast.core.admission_report import AdmissionDecision
+from openultrasast.core.admission_review import ScopeCensus
+from openultrasast.core.coverage import InputManifest, InputReadReceipt, validate_instrument
+from openultrasast.core.custody import PopulationStatus
+from openultrasast.core.family_catalog import FamilyCatalog
+from openultrasast.core.measurement import BaselineIdentity, DenominatorPolicy, IntervalMethod, ScopeKey, standard_metric_rules
+from openultrasast.core.readiness import (
+    AnalysisIdentity,
+    FixtureResult,
+    FixtureRole,
+    ReadinessResult,
+    ReadinessStage,
+    StageResult,
+)
+from openultrasast.core.verdict import derive_verdict_outputs
+
+
+@dataclass(frozen=True)
+class ReferencePathResult:
+    family_id: str
+    ecosystem: str
+    repository_bytes: int
+    opened_input_count: int
+    source_authority_digest: str
+    source_content_digests: tuple[str, ...]
+    inputs: tuple[validation.EvidenceValidationInput, ...]
+    readiness: tuple[ReadinessResult, ...]
+    admission: AdmissionReplay
+    verdicts: tuple[consumer.PushVerdict, ...]
+    population_status_at_label_read: PopulationStatus
+    label_read_count: int
+
+
+@dataclass(frozen=True)
+class _PreparedFamily:
+    contract: family.FamilyContract
+    ecosystem: str
+    inputs: tuple[validation.EvidenceValidationInput, ...]
+    resolver: InMemoryContentResolver
+    manifest: InputManifest
+    receipts: tuple[InputReadReceipt, ...]
+    authority: SourceAuthority
+
+
+def _load_object(path: Path) -> dict[str, object]:
+    payload = path.read_bytes()
+    if not payload:
+        raise ValueError(f"zero-byte fixture: {path}")
+    value = json.loads(payload)
+    if not isinstance(value, dict):
+        raise ValueError(f"fixture is not an object: {path}")
+    return cast(dict[str, object], value)
+
+
+def _prepare_family(fixture_root: Path, family_file: str, cases_file: str, repository: str) -> _PreparedFamily:
+    contract = _contract(_load_object(fixture_root / family_file))
+    raw_cases = cast(list[dict[str, object]], _load_object(fixture_root / cases_file)["cases"])
+    expected_cases = tuple((str(item["case_id"]), str(item["path"])) for item in raw_cases)
+    authority = load_source_authority(fixture_root / repository, repository, expected_cases)
+    inputs: list[validation.EvidenceValidationInput] = []
+    resolvers: list[InMemoryContentResolver] = []
+    tracked: list[tuple[str, Path]] = []
+    for raw, source in zip(raw_cases, authority.sources, strict=True):
+        case_id = str(raw["case_id"])
+        raw_record = _with_fact_set(_record(contract, raw), ())
+        base_record = bind_source_record(
+            raw_record,
+            canonical.content_digest("acquisition"),
+            canonical.content_digest("change-input"),
+        )
+        named, resolver = _named_inputs(base_record, contract, ())
+        named, resolver = _rename_named_inputs(named, resolver, case_id)
+        record = bind_source_record(base_record, authority.authority_digest, source.content_digest)
+        named, resolver = bind_source_inputs(named, resolver, authority.manifest_content, source)
+        inputs.append(validation.make_evidence_validation_input(record=record, named_inputs=named))
+        resolvers.append(resolver)
+        tracked.append((case_id, fixture_root / repository / source.path))
+    manifest, receipts = readiness_contract.read_tracked_inputs(tuple(tracked))
+    validate_instrument(manifest, receipts)
+    expected_reads = tuple((item.case_id, len(item.content), item.content_digest) for item in authority.sources)
+    actual_reads = tuple((item.input_id, item.byte_count, item.content_digest) for item in receipts)
+    if actual_reads != expected_reads:
+        raise ValueError("source authority does not match instrument read receipts")
+    ecosystem = inputs[0].record.packet.ecosystem
+    return _PreparedFamily(contract, ecosystem, tuple(inputs), _resolver(*resolvers), manifest, receipts, authority)
+
+
+def _evidence_context(base: _MetricSetup, catalog: FamilyCatalog) -> AdmissionValidationContext:
+    program_lineages = {item.lineage_manifest_digest for item in base.context.evidence.model_programs}
+    lineages = tuple(item for item in base.context.evidence.lineages if item.manifest_digest not in program_lineages)
+    evidence_context = validation.make_evidence_validation_context(
+        family_catalog=catalog,
+        fact_packs=(),
+        model_programs=(),
+        lineages=lineages,
+        fact_review_policy=base.context.evidence.fact_review_policy,
+        provenance_trust_policy=base.context.evidence.provenance_trust_policy,
+        custody_policy=base.context.evidence.custody_policy,
+    )
+    return admission_context.make_admission_validation_context(
+        evidence=evidence_context,
+        admission_review_policy=base.context.admission_review_policy,
+        proof_trust_policy=None,
+    )
+
+
+def _baseline(
+    base: _MetricSetup,
+    catalog: FamilyCatalog,
+    scope: ScopeKey,
+    denominator: DenominatorPolicy,
+    acquisition_record_digest: str,
+) -> BaselineIdentity:
+    entry = catalog.entries[0]
+    return replace(
+        base.manifest.identity,
+        claimed_scopes=(scope,),
+        family_contract_digests=(entry.contract_digest,),
+        family_contract_set_digest=catalog.catalog_digest,
+        fact_schema_digest=admission_context.derive_fact_schema_digest(),
+        fact_set_digest=validation.derive_fact_set_digest(()),
+        acquisition_record_digest=acquisition_record_digest,
+        analyzer_revision="reference-analyzer@v1",
+        configuration_digest=canonical.content_digest("configuration"),
+        model_program_digests=(),
+        model_program_set_digest=None,
+        denominator_policy_digest=denominator.policy_digest,
+        outcome_mapping_digest=admission_context.derive_outcome_mapping_digest(catalog),
+    )
+
+
+def _fixtures(inputs: tuple[validation.EvidenceValidationInput, ...]) -> tuple[FixtureResult, ...]:
+    return tuple(FixtureResult(role, item.record.record_id, True, ()) for role, item in zip(FixtureRole, inputs, strict=True))
+
+
+def _measurement(stage: ReadinessStage, prepared: _PreparedFamily) -> str:
+    records = tuple(item.record for item in prepared.inputs)
+    if stage is ReadinessStage.INSTRUMENT:
+        validate_instrument(prepared.manifest, prepared.receipts)
+        return canonical.content_digest(
+            {
+                "stage": stage.value,
+                "manifest": prepared.manifest.manifest_digest,
+                "source_authority": prepared.authority.authority_digest,
+                "reads": tuple((item.input_id, item.byte_count, item.content_digest) for item in prepared.receipts),
+            }
+        )
+    for record in records:
+        family.validate_record(prepared.contract, record)
+    if stage is ReadinessStage.RECOGNITION and any(not item.packet.security_object_id for item in records):
+        raise ValueError("reference recognition did not construct every security object")
+    if stage is ReadinessStage.TRANSITIONS and any(not item.packet.transitions for item in records):
+        raise ValueError("reference transition stage produced an empty relation set")
+    if stage is ReadinessStage.CONSTRAINTS and any(not item.packet.observations for item in records):
+        raise ValueError("reference constraint stage produced no observation")
+    outcomes = tuple(item.decision.outcome.value for item in records)
+    if stage is ReadinessStage.PAIRS and outcomes != ("violation", "satisfied", "indeterminate"):
+        raise ValueError("reference pair stage did not preserve the outcome triplet")
+    return canonical.content_digest({"stage": stage.value, "records": tuple(item.record_id for item in records)})
+
+
+def _prerequisite_chain(prepared: _PreparedFamily, identity: AnalysisIdentity) -> tuple[ReadinessResult, ...]:
+    results: list[ReadinessResult] = []
+    fixtures = _fixtures(prepared.inputs)
+    for stage in tuple(ReadinessStage)[:5]:
+        if results:
+            readiness_contract.require_stage_authorized(stage, tuple(results), identity)
+        measurement = _measurement(stage, prepared)
+        results.append(
+            readiness_contract.build_readiness_result(
+                arm_id="reference-arm",
+                family_id=prepared.contract.family_id,
+                ecosystem=prepared.ecosystem,
+                stage=stage,
+                result=StageResult.PASSED,
+                identity=identity,
+                predecessor_digests=tuple(item.result_digest for item in results),
+                fixture_results=fixtures,
+                measurement_digest=measurement,
+                correction_lever=None,
+                population_digest=None,
+                population_freeze_digest=None,
+                population_use_receipt_digests=(),
+            )
+        )
+    return tuple(results)
+
+
+def _development(
+    prepared: _PreparedFamily,
+    identity: AnalysisIdentity,
+    prior: tuple[ReadinessResult, ...],
+) -> ReadinessResult:
+    readiness_contract.require_stage_authorized(ReadinessStage.DEVELOPMENT, prior, identity)
+    observation = readiness_contract.build_development_observation(
+        arm_id="reference-arm",
+        family_id=prepared.contract.family_id,
+        ecosystem=prepared.ecosystem,
+        identity=identity,
+        population_digest=canonical.content_digest(f"{prepared.contract.family_id}:development"),
+        population_freeze_digest=canonical.content_digest(f"{prepared.contract.family_id}:development-freeze"),
+        population_use_receipt_digests=(canonical.content_digest(f"{prepared.contract.family_id}:development-use"),),
+        predecessor_digests=tuple(item.result_digest for item in prior),
+        fixture_results=_fixtures(prepared.inputs),
+        input_manifest=prepared.manifest,
+        input_read_receipts=prepared.receipts,
+    )
+    return observation.readiness_result
+
+
+def _canonical_case(
+    value: AdmissionCaseResult,
+    record: evidence.EvidenceRecord,
+    contract: family.FamilyContract,
+) -> AdmissionCaseResult:
+    pending = derive_verdict_outputs((record,), contract, canonical.content_digest("admission-coverage-pending"))
+    funnel = derive_case_funnel(value, (record,), len(pending.defects))
+    coverage_digest = derive_case_coverage_digest(value.case_id_digest, value.scope, funnel)
+    outputs = derive_verdict_outputs((record,), contract, coverage_digest)
+    changed = replace(
+        value,
+        defect_claims=outputs.defects,
+        satisfactions=outputs.satisfactions,
+        silence=outputs.silence,
+        coverage=funnel,
+        result_digest="0" * 64,
+    )
+    return replace(changed, result_digest=derive_admission_case_result_digest(changed))
+
+
+def _admission_setup(prepared: _PreparedFamily) -> _MetricSetup:
+    base = _setup()
+    catalog = _catalog(prepared.contract)
+    context = _evidence_context(base, catalog)
+    scope = ScopeKey(prepared.contract.family_id, prepared.ecosystem)
+    denominator = DenominatorPolicy(standard_metric_rules((scope,)), IntervalMethod.WILSON_95, "")
+    identity = _baseline(base, catalog, scope, denominator, prepared.authority.authority_digest)
+    current = admission_context.derive_analysis_identity(identity, catalog, scope, None)
+    prior = _prerequisite_chain(prepared, current)
+    readiness = (*prior, _development(prepared, current, prior))
+    resolver = _resolver(base.resolver, prepared.resolver)
+    pair = canonical.content_digest(f"{prepared.contract.family_id}:pair")
+    roles = (AdmissionCaseRole.VULNERABLE, AdmissionCaseRole.FIXED, AdmissionCaseRole.ORDINARY)
+    pairs = (pair, pair, None)
+    cases = []
+    for value, role, pair_id in zip(prepared.inputs, roles, pairs, strict=True):
+        case, resolver = _case_result(value, role, pair_id, resolver, prepared.contract)
+        cases.append(_canonical_case(case, value.record, prepared.contract))
+    semantic_sites = sum(item.coverage.sites_considered for item in cases)
+    row = ScopeCensus(scope, 3, 1, 1, 1, 1, semantic_sites, 0)
+    census = _resign_census(base.manifest.census, row, base.manifest.identity.population_digest)
+    manifest = _resign_manifest(
+        replace(
+            base.manifest,
+            identity=identity,
+            census=census,
+            readiness_digests=tuple(sorted(item.result_digest for item in readiness)),
+            denominator_policy_digest=denominator.policy_digest,
+            validation_context_digest=context.context_digest,
+        ),
+        census.review.conflict_disclosure,
+    )
+    return _MetricSetup(manifest, denominator, tuple(cases), context, base.custody, resolver, readiness)
+
+
+def _run_admission(setup: _MetricSetup, scratch: Path) -> tuple[AdmissionReplay, PopulationStatus, int]:
+    readiness_contract.require_stage_authorized(ReadinessStage.ADMISSION, setup.readiness, setup.readiness[0].identity)
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    label_path = scratch / f"{setup.manifest.identity.claimed_scopes[0].family_id}-labels.json"
+    label_path.write_bytes(b'{"roles":["vulnerable","fixed","ordinary"]}')
+    status_at_read: list[PopulationStatus] = []
+
+    def load_labels() -> bytes:
+        status_at_read.append(store.log.status)
+        return label_path.read_bytes()
+
+    value = V2AdmissionRunInput(
+        AdmissionSpendAuthority(setup.manifest, setup.denominator, setup.readiness, setup.context, setup.resolver),
+        custodian,
+        load_labels,
+        lambda _labels: setup.cases,
+        _reviewer(setup),
+    )
+    replay = run_v2_admission(value)
+    if replay.report.decision is not AdmissionDecision.GO:
+        raise ValueError("synthetic reference admission did not pass its reviewed threshold")
+    return replay, status_at_read[0], len(status_at_read)
+
+
+def _admission_result(
+    setup: _MetricSetup,
+    replay: AdmissionReplay,
+    prior: tuple[ReadinessResult, ...],
+) -> ReadinessResult:
+    baseline = setup.manifest.identity
+    return readiness_contract.build_readiness_result(
+        arm_id=baseline.arm_id,
+        family_id=baseline.claimed_scopes[0].family_id,
+        ecosystem=baseline.claimed_scopes[0].ecosystem,
+        stage=ReadinessStage.ADMISSION,
+        result=StageResult.PASSED,
+        identity=prior[0].identity,
+        predecessor_digests=tuple(item.result_digest for item in prior),
+        fixture_results=prior[0].fixture_results,
+        measurement_digest=replay.report.report_digest,
+        correction_lever=None,
+        population_digest=baseline.population_digest,
+        population_freeze_digest=baseline.freeze_digest,
+        population_use_receipt_digests=(replay.label_release_receipt.receipt_digest,),
+    )
+
+
+def _consume(
+    setup: _MetricSetup,
+    prepared: _PreparedFamily,
+    readiness: tuple[ReadinessResult, ...],
+) -> tuple[tuple[consumer.PushVerdict, ...], InMemoryContentResolver]:
+    from test_core_consumer_v2 import _request
+
+    readiness_contract.require_stage_authorized(ReadinessStage.INTEGRATION, readiness, readiness[0].identity)
+    resolver = setup.resolver
+    verdicts = []
+    for value in prepared.inputs:
+        request, resolver = _request(value, resolver)
+        verdicts.append(
+            consumer.consume_records(
+                request,
+                (value,),
+                setup.context.evidence,
+                setup.custody,
+                1,
+                resolver,
+            )
+        )
+    return tuple(verdicts), resolver
+
+
+def _integration_result(
+    prior: tuple[ReadinessResult, ...],
+    verdicts: tuple[consumer.PushVerdict, ...],
+) -> ReadinessResult:
+    first = prior[0]
+    measurement = canonical.content_digest(
+        {"requests": tuple(item.request_digest for item in verdicts), "records": tuple(item.input_record_digests for item in verdicts)}
+    )
+    return readiness_contract.build_readiness_result(
+        arm_id=first.arm_id,
+        family_id=first.family_id,
+        ecosystem=first.ecosystem,
+        stage=ReadinessStage.INTEGRATION,
+        result=StageResult.PASSED,
+        identity=first.identity,
+        predecessor_digests=tuple(item.result_digest for item in prior),
+        fixture_results=first.fixture_results,
+        measurement_digest=measurement,
+        correction_lever=None,
+        population_digest=None,
+        population_freeze_digest=None,
+        population_use_receipt_digests=(),
+    )
+
+
+def run_reference_family_path(
+    *,
+    fixture_root: Path,
+    family_file: str,
+    cases_file: str,
+    repository: str,
+    scratch: Path,
+) -> ReferencePathResult:
+    prepared = _prepare_family(fixture_root, family_file, cases_file, repository)
+    setup = _admission_setup(prepared)
+    replay, status_at_read, read_count = _run_admission(setup, scratch)
+    admission = _admission_result(setup, replay, setup.readiness)
+    through_admission = (*setup.readiness, admission)
+    verdicts, _resolver_value = _consume(setup, prepared, through_admission)
+    integration = _integration_result(through_admission, verdicts)
+    readiness = (*through_admission, integration)
+    return ReferencePathResult(
+        prepared.contract.family_id,
+        prepared.ecosystem,
+        sum(item.byte_count for item in prepared.receipts),
+        sum(item.opened for item in prepared.receipts),
+        prepared.authority.authority_digest,
+        tuple(item.content_digest for item in prepared.authority.sources),
+        prepared.inputs,
+        readiness,
+        replay,
+        verdicts,
+        status_at_read,
+        read_count,
+    )
+
+
+__all__ = ["ReferencePathResult", "run_reference_family_path"]
@@ -0,0 +1,291 @@
+"""Frozen source authority for detection-core end-to-end fixtures."""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+from collections.abc import Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass, replace
+from pathlib import Path, PurePosixPath
+from typing import cast
+
+from detection_core_fixtures import InMemoryContentResolver, stored_artifact
+
+from openultrasast.core import canonical, evidence, provenance
+
+_SCHEMA = "detection-core-source-authority-v1"
+_MANIFEST = "source_authority.json"
+_HEADER_KEYS = {"schema_version", "repository", "producer_identity", "reviewer_identity", "cases"}
+_CASE_KEYS = {"case_id", "path", "content_digest"}
+_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
+_FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC
+
+
+@dataclass(frozen=True)
+class ReviewedSource:
+    case_id: str
+    path: str
+    content: bytes
+    content_digest: str
+
+
+@dataclass(frozen=True)
+class SourceAuthority:
+    repository: str
+    manifest_content: bytes
+    authority_digest: str
+    sources: tuple[ReviewedSource, ...]
+
+
+def bind_source_record(
+    value: evidence.EvidenceRecord,
+    acquisition_digest: str,
+    change_input_digest: str,
+) -> evidence.EvidenceRecord:
+    stamp = replace(
+        value.packet.provenance,
+        analyzer_revision=canonical.content_digest("analyzer"),
+        acquisition_record_digest=acquisition_digest,
+        configuration_digest=canonical.content_digest("configuration"),
+        population_digest=canonical.content_digest("population"),
+        change_input_digest=change_input_digest,
+    )
+    packet = replace(value.packet, provenance=stamp)
+    decision = replace(value.decision, packet_digest=evidence.derive_packet_digest(packet))
+    return evidence.EvidenceRecord(packet, decision, "")
+
+
+def bind_source_inputs(
+    named: provenance.NamedInputs,
+    resolver: InMemoryContentResolver,
+    authority_content: bytes,
+    source: ReviewedSource,
+) -> tuple[provenance.NamedInputs, InMemoryContentResolver]:
+    artifacts = dict(resolver.artifacts)
+    items = []
+    replacements = {
+        provenance.ProvenanceInputRole.ACQUISITION: authority_content,
+        provenance.ProvenanceInputRole.CHANGE_INPUT: source.content,
+    }
+    for item in named.inputs:
+        content = replacements.get(item.role)
+        if content is not None:
+            reference, replacement = stored_artifact(content, artifact_id=item.content.artifact_id)
+            artifacts.update(replacement.artifacts)
+            item = replace(item, identity=reference.content_digest, content=reference)
+        items.append(item)
+    values = tuple(items)
+    digest = provenance.derive_named_inputs_digest(values)
+    return provenance.NamedInputs(values, digest), InMemoryContentResolver(artifacts)
+
+
+def _object(value: object, label: str) -> dict[str, object]:
+    if not isinstance(value, dict):
+        raise ValueError(f"source authority {label} must be an object")
+    return cast(dict[str, object], value)
+
+
+def _text(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"source authority {label} must be nonempty text")
+    return value
+
+
+def _canonical_path(value: object) -> str:
+    raw = _text(value, "path")
+    path = PurePosixPath(raw)
+    forbidden = "\x00" in raw or "\\" in raw or any(part in ("", ".", "..") for part in path.parts)
+    if path.is_absolute() or path.as_posix() != raw or forbidden:
+        raise ValueError(f"source authority path is not canonical: {raw}")
+    return raw
+
+
+@contextmanager
+def _repository_descriptor(repository_root: Path) -> Iterator[int]:
+    try:
+        descriptor = os.open(repository_root, _DIR_FLAGS)
+    except (OSError, ValueError) as exc:
+        raise ValueError("source authority repository root is unsafe") from exc
+    try:
+        if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            raise ValueError("source authority repository root is not a directory")
+        yield descriptor
+    finally:
+        os.close(descriptor)
+
+
+def _open_directory(parent: int, name: str) -> int:
+    try:
+        descriptor = os.open(name, _DIR_FLAGS, dir_fd=parent)
+    except (OSError, ValueError) as exc:
+        raise ValueError(f"source authority directory component is unsafe: {name}") from exc
+    try:
+        if stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            return descriptor
+        raise ValueError("source authority path component is not a directory")
+    except (OSError, ValueError) as exc:
+        if descriptor >= 0:
+            os.close(descriptor)
+        raise ValueError(f"source authority directory component is unsafe: {name}") from exc
+
+
+def _read_regular(root: int, relative: str) -> bytes:
+    parts = PurePosixPath(relative).parts
+    directory = os.dup(root)
+    try:
+        for part in parts[:-1]:
+            child = _open_directory(directory, part)
+            os.close(directory)
+            directory = child
+        return _read_final_regular(directory, parts[-1], relative)
+    finally:
+        os.close(directory)
+
+
+def _read_final_regular(parent: int, name: str, relative: str) -> bytes:
+    try:
+        descriptor = os.open(name, _FILE_FLAGS, dir_fd=parent)
+    except (OSError, ValueError) as exc:
+        raise ValueError(f"source authority entry is unsafe: {relative}") from exc
+    try:
+        before = os.fstat(descriptor)
+        if not stat.S_ISREG(before.st_mode):
+            raise ValueError(f"source authority entry is not a regular file: {relative}")
+        content = _read_descriptor(descriptor, relative)
+        after = os.fstat(descriptor)
+        if len(content) != before.st_size or after.st_size != before.st_size:
+            raise ValueError(f"source authority entry changed while reading: {relative}")
+        return content
+    finally:
+        os.close(descriptor)
+
+
+def _read_descriptor(descriptor: int, relative: str) -> bytes:
+    chunks: list[bytes] = []
+    try:
+        while chunk := os.read(descriptor, 65536):
+            chunks.append(chunk)
+    except OSError as exc:
+        raise ValueError(f"source authority entry could not be read: {relative}") from exc
+    return b"".join(chunks)
+
+
+def _read_source(root: int, case: dict[str, object]) -> ReviewedSource:
+    if set(case) != _CASE_KEYS:
+        raise ValueError("source authority case fields do not match the closed schema")
+    case_id = _text(case["case_id"], "case_id")
+    relative = _canonical_path(case["path"])
+    expected_digest = _text(case["content_digest"], "content_digest")
+    content = _read_regular(root, relative)
+    if not content:
+        raise ValueError(f"source authority input is empty: {case_id}")
+    if canonical.bytes_digest(content) != expected_digest:
+        raise ValueError(f"source authority content mismatch: {case_id}")
+    return ReviewedSource(case_id, relative, content, expected_digest)
+
+
+def _allowed_directories(files: set[str]) -> set[str]:
+    directories: set[str] = set()
+    for value in files:
+        parts = PurePosixPath(value).parts
+        directories.update("/".join(parts[:index]) for index in range(1, len(parts)))
+    return directories
+
+
+def _entry_mode(entry: os.DirEntry[str], relative: str) -> int:
+    try:
+        return entry.stat(follow_symlinks=False).st_mode
+    except OSError as exc:
+        raise ValueError(f"source authority entry cannot be inspected: {relative}") from exc
+
+
+def _scan_child(parent: int, entry: os.DirEntry[str], relative: str, files: set[str], directories: set[str]) -> set[str]:
+    if relative not in directories:
+        raise ValueError(f"source authority contains an additional directory: {relative}")
+    descriptor = _open_directory(parent, entry.name)
+    try:
+        return _scan_directory(descriptor, relative, files, directories)
+    finally:
+        os.close(descriptor)
+
+
+def _scan_entry(parent: int, entry: os.DirEntry[str], prefix: str, files: set[str], directories: set[str]) -> set[str]:
+    relative = f"{prefix}/{entry.name}" if prefix else entry.name
+    mode = _entry_mode(entry, relative)
+    if stat.S_ISLNK(mode):
+        raise ValueError(f"source authority contains a symlink: {relative}")
+    if stat.S_ISDIR(mode):
+        return _scan_child(parent, entry, relative, files, directories)
+    if stat.S_ISREG(mode) and relative in files:
+        return {relative}
+    kind = "additional file" if stat.S_ISREG(mode) else "special entry"
+    raise ValueError(f"source authority contains an {kind}: {relative}")
+
+
+def _scan_directory(root: int, prefix: str, files: set[str], directories: set[str]) -> set[str]:
+    found: set[str] = set()
+    try:
+        with os.scandir(root) as entries:
+            for entry in entries:
+                found.update(_scan_entry(root, entry, prefix, files, directories))
+    except OSError as exc:
+        raise ValueError("source authority repository census failed") from exc
+    return found
+
+
+def _validate_repository_entries(root: int, source_paths: tuple[str, ...]) -> None:
+    allowed_files = {_MANIFEST, *source_paths}
+    found = _scan_directory(root, "", allowed_files, _allowed_directories(allowed_files))
+    if found != allowed_files:
+        raise ValueError("source authority entries do not match the exact repository")
+
+
+def load_source_authority(
+    repository_root: Path,
+    repository: str,
+    expected_cases: tuple[tuple[str, str], ...],
+) -> SourceAuthority:
+    with _repository_descriptor(repository_root) as root:
+        manifest_content = _read_regular(root, _MANIFEST)
+        manifest = _validate_manifest(manifest_content, repository)
+        cases = _manifest_cases(manifest, expected_cases)
+        _validate_repository_entries(root, tuple(path for _, path in expected_cases))
+        sources = tuple(_read_source(root, item) for item in cases)
+    return SourceAuthority(repository, manifest_content, canonical.bytes_digest(manifest_content), sources)
+
+
+def _validate_manifest(content: bytes, repository: str) -> dict[str, object]:
+    if not content:
+        raise ValueError("source authority manifest is empty")
+    manifest = _object(json.loads(content), "manifest")
+    if set(manifest) != _HEADER_KEYS or manifest.get("schema_version") != _SCHEMA:
+        raise ValueError("source authority manifest does not match the closed schema")
+    if manifest.get("repository") != repository:
+        raise ValueError("source authority repository identity does not match")
+    producer = _text(manifest["producer_identity"], "producer_identity")
+    reviewer = _text(manifest["reviewer_identity"], "reviewer_identity")
+    if producer == reviewer:
+        raise ValueError("source authority requires an independent reviewer")
+    return manifest
+
+
+def _manifest_cases(manifest: dict[str, object], expected_cases: tuple[tuple[str, str], ...]) -> tuple[dict[str, object], ...]:
+    raw_cases = manifest["cases"]
+    if not isinstance(raw_cases, list):
+        raise ValueError("source authority cases must be a list")
+    cases = tuple(_object(item, "case") for item in raw_cases)
+    declared = tuple((_text(item.get("case_id"), "case_id"), _canonical_path(item.get("path"))) for item in cases)
+    if declared != expected_cases or len(set(declared)) != len(declared):
+        raise ValueError("source authority cases do not match the exact authored cases")
+    return cases
+
+
+__all__ = [
+    "ReviewedSource",
+    "SourceAuthority",
+    "bind_source_inputs",
+    "bind_source_record",
+    "load_source_authority",
+]
@@ -0,0 +1,3 @@
+def publish(request: object, store: object, policy: object) -> None:
+    if policy.permits(request):
+        store.publish(request)
@@ -0,0 +1,3 @@
+def publish(request: object, store: object, allowed: bool) -> None:
+    if allowed:
+        store.publish(request)
@@ -0,0 +1,2 @@
+def publish(request: object, store: object) -> None:
+    store.publish(request)
@@ -0,0 +1,23 @@
+{
+  "schema_version": "detection-core-source-authority-v1",
+  "repository": "obligation_repository",
+  "producer_identity": "reference-fixture-author@v1",
+  "reviewer_identity": "independent-reference-review@v1",
+  "cases": [
+    {
+      "case_id": "obligation-positive",
+      "path": "reference/obligation_positive.py",
+      "content_digest": "e29307cd0ae6d1d817c19fd79bfa4d494d602fc5daf9aad4e6da9caa1ee31e29"
+    },
+    {
+      "case_id": "obligation-negative",
+      "path": "reference/obligation_negative.py",
+      "content_digest": "223e1065ef30aa0cb1073125e2f845c1b2b5a498400ac34cdc5f78bb0621cd7a"
+    },
+    {
+      "case_id": "obligation-missing",
+      "path": "reference/obligation_missing.py",
+      "content_digest": "c119b846e6797a9e1b6c96aff73426c1f53a46053ab074f88931e7d07383652c"
+    }
+  ]
+}
@@ -0,0 +1,2 @@
+def handle(value: str, effect: object, transform: object) -> None:
+    effect.write(transform.apply(value))
@@ -0,0 +1,2 @@
+def handle(value: str, effect: object) -> None:
+    effect.write(value.strip())
@@ -0,0 +1,2 @@
+def handle(value: str, effect: object) -> None:
+    effect.write(value)
@@ -0,0 +1,23 @@
+{
+  "schema_version": "detection-core-source-authority-v1",
+  "repository": "taint_repository",
+  "producer_identity": "reference-fixture-author@v1",
+  "reviewer_identity": "independent-reference-review@v1",
+  "cases": [
+    {
+      "case_id": "flow-positive",
+      "path": "reference/flow_positive.py",
+      "content_digest": "ece3add1d60cf01527cb83746b20c3be1f9468c589677932830efbb6a5d67a48"
+    },
+    {
+      "case_id": "flow-negative",
+      "path": "reference/flow_negative.py",
+      "content_digest": "2413778080ea6626b47dff8ee82a54f88902707a90e1f41c67a7380bc8784a5f"
+    },
+    {
+      "case_id": "flow-missing",
+      "path": "reference/flow_missing.py",
+      "content_digest": "662b99c3c093b38de535ab21b17a1acfe10bb70911c8b83768844b607424c345"
+    }
+  ]
+}
@@ -0,0 +1,230 @@
+"""Two unlike families traverse the complete synthetic readiness path."""
+
+from __future__ import annotations
+
+import os
+import shutil
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+from detection_core_end_to_end import ReferencePathResult, run_reference_family_path
+
+from benchmarks.valid import readiness_contract
+from openultrasast.core import canonical
+from openultrasast.core.admission_report import AdmissionDecision
+from openultrasast.core.custody import PopulationStatus
+from openultrasast.core.evidence import NeutralOutcome
+from openultrasast.core.readiness import ReadinessEvidenceUse, ReadinessStage, StageResult
+
+FIXTURES = Path(__file__).parent / "fixtures" / "detection_core"
+
+
+def _copied_obligation_fixture(tmp_path: Path) -> Path:
+    fixture_root = tmp_path / "detection-core"
+    shutil.copytree(FIXTURES, fixture_root)
+    return fixture_root
+
+
+def _run_obligation(fixture_root: Path, scratch: Path) -> ReferencePathResult:
+    return run_reference_family_path(
+        fixture_root=fixture_root,
+        family_file="obligation_family.json",
+        cases_file="obligation_cases.json",
+        repository="obligation_repository",
+        scratch=scratch,
+    )
+
+
+def _assert_obligation_refused(fixture_root: Path, scratch: Path) -> None:
+    with pytest.raises(ValueError, match="source authority"):
+        _run_obligation(fixture_root, scratch)
+    assert not tuple(scratch.glob("*-labels.json"))
+
+
+@pytest.mark.parametrize(
+    ("family_file", "cases_file", "repository", "family_id", "ecosystem", "silence_reason"),
+    (
+        (
+            "obligation_family.json",
+            "obligation_cases.json",
+            "obligation_repository",
+            "operation-constraint",
+            "reference-obligation",
+            "missing_constraint_relation",
+        ),
+        (
+            "taint_family.json",
+            "taint_cases.json",
+            "taint_repository",
+            "ordered-taint-flow",
+            "reference-flow",
+            "missing_flow_segment",
+        ),
+    ),
+)
+def test_reference_family_completes_synthetic_readiness_path(
+    tmp_path: Path,
+    family_file: str,
+    cases_file: str,
+    repository: str,
+    family_id: str,
+    ecosystem: str,
+    silence_reason: str,
+) -> None:
+    result = run_reference_family_path(
+        fixture_root=FIXTURES,
+        family_file=family_file,
+        cases_file=cases_file,
+        repository=repository,
+        scratch=tmp_path,
+    )
+
+    _assert_instrument_and_readiness(result, family_id, ecosystem)
+    _assert_bounded_outputs(result, silence_reason)
+    _assert_reviewed_guarded_admission(result)
+
+
+def test_replaced_authored_source_is_refused_before_admission(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    source = fixture_root / "obligation_repository/reference/obligation_positive.py"
+    source.write_text("def unrelated() -> int:\n    return 7\n")
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_swapped_authored_sources_are_refused_before_admission(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    positive = fixture_root / "obligation_repository/reference/obligation_positive.py"
+    negative = fixture_root / "obligation_repository/reference/obligation_negative.py"
+    positive_bytes, negative_bytes = positive.read_bytes(), negative.read_bytes()
+    positive.write_bytes(negative_bytes)
+    negative.write_bytes(positive_bytes)
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_nested_authority_named_file_is_refused_before_admission(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    nested = fixture_root / "obligation_repository/reference/source_authority.json"
+    nested.write_bytes(b"")
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_outside_root_authority_symlink_is_refused_without_descriptor_leak(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    manifest = fixture_root / "obligation_repository/source_authority.json"
+    outside = tmp_path / "outside-authority.json"
+    manifest.rename(outside)
+    manifest.symlink_to(outside)
+    descriptor_root = Path("/proc/self/fd")
+    before = len(tuple(descriptor_root.iterdir()))
+
+    for _attempt in range(5):
+        _assert_obligation_refused(fixture_root, tmp_path)
+
+    assert len(tuple(descriptor_root.iterdir())) == before
+
+
+def test_inside_root_source_symlink_is_refused_before_admission(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    repository = fixture_root / "obligation_repository"
+    source = repository / "reference/obligation_positive.py"
+    hidden = repository / "reference/source_authority.json"
+    source.rename(hidden)
+    source.symlink_to(hidden.name)
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_outside_root_source_symlink_is_refused_before_admission(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    source = fixture_root / "obligation_repository/reference/obligation_positive.py"
+    outside = tmp_path / "outside-source.py"
+    source.rename(outside)
+    source.symlink_to(outside)
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_additional_fifo_is_refused_without_opening_it(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    fifo = fixture_root / "obligation_repository/reference/untracked.pipe"
+    os.mkfifo(fifo)
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def test_declared_fifo_is_refused_without_opening_it(tmp_path: Path) -> None:
+    fixture_root = _copied_obligation_fixture(tmp_path)
+    source = fixture_root / "obligation_repository/reference/obligation_positive.py"
+    source.unlink()
+    os.mkfifo(source)
+
+    _assert_obligation_refused(fixture_root, tmp_path)
+
+
+def _assert_instrument_and_readiness(
+    result: ReferencePathResult,
+    family_id: str,
+    ecosystem: str,
+) -> None:
+    assert result.family_id == family_id
+    assert result.ecosystem == ecosystem
+    assert result.repository_bytes > 0
+    assert result.opened_input_count == 3
+    identity = result.readiness[0].identity
+    assert identity.acquisition_record_digest == result.source_authority_digest
+    assert {item.record.packet.provenance.acquisition_record_digest for item in result.inputs} == {result.source_authority_digest}
+    assert tuple(item.record.packet.provenance.change_input_digest for item in result.inputs) == result.source_content_digests
+    assert tuple(item.stage for item in result.readiness) == tuple(ReadinessStage)
+    assert tuple(item.result for item in result.readiness) == (StageResult.PASSED,) * len(ReadinessStage)
+    assert tuple(item.evidence_use for item in result.readiness) == (
+        ReadinessEvidenceUse.REGRESSION_FIXTURE,
+        ReadinessEvidenceUse.REGRESSION_FIXTURE,
+        ReadinessEvidenceUse.REGRESSION_FIXTURE,
+        ReadinessEvidenceUse.REGRESSION_FIXTURE,
+        ReadinessEvidenceUse.REGRESSION_FIXTURE,
+        ReadinessEvidenceUse.DEVELOPMENT_POPULATION,
+        ReadinessEvidenceUse.FRESH_ADMISSION,
+        ReadinessEvidenceUse.CONSUMER_INTEGRATION,
+    )
+    for index, item in enumerate(result.readiness):
+        assert item.predecessor_digests == tuple(prior.result_digest for prior in result.readiness[:index])
+        assert item.identity == result.readiness[0].identity
+    changed_identity = replace(identity, acquisition_record_digest=canonical.content_digest("different-source-authority"))
+    with pytest.raises(readiness_contract.ReadinessBlocked, match="readiness is stale"):
+        readiness_contract.require_stage_authorized(ReadinessStage.ADMISSION, result.readiness[:6], changed_identity)
+
+
+def _assert_bounded_outputs(result: ReferencePathResult, silence_reason: str) -> None:
+    assert tuple(item.record.decision.outcome for item in result.inputs) == (
+        NeutralOutcome.VIOLATION,
+        NeutralOutcome.SATISFIED,
+        NeutralOutcome.INDETERMINATE,
+    )
+    assert tuple((len(item.defects), len(item.satisfactions), len(item.silence)) for item in result.verdicts) == (
+        (1, 0, 0),
+        (0, 1, 0),
+        (0, 0, 1),
+    )
+    assert result.verdicts[2].silence[0].reasons == (silence_reason,)
+    assert all(item.total_reply_ms > 0 for item in result.verdicts)
+
+
+def _assert_reviewed_guarded_admission(result: ReferencePathResult) -> None:
+    replay = result.admission
+    assert replay.manifest.label_isolated is True
+    assert replay.manifest.labels_opened is False
+    assert replay.manifest.review.producer_identity != replay.manifest.review.reviewer_identity
+    assert replay.label_release_receipt.status is PopulationStatus.SPENT
+    assert replay.report.decision is AdmissionDecision.GO
+    assert result.population_status_at_label_read is PopulationStatus.SPENT
+    assert result.label_read_count == 1
+    assert replay.readiness == result.readiness[:6]
+    references = tuple(
+        (reference.stage, reference.result, reference.result_digest) for reference in replay.report.measurement.prerequisite_readiness
+    )
+    expected = tuple((item.stage, item.result, item.result_digest) for item in replay.readiness)
+    assert references == expected