feat(access-control-prover): 11.5 acquisition provenance and label-isolated mode

.kiro/specs/access-control-prover/tasks.md

@@ -235,7 +235,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with
   - _Requirements: 2.9, 2.10_
   - _Boundary: Fact Initialization_
   - _Depends: 11.3_
-- [ ] 11.5 Record acquisition provenance and enforce label-isolated mode
+- [x] 11.5 Record acquisition provenance and enforce label-isolated mode
   - Persist input revision, evidence digest, accepted/rejected proposals and output digest.
   - Reject admission acquisition inputs containing vulnerability, fix-location or ordinary-change labels.
   - Done when changed output after labels are opened marks the population spent.

src/openultrasast/tools/fact_init.py

@@ -157,11 +157,70 @@ def promote_proposals(candidates: tuple[CandidateRecord, ...], promoted: tuple[C
 
 
 __all__ = [
+    "AcquisitionRecord",
+    "AdmissionLabelsExposed",
     "CandidateRecord",
     "PromotionResult",
+    "acquire",
     "extract_all",
     "extract_fact_rows",
     "extract_legacy_strategies",
+    "population_is_spent",
     "promote_proposals",
     "verify_candidate",
 ]
+
+
+class AdmissionLabelsExposed(ValueError):
+    """Raised when acquisition input carries admission evaluation labels (Req 12.2)."""
+
+
+FORBIDDEN_LABEL_MARKERS: tuple[str, ...] = ("vulnerability", "fix-location", "ordinary-change")
+
+
+@dataclass(frozen=True)
+class AcquisitionRecord:
+    """Provenance record of one acquisition pass (Req 12.1)."""
+
+    input_revision: str
+    evidence_digest: str
+    accepted_names: tuple[str, ...]
+    rejected_names: tuple[str, ...]
+    output_digest: str
+
+
+def _corpus_digest(records: tuple[CandidateRecord, ...]) -> str:
+    lines = sorted(
+        f"{record.kind}\x1f{record.name}\x1f{record.source}\x1f{record.version}\x1f{record.region}\x1f{record.payload}"
+        for record in records
+    )
+    return hashlib.sha256("\n".join(lines).encode()).hexdigest()[:12]
+
+
+def acquire(
+    candidates: tuple[CandidateRecord, ...],
+    promoted: tuple[CandidateRecord, ...],
+    *,
+    input_labels: tuple[str, ...] = (),
+) -> AcquisitionRecord:
+    """Run one label-isolated acquisition pass (Reqs 12.1, 12.2).
+
+    The label guard runs first: an input carrying admission evaluation labels is
+    rejected before any promotion happens, leaving the promoted corpus untouched.
+    """
+    for label in input_labels:
+        if any(marker in label for marker in FORBIDDEN_LABEL_MARKERS):
+            raise AdmissionLabelsExposed(label)
+    result = promote_proposals(candidates, promoted)
+    return AcquisitionRecord(
+        input_revision=_corpus_digest(promoted),
+        evidence_digest=_corpus_digest(candidates),
+        accepted_names=tuple(sorted(record.name for record in result.accepted)),
+        rejected_names=tuple(sorted(record.name for record, _ in result.rejected)),
+        output_digest=_corpus_digest(result.promoted_after),
+    )
+
+
+def population_is_spent(blind_record: AcquisitionRecord, later_output_digest: str) -> bool:
+    """A changed acquisition output after labels opened spends the population (Req 12.3)."""
+    return later_output_digest != blind_record.output_digest

tests/test_fact_init.py

@@ -14,11 +14,15 @@ from pathlib import Path
 import pytest
 
 from openultrasast.tools.fact_init import (
+    AcquisitionRecord,
+    AdmissionLabelsExposed,
     CandidateRecord,
     PromotionResult,
+    acquire,
     extract_all,
     extract_fact_rows,
     extract_legacy_strategies,
+    population_is_spent,
     promote_proposals,
     verify_candidate,
 )
@@ -238,3 +242,55 @@ def test_candidate_record_is_frozen() -> None:
     )
     with pytest.raises(Exception):  # noqa: B017
         record.name = "other"  # type: ignore[misc]
+
+
+def test_label_blind_acquire_records_provenance() -> None:
+    record = acquire((_proposal(), _contradicted_proposal()), _promoted_corpus())
+    assert isinstance(record, AcquisitionRecord)
+    assert _VERSION_PATTERN.fullmatch(record.input_revision)
+    assert _VERSION_PATTERN.fullmatch(record.evidence_digest)
+    assert _VERSION_PATTERN.fullmatch(record.output_digest)
+    assert record.accepted_names == ("proposed-read",)
+    assert record.rejected_names == ("promoted-read",)
+
+
+def test_admission_labels_are_rejected_before_promotion() -> None:
+    promoted = _promoted_corpus()
+    with pytest.raises(AdmissionLabelsExposed):
+        acquire((_proposal(),), promoted, input_labels=("vulnerability-fix-label",))
+    assert promoted == _promoted_corpus()
+
+
+def test_every_forbidden_marker_raises() -> None:
+    for label in ("fix-location:Handler.java:4", "ordinary-change:repo-a", "vulnerability:case-9"):
+        with pytest.raises(AdmissionLabelsExposed):
+            acquire((_proposal(),), _promoted_corpus(), input_labels=(label,))
+
+
+def test_acquire_is_deterministic() -> None:
+    first = acquire((_proposal(),), _promoted_corpus())
+    second = acquire((_proposal(),), _promoted_corpus())
+    assert first == second
+
+
+def test_population_is_spent_detects_output_change() -> None:
+    blind = acquire((_proposal(),), _promoted_corpus())
+    assert population_is_spent(blind, blind.output_digest) is False
+    assert population_is_spent(blind, "0" * 12) is True
+
+
+def test_acquisition_record_is_frozen() -> None:
+    record = acquire((_proposal(),), _promoted_corpus())
+    with pytest.raises(Exception):  # noqa: B017
+        record.output_digest = "0" * 12  # type: ignore[misc]
+
+
+def _contradicted_proposal() -> CandidateRecord:
+    return CandidateRecord(
+        kind="operation",
+        name="promoted-read",
+        source="proposals.toml",
+        version="a" * 12,
+        region="proposals.toml:operation:promoted-read",
+        payload=(("calls", "write"), ("kind", "protected_read")),
+    )