diff --git a/core/cli.py b/core/cli.py index 8d7ddf59..ed085a81 100644 --- a/core/cli.py +++ b/core/cli.py @@ -66,6 +66,7 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = { "tests/test_pipeline_teaching_integration.py", "tests/test_epistemic_invariants.py", "tests/test_adr_0172_w2_decomposer.py", + "tests/test_adr_0172_w5_inference_proposal.py", ), "packs": ( "tests/test_core_semantic_seed_pack.py", diff --git a/teaching/math_inference_proposal.py b/teaching/math_inference_proposal.py new file mode 100644 index 00000000..c49a7cf3 --- /dev/null +++ b/teaching/math_inference_proposal.py @@ -0,0 +1,570 @@ +"""ADR-0172 Tier 2 / W5 — MathReaderInferenceProposal schema. + +Tier 2 intensional-contemplation proposal. Records a proposed structural +equivalence (canonicalization bridge) derived from the refusal corpus. + +Each proposal carries: + +- ≥3 :class:`~teaching.math_evidence.MathReaderRefusalEvidence` pointers + (tighter than Tier 1's ≥2); +- a ``structural_claim`` naming the proposed equivalence class; +- two :class:`ArmResult` records (arm1 = held-out, arm2 = known-good) + from the two-arm self-test (W7); +- ``ratification_effect_kind`` pinned to ``"canonicalization_bridge"``; +- a ``wrong_zero_assertion`` (≥40 chars); +- a :class:`~teaching.math_reasoning_trace.ReasoningTrace` carrying ≥6 + steps including ``{abstraction, test_design, test_application, test_result}``. + +Invariants enforced by :func:`build_inference_proposal`: + +1. ``domain == "math"``. +2. ``len(evidence_pointers) >= 3``. +3. ``reasoning_trace`` carries ≥6 steps. +4. ``reasoning_trace`` steps include every kind in ``_REQUIRED_STEP_KINDS``. +5. Both arms cannot simultaneously be ``"REJECT"``. +6. Arm 2 ``"PASS"`` requires ``cases_changed_answer == 0``. +7. ``ratification_effect_kind == "canonicalization_bridge"``. +8. ``ratification_effect_payload`` is JSON-serializable. +9. ``wrong_zero_assertion`` ≥ 40 chars (stripped). + +JSONL self-containment via :func:`to_jsonl_record` / :func:`from_jsonl_record` +mirrors the post-#386 pattern from ``math_contemplation_proposal.py``. + +Trust boundary: schema-only module. No filesystem I/O, no teaching-store +writes, no runtime pipeline hooks. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Literal + +from generate.comprehension.audit import AuditRow +from teaching.math_evidence import MathReaderRefusalEvidence +from teaching.math_reasoning_trace import ReasoningStep, ReasoningTrace, build_trace + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +ArmOutcome = Literal["PASS", "NEUTRAL", "REJECT"] +ArmName = Literal["arm1_held_out", "arm2_known_good"] + +_ARM_OUTCOMES: frozenset[str] = frozenset({"PASS", "NEUTRAL", "REJECT"}) +_ARM_NAMES: frozenset[str] = frozenset({"arm1_held_out", "arm2_known_good"}) + +_EVIDENCE_FLOOR: int = 3 +_REQUIRED_STEP_KINDS: frozenset[str] = frozenset({ + "abstraction", + "test_design", + "test_application", + "test_result", +}) +_MIN_TRACE_STEPS: int = 6 +_WRONG_ZERO_MIN_LEN: int = 40 + + +# --------------------------------------------------------------------------- +# ArmResult +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class ArmResult: + """Outcome record for one self-test arm. + + Fields + ------ + arm: + ``"arm1_held_out"`` — 30% held-out refusal subset, or + ``"arm2_known_good"`` — prior admitted-with-correct-answer set. + outcome: + ``"PASS"`` | ``"NEUTRAL"`` | ``"REJECT"``. + cases_tested: + Total cases evaluated in this arm. + cases_admitted: + Cases where the bridge produced an admission result. + cases_changed_answer: + Cases where a previously-correct answer changed under the bridge. + Must be 0 when arm is ``"arm2_known_good"`` and outcome is ``"PASS"``. + """ + + arm: ArmName + outcome: ArmOutcome + cases_tested: int + cases_admitted: int + cases_changed_answer: int + + +def build_arm_result( + *, + arm: str, + outcome: str, + cases_tested: int, + cases_admitted: int, + cases_changed_answer: int, +) -> ArmResult: + """Build an :class:`ArmResult` with basic field validation.""" + if arm not in _ARM_NAMES: + raise ValueError( + f"arm must be one of {sorted(_ARM_NAMES)}; got {arm!r}" + ) + if outcome not in _ARM_OUTCOMES: + raise ValueError( + f"outcome must be one of {sorted(_ARM_OUTCOMES)}; got {outcome!r}" + ) + if cases_tested < 0: + raise ValueError(f"cases_tested must be ≥0; got {cases_tested}") + if cases_admitted < 0: + raise ValueError(f"cases_admitted must be ≥0; got {cases_admitted}") + if cases_changed_answer < 0: + raise ValueError(f"cases_changed_answer must be ≥0; got {cases_changed_answer}") + return ArmResult( + arm=arm, # type: ignore[arg-type] + outcome=outcome, # type: ignore[arg-type] + cases_tested=cases_tested, + cases_admitted=cases_admitted, + cases_changed_answer=cases_changed_answer, + ) + + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class MathReaderInferenceProposal: + """One proposed canonicalization bridge for the math domain (Tier 2). + + Construct via :func:`build_inference_proposal` — do not instantiate + directly (``inference_id`` is content-derived). + + Fields + ------ + inference_id: + ``sha256(canonical_bytes(...)).hexdigest()`` over all other fields. + domain: + Always ``"math"``. + structural_claim: + Human-readable description of the proposed equivalence class. + evidence_pointers: + ≥3 :class:`MathReaderRefusalEvidence` records. + arm1_result: + Two-arm self-test result for the held-out subset. + arm2_result: + Two-arm self-test result for the known-good set. + ratification_effect_kind: + Always ``"canonicalization_bridge"`` for Tier 2 proposals. + ratification_effect_payload: + JSON-serializable payload describing the bridge implementation. + wrong_zero_assertion: + ≥40-char statement pinning the wrong=0 invariant. + replay_equivalence_hash: + ``sha256`` digest of the replay-equivalence gate output. + reasoning_trace: + :class:`~teaching.math_reasoning_trace.ReasoningTrace` carrying ≥6 + steps, including ``{abstraction, test_design, test_application, + test_result}``. + """ + + inference_id: str + domain: Literal["math"] + structural_claim: str + evidence_pointers: tuple[MathReaderRefusalEvidence, ...] + arm1_result: ArmResult + arm2_result: ArmResult + ratification_effect_kind: Literal["canonicalization_bridge"] + ratification_effect_payload: object + wrong_zero_assertion: str + replay_equivalence_hash: str + reasoning_trace: ReasoningTrace + + +# --------------------------------------------------------------------------- +# Canonical-bytes serialization (content-hash; not round-trip) +# --------------------------------------------------------------------------- + + +def _arm_result_to_canonical(arm: ArmResult) -> dict[str, Any]: + return { + "arm": arm.arm, + "cases_admitted": arm.cases_admitted, + "cases_changed_answer": arm.cases_changed_answer, + "cases_tested": arm.cases_tested, + "outcome": arm.outcome, + } + + +def canonical_bytes(proposal: MathReaderInferenceProposal) -> bytes: + """Return deterministic canonical bytes over all fields except inference_id. + + Evidence pointers are reduced to their ``evidence_hash`` digests; + ``reasoning_trace`` is reduced to its ``trace_id``. Stable across + processes and dict insertion order. + """ + payload: dict[str, Any] = { + "arm1_result": _arm_result_to_canonical(proposal.arm1_result), + "arm2_result": _arm_result_to_canonical(proposal.arm2_result), + "domain": proposal.domain, + "evidence_pointers": sorted( + ev.evidence_hash for ev in proposal.evidence_pointers + ), + "ratification_effect_kind": proposal.ratification_effect_kind, + "ratification_effect_payload": proposal.ratification_effect_payload, + "reasoning_trace_id": proposal.reasoning_trace.trace_id, + "replay_equivalence_hash": proposal.replay_equivalence_hash, + "structural_claim": proposal.structural_claim, + "wrong_zero_assertion": proposal.wrong_zero_assertion, + } + return json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def compute_inference_id( + *, + domain: Literal["math"], + structural_claim: str, + evidence_pointers: tuple[MathReaderRefusalEvidence, ...], + arm1_result: ArmResult, + arm2_result: ArmResult, + ratification_effect_kind: Literal["canonicalization_bridge"], + ratification_effect_payload: object, + wrong_zero_assertion: str, + replay_equivalence_hash: str, + reasoning_trace: ReasoningTrace, +) -> str: + """Hash all content fields to produce a stable ``inference_id``.""" + placeholder = MathReaderInferenceProposal( + inference_id="", + domain=domain, + structural_claim=structural_claim, + evidence_pointers=evidence_pointers, + arm1_result=arm1_result, + arm2_result=arm2_result, + ratification_effect_kind=ratification_effect_kind, + ratification_effect_payload=ratification_effect_payload, + wrong_zero_assertion=wrong_zero_assertion, + replay_equivalence_hash=replay_equivalence_hash, + reasoning_trace=reasoning_trace, + ) + return hashlib.sha256(canonical_bytes(placeholder)).hexdigest() + + +# --------------------------------------------------------------------------- +# Factory +# --------------------------------------------------------------------------- + + +def build_inference_proposal( + *, + domain: Literal["math"] = "math", + structural_claim: str, + evidence_pointers: tuple[MathReaderRefusalEvidence, ...], + arm1_result: ArmResult, + arm2_result: ArmResult, + ratification_effect_kind: str, + ratification_effect_payload: object, + wrong_zero_assertion: str, + replay_equivalence_hash: str, + reasoning_trace: ReasoningTrace, +) -> MathReaderInferenceProposal: + """Build a :class:`MathReaderInferenceProposal` with all invariants enforced. + + Raises ``ValueError`` on any violation; the caller must fix the inputs. + """ + if domain != "math": + raise ValueError(f"domain must be 'math'; got {domain!r}") + + if len(evidence_pointers) < _EVIDENCE_FLOOR: + raise ValueError( + f"evidence_pointers requires ≥{_EVIDENCE_FLOOR} entries; " + f"got {len(evidence_pointers)}" + ) + + if not isinstance(reasoning_trace, ReasoningTrace): + raise ValueError( + f"reasoning_trace must be a ReasoningTrace instance; " + f"got {type(reasoning_trace).__name__}" + ) + + if len(reasoning_trace.steps) < _MIN_TRACE_STEPS: + raise ValueError( + f"reasoning_trace must carry ≥{_MIN_TRACE_STEPS} steps; " + f"got {len(reasoning_trace.steps)}" + ) + + present_kinds = {step.step_kind for step in reasoning_trace.steps} + missing_kinds = _REQUIRED_STEP_KINDS - present_kinds + if missing_kinds: + raise ValueError( + f"reasoning_trace is missing required step kind(s): " + f"{sorted(missing_kinds)}; found kinds: {sorted(present_kinds)}" + ) + + if arm1_result.outcome == "REJECT" and arm2_result.outcome == "REJECT": + raise ValueError( + "both arms cannot simultaneously be REJECT at construction; " + "proposals with two REJECT arms must not surface to the schema layer" + ) + + if arm2_result.outcome == "PASS" and arm2_result.cases_changed_answer != 0: + raise ValueError( + f"arm2 PASS requires cases_changed_answer == 0; " + f"got {arm2_result.cases_changed_answer}" + ) + + if ratification_effect_kind != "canonicalization_bridge": + raise ValueError( + f"ratification_effect_kind must be 'canonicalization_bridge'; " + f"got {ratification_effect_kind!r}" + ) + + try: + json.dumps( + ratification_effect_payload, + ensure_ascii=False, + separators=(",", ":"), + ) + except (TypeError, ValueError) as exc: + raise ValueError( + f"ratification_effect_payload is not JSON-serializable: {exc}" + ) from exc + + if not wrong_zero_assertion or len(wrong_zero_assertion.strip()) < _WRONG_ZERO_MIN_LEN: + raise ValueError( + f"wrong_zero_assertion must be ≥{_WRONG_ZERO_MIN_LEN} chars (non-empty); " + f"got {len(wrong_zero_assertion)!r}" + ) + + iid = compute_inference_id( + domain=domain, + structural_claim=structural_claim, + evidence_pointers=evidence_pointers, + arm1_result=arm1_result, + arm2_result=arm2_result, + ratification_effect_kind="canonicalization_bridge", + ratification_effect_payload=ratification_effect_payload, + wrong_zero_assertion=wrong_zero_assertion, + replay_equivalence_hash=replay_equivalence_hash, + reasoning_trace=reasoning_trace, + ) + + return MathReaderInferenceProposal( + inference_id=iid, + domain=domain, + structural_claim=structural_claim, + evidence_pointers=tuple(evidence_pointers), + arm1_result=arm1_result, + arm2_result=arm2_result, + ratification_effect_kind="canonicalization_bridge", + ratification_effect_payload=ratification_effect_payload, + wrong_zero_assertion=wrong_zero_assertion, + replay_equivalence_hash=replay_equivalence_hash, + reasoning_trace=reasoning_trace, + ) + + +# --------------------------------------------------------------------------- +# Self-contained JSONL persistence serializer +# --------------------------------------------------------------------------- +# +# canonical_bytes() is the content-hash function; it reduces evidence_pointers +# to evidence_hashes and reasoning_trace to its trace_id. That is correct +# for inference_id derivation but not for round-tripping through disk. +# +# to_jsonl_record() / from_jsonl_record() emit a self-contained record so the +# workbench and HITL queue can read proposals.jsonl without re-running the +# two-arm loop (W7). +# +# Determinism contract: sort_keys=True, compact separators, no floats. + + +def _audit_row_to_dict(row: AuditRow) -> dict[str, Any]: + return { + "case_id": row.case_id, + "sentence_index": row.sentence_index, + "token_index": row.token_index, + "token_text": row.token_text, + "recognized_terms": list(row.recognized_terms), + "skipped_frame": row.skipped_frame, + "missing_operator": row.missing_operator, + "refusal_reason": row.refusal_reason, + "refusal_detail": row.refusal_detail, + } + + +def _audit_row_from_dict(data: dict[str, Any]) -> AuditRow: + return AuditRow( + case_id=str(data["case_id"]), + sentence_index=int(data["sentence_index"]), + token_index=int(data["token_index"]), + token_text=str(data["token_text"]), + recognized_terms=tuple(data.get("recognized_terms") or ()), + skipped_frame=data.get("skipped_frame"), + missing_operator=data.get("missing_operator"), + refusal_reason=str(data.get("refusal_reason", "")), + refusal_detail=str(data.get("refusal_detail", "")), + ) + + +def _evidence_to_dict(ev: MathReaderRefusalEvidence) -> dict[str, Any]: + return { + "case_id": ev.case_id, + "sentence_index": ev.sentence_index, + "token_index": ev.token_index, + "refusal_reason": ev.refusal_reason, + "missing_operator": ev.missing_operator, + "claim_signature": ev.claim_signature, + "evidence_hash": ev.evidence_hash, + "sub_type": ev.sub_type, + "audit_row": _audit_row_to_dict(ev.audit_row), + } + + +def _evidence_from_dict(data: dict[str, Any]) -> MathReaderRefusalEvidence: + return MathReaderRefusalEvidence( + case_id=str(data["case_id"]), + sentence_index=int(data["sentence_index"]), + token_index=int(data["token_index"]), + refusal_reason=str(data["refusal_reason"]), + missing_operator=data.get("missing_operator"), + claim_signature=str(data.get("claim_signature", "")), + evidence_hash=str(data["evidence_hash"]), + audit_row=_audit_row_from_dict(data["audit_row"]), + sub_type=data["sub_type"], + ) + + +def _arm_result_to_dict(arm: ArmResult) -> dict[str, Any]: + return { + "arm": arm.arm, + "outcome": arm.outcome, + "cases_tested": arm.cases_tested, + "cases_admitted": arm.cases_admitted, + "cases_changed_answer": arm.cases_changed_answer, + } + + +def _arm_result_from_dict(data: dict[str, Any]) -> ArmResult: + return ArmResult( + arm=data["arm"], + outcome=data["outcome"], + cases_tested=int(data["cases_tested"]), + cases_admitted=int(data["cases_admitted"]), + cases_changed_answer=int(data["cases_changed_answer"]), + ) + + +def _step_to_dict(step: ReasoningStep) -> dict[str, Any]: + return { + "step_index": step.step_index, + "step_kind": step.step_kind, + "input_pointers": list(step.input_pointers), + "claim": step.claim, + "justification": step.justification, + "output_payload": step.output_payload, + } + + +def _step_from_dict(data: dict[str, Any]) -> ReasoningStep: + return ReasoningStep( + step_index=int(data["step_index"]), + step_kind=data["step_kind"], + input_pointers=tuple(str(p) for p in data.get("input_pointers", ())), + claim=str(data.get("claim", "")), + justification=str(data.get("justification", "")), + output_payload=data.get("output_payload"), + ) + + +def to_jsonl_record(proposal: MathReaderInferenceProposal) -> dict[str, Any]: + """Return a self-contained dict representation suitable for JSONL persistence. + + Unlike :func:`canonical_bytes`, this record includes: + - ``inference_id`` (so consumers don't need to recompute it) + - full ``evidence_pointers`` (nested dicts — not just hashes) + - full ``arm1_result`` and ``arm2_result`` + - full ``reasoning_trace.steps`` (inline — not just trace_id) + + The output is JSON-serializable. Encoding via + ``json.dumps(record, sort_keys=True, separators=(",", ":"), + ensure_ascii=False)`` produces deterministic byte-identical output. + """ + return { + "inference_id": proposal.inference_id, + "domain": proposal.domain, + "structural_claim": proposal.structural_claim, + "evidence_pointers": [ + _evidence_to_dict(ev) for ev in proposal.evidence_pointers + ], + "arm1_result": _arm_result_to_dict(proposal.arm1_result), + "arm2_result": _arm_result_to_dict(proposal.arm2_result), + "ratification_effect_kind": proposal.ratification_effect_kind, + "ratification_effect_payload": proposal.ratification_effect_payload, + "wrong_zero_assertion": proposal.wrong_zero_assertion, + "replay_equivalence_hash": proposal.replay_equivalence_hash, + "reasoning_trace": { + "trace_id": proposal.reasoning_trace.trace_id, + "steps": [_step_to_dict(s) for s in proposal.reasoning_trace.steps], + }, + } + + +def from_jsonl_record(record: dict[str, Any]) -> MathReaderInferenceProposal: + """Reconstruct a proposal from a :func:`to_jsonl_record` dict. + + Goes through :func:`build_inference_proposal` so all invariants are + re-validated. The reconstructed ``inference_id`` must match the + persisted one — mismatch indicates tampering or schema drift and + raises :class:`ValueError`. + """ + evidence_records = tuple( + _evidence_from_dict(d) for d in record.get("evidence_pointers", ()) + ) + steps = tuple(_step_from_dict(d) for d in record["reasoning_trace"]["steps"]) + trace = build_trace(steps) + + arm1 = _arm_result_from_dict(record["arm1_result"]) + arm2 = _arm_result_from_dict(record["arm2_result"]) + + proposal = build_inference_proposal( + domain=record.get("domain", "math"), + structural_claim=str(record["structural_claim"]), + evidence_pointers=evidence_records, + arm1_result=arm1, + arm2_result=arm2, + ratification_effect_kind=str(record["ratification_effect_kind"]), + ratification_effect_payload=record.get("ratification_effect_payload"), + wrong_zero_assertion=str(record["wrong_zero_assertion"]), + replay_equivalence_hash=str(record["replay_equivalence_hash"]), + reasoning_trace=trace, + ) + + persisted_id = str(record.get("inference_id", "")) + if persisted_id and persisted_id != proposal.inference_id: + raise ValueError( + "inference_id mismatch on JSONL round-trip: persisted " + f"{persisted_id!r} != recomputed {proposal.inference_id!r}" + ) + return proposal + + +__all__ = [ + "ArmName", + "ArmOutcome", + "ArmResult", + "MathReaderInferenceProposal", + "build_arm_result", + "build_inference_proposal", + "canonical_bytes", + "compute_inference_id", + "from_jsonl_record", + "to_jsonl_record", +] diff --git a/tests/test_adr_0172_w5_inference_proposal.py b/tests/test_adr_0172_w5_inference_proposal.py new file mode 100644 index 00000000..7b54b349 --- /dev/null +++ b/tests/test_adr_0172_w5_inference_proposal.py @@ -0,0 +1,448 @@ +"""ADR-0172 Tier 2 / W5 — MathReaderInferenceProposal schema tests. + +11 obligations tested: + +1. evidence floor: ≥3 evidence rows required. +2. canonical_bytes stability across independent calls. +3. inference_id determinism. +4. ratification_effect_kind Literal enforced ("canonicalization_bridge" only). +5. both arms cannot simultaneously be REJECT at construction. +6. arm2 PASS requires cases_changed_answer == 0. +7. reasoning_trace must carry ≥6 steps. +8. reasoning_trace must include {abstraction, test_design, test_application, + test_result}. +9. wrong_zero_assertion ≥40 chars enforced. +10. proposal is frozen (immutable dataclass). +11. to_jsonl_record / from_jsonl_record self-contained round-trip. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +from typing import Any + +import pytest + +from generate.comprehension.audit import AuditRow +from teaching.math_evidence import MathReaderRefusalEvidence, from_audit_row +from teaching.math_reasoning_trace import ReasoningStep, build_trace +from teaching.math_inference_proposal import ( + ArmResult, + MathReaderInferenceProposal, + build_arm_result, + build_inference_proposal, + canonical_bytes, + from_jsonl_record, + to_jsonl_record, +) + + +# --------------------------------------------------------------------------- +# Shared fixtures / helpers +# --------------------------------------------------------------------------- + + +def _make_audit_row(case_id: str = "test-001", token_text: str = "stub") -> AuditRow: + return AuditRow( + case_id=case_id, + sentence_index=0, + token_index=0, + token_text=token_text, + recognized_terms=(), + skipped_frame=None, + missing_operator="lexicon_entry", + refusal_reason="unknown_word", + refusal_detail=f"unknown word: {token_text}", + ) + + +def _make_evidence( + case_id: str = "test-001", + token_text: str = "stub", +) -> MathReaderRefusalEvidence: + return from_audit_row( + _make_audit_row(case_id=case_id, token_text=token_text), + "lexical", + ) + + +def _three_evidences() -> tuple[MathReaderRefusalEvidence, ...]: + return ( + _make_evidence("ev-001", "collected"), + _make_evidence("ev-002", "acquired"), + _make_evidence("ev-003", "received"), + ) + + +_STEP_KINDS_TIER2 = ( + "observation", + "grouping", + "abstraction", + "hypothesis", + "test_design", + "test_application", + "test_result", + "conclusion", +) + +_VALID_ASSERTION = ( + "Canonicalization bridge maps acquisition verbs to initial-state " + "without altering admissibility gates; wrong=0 preserved." +) + + +def _make_trace( + kinds: tuple[str, ...] = _STEP_KINDS_TIER2, +) -> Any: + steps = tuple( + ReasoningStep( + step_index=i, + step_kind=kind, # type: ignore[arg-type] + input_pointers=("ev-001", "ev-002", "ev-003"), + claim=f"step {i} claim", + justification=f"step {i} justification", + output_payload={"i": i}, + ) + for i, kind in enumerate(kinds) + ) + return build_trace(steps) + + +def _pass_arm1() -> ArmResult: + return build_arm_result( + arm="arm1_held_out", + outcome="PASS", + cases_tested=10, + cases_admitted=7, + cases_changed_answer=0, + ) + + +def _pass_arm2() -> ArmResult: + return build_arm_result( + arm="arm2_known_good", + outcome="PASS", + cases_tested=20, + cases_admitted=20, + cases_changed_answer=0, + ) + + +def _build(**overrides: Any) -> MathReaderInferenceProposal: + kwargs: dict[str, Any] = dict( + structural_claim=( + "' ' " + "canonicalizes to ' has '" + ), + evidence_pointers=_three_evidences(), + arm1_result=_pass_arm1(), + arm2_result=_pass_arm2(), + ratification_effect_kind="canonicalization_bridge", + ratification_effect_payload={"bridge": "acquisition_to_initial_state_v1"}, + wrong_zero_assertion=_VALID_ASSERTION, + replay_equivalence_hash="c" * 64, + reasoning_trace=_make_trace(), + ) + kwargs.update(overrides) + return build_inference_proposal(**kwargs) + + +# --------------------------------------------------------------------------- +# Test 1 — evidence floor: ≥3 rows required +# --------------------------------------------------------------------------- + + +def test_minimum_three_evidence_rows_rejects_two() -> None: + """Passing exactly two evidence rows raises ValueError (floor is ≥3).""" + two_ev = ( + _make_evidence("ev-001", "alpha"), + _make_evidence("ev-002", "beta"), + ) + with pytest.raises(ValueError, match="≥3"): + _build(evidence_pointers=two_ev) + + +def test_minimum_three_evidence_rows_rejects_one() -> None: + """Passing a single evidence row raises ValueError.""" + one_ev = (_make_evidence("ev-001", "alpha"),) + with pytest.raises(ValueError, match="≥3"): + _build(evidence_pointers=one_ev) + + +# --------------------------------------------------------------------------- +# Test 2 — canonical_bytes stability +# --------------------------------------------------------------------------- + + +def test_canonical_bytes_stable() -> None: + """Same inputs produce byte-identical canonical_bytes across two calls.""" + p1 = _build() + p2 = _build() + assert canonical_bytes(p1) == canonical_bytes(p2) + + +# --------------------------------------------------------------------------- +# Test 3 — inference_id determinism +# --------------------------------------------------------------------------- + + +def test_inference_id_determinism() -> None: + """Two independently built proposals from identical inputs share an inference_id.""" + p1 = _build() + p2 = _build() + assert p1.inference_id == p2.inference_id + assert p1.inference_id == hashlib.sha256(canonical_bytes(p1)).hexdigest() + + +# --------------------------------------------------------------------------- +# Test 4 — ratification_effect_kind Literal enforced +# --------------------------------------------------------------------------- + + +def test_ratification_effect_kind_must_be_canonicalization_bridge() -> None: + """Any value other than 'canonicalization_bridge' raises ValueError.""" + with pytest.raises(ValueError, match="canonicalization_bridge"): + _build(ratification_effect_kind="matcher_extension") + + +def test_ratification_effect_kind_empty_rejected() -> None: + """Empty string raises ValueError.""" + with pytest.raises(ValueError, match="canonicalization_bridge"): + _build(ratification_effect_kind="") + + +# --------------------------------------------------------------------------- +# Test 5 — both arms cannot simultaneously be REJECT +# --------------------------------------------------------------------------- + + +def test_both_arms_reject_raises() -> None: + """arm1=REJECT + arm2=REJECT raises ValueError.""" + reject_arm1 = build_arm_result( + arm="arm1_held_out", + outcome="REJECT", + cases_tested=5, + cases_admitted=2, + cases_changed_answer=0, + ) + reject_arm2 = build_arm_result( + arm="arm2_known_good", + outcome="REJECT", + cases_tested=10, + cases_admitted=1, + cases_changed_answer=3, + ) + with pytest.raises(ValueError, match="both arms"): + _build(arm1_result=reject_arm1, arm2_result=reject_arm2) + + +def test_single_reject_arm_is_accepted() -> None: + """arm1=REJECT with arm2=PASS is a valid construction.""" + reject_arm1 = build_arm_result( + arm="arm1_held_out", + outcome="REJECT", + cases_tested=5, + cases_admitted=2, + cases_changed_answer=0, + ) + # Should not raise — only both-REJECT is forbidden at schema level. + proposal = _build(arm1_result=reject_arm1, arm2_result=_pass_arm2()) + assert proposal.arm1_result.outcome == "REJECT" + assert proposal.arm2_result.outcome == "PASS" + + +# --------------------------------------------------------------------------- +# Test 6 — arm2 PASS requires cases_changed_answer == 0 +# --------------------------------------------------------------------------- + + +def test_arm2_pass_with_changed_answers_raises() -> None: + """arm2 outcome=PASS but cases_changed_answer>0 raises ValueError.""" + bad_arm2 = build_arm_result( + arm="arm2_known_good", + outcome="PASS", + cases_tested=20, + cases_admitted=20, + cases_changed_answer=1, + ) + with pytest.raises(ValueError, match="cases_changed_answer"): + _build(arm2_result=bad_arm2) + + +def test_arm2_neutral_allows_changed_answers() -> None: + """arm2 outcome=NEUTRAL with cases_changed_answer>0 is valid.""" + neutral_arm2 = build_arm_result( + arm="arm2_known_good", + outcome="NEUTRAL", + cases_tested=20, + cases_admitted=0, + cases_changed_answer=2, + ) + proposal = _build(arm2_result=neutral_arm2) + assert proposal.arm2_result.outcome == "NEUTRAL" + + +# --------------------------------------------------------------------------- +# Test 7 — reasoning_trace must carry ≥6 steps +# --------------------------------------------------------------------------- + + +def test_reasoning_trace_fewer_than_six_steps_raises() -> None: + """A trace with only 5 steps raises ValueError (floor is ≥6).""" + five_step_kinds = ( + "observation", + "abstraction", + "test_design", + "test_application", + "test_result", + ) + short_trace = _make_trace(kinds=five_step_kinds) + with pytest.raises(ValueError, match="≥6"): + _build(reasoning_trace=short_trace) + + +# --------------------------------------------------------------------------- +# Test 8 — reasoning_trace must include required step kinds +# --------------------------------------------------------------------------- + + +def test_reasoning_trace_missing_required_kind_raises() -> None: + """Omitting 'abstraction' from the trace raises ValueError.""" + kinds_without_abstraction = ( + "observation", + "grouping", + "hypothesis", + "test_design", + "test_application", + "test_result", + ) + trace = _make_trace(kinds=kinds_without_abstraction) + with pytest.raises(ValueError, match="abstraction"): + _build(reasoning_trace=trace) + + +def test_reasoning_trace_missing_test_design_raises() -> None: + """Omitting 'test_design' from the trace raises ValueError.""" + kinds_without_test_design = ( + "observation", + "grouping", + "abstraction", + "hypothesis", + "test_application", + "test_result", + ) + trace = _make_trace(kinds=kinds_without_test_design) + with pytest.raises(ValueError, match="test_design"): + _build(reasoning_trace=trace) + + +# --------------------------------------------------------------------------- +# Test 9 — wrong_zero_assertion ≥40 chars enforced +# --------------------------------------------------------------------------- + + +def test_wrong_zero_assertion_empty_raises() -> None: + """Empty wrong_zero_assertion raises ValueError.""" + with pytest.raises(ValueError, match="wrong_zero_assertion"): + _build(wrong_zero_assertion="") + + +def test_wrong_zero_assertion_too_short_raises() -> None: + """An assertion shorter than 40 chars raises ValueError.""" + with pytest.raises(ValueError, match="wrong_zero_assertion"): + _build(wrong_zero_assertion="short") + + +# --------------------------------------------------------------------------- +# Test 10 — proposal is frozen +# --------------------------------------------------------------------------- + + +def test_proposal_is_frozen() -> None: + """MathReaderInferenceProposal instances are immutable.""" + p = _build() + with pytest.raises(dataclasses.FrozenInstanceError): + p.inference_id = "mutated" # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# Test 11 — JSONL self-contained round-trip +# --------------------------------------------------------------------------- + + +def test_to_jsonl_record_self_contained() -> None: + """to_jsonl_record emits inference_id, full evidence_pointers, arm results, + and full trace steps inline.""" + p = _build() + record = to_jsonl_record(p) + + # inference_id present + assert record["inference_id"] == p.inference_id + + # full evidence dicts (not just hashes) + assert isinstance(record["evidence_pointers"], list) + assert len(record["evidence_pointers"]) == 3 + assert isinstance(record["evidence_pointers"][0], dict) + assert "evidence_hash" in record["evidence_pointers"][0] + assert "audit_row" in record["evidence_pointers"][0] + + # arm results present + assert record["arm1_result"]["arm"] == "arm1_held_out" + assert record["arm2_result"]["arm"] == "arm2_known_good" + + # full trace steps + assert isinstance(record["reasoning_trace"], dict) + assert record["reasoning_trace"]["trace_id"] == p.reasoning_trace.trace_id + assert len(record["reasoning_trace"]["steps"]) == len(_STEP_KINDS_TIER2) + + +def test_jsonl_record_round_trip() -> None: + """to_jsonl_record → from_jsonl_record returns an equivalent proposal.""" + p = _build() + record = to_jsonl_record(p) + restored = from_jsonl_record(record) + + assert restored.inference_id == p.inference_id + assert restored.domain == p.domain + assert restored.structural_claim == p.structural_claim + assert restored.ratification_effect_kind == p.ratification_effect_kind + assert restored.replay_equivalence_hash == p.replay_equivalence_hash + assert restored.reasoning_trace.trace_id == p.reasoning_trace.trace_id + assert len(restored.evidence_pointers) == len(p.evidence_pointers) + assert restored.arm1_result == p.arm1_result + assert restored.arm2_result == p.arm2_result + + +def test_jsonl_record_byte_stability() -> None: + """Same proposal → byte-identical JSON line across reruns.""" + p1 = _build() + p2 = _build() + line1 = json.dumps( + to_jsonl_record(p1), sort_keys=True, separators=(",", ":"), ensure_ascii=False + ) + line2 = json.dumps( + to_jsonl_record(p2), sort_keys=True, separators=(",", ":"), ensure_ascii=False + ) + assert line1 == line2 + + +def test_jsonl_record_inference_id_mismatch_rejected() -> None: + """A tampered inference_id in the persisted record raises ValueError.""" + p = _build() + record = to_jsonl_record(p) + record["inference_id"] = "0" * 64 # tamper + + with pytest.raises(ValueError, match="inference_id mismatch"): + from_jsonl_record(record) + + +def test_canonical_bytes_excludes_inference_id() -> None: + """canonical_bytes (the content-hash function) omits inference_id and + uses hashes for evidence_pointers — not full dicts.""" + p = _build() + raw = canonical_bytes(p) + decoded = json.loads(raw.decode("utf-8")) + assert "inference_id" not in decoded + assert all(isinstance(ptr, str) for ptr in decoded["evidence_pointers"])