diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index f845406..3a1cabc 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -229,7 +229,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 2.9_ - _Boundary: Fact Initialization_ - _Depends: 7.1_ -- [ ] 11.4 Verify and promote proposed target knowledge +- [x] 11.4 Verify and promote proposed target knowledge - Accept source- or approved-documentation proposals, reject contradictions, and write only verified outputs. - Done when matching proposals promote and contradicted or evidence-free proposals leave promoted knowledge unchanged. - _Requirements: 2.9, 2.10_ diff --git a/src/openultrasast/tools/fact_init.py b/src/openultrasast/tools/fact_init.py index 52b2c72..1820fc3 100644 --- a/src/openultrasast/tools/fact_init.py +++ b/src/openultrasast/tools/fact_init.py @@ -1,10 +1,15 @@ -"""Task 11.3: extract existing target knowledge into candidate records (Req 2.9). +"""Tasks 11.3 + 11.4: extraction and verified promotion of target knowledge (Reqs 2.9, 2.10). Extraction mode of the initialization job: the legacy recognition strategies in ``mapping/legacy.py`` and the promoted fact rows in ``ruleset/obligations/*.toml`` are read -- never modified -- and returned as deterministic candidate records -carrying source, version and region evidence. Verification and promotion are the -acquisition workflow's job (tasks 11.4+), not extraction's. +carrying source, version and region evidence. + +Verification and promotion (task 11.4): a proposal is verified when its evidence +is complete (source, version, region) and it does not contradict promoted +knowledge. Promotion accepts only verified proposals; contradicted or +evidence-free proposals are rejected with a reason and leave the promoted +corpus unchanged (Req 2.10). This is a tools module, not an engine module: naming recognition targets here is legitimate (the whole point is to extract their vocabulary into data). @@ -109,4 +114,54 @@ def _symbol_spans(text: str) -> Iterable[tuple[str, int, int]]: yield symbol, start, end_line - 1 -__all__ = ["CandidateRecord", "extract_all", "extract_fact_rows", "extract_legacy_strategies"] +@dataclass(frozen=True) +class PromotionResult: + """Outcome of promoting a batch of proposals (Reqs 2.9, 2.10).""" + + accepted: tuple[CandidateRecord, ...] + rejected: tuple[tuple[CandidateRecord, str], ...] # (record, reason) + promoted_after: tuple[CandidateRecord, ...] + + +def verify_candidate(record: CandidateRecord, promoted: tuple[CandidateRecord, ...]) -> tuple[bool, str | None]: + """Verify one proposal against the promoted corpus. + + Evidence must be complete (source, version, region) and the proposal must not + contradict promoted knowledge. Rejections carry a reason (Req 2.10). + """ + if not record.source or not record.region or not re.fullmatch(r"[0-9a-f]{12}", record.version): + return False, "missing-evidence" + for existing in promoted: + if existing.kind == record.kind and existing.name == record.name: + if existing.payload != record.payload: + return False, f"contradicts-promoted:{record.kind}:{record.name}" + return True, None + return True, None + + +def promote_proposals(candidates: tuple[CandidateRecord, ...], promoted: tuple[CandidateRecord, ...]) -> PromotionResult: + """Promote only verified proposals; rejects never modify the corpus (Req 2.10).""" + accepted: list[CandidateRecord] = [] + rejected: list[tuple[CandidateRecord, str]] = [] + promoted_after = list(promoted) + for record in candidates: + is_valid, reason = verify_candidate(record, tuple(promoted_after)) + if not is_valid or reason is not None: + rejected.append((record, reason or "rejected")) + continue + accepted.append(record) + already_promoted = any(existing.kind == record.kind and existing.name == record.name for existing in promoted_after) + if not already_promoted: + promoted_after.append(record) + return PromotionResult(tuple(accepted), tuple(rejected), tuple(promoted_after)) + + +__all__ = [ + "CandidateRecord", + "PromotionResult", + "extract_all", + "extract_fact_rows", + "extract_legacy_strategies", + "promote_proposals", + "verify_candidate", +] diff --git a/tests/test_fact_init.py b/tests/test_fact_init.py index df22d0e..51b2cde 100644 --- a/tests/test_fact_init.py +++ b/tests/test_fact_init.py @@ -1,8 +1,10 @@ -"""Task 11.3: extraction of existing target knowledge into candidate records (Req 2.9). +"""Tasks 11.3 + 11.4: extraction and verified promotion of target knowledge (Reqs 2.9, 2.10). Extraction reads the legacy recognition strategies and the promoted fact rows and returns deterministic candidate records carrying source, version and region -evidence. It never modifies promoted outputs. +evidence. It never modifies promoted outputs. Verification and promotion accept +only evidence-complete, non-contradicting proposals; rejected proposals leave the +promoted corpus unchanged. """ import hashlib @@ -13,9 +15,12 @@ import pytest from openultrasast.tools.fact_init import ( CandidateRecord, + PromotionResult, extract_all, extract_fact_rows, extract_legacy_strategies, + promote_proposals, + verify_candidate, ) REPO_ROOT = Path(__file__).parents[1] @@ -33,6 +38,122 @@ def _digest_dir(directory: Path) -> dict[str, str]: return {path.name: _file_digest(path) for path in sorted(directory.glob("*.toml"))} +def _proposal( + name: str = "proposed-read", payload: tuple[tuple[str, str], ...] = (("calls", "read"), ("kind", "protected_read")) +) -> CandidateRecord: + return CandidateRecord( + kind="operation", + name=name, + source="docs/target.md", + version="a1b2c3d4e5f6", + region="docs/target.md:proposed-read", + payload=payload, + ) + + +def _promoted_corpus() -> tuple[CandidateRecord, ...]: + return ( + CandidateRecord( + kind="operation", + name="promoted-read", + source="src/openultrasast/ruleset/obligations/java.toml", + version="0123456789ab", + region="java.toml:operation:promoted-read", + payload=(("calls", "read"), ("kind", "protected_read")), + ), + ) + + +def test_verified_novel_proposal_is_accepted_and_promoted() -> None: + promoted = _promoted_corpus() + is_valid, reason = verify_candidate(_proposal(), promoted) + assert is_valid is True + assert reason is None + result = promote_proposals((_proposal(),), promoted) + assert result.accepted == (_proposal(),) + assert result.rejected == () + assert _proposal() in result.promoted_after + assert promoted[0] in result.promoted_after + + +def test_contradicted_proposal_is_rejected_and_promoted_knowledge_unchanged() -> None: + promoted = _promoted_corpus() + contradicted = CandidateRecord( + kind="operation", + name="promoted-read", + source="docs/target.md", + version="a1b2c3d4e5f6", + region="docs/target.md:promoted-read", + payload=(("calls", "write"),), + ) + is_valid, reason = verify_candidate(contradicted, promoted) + assert is_valid is False + assert reason is not None and reason.startswith("contradicts-promoted") + result = promote_proposals((contradicted,), promoted) + assert result.rejected[0][0] == contradicted + assert result.promoted_after == promoted + + +def test_evidence_free_proposal_is_rejected() -> None: + promoted = _promoted_corpus() + for broken in ( + _proposal().__class__(**{**_proposal().__dict__, "source": ""}), + CandidateRecord(kind="operation", name="x", source="s", version="short", region="r"), + CandidateRecord(kind="operation", name="y", source="s", version="a1b2c3d4e5f6", region=""), + ): + is_valid, reason = verify_candidate(broken, promoted) + assert is_valid is False + assert reason is not None and reason.startswith("missing-evidence") + result = promote_proposals( + (CandidateRecord(kind="operation", name="z", source="", version="a1b2c3d4e5f6", region="r"),), + promoted, + ) + assert result.promoted_after == promoted + + +def test_exact_match_proposal_is_accepted_without_duplicate() -> None: + promoted = _promoted_corpus() + twin = CandidateRecord( + kind="operation", + name="promoted-read", + source="docs/target.md", + version="a1b2c3d4e5f6", + region="docs/target.md:promoted-read", + payload=promoted[0].payload, + ) + result = promote_proposals((twin,), promoted) + assert result.accepted == (twin,) + assert result.rejected == () + same_knowledge = [record for record in result.promoted_after if record.kind == twin.kind and record.name == twin.name] + assert len(same_knowledge) == 1 + assert len(result.promoted_after) == len(promoted) + + +def test_rejected_proposals_never_modify_promoted_corpus() -> None: + promoted = _promoted_corpus() + contradicted = CandidateRecord( + kind="operation", + name="promoted-read", + source="docs/target.md", + version="a1b2c3d4e5f6", + region="docs/target.md:promoted-read", + payload=(("calls", "purge"),), + ) + result = promote_proposals((_proposal(), contradicted), promoted) + assert result.accepted == (_proposal(),) + assert len(result.rejected) == 1 + assert contradicted not in result.promoted_after + assert promoted[0] in result.promoted_after + + +def test_promotion_is_deterministic() -> None: + promoted = _promoted_corpus() + first = promote_proposals((_proposal(),), promoted) + second = promote_proposals((_proposal(),), promoted) + assert first == second + assert isinstance(first, PromotionResult) + + def test_fact_rows_extraction_yields_operation_and_discharger_records() -> None: records = extract_fact_rows(RULESET_DIR) kinds = {record.kind for record in records}