From 134a91c48aee80479a313431aa3a409ab76d0fdd Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 4 Jun 2026 14:12:18 -0700 Subject: [PATCH] feat: add tier2 reasoning evidence spine --- core/cognition/pipeline.py | 3 +- core/contemplation/miners/__init__.py | 2 + core/contemplation/miners/learning_arena.py | 111 ++++++++++ core/learning_arena/__init__.py | 2 + core/learning_arena/engine.py | 36 +++- core/learning_arena/protocols.py | 11 + core/learning_arena/report.py | 11 +- core/reasoning/__init__.py | 33 +++ core/reasoning/adapters.py | 117 +++++++++++ core/reasoning/evidence.py | 205 +++++++++++++++++++ tests/test_adr_0199_learning_arena_engine.py | 48 ++++- tests/test_deductive_logic_entail.py | 8 + tests/test_gsm8k_r1_reconstruction.py | 19 +- tests/test_reasoning_evidence.py | 128 ++++++++++++ 14 files changed, 719 insertions(+), 15 deletions(-) create mode 100644 core/contemplation/miners/learning_arena.py create mode 100644 core/reasoning/__init__.py create mode 100644 core/reasoning/adapters.py create mode 100644 core/reasoning/evidence.py create mode 100644 tests/test_reasoning_evidence.py diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index a2541d3b..f9c041d4 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -22,6 +22,7 @@ from field.state import FieldState from core.cognition.result import CognitiveTurnResult from core.cognition.surface_resolution import resolve_surface from core.cognition.trace import compute_trace_hash, hash_admissibility_trace +from core.reasoning.adapters import evidence_from_entailment_trace from generate.intent import classify_compound_intent from generate.intent_bridge import _is_useful_surface from generate.intent_ratifier import ( @@ -762,7 +763,7 @@ class CognitiveTurnPipeline: def _serialize_entailment_trace(trace: EntailmentTrace | None) -> str: if trace is None: return "" - return f"entailment:{trace.canonical_json()}" + return f"entailment:{evidence_from_entailment_trace(trace).canonical_json()}" @staticmethod def _proof_atom(text: str) -> str: diff --git a/core/contemplation/miners/__init__.py b/core/contemplation/miners/__init__.py index 77fed8cf..346746f7 100644 --- a/core/contemplation/miners/__init__.py +++ b/core/contemplation/miners/__init__.py @@ -2,8 +2,10 @@ from .contradiction_detection import mine_contradiction_detection_report from .frontier_compare import mine_frontier_compare_report +from .learning_arena import mine_learning_arena_report __all__ = [ "mine_contradiction_detection_report", "mine_frontier_compare_report", + "mine_learning_arena_report", ] diff --git a/core/contemplation/miners/learning_arena.py b/core/contemplation/miners/learning_arena.py new file mode 100644 index 00000000..7f78dccd --- /dev/null +++ b/core/contemplation/miners/learning_arena.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from core.contemplation.schema import ( + ContemplationEvidenceRef, + ContemplationFinding, + FindingKind, +) + + +def mine_learning_arena_report( + report_path: str | Path, + *, + substrate_hash: str, + min_coverage: float = 0.25, +) -> tuple[ContemplationFinding, ...]: + """Convert learning-arena report weaknesses into speculative findings. + + Read-only: parses ``report_path`` and returns immutable findings. It never + writes packs, teaching examples, proposals, or runtime state. + """ + path = Path(report_path) + report = json.loads(path.read_text(encoding="utf-8")) + source_id = str(path) + findings: list[ContemplationFinding] = [] + + for class_name, row in sorted(_per_class(report).items()): + committed = int(row.get("committed", 0) or 0) + coverage = float(row.get("coverage", 0.0) or 0.0) + t2_verified = int(row.get("t2_verified", 0) or 0) + if coverage < min_coverage: + evidence = ContemplationEvidenceRef( + source_type="learning_arena_report", + source_id=source_id, + pointer=f"class={class_name}", + summary=f"coverage={coverage};committed={committed};min={min_coverage}", + ) + findings.append( + ContemplationFinding( + kind=FindingKind.COVERAGE_GAP, + subject=class_name, + predicate="weak_coverage", + object=None, + evidence_refs=(evidence,), + proposed_action=( + "Inspect refusal diagnoses for this capability class and add " + "practice or reviewed operators only where the missing piece is named." + ), + substrate_hash=substrate_hash, + ) + ) + if committed > 0 and t2_verified == 0: + evidence = ContemplationEvidenceRef( + source_type="learning_arena_report", + source_id=source_id, + pointer=f"class={class_name}", + summary=f"committed={committed};t2_verified=0", + ) + findings.append( + ContemplationFinding( + kind=FindingKind.UNPROVED_RELATION, + subject=class_name, + predicate="missing_tier2_evidence", + object=None, + evidence_refs=(evidence,), + proposed_action=( + "Add convergent self-verification evidence before promoting this " + "class beyond gold-scored practice." + ), + substrate_hash=substrate_hash, + ) + ) + + for record in report.get("elimination_records", ()) or (): + if not isinstance(record, dict): + continue + case_id = str(record.get("case_id", "unknown_case")) + class_name = str(record.get("class_name", "unknown_class")) + evidence = ContemplationEvidenceRef( + source_type="learning_arena_report", + source_id=source_id, + pointer=f"case={case_id}", + summary=str(record.get("reason", "wrong attempt")), + ) + findings.append( + ContemplationFinding( + kind=FindingKind.BENCHMARK_CASE, + subject=f"{class_name}/{case_id}", + predicate="wrong_attempt", + object=str(record.get("gold", "")), + evidence_refs=(evidence,), + proposed_action=( + "Use this gold-caught wrong attempt as elimination evidence; do not " + "promote the class until the faulty derivation shape is pruned." + ), + substrate_hash=substrate_hash, + ) + ) + + return tuple(findings) + + +def _per_class(report: dict[str, Any]) -> dict[str, dict[str, Any]]: + rows = report.get("per_class", {}) + return {str(k): v for k, v in rows.items() if isinstance(v, dict)} + + +__all__ = ["mine_learning_arena_report"] diff --git a/core/learning_arena/__init__.py b/core/learning_arena/__init__.py index 8bd4e393..7ac28cfb 100644 --- a/core/learning_arena/__init__.py +++ b/core/learning_arena/__init__.py @@ -15,6 +15,7 @@ from core.learning_arena.protocols import ( DomainSolver, GoldTether, Problem, + Tier2Verifier, ) from core.learning_arena.report import ( REFUSAL_DIAGNOSES, @@ -31,6 +32,7 @@ __all__ = [ "DomainSolver", "GoldTether", "Problem", + "Tier2Verifier", "REFUSAL_DIAGNOSES", "EliminationRecord", "PracticeReport", diff --git a/core/learning_arena/engine.py b/core/learning_arena/engine.py index c5b16583..577d6928 100644 --- a/core/learning_arena/engine.py +++ b/core/learning_arena/engine.py @@ -15,14 +15,20 @@ Invariants (the load-bearing ADR-0199 mandates, enforced structurally here): never touches a serving path or the active teaching corpus. Promotion is the caller's separate ``propose_from_ledger`` step into the reviewed corridor. - **L-4 (determinism).** Pure fold over the input order; identical - (problems, solver, tether, diagnose) -> identical report. + (problems, solver, tether, diagnose, tier2_verifier) -> identical report. """ from __future__ import annotations from typing import Callable, Sequence -from core.learning_arena.protocols import Attempt, DomainProblem, DomainSolver, GoldTether +from core.learning_arena.protocols import ( + Attempt, + DomainProblem, + DomainSolver, + GoldTether, + Tier2Verifier, +) from core.learning_arena.report import EliminationRecord, PracticeReport from core.reliability_gate import ClassTally @@ -43,6 +49,7 @@ def run_practice( tether: GoldTether, *, diagnose: Callable[[str], str] = _default_diagnose, + tier2_verifier: Tier2Verifier | None = None, ) -> PracticeReport: """Sealed practice: attempt -> gold-tether score -> per-class ledger. @@ -60,21 +67,36 @@ def run_practice( for problem in problems: cls = problem.class_name attempt: Attempt = solver.attempt(problem) + gold_correct = False if not attempt.committed: verdict = "refused" - elif tether.is_correct(attempt, problem): - verdict = "correct" else: - verdict = "wrong" + gold_correct = tether.is_correct(attempt, problem) + verdict = "correct" if gold_correct else "wrong" counts[verdict] = counts.get(verdict, 0) + 1 tally = ledger.get(cls) or ClassTally(cls) + t2_verified = 0 + t2_agrees_gold = 0 + if attempt.committed and tier2_verifier is not None: + t2_verdict = tier2_verifier.verify(attempt, problem) + if t2_verdict.verified: + t2_verified = 1 + t2_agrees_gold = 1 if gold_correct else 0 if verdict == "correct": - tally = tally.record(correct=1) + tally = tally.record( + correct=1, + t2_verified=t2_verified, + t2_agrees_gold=t2_agrees_gold, + ) elif verdict == "wrong": - tally = tally.record(wrong=1) + tally = tally.record( + wrong=1, + t2_verified=t2_verified, + t2_agrees_gold=t2_agrees_gold, + ) elims.append( EliminationRecord( case_id=attempt.case_id, diff --git a/core/learning_arena/protocols.py b/core/learning_arena/protocols.py index 8ff1637a..b80ff52a 100644 --- a/core/learning_arena/protocols.py +++ b/core/learning_arena/protocols.py @@ -21,6 +21,8 @@ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Protocol, runtime_checkable +from core.reasoning.evidence import Tier2Verdict + @runtime_checkable class DomainProblem(Protocol): @@ -86,6 +88,15 @@ class GoldTether(Protocol): def gold_answer(self, problem: DomainProblem) -> Any: ... +@runtime_checkable +class Tier2Verifier(Protocol): + """Optional convergent self-verifier for a domain attempt.""" + + domain_id: str + + def verify(self, attempt: Attempt, problem: DomainProblem) -> Tier2Verdict: ... + + @dataclass(frozen=True, slots=True) class Problem: """Concrete :class:`DomainProblem` a domain adapter can build directly.""" diff --git a/core/learning_arena/report.py b/core/learning_arena/report.py index f4ff2d40..5a7f107e 100644 --- a/core/learning_arena/report.py +++ b/core/learning_arena/report.py @@ -1,9 +1,9 @@ """ADR-0199 / ADR-0175 — the domain-agnostic practice report. -Extracted verbatim (schema-preserving) from -``evals/gsm8k_math/practice/v1/runner.py`` so every subject's arena emits the -same report shape. ``PracticeReport.as_dict`` is byte-stable with the original -GSM8K report so existing goldens and ``report.json`` are unaffected. +Extracted from ``evals/gsm8k_math/practice/v1/runner.py`` so every subject's +arena emits the same report shape. The report now also exposes the existing +Tier-2 ledger counts when a domain supplies convergent self-verification +evidence; domains without a Tier-2 verifier report zeros. The three refusal-diagnosis axes are the universal ADR-0175 §8 router (skill / knowledge / ambiguity), not a domain quantity — so they live here. @@ -59,6 +59,9 @@ class PracticeReport: "committed": t.committed, "reliability": t.reliability, "coverage": t.coverage, + "t2_verified": t.t2_verified, + "t2_agrees_gold": t.t2_agrees_gold, + "t2_precision": t.t2_precision, } for cls, t in sorted(self.ledger.items()) }, diff --git a/core/reasoning/__init__.py b/core/reasoning/__init__.py new file mode 100644 index 00000000..ccba2161 --- /dev/null +++ b/core/reasoning/__init__.py @@ -0,0 +1,33 @@ +"""Shared deterministic reasoning evidence contracts.""" + +from __future__ import annotations + +from core.reasoning.evidence import ( + COMMITMENT_DISAGREEMENT, + DUPLICATE_STRUCTURAL_SIGNATURE, + INSUFFICIENT_EVIDENCE, + MISSING_COMMITMENT, + TIER2_VERIFIED, + EvidenceBundle, + OperatorEvidence, + Tier2Verdict, + verify_tier2_agreement, +) +from core.reasoning.adapters import ( + evidence_from_entailment_trace, + evidence_from_math_solution, +) + +__all__ = [ + "COMMITMENT_DISAGREEMENT", + "DUPLICATE_STRUCTURAL_SIGNATURE", + "INSUFFICIENT_EVIDENCE", + "MISSING_COMMITMENT", + "TIER2_VERIFIED", + "EvidenceBundle", + "OperatorEvidence", + "Tier2Verdict", + "verify_tier2_agreement", + "evidence_from_entailment_trace", + "evidence_from_math_solution", +] diff --git a/core/reasoning/adapters.py b/core/reasoning/adapters.py new file mode 100644 index 00000000..2b9cd1ff --- /dev/null +++ b/core/reasoning/adapters.py @@ -0,0 +1,117 @@ +"""Adapters from existing operator traces into shared reasoning evidence.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from core.reasoning.evidence import OperatorEvidence +from generate.math_problem_graph import MathProblemGraph +from generate.math_solver import SolutionTrace +from generate.proof_chain import Entailment, EntailmentTrace + + +def evidence_from_entailment_trace(trace: EntailmentTrace) -> OperatorEvidence: + """Convert propositional entailment trace evidence to the shared contract.""" + query_key = trace.query_key or "" + check_keys = tuple( + key for key in ( + trace.conjunction_key, + trace.entailment_check_key, + trace.refutation_check_key, + ) + if key + ) + if trace.outcome is Entailment.REFUSED: + commitment_key = "" + else: + commitment_key = f"entailment:{trace.outcome.value}:{query_key}" + structural_signature = _sha256_text( + json.dumps( + { + "operator": "propositional_entailment", + "premise_keys": list(trace.premise_keys), + "check_keys": list(check_keys), + }, + sort_keys=True, + separators=(",", ":"), + ) + ) + return OperatorEvidence( + domain="mathematics_logic", + operator="propositional_entailment", + outcome=trace.outcome.value, + reason=trace.reason, + input_keys=(*trace.premise_keys, query_key), + check_keys=check_keys, + commitment_key=commitment_key, + structural_signature=structural_signature, + payload={"entailment_trace": trace.as_dict()}, + ) + + +def evidence_from_math_solution( + *, + graph: MathProblemGraph, + trace: SolutionTrace, + reader_trace: tuple[str, ...] = (), + operator: str = "math_problem_graph_solve_verify", + reason: str = "solver_verifier_passed", +) -> OperatorEvidence: + """Convert a verified MathProblemGraph solution to shared evidence.""" + graph_hash = hashlib.sha256(graph.canonical_bytes()).hexdigest() + trace_hash = hashlib.sha256(trace.canonical_bytes()).hexdigest() + operation_kinds = tuple(step.operation_kind for step in trace.steps) + commitment_key = json.dumps( + { + "answer_entity": trace.answer_entity, + "answer_unit": trace.answer_unit, + "answer_value": trace.answer_value, + }, + sort_keys=True, + separators=(",", ":"), + ) + structural_signature = _sha256_text( + json.dumps( + { + "graph_hash": graph_hash, + "operation_kinds": list(operation_kinds), + "operator": operator, + "pack_id": trace.pack_id, + }, + sort_keys=True, + separators=(",", ":"), + ) + ) + return OperatorEvidence( + domain="mathematics_logic", + operator=operator, + outcome="verified", + reason=reason, + input_keys=(graph_hash,), + check_keys=(trace_hash, trace.graph_canonical_hash), + commitment_key=commitment_key, + structural_signature=structural_signature, + payload={ + "answer_entity": trace.answer_entity, + "answer_unit": trace.answer_unit, + "answer_value": trace.answer_value, + "graph_hash": graph_hash, + "operation_kinds": list(operation_kinds), + "pack_id": trace.pack_id, + "reader_trace": tuple(_reader_event(ev) for ev in reader_trace), + "trace_hash": trace_hash, + }, + ) + + +def _reader_event(event: str) -> Any: + try: + return json.loads(event) + except json.JSONDecodeError: + return event + + +def _sha256_text(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() diff --git a/core/reasoning/evidence.py b/core/reasoning/evidence.py new file mode 100644 index 00000000..c0ea18bd --- /dev/null +++ b/core/reasoning/evidence.py @@ -0,0 +1,205 @@ +"""Domain-neutral reasoning evidence and Tier-2 agreement checks. + +This module is deliberately pure data plus deterministic serialization. It is +the shared evidence shape for proof, reconstruction, contemplation, and sealed +learning arenas; it does not authorize serving behavior by itself. +""" + +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from collections.abc import Mapping +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import Any, Final + +TIER2_VERIFIED: Final[str] = "tier2_verified" +INSUFFICIENT_EVIDENCE: Final[str] = "insufficient_evidence" +DUPLICATE_STRUCTURAL_SIGNATURE: Final[str] = "duplicate_structural_signature" +COMMITMENT_DISAGREEMENT: Final[str] = "commitment_disagreement" +MISSING_COMMITMENT: Final[str] = "missing_commitment" + +TIER2_REASONS: Final[frozenset[str]] = frozenset({ + TIER2_VERIFIED, + INSUFFICIENT_EVIDENCE, + DUPLICATE_STRUCTURAL_SIGNATURE, + COMMITMENT_DISAGREEMENT, + MISSING_COMMITMENT, +}) + + +def _freeze_json_value(value: Any) -> Any: + """Recursively freeze JSON-like payloads for immutable evidence storage.""" + if isinstance(value, Mapping): + frozen = {str(k): _freeze_json_value(v) for k, v in value.items()} + return MappingProxyType(frozen) + if isinstance(value, list | tuple): + return tuple(_freeze_json_value(v) for v in value) + if value is None or isinstance(value, str | int | float | bool): + return value + raise TypeError(f"unsupported evidence payload value: {type(value).__name__}") + + +def _json_value(value: Any) -> Any: + """Return a JSON-serializable copy of a frozen payload value.""" + if isinstance(value, Mapping): + return {str(k): _json_value(v) for k, v in value.items()} + if isinstance(value, list | tuple): + return [_json_value(v) for v in value] + if value is None or isinstance(value, str | int | float | bool): + return value + raise TypeError(f"unsupported frozen evidence value: {type(value).__name__}") + + +def _canonical_json(payload: Mapping[str, Any]) -> str: + return json.dumps( + _json_value(payload), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + +@dataclass(frozen=True, slots=True) +class OperatorEvidence: + """Replayable evidence for one deterministic operator invocation.""" + + domain: str + operator: str + outcome: str + reason: str + input_keys: tuple[str, ...] + check_keys: tuple[str, ...] + commitment_key: str + structural_signature: str + payload: Mapping[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + for field_name in ( + "domain", + "operator", + "outcome", + "reason", + "structural_signature", + ): + value = getattr(self, field_name) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"OperatorEvidence.{field_name} is required") + object.__setattr__(self, field_name, value.strip()) + if not isinstance(self.commitment_key, str): + raise ValueError("OperatorEvidence.commitment_key must be a string") + object.__setattr__(self, "input_keys", tuple(str(k) for k in self.input_keys)) + object.__setattr__(self, "check_keys", tuple(str(k) for k in self.check_keys)) + object.__setattr__(self, "payload", _freeze_json_value(dict(self.payload))) + + def as_dict(self) -> dict[str, Any]: + return { + "domain": self.domain, + "operator": self.operator, + "outcome": self.outcome, + "reason": self.reason, + "input_keys": list(self.input_keys), + "check_keys": list(self.check_keys), + "commitment_key": self.commitment_key, + "structural_signature": self.structural_signature, + "payload": _json_value(self.payload), + } + + def canonical_json(self) -> str: + return _canonical_json(self.as_dict()) + + @property + def evidence_hash(self) -> str: + return hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest() + + +@dataclass(frozen=True, slots=True) +class EvidenceBundle: + """Ordered collection of operator evidence with stable serialization.""" + + evidences: tuple[OperatorEvidence, ...] + + def __post_init__(self) -> None: + object.__setattr__(self, "evidences", tuple(self.evidences)) + if not all(isinstance(ev, OperatorEvidence) for ev in self.evidences): + raise ValueError("EvidenceBundle.evidences must contain OperatorEvidence") + + def as_dict(self) -> dict[str, Any]: + return {"evidences": [ev.as_dict() for ev in self.evidences]} + + def canonical_json(self) -> str: + return _canonical_json(self.as_dict()) + + @property + def evidence_hash(self) -> str: + return hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest() + + +@dataclass(frozen=True, slots=True) +class Tier2Verdict: + """Result of a domain-neutral convergent self-verification check.""" + + verified: bool + reason: str + commitment_key: str = "" + evidence_hash: str = "" + structural_signatures: tuple[str, ...] = () + + def __post_init__(self) -> None: + if self.reason not in TIER2_REASONS: + raise ValueError(f"unknown Tier2Verdict.reason: {self.reason!r}") + object.__setattr__( + self, + "structural_signatures", + tuple(str(s) for s in self.structural_signatures), + ) + + def as_dict(self) -> dict[str, Any]: + return { + "verified": self.verified, + "reason": self.reason, + "commitment_key": self.commitment_key, + "evidence_hash": self.evidence_hash, + "structural_signatures": list(self.structural_signatures), + } + + +def verify_tier2_agreement( + evidences: tuple[OperatorEvidence, ...] | list[OperatorEvidence], +) -> Tier2Verdict: + """Require two distinct structures converging on one non-empty commitment.""" + bundle = EvidenceBundle(tuple(evidences)) + if len(bundle.evidences) < 2: + return Tier2Verdict(False, INSUFFICIENT_EVIDENCE) + + if any(not ev.commitment_key for ev in bundle.evidences): + return Tier2Verdict(False, MISSING_COMMITMENT, evidence_hash=bundle.evidence_hash) + + signatures = tuple(ev.structural_signature for ev in bundle.evidences) + if len(set(signatures)) < 2: + return Tier2Verdict( + False, + DUPLICATE_STRUCTURAL_SIGNATURE, + evidence_hash=bundle.evidence_hash, + structural_signatures=tuple(sorted(set(signatures))), + ) + + commitments = Counter(ev.commitment_key for ev in bundle.evidences) + shared = [key for key, count in commitments.items() if count >= 2] + if len(shared) != 1 or len(commitments) != 1: + return Tier2Verdict( + False, + COMMITMENT_DISAGREEMENT, + evidence_hash=bundle.evidence_hash, + structural_signatures=tuple(sorted(set(signatures))), + ) + + return Tier2Verdict( + True, + TIER2_VERIFIED, + commitment_key=shared[0], + evidence_hash=bundle.evidence_hash, + structural_signatures=tuple(sorted(set(signatures))), + ) diff --git a/tests/test_adr_0199_learning_arena_engine.py b/tests/test_adr_0199_learning_arena_engine.py index c75e9412..0878dcf0 100644 --- a/tests/test_adr_0199_learning_arena_engine.py +++ b/tests/test_adr_0199_learning_arena_engine.py @@ -2,8 +2,8 @@ Proves exactly the PR-2 gate: the extracted engine reuses the single pinned floor (L-1), holds the seal (L-3), is a deterministic fold (L-4), and the GSM8K -math instance behaves byte-identically to before (the committed golden queue). -Tier-2 scoring / L-5 are deferred to the PR that wires t2 verification. +math instance preserves its committed golden queue. Tier-2 scoring is optional +and remains inert unless a domain supplies a verifier. """ from __future__ import annotations @@ -19,6 +19,7 @@ from core.learning_arena import ( Problem, run_practice, ) +from core.reasoning import TIER2_VERIFIED, Tier2Verdict from core.reliability_gate import conservative_floor _REPO = Path(__file__).resolve().parents[1] @@ -54,6 +55,22 @@ class _StubTether: return float(problem.payload["gold"]) +@dataclass(frozen=True, slots=True) +class _StubTier2Verifier: + domain_id: str = "stub" + + def verify(self, attempt: BaseAttempt, problem: Problem) -> Tier2Verdict: + if not problem.payload.get("t2", False): + return Tier2Verdict(False, "insufficient_evidence") + return Tier2Verdict( + True, + TIER2_VERIFIED, + commitment_key=str(attempt.answer), + evidence_hash=f"hash:{attempt.case_id}", + structural_signatures=("shape-a", "shape-b"), + ) + + def _problems() -> list[Problem]: return [ Problem("c1", "alpha", {"verdict": "correct", "answer": 9.0, "gold": 9.0}), @@ -77,6 +94,7 @@ def test_engine_counts_ledger_eliminations_diagnoses(): alpha = rep.ledger["alpha"] assert (alpha.correct, alpha.wrong, alpha.refused) == (2, 1, 0) + assert (alpha.t2_verified, alpha.t2_agrees_gold) == (0, 0) beta = rep.ledger["beta"] assert (beta.correct, beta.wrong, beta.refused) == (0, 0, 1) @@ -94,6 +112,32 @@ def test_engine_is_deterministic(): assert json.dumps(a.as_dict(), sort_keys=True) == json.dumps(b.as_dict(), sort_keys=True) +def test_optional_tier2_verifier_populates_anchor_precision() -> None: + problems = [ + Problem("c1", "alpha", {"verdict": "correct", "answer": 9.0, "gold": 9.0, "t2": True}), + Problem("c2", "alpha", {"verdict": "wrong", "answer": 7.0, "gold": 10.0, "t2": True}), + Problem("c3", "alpha", {"verdict": "correct", "answer": 4.0, "gold": 4.0, "t2": False}), + Problem("c4", "alpha", {"verdict": "refused", "reason": "no graph", "t2": True}), + ] + + rep = run_practice( + problems, + _StubSolver(), + _StubTether(), + diagnose=_diagnose, + tier2_verifier=_StubTier2Verifier(), + ) + + alpha = rep.ledger["alpha"] + assert (alpha.correct, alpha.wrong, alpha.refused) == (2, 1, 1) + assert alpha.t2_verified == 2 + assert alpha.t2_agrees_gold == 1 + per_class = rep.as_dict()["per_class"]["alpha"] + assert per_class["t2_verified"] == 2 + assert per_class["t2_agrees_gold"] == 1 + assert per_class["t2_precision"] == alpha.t2_precision + + # --- L-1: one shared pinned floor; no per-arena pessimism constant ------------ diff --git a/tests/test_deductive_logic_entail.py b/tests/test_deductive_logic_entail.py index c12e72f1..e34d0e10 100644 --- a/tests/test_deductive_logic_entail.py +++ b/tests/test_deductive_logic_entail.py @@ -19,6 +19,7 @@ import random import pytest +from core.reasoning import evidence_from_entailment_trace from evals.deductive_logic.generate import _make_case from evals.deductive_logic.oracle import oracle_entailment from evals.deductive_logic.runner import _ROOT, _load, build_report @@ -109,6 +110,13 @@ def test_entailment_trace_is_deterministic_evidence() -> None: assert t1.canonical_json() == t2.canonical_json() assert "premise_keys" in t1.canonical_json() + evidence = evidence_from_entailment_trace(t1) + assert evidence.domain == "mathematics_logic" + assert evidence.operator == "propositional_entailment" + assert evidence.outcome == "entailed" + assert evidence.commitment_key.startswith("entailment:entailed:") + assert evidence.evidence_hash == evidence_from_entailment_trace(t2).evidence_hash + def test_refused_trace_preserves_available_canonical_evidence() -> None: trace = evaluate_entailment_with_trace(("P",), "P & & Q") diff --git a/tests/test_gsm8k_r1_reconstruction.py b/tests/test_gsm8k_r1_reconstruction.py index 9af1f973..b404422d 100644 --- a/tests/test_gsm8k_r1_reconstruction.py +++ b/tests/test_gsm8k_r1_reconstruction.py @@ -9,6 +9,7 @@ from __future__ import annotations import pytest +from core.reasoning import evidence_from_math_solution from generate.derivation import reconstruct_r1_total from generate.math_candidate_graph import parse_and_solve from generate.math_solver import solve @@ -74,6 +75,23 @@ def test_r1_reconstruction_solves_with_verified_graph(case_id: str) -> None: assert verdict.passed is True assert trace.answer_value == expected + evidence = evidence_from_math_solution( + graph=result.graph, + trace=trace, + reader_trace=result.reader_trace, + operator="gsm8k_r1_reconstruction", + ) + assert evidence.domain == "mathematics_logic" + assert evidence.operator == "gsm8k_r1_reconstruction" + assert evidence.outcome == "verified" + assert evidence.commitment_key + assert evidence.evidence_hash == evidence_from_math_solution( + graph=result.graph, + trace=trace, + reader_trace=result.reader_trace, + operator="gsm8k_r1_reconstruction", + ).evidence_hash + @pytest.mark.parametrize("case_id", sorted(R1_POSITIVES)) def test_parse_and_solve_wires_r1_only_as_verified_graph(case_id: str) -> None: @@ -113,4 +131,3 @@ def test_parse_and_solve_wires_r1_only_as_verified_graph(case_id: str) -> None: def test_r1_reconstruction_refuses_out_of_scope_shapes(text: str) -> None: result = parse_and_solve(text) assert not result.is_admitted - diff --git a/tests/test_reasoning_evidence.py b/tests/test_reasoning_evidence.py new file mode 100644 index 00000000..f23277aa --- /dev/null +++ b/tests/test_reasoning_evidence.py @@ -0,0 +1,128 @@ +"""Tier-2 reasoning evidence spine tests.""" + +from __future__ import annotations + +import dataclasses +import json + +import pytest + +from core.contemplation.miners.learning_arena import mine_learning_arena_report +from core.reasoning import ( + COMMITMENT_DISAGREEMENT, + DUPLICATE_STRUCTURAL_SIGNATURE, + MISSING_COMMITMENT, + TIER2_VERIFIED, + EvidenceBundle, + OperatorEvidence, + verify_tier2_agreement, +) + + +def _ev( + *, + signature: str = "shape-a", + commitment: str = "answer:42", + outcome: str = "verified", +) -> OperatorEvidence: + return OperatorEvidence( + domain="mathematics_logic", + operator="unit_test_operator", + outcome=outcome, + reason="test", + input_keys=("input-a",), + check_keys=("check-a",), + commitment_key=commitment, + structural_signature=signature, + payload={"nested": {"values": [1, 2, "x"]}}, + ) + + +def test_operator_evidence_is_frozen_and_payload_is_immutable() -> None: + ev = _ev() + with pytest.raises(dataclasses.FrozenInstanceError): + ev.domain = "mutated" # type: ignore[misc] + with pytest.raises(TypeError): + ev.payload["nested"] = {} # type: ignore[index] + + +def test_canonical_json_and_hash_are_deterministic() -> None: + ev1 = _ev() + ev2 = _ev() + assert ev1.canonical_json() == ev2.canonical_json() + assert ev1.evidence_hash == ev2.evidence_hash + assert json.loads(ev1.canonical_json())["payload"]["nested"]["values"] == [1, 2, "x"] + + +def test_evidence_bundle_hash_is_ordered_and_stable() -> None: + bundle = EvidenceBundle((_ev(signature="a"), _ev(signature="b"))) + same = EvidenceBundle((_ev(signature="a"), _ev(signature="b"))) + swapped = EvidenceBundle((_ev(signature="b"), _ev(signature="a"))) + assert bundle.evidence_hash == same.evidence_hash + assert bundle.evidence_hash != swapped.evidence_hash + + +def test_tier2_verifies_two_distinct_structures_same_commitment() -> None: + verdict = verify_tier2_agreement(( + _ev(signature="shape-a"), + _ev(signature="shape-b"), + )) + assert verdict.verified is True + assert verdict.reason == TIER2_VERIFIED + assert verdict.commitment_key == "answer:42" + assert verdict.structural_signatures == ("shape-a", "shape-b") + + +def test_tier2_refuses_duplicate_signature() -> None: + verdict = verify_tier2_agreement(( + _ev(signature="same"), + _ev(signature="same"), + )) + assert verdict.verified is False + assert verdict.reason == DUPLICATE_STRUCTURAL_SIGNATURE + + +def test_tier2_refuses_disagreeing_commitments() -> None: + verdict = verify_tier2_agreement(( + _ev(signature="shape-a", commitment="answer:42"), + _ev(signature="shape-b", commitment="answer:41"), + )) + assert verdict.verified is False + assert verdict.reason == COMMITMENT_DISAGREEMENT + + +def test_tier2_refuses_missing_commitment() -> None: + verdict = verify_tier2_agreement(( + _ev(signature="shape-a", commitment=""), + _ev(signature="shape-b", commitment="answer:42"), + )) + assert verdict.verified is False + assert verdict.reason == MISSING_COMMITMENT + + +def test_learning_arena_miner_emits_speculative_findings(tmp_path) -> None: + report = { + "per_class": { + "alpha": { + "committed": 2, + "coverage": 0.2, + "t2_verified": 0, + } + }, + "elimination_records": [ + { + "case_id": "c-wrong", + "class_name": "alpha", + "gold": 10.0, + "reason": "answer mismatch", + } + ], + } + path = tmp_path / "learning_report.json" + path.write_text(json.dumps(report), encoding="utf-8") + + findings = mine_learning_arena_report(path, substrate_hash="substrate") + + predicates = {finding.predicate for finding in findings} + assert predicates == {"weak_coverage", "missing_tier2_evidence", "wrong_attempt"} + assert all(finding.epistemic_status.value == "speculative" for finding in findings)