diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 7a8f11f..6f78144 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -248,7 +248,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 1.1, 1.7_ - _Boundary: Mapping Registry_ - _Depends: 11.1_ -- [ ] 11.7 Run the first label-blind development acquisition +- [x] 11.7 Run the first label-blind development acquisition - Produce registration, handler-operation, enforcement, resource and authority knowledge through the gate. - Treat the output as development data and record its acquisition digest; do not encode names in engine modules. - Done when promoted outputs load, rejected proposals have reasons, and the old unfinished task 3.2 has no residue. diff --git a/src/openultrasast/tools/fact_init.py b/src/openultrasast/tools/fact_init.py index c1a5b26..ccfc69b 100644 --- a/src/openultrasast/tools/fact_init.py +++ b/src/openultrasast/tools/fact_init.py @@ -16,10 +16,11 @@ legitimate (the whole point is to extract their vocabulary into data). """ import hashlib +import json import re import tomllib from collections.abc import Iterable, Iterator -from dataclasses import dataclass +from dataclasses import asdict, dataclass from pathlib import Path _RECOGNISER_PATTERN = re.compile( @@ -162,6 +163,8 @@ __all__ = [ "CandidateRecord", "PromotionResult", "acquire", + "development_acquisition", + "engine_purity_of", "extract_all", "extract_fact_rows", "extract_legacy_strategies", @@ -224,3 +227,55 @@ def acquire( 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 + + +def engine_purity_of(candidates: tuple[CandidateRecord, ...]) -> tuple[str, ...]: + """Return candidate sources that live inside engine modules (task 11.7). + + Engine modules must stay free of target knowledge; a candidate sourced from + ``src/openultrasast/semantic/`` would mean target vocabulary leaked into the + engine, so the development acquisition refuses to run over such a corpus. + """ + return tuple(sorted({record.source for record in candidates if record.source.startswith("src/openultrasast/semantic/")})) + + +def development_acquisition( + repo_root: Path, + *, + artifact_dir: Path | None = None, + input_labels: tuple[str, ...] = (), +) -> tuple[AcquisitionRecord, Path]: + """Run the first label-blind development acquisition (tasks 11.7, Reqs 5.7, 12.9). + + Candidates come from extraction only (repository code and the promoted + ruleset -- never pool labels, never generated content). The pass runs + through the verification gate; the artifact is written only after the gate + returns, so a label-guard rejection leaves the artifact directory empty. + """ + candidates = extract_all(repo_root) + impure_sources = engine_purity_of(candidates) + if impure_sources: + raise ValueError(f"engine modules carry target knowledge: {impure_sources}") + promoted = extract_fact_rows(repo_root / "src" / "openultrasast" / "ruleset" / "obligations") + record = acquire(candidates, promoted, input_labels=input_labels) + result = promote_proposals(candidates, promoted) + artifact_path = (artifact_dir or repo_root / ".acquisition") / (f"development-{record.output_digest[:6]}.json") + artifact_path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "generated_by": "fact_init.development_acquisition", + "label_blind": True, + "record": asdict(record), + "accepted": [ + { + "kind": candidate.kind, + "name": candidate.name, + "source": candidate.source, + "version": candidate.version, + "region": candidate.region, + } + for candidate in result.accepted + ], + "rejected": [{"name": candidate.name, "reason": reason} for candidate, reason in result.rejected], + } + artifact_path.write_text(json.dumps(payload, indent=2, sort_keys=True)) + return record, artifact_path diff --git a/tests/test_fact_init.py b/tests/test_fact_init.py index 083d82a..c3444f0 100644 --- a/tests/test_fact_init.py +++ b/tests/test_fact_init.py @@ -1,4 +1,5 @@ -"""Tasks 11.3 + 11.4: extraction and verified promotion of target knowledge (Reqs 2.9, 2.10). +"""Tasks 11.3 + 11.4 + 11.7: extraction, verified promotion and the first +label-blind development acquisition (Reqs 2.9, 2.10, 5.7, 12.9). Extraction reads the legacy recognition strategies and the promoted fact rows and returns deterministic candidate records carrying source, version and region @@ -8,6 +9,7 @@ promoted corpus unchanged. """ import hashlib +import json import re from pathlib import Path @@ -19,6 +21,8 @@ from openultrasast.tools.fact_init import ( CandidateRecord, PromotionResult, acquire, + development_acquisition, + engine_purity_of, extract_all, extract_fact_rows, extract_legacy_strategies, @@ -295,3 +299,49 @@ def _contradicted_proposal() -> CandidateRecord: region="proposals.toml:operation:promoted-read", payload=(("calls", "write"), ("kind", "protected_read")), ) + + +def _development_record(tmp_path: Path) -> tuple[AcquisitionRecord, Path]: + return development_acquisition(REPO_ROOT, artifact_dir=tmp_path) + + +def test_development_acquisition_writes_a_digest_artifact(tmp_path: Path) -> None: + record, artifact = _development_record(tmp_path) + assert re.fullmatch(r"[0-9a-f]{12}", record.input_revision) + assert re.fullmatch(r"[0-9a-f]{12}", record.evidence_digest) + assert re.fullmatch(r"[0-9a-f]{12}", record.output_digest) + assert artifact.parent == tmp_path + assert artifact.exists() + payload = json.loads(artifact.read_text()) + assert payload["generated_by"] == "fact_init.development_acquisition" + assert payload["record"]["output_digest"] == record.output_digest + assert payload["label_blind"] is True + + +def test_development_acquisition_is_reproducible(tmp_path: Path) -> None: + first_record, first_artifact = _development_record(tmp_path) + second_record, second_artifact = _development_record(tmp_path) + assert first_record == second_record + assert first_artifact.read_bytes() == second_artifact.read_bytes() + + +def test_development_acquisition_rejects_admission_labels(tmp_path: Path) -> None: + with pytest.raises(AdmissionLabelsExposed): + development_acquisition(REPO_ROOT, artifact_dir=tmp_path, input_labels=("vulnerability:case-9",)) + assert list(tmp_path.iterdir()) == [] + + +def test_engine_purity_of_extraction_candidates() -> None: + candidates = extract_all(REPO_ROOT) + assert engine_purity_of(candidates) == () + + +def test_engine_purity_of_flags_engine_sources() -> None: + engine_candidate = CandidateRecord( + kind="operation", + name="engine-row", + source="src/openultrasast/semantic/obligations/facts.py", + version="a" * 12, + region="facts.py:1-2", + ) + assert engine_purity_of((engine_candidate,)) == (engine_candidate.source,)