feat(detection-core): enforce append-only population custody

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

@@ -50,7 +50,7 @@
   - _Requirements: 10.1, 10.4_
   - _Boundary: Provenance and Lineage_
   - _Depends: 1.2_
-- [ ] 2.3 Enforce append-only population custody
+- [x] 2.3 Enforce append-only population custody
   - Construct and validate contiguous reviewed-custodian receipt chains against the complete current log.
   - Require compare-and-swap requests to name the expected log head and derive the next sequence and predecessor.
   - Make spent status monotonic and reject old prefixes, omitted receipts, broken chains, and later availability claims.

src/openultrasast/core/custody.py

@@ -0,0 +1,363 @@
+"""Pure append-only population custody contracts."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, replace
+from enum import Enum
+from typing import Protocol
+
+from openultrasast.core import canonical
+
+CUSTODY_LOG_SCHEMA = "openultrasast.detection-core.population-custody-log.v2"
+CUSTODY_POLICY_SCHEMA = "openultrasast.detection-core.population-custody-policy.v2"
+CUSTODY_REQUEST_SCHEMA = "openultrasast.detection-core.population-use-request.v2"
+CUSTODY_RECEIPT_SCHEMA = "openultrasast.detection-core.population-use-receipt.v2"
+_DIGEST = re.compile(r"[0-9a-f]{64}")
+
+
+def _require_digest(value: object, field_name: str) -> None:
+    if not isinstance(value, str) or _DIGEST.fullmatch(value) is None:
+        raise canonical.DigestFormatError(f"{field_name} must be a full lowercase SHA-256 digest")
+
+
+def _require_text(value: object, field_name: str) -> None:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{field_name} must be nonempty text")
+
+
+class PopulationCustodyError(ValueError):
+    """A population custody chain or compare-and-swap request is invalid."""
+
+
+class LineagePurpose(Enum):
+    PROMPT = "prompt"
+    EXAMPLE = "example"
+    THRESHOLD = "threshold"
+    FITTED_MODEL = "fitted_model"
+    SEMANTIC_FACT = "semantic_fact"
+    MATCHING_BEHAVIOR = "matching_behavior"
+    DECISION_MAPPING = "decision_mapping"
+    PROMOTION = "promotion"
+    ADMISSION_LABEL_OPEN = "admission_label_open"
+
+
+class PopulationStatus(Enum):
+    AVAILABLE = "available"
+    SPENT = "spent"
+
+
+@dataclass(frozen=True)
+class PopulationUseRequest:
+    population_digest: str
+    arm_id: str
+    purpose: LineagePurpose
+    labels_accessed: bool
+    influenced_artifact_digest: str | None
+    expected_log_digest: str
+
+    def __post_init__(self) -> None:
+        _require_digest(self.population_digest, "population_digest")
+        _require_text(self.arm_id, "arm_id")
+        if not isinstance(self.purpose, LineagePurpose):
+            raise TypeError("purpose must be a LineagePurpose")
+        if not isinstance(self.labels_accessed, bool):
+            raise TypeError("labels_accessed must be boolean")
+        if self.influenced_artifact_digest is not None:
+            _require_digest(self.influenced_artifact_digest, "influenced_artifact_digest")
+        _require_digest(self.expected_log_digest, "expected_log_digest")
+
+
+@dataclass(frozen=True)
+class PopulationUseReceipt:
+    request: PopulationUseRequest
+    request_digest: str
+    population_digest: str
+    custodian_identity: str
+    sequence: int
+    predecessor_digest: str | None
+    prior_status: PopulationStatus
+    status: PopulationStatus
+    reasons: tuple[str, ...]
+    receipt_digest: str
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.request, PopulationUseRequest):
+            raise TypeError("request must be a PopulationUseRequest")
+        for name in ("request_digest", "population_digest", "custodian_identity", "receipt_digest"):
+            _require_digest(getattr(self, name), name)
+        if self.predecessor_digest is not None:
+            _require_digest(self.predecessor_digest, "predecessor_digest")
+        if isinstance(self.sequence, bool) or not isinstance(self.sequence, int) or self.sequence <= 0:
+            raise ValueError("sequence must be a positive integer")
+        if not isinstance(self.prior_status, PopulationStatus) or not isinstance(self.status, PopulationStatus):
+            raise TypeError("receipt status must be a PopulationStatus")
+        if not isinstance(self.reasons, tuple) or not all(isinstance(reason, str) and reason for reason in self.reasons):
+            raise TypeError("reasons must be a tuple of nonempty text")
+
+
+@dataclass(frozen=True)
+class PopulationCustodyPolicy:
+    reviewed_custodian_digests: tuple[str, ...]
+    policy_digest: str
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.reviewed_custodian_digests, tuple) or not self.reviewed_custodian_digests:
+            raise ValueError("reviewed_custodian_digests must be a nonempty tuple")
+        for digest in self.reviewed_custodian_digests:
+            _require_digest(digest, "reviewed_custodian_digest")
+        if len(set(self.reviewed_custodian_digests)) != len(self.reviewed_custodian_digests):
+            raise ValueError("reviewed_custodian_digests must be unique")
+        _require_digest(self.policy_digest, "policy_digest")
+
+
+@dataclass(frozen=True)
+class PopulationCustodyLog:
+    population_digest: str
+    custodian_identity: str
+    receipts: tuple[PopulationUseReceipt, ...]
+    receipt_count: int
+    head_receipt_digest: str | None
+    status: PopulationStatus
+    log_digest: str
+
+    def __post_init__(self) -> None:
+        _require_digest(self.population_digest, "population_digest")
+        _require_digest(self.custodian_identity, "custodian_identity")
+        if not isinstance(self.receipts, tuple) or not all(isinstance(item, PopulationUseReceipt) for item in self.receipts):
+            raise TypeError("receipts must be a tuple of PopulationUseReceipt values")
+        if isinstance(self.receipt_count, bool) or not isinstance(self.receipt_count, int) or self.receipt_count < 0:
+            raise ValueError("receipt_count must be a nonnegative integer")
+        if self.head_receipt_digest is not None:
+            _require_digest(self.head_receipt_digest, "head_receipt_digest")
+        if not isinstance(self.status, PopulationStatus):
+            raise TypeError("status must be a PopulationStatus")
+        _require_digest(self.log_digest, "log_digest")
+
+
+class PopulationCustodyPort(Protocol):
+    def current_log(self, population_digest: str) -> PopulationCustodyLog: ...
+
+
+def _request_value(request: PopulationUseRequest) -> canonical.JsonValue:
+    return {
+        "schema_version": CUSTODY_REQUEST_SCHEMA,
+        "population_digest": request.population_digest,
+        "arm_id": request.arm_id,
+        "purpose": request.purpose.value,
+        "labels_accessed": request.labels_accessed,
+        "influenced_artifact_digest": request.influenced_artifact_digest,
+        "expected_log_digest": request.expected_log_digest,
+    }
+
+
+def derive_population_use_request_digest(request: PopulationUseRequest) -> str:
+    return canonical.content_digest(_request_value(request))
+
+
+def _receipt_value(receipt: PopulationUseReceipt) -> canonical.JsonValue:
+    return {
+        "schema_version": CUSTODY_RECEIPT_SCHEMA,
+        "request": _request_value(receipt.request),
+        "request_digest": receipt.request_digest,
+        "population_digest": receipt.population_digest,
+        "custodian_identity": receipt.custodian_identity,
+        "sequence": receipt.sequence,
+        "predecessor_digest": receipt.predecessor_digest,
+        "prior_status": receipt.prior_status.value,
+        "status": receipt.status.value,
+        "reasons": receipt.reasons,
+    }
+
+
+def derive_population_use_receipt_digest(receipt: PopulationUseReceipt) -> str:
+    return canonical.content_digest(_receipt_value(receipt))
+
+
+def derive_population_custody_policy_digest(reviewed_custodian_digests: tuple[str, ...]) -> str:
+    return canonical.content_digest(
+        {"schema_version": CUSTODY_POLICY_SCHEMA, "reviewed_custodian_digests": tuple(sorted(reviewed_custodian_digests))}
+    )
+
+
+def _log_digest(population: str, custodian: str, receipts: tuple[PopulationUseReceipt, ...], status: PopulationStatus) -> str:
+    return canonical.content_digest(
+        {
+            "schema_version": CUSTODY_LOG_SCHEMA,
+            "population_digest": population,
+            "custodian_identity": custodian,
+            "receipt_digests": tuple(receipt.receipt_digest for receipt in receipts),
+            "receipt_count": len(receipts),
+            "head_receipt_digest": receipts[-1].receipt_digest if receipts else None,
+            "status": status.value,
+        }
+    )
+
+
+def empty_population_custody_log(population_digest: str, custodian_identity: str) -> PopulationCustodyLog:
+    return PopulationCustodyLog(
+        population_digest,
+        custodian_identity,
+        (),
+        0,
+        None,
+        PopulationStatus.AVAILABLE,
+        _log_digest(population_digest, custodian_identity, (), PopulationStatus.AVAILABLE),
+    )
+
+
+def _validate_custody_policy(policy: PopulationCustodyPolicy) -> None:
+    expected = derive_population_custody_policy_digest(policy.reviewed_custodian_digests)
+    if policy.policy_digest != expected:
+        raise PopulationCustodyError("population custody policy digest cannot be re-derived")
+
+
+def _validate_receipt_link(
+    log: PopulationCustodyLog,
+    receipt: PopulationUseReceipt,
+    sequence: int,
+    prefix: tuple[PopulationUseReceipt, ...],
+    status: PopulationStatus,
+) -> None:
+    predecessor = prefix[-1].receipt_digest if prefix else None
+    if receipt.sequence != sequence:
+        raise PopulationCustodyError("receipt sequence is not contiguous")
+    if receipt.predecessor_digest != predecessor:
+        raise PopulationCustodyError("receipt predecessor does not match complete chain")
+    if receipt.population_digest != log.population_digest or receipt.request.population_digest != log.population_digest:
+        raise PopulationCustodyError("receipt population does not match log")
+    if receipt.custodian_identity != log.custodian_identity:
+        raise PopulationCustodyError("receipt custodian does not match log")
+    expected = _log_digest(log.population_digest, log.custodian_identity, prefix, status)
+    if receipt.request.expected_log_digest != expected:
+        raise PopulationCustodyError("receipt expected log does not match prefix")
+
+
+def _validate_receipt_identity(receipt: PopulationUseReceipt, status: PopulationStatus) -> PopulationStatus:
+    expected_status = PopulationStatus.SPENT if status is PopulationStatus.SPENT or receipt.request.labels_accessed else status
+    if receipt.prior_status is not status or receipt.status is not expected_status:
+        raise PopulationCustodyError("receipt status is not monotonic")
+    expected_reasons = ("labels_accessed",) if receipt.request.labels_accessed else ()
+    if status is PopulationStatus.SPENT and not receipt.request.labels_accessed:
+        expected_reasons = ("population_already_spent",)
+    if receipt.reasons != expected_reasons:
+        raise PopulationCustodyError("receipt reasons do not match status transition")
+    if receipt.request_digest != derive_population_use_request_digest(receipt.request):
+        raise PopulationCustodyError("request digest cannot be re-derived")
+    if receipt.receipt_digest != derive_population_use_receipt_digest(receipt):
+        raise PopulationCustodyError("receipt digest cannot be re-derived")
+    return expected_status
+
+
+def _validate_log_metadata(log: PopulationCustodyLog, status: PopulationStatus) -> None:
+    if log.receipt_count != len(log.receipts):
+        raise PopulationCustodyError("receipt count does not match complete chain")
+    head = log.receipts[-1].receipt_digest if log.receipts else None
+    if log.head_receipt_digest != head:
+        raise PopulationCustodyError("head receipt does not match complete chain")
+    if log.status is not status:
+        raise PopulationCustodyError("log status does not match receipt chain")
+    if log.log_digest != _log_digest(log.population_digest, log.custodian_identity, log.receipts, status):
+        raise PopulationCustodyError("log digest cannot be re-derived")
+
+
+def validate_population_custody_log(
+    log: PopulationCustodyLog,
+    policy: PopulationCustodyPolicy,
+    *,
+    current: PopulationCustodyPort | None = None,
+) -> None:
+    _validate_custody_policy(policy)
+    if log.custodian_identity not in policy.reviewed_custodian_digests:
+        raise PopulationCustodyError("population custodian is not reviewed")
+    status = PopulationStatus.AVAILABLE
+    prefix: tuple[PopulationUseReceipt, ...] = ()
+    for sequence, receipt in enumerate(log.receipts, 1):
+        _validate_receipt_link(log, receipt, sequence, prefix, status)
+        status = _validate_receipt_identity(receipt, status)
+        prefix = (*prefix, receipt)
+    _validate_log_metadata(log, status)
+    if current is not None:
+        current_log = current.current_log(log.population_digest)
+        validate_population_custody_log(current_log, policy)
+        if current_log.log_digest != log.log_digest:
+            raise PopulationCustodyError("candidate is not the current complete log")
+
+
+def record_population_use(
+    request: PopulationUseRequest,
+    prior: PopulationCustodyLog,
+    custodian_identity: str,
+    policy: PopulationCustodyPolicy,
+) -> PopulationUseReceipt:
+    validate_population_custody_log(prior, policy)
+    if custodian_identity not in policy.reviewed_custodian_digests:
+        raise PopulationCustodyError("population custodian is not reviewed")
+    if custodian_identity != prior.custodian_identity:
+        raise PopulationCustodyError("population custodian does not match log")
+    if request.population_digest != prior.population_digest:
+        raise PopulationCustodyError("request population does not match log")
+    if request.expected_log_digest != prior.log_digest:
+        raise PopulationCustodyError("request expected log does not match current head")
+    status = PopulationStatus.SPENT if prior.status is PopulationStatus.SPENT or request.labels_accessed else prior.status
+    reasons: tuple[str, ...]
+    if request.labels_accessed:
+        reasons = ("labels_accessed",)
+    else:
+        reasons = ("population_already_spent",) if prior.status is PopulationStatus.SPENT else ()
+    provisional = PopulationUseReceipt(
+        request,
+        derive_population_use_request_digest(request),
+        prior.population_digest,
+        custodian_identity,
+        prior.receipt_count + 1,
+        prior.head_receipt_digest,
+        prior.status,
+        status,
+        reasons,
+        "0" * 64,
+    )
+    return replace(provisional, receipt_digest=derive_population_use_receipt_digest(provisional))
+
+
+def append_population_use(
+    prior: PopulationCustodyLog, receipt: PopulationUseReceipt, policy: PopulationCustodyPolicy
+) -> PopulationCustodyLog:
+    validate_population_custody_log(prior, policy)
+    if receipt.request.expected_log_digest != prior.log_digest:
+        raise PopulationCustodyError("receipt expected log does not match prior log")
+    receipts = (*prior.receipts, receipt)
+    provisional = PopulationCustodyLog(
+        prior.population_digest,
+        prior.custodian_identity,
+        receipts,
+        len(receipts),
+        receipt.receipt_digest,
+        receipt.status,
+        "0" * 64,
+    )
+    result = replace(
+        provisional,
+        log_digest=_log_digest(provisional.population_digest, provisional.custodian_identity, receipts, provisional.status),
+    )
+    validate_population_custody_log(result, policy)
+    return result
+
+
+__all__ = [
+    "LineagePurpose",
+    "PopulationCustodyError",
+    "PopulationCustodyLog",
+    "PopulationCustodyPolicy",
+    "PopulationCustodyPort",
+    "PopulationStatus",
+    "PopulationUseReceipt",
+    "PopulationUseRequest",
+    "append_population_use",
+    "derive_population_custody_policy_digest",
+    "derive_population_use_receipt_digest",
+    "derive_population_use_request_digest",
+    "empty_population_custody_log",
+    "record_population_use",
+    "validate_population_custody_log",
+]

src/openultrasast/core/provenance.py

@@ -11,6 +11,27 @@ from enum import Enum
 
 from openultrasast.core import canonical
 from openultrasast.core.canonical import ContentRef, ContentResolver
+from openultrasast.core.custody import (
+    CUSTODY_LOG_SCHEMA,
+    CUSTODY_POLICY_SCHEMA,
+    CUSTODY_RECEIPT_SCHEMA,
+    CUSTODY_REQUEST_SCHEMA,
+    LineagePurpose,
+    PopulationCustodyError,
+    PopulationCustodyLog,
+    PopulationCustodyPolicy,
+    PopulationCustodyPort,
+    PopulationStatus,
+    PopulationUseReceipt,
+    PopulationUseRequest,
+    append_population_use,
+    derive_population_custody_policy_digest,
+    derive_population_use_receipt_digest,
+    derive_population_use_request_digest,
+    empty_population_custody_log,
+    record_population_use,
+    validate_population_custody_log,
+)
 
 NAMED_INPUTS_SCHEMA = "openultrasast.detection-core.named-inputs.v2"
 PROVENANCE_POLICY_SCHEMA = "openultrasast.detection-core.provenance-policy.v2"
@@ -291,6 +312,25 @@ def validate_provenance(
 
 
 __all__ = [
+    "CUSTODY_LOG_SCHEMA",
+    "CUSTODY_POLICY_SCHEMA",
+    "CUSTODY_RECEIPT_SCHEMA",
+    "CUSTODY_REQUEST_SCHEMA",
+    "LineagePurpose",
+    "PopulationCustodyError",
+    "PopulationCustodyLog",
+    "PopulationCustodyPolicy",
+    "PopulationCustodyPort",
+    "PopulationStatus",
+    "PopulationUseReceipt",
+    "PopulationUseRequest",
+    "append_population_use",
+    "derive_population_custody_policy_digest",
+    "derive_population_use_receipt_digest",
+    "derive_population_use_request_digest",
+    "empty_population_custody_log",
+    "record_population_use",
+    "validate_population_custody_log",
     "DecisionProvenance",
     "ModelCallIdentity",
     "NAMED_INPUTS_SCHEMA",

tests/detection_core_fixtures.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 from collections.abc import Mapping
 from dataclasses import dataclass
+from threading import Lock
 from types import MappingProxyType
 
-from openultrasast.core import canonical
+from openultrasast.core import canonical, provenance
 
 
 @dataclass(frozen=True)
@@ -59,3 +60,40 @@ def stored_artifact(
     )
     resolver = InMemoryContentResolver({artifact_id: canonical.ResolvedContent(content=content, receipt=receipt)})
     return reference, resolver
+
+
+class InMemoryPopulationCustody:
+    """Deterministic compare-and-swap fixture; production persistence stays external."""
+
+    def __init__(self, initial: provenance.PopulationCustodyLog, policy: provenance.PopulationCustodyPolicy) -> None:
+        provenance.validate_population_custody_log(initial, policy)
+        self._log = initial
+        self._policy = policy
+        self._lock = Lock()
+
+    def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog:
+        if population_digest != self._log.population_digest:
+            raise provenance.PopulationCustodyError("population custody log not found")
+        return self._log
+
+    def append(self, request: provenance.PopulationUseRequest, custodian: str) -> provenance.PopulationUseReceipt:
+        with self._lock:
+            receipt = provenance.record_population_use(request, self._log, custodian, self._policy)
+            self._log = provenance.append_population_use(self._log, receipt, self._policy)
+            return receipt
+
+
+def custody_contract() -> tuple[
+    provenance.PopulationCustodyLog,
+    provenance.PopulationCustodyPolicy,
+    str,
+]:
+    """Create one reviewed custodian and an empty reproducible population log."""
+    population = canonical.content_digest("population")
+    custodian = canonical.content_digest("reviewed-custodian")
+    reviewed = (custodian,)
+    policy = provenance.PopulationCustodyPolicy(
+        reviewed_custodian_digests=reviewed,
+        policy_digest=provenance.derive_population_custody_policy_digest(reviewed),
+    )
+    return provenance.empty_population_custody_log(population, custodian), policy, custodian

tests/test_core_custody.py

@@ -0,0 +1,211 @@
+"""Append-only population custody contracts (Req 12.2, 13.1, 13.4)."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+
+import detection_core_fixtures as fixtures
+import pytest
+
+from openultrasast.core import canonical, provenance
+
+
+def _request(
+    log: provenance.PopulationCustodyLog,
+    *,
+    labels_accessed: bool = False,
+    purpose: provenance.LineagePurpose | None = None,
+) -> provenance.PopulationUseRequest:
+    return provenance.PopulationUseRequest(
+        population_digest=log.population_digest,
+        arm_id="arm-v2",
+        purpose=purpose or provenance.LineagePurpose.SEMANTIC_FACT,
+        labels_accessed=labels_accessed,
+        influenced_artifact_digest=canonical.content_digest("artifact"),
+        expected_log_digest=log.log_digest,
+    )
+
+
+def test_empty_log_and_policy_have_deterministic_identities() -> None:
+    first, policy, _ = fixtures.custody_contract()
+    second, _, _ = fixtures.custody_contract()
+    assert first == second
+    assert first.receipts == ()
+    assert first.receipt_count == 0
+    assert first.head_receipt_digest is None
+    assert first.status is provenance.PopulationStatus.AVAILABLE
+    provenance.validate_population_custody_log(first, policy)
+
+
+def test_record_derives_cas_sequence_predecessor_and_receipt_identity() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    receipt = provenance.record_population_use(_request(prior), prior, custodian, policy)
+    assert receipt.sequence == 1
+    assert receipt.predecessor_digest is None
+    assert receipt.prior_status is provenance.PopulationStatus.AVAILABLE
+    assert receipt.status is provenance.PopulationStatus.AVAILABLE
+    assert receipt.request_digest == provenance.derive_population_use_request_digest(receipt.request)
+    assert receipt.receipt_digest == provenance.derive_population_use_receipt_digest(receipt)
+
+
+def test_append_constructs_and_validates_a_contiguous_complete_chain() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    first = provenance.record_population_use(_request(prior), prior, custodian, policy)
+    one = provenance.append_population_use(prior, first, policy)
+    second = provenance.record_population_use(_request(one), one, custodian, policy)
+    complete = provenance.append_population_use(one, second, policy)
+    provenance.validate_population_custody_log(complete, policy)
+    assert complete.receipt_count == 2
+    assert complete.head_receipt_digest == second.receipt_digest
+    assert second.predecessor_digest == first.receipt_digest
+
+
+def test_compare_and_swap_refuses_an_old_log_head() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    stale = _request(prior)
+    first = provenance.record_population_use(stale, prior, custodian, policy)
+    current = provenance.append_population_use(prior, first, policy)
+    with pytest.raises(provenance.PopulationCustodyError, match="expected log"):
+        provenance.record_population_use(stale, current, custodian, policy)
+
+
+def test_label_access_spends_population_and_status_is_monotonic() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    spent_receipt = provenance.record_population_use(_request(prior, labels_accessed=True), prior, custodian, policy)
+    spent = provenance.append_population_use(prior, spent_receipt, policy)
+    later = provenance.record_population_use(_request(spent), spent, custodian, policy)
+    complete = provenance.append_population_use(spent, later, policy)
+    assert spent_receipt.status is provenance.PopulationStatus.SPENT
+    assert later.prior_status is provenance.PopulationStatus.SPENT
+    assert later.status is provenance.PopulationStatus.SPENT
+    assert complete.status is provenance.PopulationStatus.SPENT
+
+
+def test_broken_or_omitted_receipts_are_refused() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    first = provenance.record_population_use(_request(prior), prior, custodian, policy)
+    one = provenance.append_population_use(prior, first, policy)
+    second = provenance.record_population_use(_request(one), one, custodian, policy)
+    complete = provenance.append_population_use(one, second, policy)
+    broken = replace(second, predecessor_digest=None)
+    with pytest.raises(provenance.PopulationCustodyError, match="predecessor"):
+        provenance.validate_population_custody_log(replace(complete, receipts=(first, broken)), policy)
+    with pytest.raises(provenance.PopulationCustodyError, match="sequence"):
+        provenance.validate_population_custody_log(
+            replace(complete, receipts=(second,), receipt_count=1, head_receipt_digest=second.receipt_digest), policy
+        )
+
+
+def test_log_metadata_and_receipt_identities_are_rederived() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    receipt = provenance.record_population_use(_request(prior), prior, custodian, policy)
+    current = provenance.append_population_use(prior, receipt, policy)
+    for changed, message in (
+        (replace(current, receipt_count=2), "count"),
+        (replace(current, head_receipt_digest=None), "head"),
+        (replace(current, log_digest="0" * 64), "log digest"),
+        (replace(current, status=provenance.PopulationStatus.SPENT), "status"),
+    ):
+        with pytest.raises(provenance.PopulationCustodyError, match=message):
+            provenance.validate_population_custody_log(changed, policy)
+    with pytest.raises(provenance.PopulationCustodyError, match="receipt digest"):
+        provenance.validate_population_custody_log(replace(current, receipts=(replace(receipt, receipt_digest="0" * 64),)), policy)
+
+
+def test_unreviewed_custodian_and_tampered_policy_are_refused() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    with pytest.raises(provenance.PopulationCustodyError, match="reviewed"):
+        provenance.record_population_use(_request(prior), prior, canonical.content_digest("other"), policy)
+    with pytest.raises(provenance.PopulationCustodyError, match="policy"):
+        provenance.record_population_use(_request(prior), prior, custodian, replace(policy, policy_digest="0" * 64))
+
+
+def test_reviewed_custodian_cannot_take_over_an_existing_log() -> None:
+    prior, _, custodian = fixtures.custody_contract()
+    other = canonical.content_digest("other-reviewed-custodian")
+    reviewed = (custodian, other)
+    policy = provenance.PopulationCustodyPolicy(
+        reviewed,
+        provenance.derive_population_custody_policy_digest(reviewed),
+    )
+    with pytest.raises(provenance.PopulationCustodyError, match="does not match log"):
+        provenance.record_population_use(_request(prior), prior, other, policy)
+
+
+def test_receipt_reasons_are_derived_from_the_status_transition() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    receipt = provenance.record_population_use(_request(prior), prior, custodian, policy)
+    tampered = replace(receipt, reasons=("labels_accessed",), receipt_digest="0" * 64)
+    tampered = replace(tampered, receipt_digest=provenance.derive_population_use_receipt_digest(tampered))
+    with pytest.raises(provenance.PopulationCustodyError, match="reasons"):
+        provenance.append_population_use(prior, tampered, policy)
+
+
+def test_current_log_check_refuses_an_old_but_internally_valid_prefix() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    port = fixtures.InMemoryPopulationCustody(prior, policy)
+    port.append(_request(prior), custodian)
+    provenance.validate_population_custody_log(port.current_log(prior.population_digest), policy, current=port)
+    with pytest.raises(provenance.PopulationCustodyError, match="current complete log"):
+        provenance.validate_population_custody_log(prior, policy, current=port)
+
+
+def test_current_log_check_validates_the_ports_complete_chain() -> None:
+    prior, policy, _ = fixtures.custody_contract()
+
+    class InvalidPort:
+        def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog:
+            assert population_digest == prior.population_digest
+            return replace(prior, log_digest="0" * 64)
+
+    with pytest.raises(provenance.PopulationCustodyError, match="log digest"):
+        provenance.validate_population_custody_log(prior, policy, current=InvalidPort())
+
+
+def test_concurrent_requests_have_exactly_one_cas_successor() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    port = fixtures.InMemoryPopulationCustody(prior, policy)
+    left = _request(prior, labels_accessed=True, purpose=provenance.LineagePurpose.ADMISSION_LABEL_OPEN)
+    right = replace(left, arm_id="other-arm")
+
+    def attempt(
+        request: provenance.PopulationUseRequest,
+    ) -> provenance.PopulationUseReceipt | provenance.PopulationCustodyError:
+        try:
+            return port.append(request, custodian)
+        except provenance.PopulationCustodyError as exc:
+            return exc
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = tuple(pool.map(attempt, (left, right)))
+    winners = tuple(result for result in results if isinstance(result, provenance.PopulationUseReceipt))
+    failures = tuple(result for result in results if isinstance(result, provenance.PopulationCustodyError))
+    assert len(winners) == len(failures) == 1
+    assert "expected log" in str(failures[0])
+    current = port.current_log(prior.population_digest)
+    assert current.receipts == winners
+    assert current.status is provenance.PopulationStatus.SPENT
+
+
+def test_fixture_cannot_restore_a_spent_population() -> None:
+    prior, policy, custodian = fixtures.custody_contract()
+    port = fixtures.InMemoryPopulationCustody(prior, policy)
+    port.append(_request(prior, labels_accessed=True), custodian)
+    spent = port.current_log(prior.population_digest)
+    port.append(_request(spent), custodian)
+    current = port.current_log(prior.population_digest)
+    assert current.status is provenance.PopulationStatus.SPENT
+    restored = replace(current, status=provenance.PopulationStatus.AVAILABLE)
+    with pytest.raises(provenance.PopulationCustodyError, match="status"):
+        provenance.validate_population_custody_log(restored, policy)
+
+
+@pytest.mark.parametrize("field", ["population_digest", "expected_log_digest"])
+def test_request_digest_fields_are_runtime_validated(field: str) -> None:
+    prior, _, _ = fixtures.custody_contract()
+    with pytest.raises(canonical.DigestFormatError):
+        if field == "population_digest":
+            replace(_request(prior), population_digest="short")
+        else:
+            replace(_request(prior), expected_log_digest="short")