feat(detection-core): integrate guarded admission custody

.kiro/specs/detection-core/tasks.md

@@ -343,7 +343,7 @@
   - _Requirements: 8.1, 8.3, 8.4, 8.5, 8.6, 14.3, 15.5_
   - _Boundary: Benchmark Readiness Adapter_
   - _Depends: 4.4, 8.2_
-- [ ] 9.6 Integrate benchmark-owned admission scoring and guarded custody
+- [x] 9.6 Integrate benchmark-owned admission scoring and guarded custody
   - Delegate neutral scoring to core while retaining population persistence and run orchestration in the benchmark.
   - Make the custodian the sole append-only writer and combine durable compare-and-swap spend plus label release as one guarded operation.
   - After the winning spend, produce and independently attest the canonical case release over the exact immutable result, evidence, change, and input identities supplied to scoring.

benchmarks/valid/admission_run.py

@@ -18,13 +18,19 @@ from __future__ import annotations
 import hashlib
 import json
 import tomllib
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 from benchmarks.unseen.replay import canonical
-from benchmarks.valid.admission_scoring import AdmissionCase, OrdinaryChange, admission_report
+from benchmarks.valid.admission_scoring import AdmissionCase, OrdinaryChange, admission_report, score_v2_admission
 from benchmarks.valid.arm_regression import BundleDigests, _input_digest, _prove_one
+from benchmarks.valid.population_custody import AdmissionSpendAuthority, GuardedPopulationCustodian, ReleaseReview
 from benchmarks.valid.prover_admission import AdmissionManifest, validate_admission_manifest
+from openultrasast.core.admission_case import AdmissionCaseResult
+from openultrasast.core.admission_report import AdmissionReplay
+from openultrasast.core.admission_spend import authorize_admission
 from openultrasast.preprocess import detect_language
 from openultrasast.source import canonical_language, ranges
 
@@ -37,12 +43,57 @@ __all__ = [
     "load_admission_population",
     "locate_function",
     "run_admission",
+    "run_v2_admission",
+    "V2AdmissionRunInput",
     "site_judgment",
 ]
 
 ARM_NAME = "access-control-admission"
 
 
+@dataclass(frozen=True)
+class V2AdmissionRunInput:
+    authority: AdmissionSpendAuthority
+    custodian: GuardedPopulationCustodian
+    load_labels: Callable[[], bytes]
+    analyze_labels: Callable[[bytes], Sequence[AdmissionCaseResult]]
+    review_release: ReleaseReview
+
+
+def run_v2_admission(value: V2AdmissionRunInput) -> AdmissionReplay:
+    """Run one synthetic v2 admission through guarded custody and core scoring."""
+    authority = value.authority
+    preparation = authorize_admission(
+        authority.manifest,
+        authority.denominator,
+        authority.readiness,
+        authority.context,
+        value.custodian,
+        authority.resolver,
+    )
+    guarded = value.custodian.spend_and_release_labels(authority, preparation, value.load_labels)
+    cases = tuple(value.analyze_labels(guarded.labels))
+    release = value.custodian.release_cases(
+        authority,
+        preparation,
+        guarded.receipt,
+        cases,
+        value.review_release,
+    )
+    return score_v2_admission(
+        authority.manifest,
+        authority.denominator,
+        authority.readiness,
+        preparation,
+        guarded.receipt,
+        release,
+        cases,
+        authority.context,
+        value.custodian,
+        authority.resolver,
+    )
+
+
 def _site_traces(record: dict[str, Any], path: str, start: int, end: int) -> list[dict[str, Any]]:
     """Traces whose operation site falls inside the labelled range."""
     matched = []

benchmarks/valid/admission_scoring.py

@@ -10,13 +10,23 @@ indeterminate and never converted to clean results; ordinary changes whose
 review found a real missing-authorization defect are refused before scoring.
 """
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any
 
 from benchmarks.unseen import score
 from benchmarks.valid.prover_admission import AdmissionManifest
+from openultrasast.core import canonical
+from openultrasast.core.admission_case import AdmissionCaseRelease, AdmissionCaseResult
+from openultrasast.core.admission_context import AdmissionValidationContext
+from openultrasast.core.admission_report import AdmissionReplay, score_admission, validate_admission_replay
+from openultrasast.core.admission_review import AdmissionManifestV2
+from openultrasast.core.admission_spend import AdmissionPreparation
+from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt
+from openultrasast.core.measurement import DenominatorPolicy
+from openultrasast.core.readiness import ReadinessResult
 
-__all__ = ["AdmissionCase", "AdmissionScoringError", "OrdinaryChange", "admission_report"]
+__all__ = ["AdmissionCase", "AdmissionScoringError", "OrdinaryChange", "admission_report", "score_v2_admission"]
 
 _JUDGMENTS = ("unguarded", "discharged", "indeterminate")
 
@@ -45,6 +55,35 @@ class OrdinaryChange:
     mislabeled_defect: bool
 
 
+def score_v2_admission(
+    manifest: AdmissionManifestV2,
+    denominator: DenominatorPolicy,
+    readiness: Sequence[ReadinessResult],
+    preparation: AdmissionPreparation,
+    receipt: PopulationUseReceipt,
+    release: AdmissionCaseRelease,
+    cases: Sequence[AdmissionCaseResult],
+    context: AdmissionValidationContext,
+    custody: PopulationCustodyPort,
+    resolver: canonical.ContentResolver,
+) -> AdmissionReplay:
+    """Delegate v2 scoring and authoritative replay to detection core."""
+    replay = score_admission(
+        manifest,
+        denominator,
+        tuple(readiness),
+        preparation,
+        receipt,
+        release,
+        tuple(cases),
+        context,
+        custody,
+        resolver,
+    )
+    validate_admission_replay(replay, context, custody, resolver)
+    return replay
+
+
 def _refuse_inputs(
     manifest: AdmissionManifest,
     cases: tuple[AdmissionCase, ...],

benchmarks/valid/population_custody.py

@@ -0,0 +1,243 @@
+"""Benchmark-owned guarded population custody for v2 admission."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, replace
+from typing import Protocol
+
+from openultrasast.core import canonical
+from openultrasast.core.admission_case import (
+    ADMISSION_CASE_RELEASE_SCHEMA,
+    AdmissionCaseRelease,
+    AdmissionCaseResult,
+    ReleasedCaseBinding,
+    derive_admission_case_release_digest,
+    derive_admission_case_release_payload_digest,
+    derive_released_case_binding_digest,
+    validate_admission_case_release,
+)
+from openultrasast.core.admission_context import AdmissionValidationContext
+from openultrasast.core.admission_review import AdmissionManifestV2
+from openultrasast.core.admission_spend import (
+    AdmissionPreparation,
+    AdmissionSpendError,
+    authorize_admission,
+    validate_pre_label_spend,
+)
+from openultrasast.core.custody import (
+    PopulationCustodyError,
+    PopulationCustodyLog,
+    PopulationCustodyPolicy,
+    PopulationCustodyPort,
+    PopulationUseReceipt,
+    append_population_use,
+    record_population_use,
+    validate_population_custody_log,
+)
+from openultrasast.core.factpack import IndependenceAttestation
+from openultrasast.core.measurement import DenominatorPolicy
+from openultrasast.core.readiness import ReadinessResult
+
+ReleaseReview = Callable[[str, str], IndependenceAttestation]
+
+
+class GuardedCustodyError(RuntimeError):
+    """A guarded append or label release did not complete."""
+
+
+class CompareAndSwapCustodyStore(Protocol):
+    """Durable storage seam owned by the benchmark custodian."""
+
+    def current_log(self, population_digest: str) -> PopulationCustodyLog: ...
+
+    def compare_and_swap(self, expected_log_digest: str, replacement: PopulationCustodyLog) -> bool: ...
+
+
+@dataclass(frozen=True)
+class GuardedLabelRelease:
+    receipt: PopulationUseReceipt
+    labels: bytes
+
+
+@dataclass(frozen=True)
+class AdmissionSpendAuthority:
+    manifest: AdmissionManifestV2
+    denominator: DenominatorPolicy
+    readiness: tuple[ReadinessResult, ...]
+    context: AdmissionValidationContext
+    resolver: canonical.ContentResolver
+
+
+class GuardedPopulationCustodian:
+    """Sole writer for one admission population."""
+
+    def __init__(
+        self,
+        population_digest: str,
+        custodian_identity: str,
+        policy: PopulationCustodyPolicy,
+        store: CompareAndSwapCustodyStore,
+        background: PopulationCustodyPort,
+    ) -> None:
+        self.population_digest = population_digest
+        self.custodian_identity = custodian_identity
+        self.policy = policy
+        self.store = store
+        self.background = background
+
+    def current_log(self, population_digest: str) -> PopulationCustodyLog:
+        if population_digest == self.population_digest:
+            return self.store.current_log(population_digest)
+        return self.background.current_log(population_digest)
+
+    def spend_and_release_labels(
+        self,
+        authority: AdmissionSpendAuthority,
+        preparation: AdmissionPreparation,
+        load_labels: Callable[[], bytes],
+    ) -> GuardedLabelRelease:
+        expected = self._authorize(authority)
+        if preparation != expected:
+            raise GuardedCustodyError("supplied preparation is not the authoritative core preparation")
+        request = preparation.label_release_request
+        if request.population_digest != self.population_digest:
+            raise GuardedCustodyError("label-release request names another population")
+        prior = self.store.current_log(self.population_digest)
+        try:
+            validate_population_custody_log(prior, self.policy)
+            receipt = record_population_use(request, prior, self.custodian_identity, self.policy)
+            replacement = append_population_use(prior, receipt, self.policy)
+        except PopulationCustodyError as exc:
+            raise GuardedCustodyError(f"custody successor refused: {exc}") from exc
+        if not self.store.compare_and_swap(prior.log_digest, replacement):
+            raise GuardedCustodyError("compare-and-swap lost the current custody head")
+        committed = self.store.current_log(self.population_digest)
+        validate_population_custody_log(committed, self.policy)
+        if committed != replacement:
+            raise GuardedCustodyError("durable custody append does not match the committed successor")
+        self._validate_spend(authority, preparation, receipt)
+        labels = load_labels()
+        if type(labels) is not bytes or not labels:
+            raise GuardedCustodyError("label release returned a zero-byte or invalid input")
+        return GuardedLabelRelease(receipt, labels)
+
+    def _authorize(self, authority: AdmissionSpendAuthority) -> AdmissionPreparation:
+        try:
+            return authorize_admission(
+                authority.manifest,
+                authority.denominator,
+                authority.readiness,
+                authority.context,
+                self,
+                authority.resolver,
+            )
+        except AdmissionSpendError as exc:
+            raise GuardedCustodyError(f"admission preparation refused: {exc}") from exc
+
+    def _validate_spend(
+        self,
+        authority: AdmissionSpendAuthority,
+        preparation: AdmissionPreparation,
+        receipt: PopulationUseReceipt,
+    ) -> None:
+        try:
+            validate_pre_label_spend(
+                authority.manifest,
+                authority.denominator,
+                authority.readiness,
+                authority.context,
+                preparation,
+                receipt,
+                self,
+                authority.resolver,
+            )
+        except AdmissionSpendError as exc:
+            raise GuardedCustodyError(f"committed admission spend refused: {exc}") from exc
+
+    def release_cases(
+        self,
+        authority: AdmissionSpendAuthority,
+        preparation: AdmissionPreparation,
+        receipt: PopulationUseReceipt,
+        cases: Sequence[AdmissionCaseResult],
+        review_release: ReleaseReview,
+    ) -> AdmissionCaseRelease:
+        selected = tuple(cases)
+        release = _build_release(
+            authority.manifest,
+            preparation,
+            receipt,
+            selected,
+            review_release,
+            self.custodian_identity,
+        )
+        return validate_admission_case_release(
+            authority.manifest,
+            authority.denominator,
+            authority.readiness,
+            authority.context,
+            preparation,
+            receipt,
+            release,
+            selected,
+            self,
+            authority.resolver,
+        )
+
+
+def _binding(case: AdmissionCaseResult) -> ReleasedCaseBinding:
+    evidence = tuple(sorted(item.input_digest for item in case.evidence_inputs))
+    changes = tuple(sorted({item.record.packet.provenance.change_input_digest for item in case.evidence_inputs}))
+    draft = ReleasedCaseBinding(
+        case.case_id_digest,
+        case.scope,
+        case.role,
+        case.pair_id_digest,
+        case.result_digest,
+        evidence,
+        changes,
+        case.input_manifest.manifest_digest,
+        "0" * 64,
+    )
+    return replace(draft, binding_digest=derive_released_case_binding_digest(draft))
+
+
+def _build_release(
+    manifest: AdmissionManifestV2,
+    preparation: AdmissionPreparation,
+    receipt: PopulationUseReceipt,
+    cases: tuple[AdmissionCaseResult, ...],
+    review_release: ReleaseReview,
+    custodian_identity: str,
+) -> AdmissionCaseRelease:
+    bindings = tuple(sorted((_binding(case) for case in cases), key=lambda item: item.case_id_digest))
+    draft = AdmissionCaseRelease(
+        ADMISSION_CASE_RELEASE_SCHEMA,
+        manifest.identity.population_digest,
+        manifest.manifest_digest,
+        preparation.authorization.authorization_digest,
+        preparation.preparation_digest,
+        receipt.receipt_digest,
+        custodian_identity,
+        bindings,
+        "0" * 64,
+        manifest.review,
+        "0" * 64,
+    )
+    payload = derive_admission_case_release_payload_digest(draft)
+    review = review_release(payload, custodian_identity)
+    if type(review) is not IndependenceAttestation:
+        raise GuardedCustodyError("release reviewer returned an invalid attestation")
+    reviewed = replace(draft, release_payload_digest=payload, review=review)
+    return replace(reviewed, release_digest=derive_admission_case_release_digest(reviewed))
+
+
+__all__ = [
+    "AdmissionSpendAuthority",
+    "CompareAndSwapCustodyStore",
+    "GuardedCustodyError",
+    "GuardedLabelRelease",
+    "GuardedPopulationCustodian",
+    "ReleaseReview",
+]

tests/test_detection_core_admission_adapter.py

@@ -0,0 +1,302 @@
+"""Benchmark-owned guarded admission integration for detection-core Task 9.6."""
+
+from __future__ import annotations
+
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+from pathlib import Path
+from threading import Lock
+
+import pytest
+from test_core_admission_metrics import _setup
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from benchmarks.valid.admission_run import V2AdmissionRunInput, run_v2_admission
+from benchmarks.valid.admission_scoring import score_v2_admission
+from benchmarks.valid.population_custody import (
+    AdmissionSpendAuthority,
+    GuardedCustodyError,
+    GuardedPopulationCustodian,
+    ReleaseReview,
+)
+from openultrasast.core.admission_case import AdmissionCaseResult, derive_admission_case_result_digest
+from openultrasast.core.admission_report import AdmissionDecision
+from openultrasast.core.admission_spend import AdmissionPreparation, authorize_admission
+from openultrasast.core.canonical import content_digest
+from openultrasast.core.custody import LineagePurpose, PopulationCustodyLog, PopulationStatus
+from openultrasast.core.factpack import IndependenceAttestation, derive_independence_attestation_digest
+
+
+class SimulatedCrash(RuntimeError):
+    pass
+
+
+class _Store:
+    def __init__(self, initial: PopulationCustodyLog, *, fail_before_append: bool = False) -> None:
+        self.log = initial
+        self.fail_before_append = fail_before_append
+        self.lock = Lock()
+
+    def current_log(self, population_digest: str) -> PopulationCustodyLog:
+        if population_digest != self.log.population_digest:
+            raise LookupError("population not found")
+        return self.log
+
+    def compare_and_swap(self, expected_log_digest: str, replacement: PopulationCustodyLog) -> bool:
+        with self.lock:
+            if self.fail_before_append:
+                raise SimulatedCrash("before durable append")
+            if self.log.log_digest != expected_log_digest:
+                return False
+            self.log = replacement
+            return True
+
+
+def _custodian(setup: object, store: _Store) -> GuardedPopulationCustodian:
+    population = setup.manifest.identity.population_digest  # type: ignore[attr-defined]
+    current = setup.custody.current_log(population)  # type: ignore[attr-defined]
+    return GuardedPopulationCustodian(
+        population,
+        current.custodian_identity,
+        setup.context.evidence.custody_policy,  # type: ignore[attr-defined]
+        store,
+        setup.custody,  # type: ignore[attr-defined]
+    )
+
+
+def _label_file(tmp_path: Path) -> Path:
+    path = tmp_path / "labels.json"
+    path.write_bytes(b'{"labels":["vulnerable","fixed","ordinary"]}')
+    assert path.stat().st_size > 0
+    return path
+
+
+def _reviewer(setup: object) -> ReleaseReview:
+    def review(payload: str, producer: str) -> IndependenceAttestation:
+        draft = IndependenceAttestation(
+            payload,
+            producer,
+            setup.manifest.review.reviewer_identity,  # type: ignore[attr-defined]
+            setup.manifest.review.conflict_disclosure,  # type: ignore[attr-defined]
+            "0" * 64,
+        )
+        return replace(draft, attestation_digest=derive_independence_attestation_digest(draft))
+
+    return review
+
+
+def _authority(setup: object) -> AdmissionSpendAuthority:
+    return AdmissionSpendAuthority(
+        setup.manifest,  # type: ignore[attr-defined]
+        setup.denominator,  # type: ignore[attr-defined]
+        setup.readiness,  # type: ignore[attr-defined]
+        setup.context,  # type: ignore[attr-defined]
+        setup.resolver,  # type: ignore[attr-defined]
+    )
+
+
+def _request(setup: object, custodian: GuardedPopulationCustodian, path: Path) -> V2AdmissionRunInput:
+    def analyze(labels: bytes) -> tuple[AdmissionCaseResult, ...]:
+        assert labels == path.read_bytes()
+        return setup.cases  # type: ignore[attr-defined,no-any-return]
+
+    return V2AdmissionRunInput(
+        _authority(setup),
+        custodian,
+        path.read_bytes,
+        analyze,
+        _reviewer(setup),
+    )
+
+
+def _prepare(setup: object, custodian: GuardedPopulationCustodian) -> AdmissionPreparation:
+    authority = _authority(setup)
+    return authorize_admission(
+        authority.manifest,
+        authority.denominator,
+        authority.readiness,
+        authority.context,
+        custodian,
+        authority.resolver,
+    )
+
+
+def test_synthetic_admission_runs_end_to_end_after_nonzero_label_read(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+
+    replay = run_v2_admission(_request(setup, custodian, _label_file(tmp_path)))
+
+    assert replay.report.decision in (AdmissionDecision.GO, AdmissionDecision.NO_GO)
+    assert replay.label_release_receipt.status is PopulationStatus.SPENT
+    assert replay.case_release.review.producer_identity == custodian.custodian_identity
+    assert replay.case_release.review.reviewer_identity != custodian.custodian_identity
+    assert store.log.status is PopulationStatus.SPENT
+
+
+def test_crash_before_append_releases_no_labels_and_preserves_available(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population), fail_before_append=True)
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+    reads = 0
+
+    def labels() -> bytes:
+        nonlocal reads
+        reads += 1
+        return _label_file(tmp_path).read_bytes()
+
+    with pytest.raises(SimulatedCrash, match="before durable append"):
+        custodian.spend_and_release_labels(_authority(setup), preparation, labels)
+    assert reads == 0
+    assert store.log.status is PopulationStatus.AVAILABLE
+
+
+@pytest.mark.parametrize("mutation", ("preparation", "authorization", "purpose", "influenced"))
+def test_invalid_preparation_never_spends_or_releases_labels(mutation: str) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+    if mutation == "preparation":
+        changed = replace(preparation, preparation_digest="0" * 64)
+    elif mutation == "authorization":
+        authorization = replace(preparation.authorization, authorization_digest="0" * 64)
+        changed = replace(preparation, authorization=authorization)
+    elif mutation == "purpose":
+        request = replace(preparation.label_release_request, purpose=LineagePurpose.PROMOTION)
+        changed = replace(preparation, label_release_request=request)
+    else:
+        request = replace(preparation.label_release_request, influenced_artifact_digest=content_digest("substituted"))
+        changed = replace(preparation, label_release_request=request)
+    reads = 0
+
+    def labels() -> bytes:
+        nonlocal reads
+        reads += 1
+        return b"escaped-labels"
+
+    with pytest.raises(GuardedCustodyError):
+        custodian.spend_and_release_labels(_authority(setup), changed, labels)
+    assert reads == 0
+    assert store.log.status is PopulationStatus.AVAILABLE
+
+
+def test_crash_after_append_leaves_population_spent(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+
+    def crash() -> bytes:
+        _label_file(tmp_path).read_bytes()
+        raise SimulatedCrash("after durable append")
+
+    with pytest.raises(SimulatedCrash, match="after durable append"):
+        custodian.spend_and_release_labels(_authority(setup), preparation, crash)
+    assert store.log.status is PopulationStatus.SPENT
+    assert store.log.receipt_count > 0
+
+
+def test_concurrent_preparations_release_labels_to_one_cas_winner(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+    path = _label_file(tmp_path)
+    reads = 0
+    lock = Lock()
+
+    def labels() -> bytes:
+        nonlocal reads
+        with lock:
+            reads += 1
+        return path.read_bytes()
+
+    def attempt() -> object:
+        try:
+            return custodian.spend_and_release_labels(_authority(setup), preparation, labels)
+        except GuardedCustodyError as exc:
+            return exc
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = tuple(pool.map(lambda _: attempt(), range(2)))
+    assert sum(not isinstance(item, GuardedCustodyError) for item in results) == 1
+    assert reads == 1
+    assert store.log.status is PopulationStatus.SPENT
+
+
+def test_swapped_operational_payload_is_refused_before_metrics(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+    authority = _authority(setup)
+    guarded = custodian.spend_and_release_labels(authority, preparation, _label_file(tmp_path).read_bytes)
+    release = custodian.release_cases(
+        authority,
+        preparation,
+        guarded.receipt,
+        setup.cases,
+        _reviewer(setup),
+    )
+    changed = replace(setup.cases[0], total_reply_ms=setup.cases[0].total_reply_ms + 1)
+    changed = replace(changed, result_digest=derive_admission_case_result_digest(changed))
+
+    with pytest.raises(ValueError, match="release|binding|case"):
+        score_v2_admission(
+            setup.manifest,
+            setup.denominator,
+            setup.readiness,
+            preparation,
+            guarded.receipt,
+            release,
+            (changed, *setup.cases[1:]),
+            setup.context,
+            custodian,
+            setup.resolver,
+        )
+
+
+def test_case_release_requires_an_independent_attestation(tmp_path: Path) -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+    authority = _authority(setup)
+    guarded = custodian.spend_and_release_labels(authority, preparation, _label_file(tmp_path).read_bytes)
+
+    def self_review(payload: str, producer: str) -> IndependenceAttestation:
+        draft = IndependenceAttestation(payload, producer, producer, setup.manifest.review.conflict_disclosure, "0" * 64)
+        return replace(draft, attestation_digest=derive_independence_attestation_digest(draft))
+
+    with pytest.raises(ValueError, match="independent"):
+        custodian.release_cases(
+            authority,
+            preparation,
+            guarded.receipt,
+            setup.cases,
+            self_review,
+        )
+
+
+def test_zero_byte_label_input_fails_after_monotonic_spend() -> None:
+    setup = _setup()
+    population = setup.manifest.identity.population_digest
+    store = _Store(setup.custody.current_log(population))
+    custodian = _custodian(setup, store)
+    preparation = _prepare(setup, custodian)
+
+    with pytest.raises(GuardedCustodyError, match="zero-byte"):
+        custodian.spend_and_release_labels(_authority(setup), preparation, lambda: b"")
+    assert store.log.status is PopulationStatus.SPENT