feat(access-control-prover): 11.7 first label-blind development acquisition

This commit is contained in:
Marius Ciepluch
2026-10-29 21:04:51 +01:00
parent acadf04109
commit 85aa3b1484
3 changed files with 108 additions and 3 deletions
+1 -1
View File
@@ -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.
+56 -1
View File
@@ -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
+51 -1
View File
@@ -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,)