core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat: add tier2 reasoning evidence spine

Browse source
This commit is contained in:
Shay 2026-06-04 14:12:18 -07:00
parent efbd646ba1
commit 134a91c48a
14 changed files with 719 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
core/cognition/pipeline.py
View file

@ -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:

2
core/contemplation/miners/__init__.py
View file

@ -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",
]

111
core/contemplation/miners/learning_arena.py Normal file
View file

@ -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"]

2
core/learning_arena/__init__.py
View file

@ -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",

36
core/learning_arena/engine.py
View file

@ -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,

11
core/learning_arena/protocols.py
View file

@ -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."""

11
core/learning_arena/report.py
View file

@ -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())
},

33
core/reasoning/__init__.py Normal file
View file

@ -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",
]

117
core/reasoning/adapters.py Normal file
View file

@ -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()

205
core/reasoning/evidence.py Normal file
View file

@ -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))),
)

48
tests/test_adr_0199_learning_arena_engine.py
View file

@ -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 ------------

8
tests/test_deductive_logic_entail.py
View file

@ -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")

19
tests/test_gsm8k_r1_reconstruction.py
View file

@ -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

128
tests/test_reasoning_evidence.py Normal file
View file

@ -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)