diff --git a/core/cli.py b/core/cli.py index 2a6c4755..9441127a 100644 --- a/core/cli.py +++ b/core/cli.py @@ -1818,6 +1818,14 @@ def cmd_demo(args: argparse.Namespace) -> int: print(f" {k}: {v}") return 0 + if target == "proof-carrying-promotion": + from evals.proof_carrying_promotion_demo.run_tour import run_tour + + result = run_tour(emit_json=args.json) + if args.json: + print(json.dumps(result, indent=2, sort_keys=True, default=str)) + return 0 if result.get("all_passed", False) else 1 + if target == "audit-tour": from evals.audit_tour.run_tour import run_tour @@ -2192,7 +2200,7 @@ def _run_demo_all(emit_json: bool) -> int: passed["learning_arc"] = bool(arc_report.get("learning_arc_closed", False)) # 9. articulation - _section("9/9 articulation — discourse-planner spine") + _section("9/10 articulation — discourse-planner spine") from evals.articulation.run_demo import run_demo as run_art if not emit_json: @@ -2202,6 +2210,15 @@ def _run_demo_all(emit_json: bool) -> int: consolidated["articulation"] = art_report passed["articulation"] = bool(art_report.get("all_claims_supported", False)) + # 10. proof-carrying-promotion + _section("10/10 proof-carrying-promotion — deductive engine pipeline") + from evals.proof_carrying_promotion_demo.run_tour import run_tour as run_pcp + + with _maybe_suppress(): + pcp_report = run_pcp(emit_json=emit_json) + consolidated["proof_carrying_promotion"] = pcp_report + passed["proof_carrying_promotion"] = bool(pcp_report.get("all_passed", False)) + all_passed = all(passed.values()) consolidated["passed"] = passed consolidated["all_demos_passed"] = all_passed @@ -3632,6 +3649,7 @@ def build_parser() -> argparse.ArgumentParser: "phase5", "phase6", "adr-0024-chain", + "proof-carrying-promotion", "audit-tour", "register-tour", "anchor-lens-tour", @@ -3653,8 +3671,9 @@ def build_parser() -> argparse.ArgumentParser: "phase5: stratified 5-family mechanism-isolation. " "phase6: 3-condition head-to-head vs in-system baseline. " "adr-0024-chain: phase5 + phase6 combined evidence. " - "all: run every demo (eight in total) and print a " + "all: run every demo and print a " "consolidated PASS/FAIL table; exits non-zero if any demo fails. " + "proof-carrying-promotion: ADR-0218 PR D deterministic demo. " "audit-tour: ADR-0027..0041 pack-layer architecture in four " "scenes (identity / safety / ethics / replay). " "register-tour: ADR-0068..0072 presentation-axis seam — same " diff --git a/evals/proof_carrying_promotion_demo/__init__.py b/evals/proof_carrying_promotion_demo/__init__.py new file mode 100644 index 00000000..7a2667a2 --- /dev/null +++ b/evals/proof_carrying_promotion_demo/__init__.py @@ -0,0 +1 @@ +"""Proof-Carrying Coherence Promotion deterministic demo.""" diff --git a/evals/proof_carrying_promotion_demo/run_tour.py b/evals/proof_carrying_promotion_demo/run_tour.py new file mode 100644 index 00000000..16d5e794 --- /dev/null +++ b/evals/proof_carrying_promotion_demo/run_tour.py @@ -0,0 +1,245 @@ +"""Proof-carrying coherence promotion demo (PR D of ADR-0218). + +Demonstrates the pure proof-carrying promotion decider and the mutation owner. +- Scene 1: Valid entailment from COHERENT premises promotes successfully. +- Scene 2: Unentailed consistency stays SPECULATIVE. +- Scene 3: Uncertified reading fails closed. +- Scene 4: Proposer payload (data) is ignored. +""" + +from __future__ import annotations + +import json +from typing import Any + +import numpy as np + +from algebra.cga import embed_point +from teaching.epistemic import EpistemicStatus +from teaching.proof_promotion import certify_promotion +from vault.store import VaultStore + +_VERBOSE = True + + +def _say(*args, **kwargs) -> None: + if _VERBOSE: + print(*args, **kwargs) + + +def _print_header(title: str, claim: str) -> None: + _say() + _say("─" * 72) + _say(f" {title}") + _say("─" * 72) + _say(f" CLAIM: {claim}") + _say() + + +def _versor(seed: int) -> np.ndarray: + rng = np.random.default_rng(seed) + return embed_point(rng.standard_normal(3).astype(np.float32)) + + +def _store( + vault: VaultStore, + seed: int, + form: str, + status: EpistemicStatus, + *, + certified: bool = True, +) -> int: + metadata: dict = {"reading_certified": certified, "propositional_form": form} + return vault.store(_versor(seed), metadata, epistemic_status=status) + + +class _PoisonedPayload(dict): + """A proposer payload that detonates on ANY read.""" + def _boom(self, *args, **kwargs): + raise AssertionError("certify_promotion read the proposer payload — D3.5 violated") + __getitem__ = _boom + __iter__ = _boom + __len__ = _boom + __contains__ = _boom + get = _boom + keys = _boom + values = _boom + items = _boom + + +def _scene_1_valid_promotion() -> dict[str, Any]: + _print_header( + "Scene 1 — Valid deductive entailment promotes.", + "A SPECULATIVE claim deductively entailed by COHERENT premises is certified and promoted." + ) + vault = VaultStore(reproject_interval=0) + p1 = _store(vault, 101, "p", EpistemicStatus.COHERENT) + p2 = _store(vault, 102, "p -> q", EpistemicStatus.COHERENT) + c1 = _store(vault, 7, "q", EpistemicStatus.SPECULATIVE) + + _say(f" [Setup] Premise 1 (COHERENT): 'p' (entry {p1})") + _say(f" [Setup] Premise 2 (COHERENT): 'p -> q' (entry {p2})") + _say(f" [Setup] Claim (SPECULATIVE): 'q' (entry {c1})") + + decision = certify_promotion( + claim_entry_index=c1, + premise_entry_indices=(p1, p2), + vault=vault, + ) + _say(f" [Decider] Promoted? {decision.promoted}") + _say(f" [Decider] Reason: {decision.reason}") + + mutation_applied = False + if decision.promoted and decision.certificate: + result = vault.apply_certified_promotion(c1, decision.certificate) + _say(f" [Mutation] applied: {result.applied}, reason: {result.reason}") + mutation_applied = result.applied + + # Verify store state + meta = next((m for i, m in vault.iter_metadata() if i == c1), None) + if meta: + _say(f" [Verify] Final status in vault: {meta['epistemic_status']}") + + return { + "promoted": decision.promoted, + "reason": decision.reason, + "mutation_applied": mutation_applied + } + + +def _scene_2_not_entailed() -> dict[str, Any]: + _print_header( + "Scene 2 — Unentailed consistency stays SPECULATIVE.", + "A claim consistent with premises but not entailed refuses to promote." + ) + vault = VaultStore(reproject_interval=0) + p1 = _store(vault, 101, "p", EpistemicStatus.COHERENT) + c1 = _store(vault, 7, "q", EpistemicStatus.SPECULATIVE) + + _say(f" [Setup] Premise 1 (COHERENT): 'p' (entry {p1})") + _say(f" [Setup] Claim (SPECULATIVE): 'q' (entry {c1})") + + decision = certify_promotion( + claim_entry_index=c1, + premise_entry_indices=(p1,), + vault=vault, + ) + _say(f" [Decider] Promoted? {decision.promoted}") + _say(f" [Decider] Reason: {decision.reason}") + + return { + "promoted": decision.promoted, + "reason": decision.reason, + } + + +def _scene_3_missing_reading() -> dict[str, Any]: + _print_header( + "Scene 3 — Uncertified reading fails closed.", + "A valid entailment fails closed if a premise lacks curator reading certification." + ) + vault = VaultStore(reproject_interval=0) + p1 = _store(vault, 101, "p", EpistemicStatus.COHERENT, certified=False) + p2 = _store(vault, 102, "p -> q", EpistemicStatus.COHERENT) + c1 = _store(vault, 7, "q", EpistemicStatus.SPECULATIVE) + + _say(f" [Setup] Premise 1 (COHERENT, NO CERT): 'p' (entry {p1})") + _say(f" [Setup] Premise 2 (COHERENT): 'p -> q' (entry {p2})") + _say(f" [Setup] Claim (SPECULATIVE): 'q' (entry {c1})") + + decision = certify_promotion( + claim_entry_index=c1, + premise_entry_indices=(p1, p2), + vault=vault, + ) + _say(f" [Decider] Promoted? {decision.promoted}") + _say(f" [Decider] Reason: {decision.reason}") + + return { + "promoted": decision.promoted, + "reason": decision.reason, + } + + +def _scene_4_proposer_payload_ignored() -> dict[str, Any]: + _print_header( + "Scene 4 — Proposer payload is data, not authority.", + "The decider provably ignores proposer_payload. If it reads it, the demo crashes." + ) + vault = VaultStore(reproject_interval=0) + p1 = _store(vault, 101, "p", EpistemicStatus.COHERENT) + p2 = _store(vault, 102, "p -> q", EpistemicStatus.COHERENT) + c1 = _store(vault, 7, "q", EpistemicStatus.SPECULATIVE) + + _say(f" [Setup] Premise 1 (COHERENT): 'p' (entry {p1})") + _say(f" [Setup] Premise 2 (COHERENT): 'p -> q' (entry {p2})") + _say(f" [Setup] Claim (SPECULATIVE): 'q' (entry {c1})") + _say(f" [Setup] Proposer payload: detonating dictionary") + + payload = _PoisonedPayload() + decision = certify_promotion( + claim_entry_index=c1, + premise_entry_indices=(p1, p2), + vault=vault, + proposer_payload=payload, + ) + _say(f" [Decider] Promoted? {decision.promoted}") + _say(f" [Decider] Reason: {decision.reason}") + _say(f" [EVIDENCE] Demo did not crash; payload was ignored.") + + return { + "promoted": decision.promoted, + "reason": decision.reason, + } + + +def run_tour(*, emit_json: bool = False) -> dict[str, Any]: + global _VERBOSE + _VERBOSE = not emit_json + if not emit_json: + _say() + _say("=" * 72) + _say(" CORE Proof-Carrying Coherence Promotion Demo") + _say("=" * 72) + _say(" Demonstrates ADR-0218 PR D: Local deterministic demo.") + _say(" Proposer submits claim + proof candidate; CORE recomputes,") + _say(" promotes or refuses on pinned verification only.") + + s1 = _scene_1_valid_promotion() + s2 = _scene_2_not_entailed() + s3 = _scene_3_missing_reading() + s4 = _scene_4_proposer_payload_ignored() + + all_passed = ( + s1["promoted"] is True and s1["mutation_applied"] is True and + s2["promoted"] is False and s2["reason"] == "refused_not_entailed" and + s3["promoted"] is False and s3["reason"] == "refused_premise_reading_uncertified" and + s4["promoted"] is True and s4["reason"] == "promoted_entailed_from_coherent_premises" + ) + + if not emit_json: + _say() + _say("=" * 72) + _say(" Summary") + _say("=" * 72) + _say(f" Valid entailment promoted: {s1['promoted']}") + _say(f" Unentailed claim refused: {not s2['promoted']}") + _say(f" Uncertified reading refused: {not s3['promoted']}") + _say(f" Proposer payload ignored safely: {s4['promoted']}") + _say() + + return { + "scene_1_valid_promotion": s1, + "scene_2_not_entailed": s2, + "scene_3_missing_reading": s3, + "scene_4_proposer_payload_ignored": s4, + "all_passed": all_passed, + } + + +if __name__ == "__main__": + import sys + emit_json = "--json" in sys.argv + result = run_tour(emit_json=emit_json) + if emit_json: + print(json.dumps(result, indent=2))