feat: add deductive proof evidence gates

This commit is contained in:
Shay 2026-06-04 08:37:51 -07:00
parent cc885157a5
commit 3389cc68c3
15 changed files with 432 additions and 37 deletions

View file

@ -165,6 +165,9 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = {
"math": ( "math": (
"tests/test_adr_0126_train_sample_runner.py", "tests/test_adr_0126_train_sample_runner.py",
), ),
"deductive": (
"tests/test_deductive_logic_entail.py",
),
"full": ("tests/",), "full": ("tests/",),
} }

View file

@ -47,6 +47,7 @@ from generate.operators import (
multi_relation_walk, multi_relation_walk,
transitive_walk, transitive_walk,
) )
from generate.proof_chain import EntailmentTrace, evaluate_entailment_with_trace
from teaching.correction import CorrectionCandidate, extract_correction from teaching.correction import CorrectionCandidate, extract_correction
from teaching.epistemic import EpistemicStatus from teaching.epistemic import EpistemicStatus
from teaching.review import ReviewedTeachingExample, review_correction from teaching.review import ReviewedTeachingExample, review_correction
@ -122,7 +123,7 @@ class CognitiveTurnPipeline:
) -> None: # runtime: ChatRuntime (no import cycle) ) -> None: # runtime: ChatRuntime (no import cycle)
self.runtime = runtime self.runtime = runtime
self._last_node_id: str | None = None self._last_node_id: str | None = None
self.teaching_store = teaching_store or TeachingStore() self.teaching_store = teaching_store if teaching_store is not None else TeachingStore()
if recognizer is not None: if recognizer is not None:
self._recognizer = recognizer self._recognizer = recognizer
elif hasattr(runtime, "first_admitted_recognizer"): elif hasattr(runtime, "first_admitted_recognizer"):
@ -304,6 +305,8 @@ class CognitiveTurnPipeline:
): ):
compose_surface = CognitiveTurnPipeline._render_compose_surface(compose_result) compose_surface = CognitiveTurnPipeline._render_compose_surface(compose_result)
entailment_trace = self._maybe_entailment_trace(intent, triples)
resolved = resolve_surface( resolved = resolve_surface(
canonical_surface=canonical, canonical_surface=canonical,
pre_decoration_surface=pre_decoration, pre_decoration_surface=pre_decoration,
@ -393,13 +396,14 @@ class CognitiveTurnPipeline:
epistemic_status = proposal.epistemic_status.value if proposal is not None else "" epistemic_status = proposal.epistemic_status.value if proposal is not None else ""
walk_serialised = CognitiveTurnPipeline._serialize_operator(walk_result) walk_serialised = CognitiveTurnPipeline._serialize_operator(walk_result)
compose_serialised = CognitiveTurnPipeline._serialize_operator(compose_result) compose_serialised = CognitiveTurnPipeline._serialize_operator(compose_result)
# Deterministic concatenation: walk record, then compose record. entailment_serialised = CognitiveTurnPipeline._serialize_entailment_trace(
# Empty strings are dropped so single-operator turns keep their entailment_trace
# existing trace_hash byte-for-byte. )
operator_invocation = ( # Deterministic concatenation: walk, compose, then entailment. Empty
f"{walk_serialised}|{compose_serialised}" # strings are dropped so unaffected turns keep existing trace bytes.
if compose_serialised operator_invocation = "|".join(
else walk_serialised s for s in (walk_serialised, compose_serialised, entailment_serialised)
if s
) )
# ADR-0023 — admissibility trace + ratification provenance. # ADR-0023 — admissibility trace + ratification provenance.
# Comb pass 2026-05-21 — direct attribute access; the fields # Comb pass 2026-05-21 — direct attribute access; the fields
@ -690,6 +694,39 @@ class CognitiveTurnPipeline:
relation=intent.relation, relation=intent.relation,
) )
def _maybe_entailment_trace(
self,
intent,
triples: tuple[tuple[str, str, str], ...],
) -> EntailmentTrace | None:
"""Compile exact verification triples into propositional entailment.
Telemetry-only v1: the result is folded into ``operator_invocation`` and
never changes the user-facing surface. Runs only when classification
exposes a precise positive ``subject relation object`` shape.
"""
if intent.tag is not IntentTag.VERIFICATION:
return None
if intent.negated or not intent.relation or not intent.object:
return None
head = self._proof_atom(intent.subject)
tail = self._proof_atom(intent.object)
if not head or not tail:
return None
relation = intent.relation.strip().lower()
premises: list[str] = []
for h, r, t in triples:
if r.strip().lower() != relation:
continue
h_atom = self._proof_atom(h)
t_atom = self._proof_atom(t)
if h_atom and t_atom:
premises.append(f"{h_atom} -> {t_atom}")
if not premises:
return None
return evaluate_entailment_with_trace(tuple(premises), f"{head} -> {tail}")
@staticmethod @staticmethod
def _render_compose_surface(compose: FrameComposeResult) -> str: def _render_compose_surface(compose: FrameComposeResult) -> str:
"""Render a frame-transfer composition suffix without selecting authority.""" """Render a frame-transfer composition suffix without selecting authority."""
@ -721,6 +758,19 @@ class CognitiveTurnPipeline:
return "" return ""
return json.dumps(op.as_dict(), sort_keys=True, ensure_ascii=False) return json.dumps(op.as_dict(), sort_keys=True, ensure_ascii=False)
@staticmethod
def _serialize_entailment_trace(trace: EntailmentTrace | None) -> str:
if trace is None:
return ""
return f"entailment:{trace.canonical_json()}"
@staticmethod
def _proof_atom(text: str) -> str:
parts = [p for p in _SUBJECT_SPLIT_RE.split(text.lower()) if p]
if not parts:
return ""
return "atom_" + "_".join(parts)
@staticmethod @staticmethod
def _render_walk_surface(walk: WalkResult) -> str: def _render_walk_surface(walk: WalkResult) -> str:
"""Render a chain-aware walk suffix without selecting authority.""" """Render a chain-aware walk suffix without selecting authority."""

View file

@ -23,14 +23,16 @@ make stochastic.
## The numbers (2026-06-04) ## The numbers (2026-06-04)
| Set | n | correct | wrong | of which non-trivial (entailed + refuted) | | Set | n | correct | wrong | refused | of which non-trivial (entailed + refuted) |
|---|---:|---:|---:|---:| |---|---:|---:|---:|---:|---:|
| dev | 200 | 200 | **0** | 74 | | dev | 200 | 200 | **0** | **0** | 74 |
| **holdout v1** | 500 | **500** | **0** | **227** (117 entailed + 110 refuted) | | **holdout v1** | 500 | **500** | **0** | **0** | **227** (117 entailed + 110 refuted) |
| fuzz (fresh random, engine vs oracle) | 7,340 | 7,340 | **0** | 3,360 | | external mirror v1 | 16 | **16** | **0** | **0** | 12 |
| fuzz (fresh random, engine vs oracle) | 7,340 | 7,340 | **0** | **0** | 3,360 |
**100% correct with `wrong = 0`**, and the correct answers include hundreds of **100% correct with `wrong = 0` and `refused = 0` on committed in-regime
non-trivial multi-hop entailments and refutations — not just `unknown` pass-throughs. cases**, and the correct answers include hundreds of non-trivial multi-hop
entailments and refutations — not just `unknown` pass-throughs.
## Why the gold is trustworthy (the GSM8K lesson applied) ## Why the gold is trustworthy (the GSM8K lesson applied)
@ -40,7 +42,9 @@ enumeration over all 2^k assignments, sharing **no code** with the ROBDD engine.
independently-coded sound procedures agreeing on **8,000** fresh random problems with independently-coded sound procedures agreeing on **8,000** fresh random problems with
**zero disagreements** is real soundness evidence — exactly the property the GSM8K **zero disagreements** is real soundness evidence — exactly the property the GSM8K
composer could never establish (it could not tell its 2 right answers from its 87 composer could never establish (it could not tell its 2 right answers from its 87
wrong ones). A single engine/oracle disagreement fails the test suite. wrong ones). A single engine/oracle disagreement fails the test suite. A refusal
on a committed in-regime case is also a lane failure; refusal-boundary cases are
tested separately.
## Honesty boundary (load-bearing — do not overclaim) ## Honesty boundary (load-bearing — do not overclaim)
@ -70,7 +74,7 @@ wrong ones). A single engine/oracle disagreement fails the test suite.
## Run ## Run
```bash ```bash
PYTHONPATH=. .venv/bin/python -m evals.deductive_logic.runner # dev + holdout, exits 1 if wrong>0 PYTHONPATH=. .venv/bin/python -m evals.deductive_logic.runner # dev + holdout + external; exits 1 unless every in-regime case is correct
PYTHONPATH=. .venv/bin/python -m evals.deductive_logic.generate # regenerate committed cases (deterministic) PYTHONPATH=. .venv/bin/python -m evals.deductive_logic.generate # regenerate committed cases (deterministic)
PYTHONPATH=. .venv/bin/python -m pytest tests/test_deductive_logic_entail.py -q PYTHONPATH=. .venv/bin/python -m pytest tests/test_deductive_logic_entail.py -q
``` ```
@ -85,4 +89,6 @@ evals/deductive_logic/
runner.py # engine vs gold; correct/wrong/refused + class breakdown runner.py # engine vs gold; correct/wrong/refused + class breakdown
dev/cases.jsonl # seed 20260604, 200 cases dev/cases.jsonl # seed 20260604, 200 cases
holdout/v1/cases.jsonl # seed 70260604, 500 cases (disjoint) holdout/v1/cases.jsonl # seed 70260604, 500 cases (disjoint)
external/v1/cases.jsonl # frozen finite-entity/proposition mirror cases
refusal/v1/cases.jsonl # refusal-boundary cases, tested separately
``` ```

View file

@ -0,0 +1,16 @@
{"gold":"entailed","id":"dl-external-v1-0001","premises":["cat_furry -> mammal_warm","mammal_warm -> needs_food","cat_furry"],"query":"needs_food"}
{"gold":"refuted","id":"dl-external-v1-0002","premises":["penguin -> ~flies","penguin"],"query":"flies"}
{"gold":"unknown","id":"dl-external-v1-0003","premises":["student -> studies"],"query":"graduates"}
{"gold":"entailed","id":"dl-external-v1-0004","premises":["red | blue","~red"],"query":"blue"}
{"gold":"entailed","id":"dl-external-v1-0005","premises":["p","q","(p & q) -> r"],"query":"r"}
{"gold":"unknown","id":"dl-external-v1-0006","premises":["p","(p & q) -> r"],"query":"r"}
{"gold":"entailed","id":"dl-external-v1-0007","premises":["a -> b","b -> c"],"query":"a -> c"}
{"gold":"refuted","id":"dl-external-v1-0008","premises":["a -> ~b","a"],"query":"a & b"}
{"gold":"entailed","id":"dl-external-v1-0009","premises":["~(a | b)"],"query":"~a & ~b"}
{"gold":"unknown","id":"dl-external-v1-0010","premises":["a | b"],"query":"a"}
{"gold":"entailed","id":"dl-external-v1-0011","premises":["rough_cat -> animal_cat","animal_cat -> living_cat","rough_cat"],"query":"living_cat"}
{"gold":"refuted","id":"dl-external-v1-0012","premises":["round_rock -> heavy_rock","heavy_rock -> ~floating_rock","round_rock"],"query":"floating_rock"}
{"gold":"entailed","id":"dl-external-v1-0013","premises":["green_frog | brown_frog","~green_frog"],"query":"brown_frog"}
{"gold":"entailed","id":"dl-external-v1-0014","premises":["teaches_lee -> helps_lee","helps_lee -> learns_lee","teaches_lee"],"query":"learns_lee"}
{"gold":"unknown","id":"dl-external-v1-0015","premises":["kind_ada -> trusted_ada"],"query":"trusted_ada"}
{"gold":"refuted","id":"dl-external-v1-0016","premises":["open_gate -> safe_room","~safe_room"],"query":"open_gate"}

View file

@ -0,0 +1,4 @@
{"gold":"refused","id":"dl-refusal-v1-0001","premises":["p","~p"],"query":"q"}
{"gold":"refused","id":"dl-refusal-v1-0002","premises":["forall x. p"],"query":"p"}
{"gold":"refused","id":"dl-refusal-v1-0003","premises":["rough(x)"],"query":"p"}
{"gold":"refused","id":"dl-refusal-v1-0004","premises":["p"],"query":"p & & q"}

View file

@ -9,14 +9,15 @@ Counts:
``unknown``). ``unknown``).
* ``wrong`` engine outcome != gold and engine did not refuse (a confabulated * ``wrong`` engine outcome != gold and engine did not refuse (a confabulated
deduction this MUST stay 0). deduction this MUST stay 0).
* ``refused`` engine returned ``refused`` (should not happen on these * ``refused`` engine returned ``refused`` on a committed in-regime case. This
well-formed, consistent, propositional cases; counted separately, never as wrong). is a capability failure for this lane, not a safety success.
A breakdown by gold class (entailed / refuted / unknown) is reported so the A breakdown by gold class (entailed / refuted / unknown) is reported so the
"sizeable numbers" are visible: how many non-trivial entailments/refutations the "sizeable numbers" are visible: how many non-trivial entailments/refutations the
engine decides correctly, not just how many ``unknown``s it passes through. engine decides correctly, not just how many ``unknown``s it passes through.
Exits non-zero if ``wrong > 0`` (the floor). Exits non-zero unless every committed in-regime case is correct. Refusal-boundary
cases live in unit tests, not in these dev/holdout formula splits.
""" """
from __future__ import annotations from __future__ import annotations
@ -41,6 +42,7 @@ def build_report(cases: list[dict]) -> dict:
by_gold: Counter[str] = Counter() by_gold: Counter[str] = Counter()
correct_by_gold: Counter[str] = Counter() correct_by_gold: Counter[str] = Counter()
wrong_examples: list[dict] = [] wrong_examples: list[dict] = []
refused_examples: list[dict] = []
for case in cases: for case in cases:
gold = case["gold"] gold = case["gold"]
@ -52,6 +54,11 @@ def build_report(cases: list[dict]) -> dict:
correct_by_gold[gold] += 1 correct_by_gold[gold] += 1
elif verdict.outcome is Entailment.REFUSED: elif verdict.outcome is Entailment.REFUSED:
counts["refused"] += 1 counts["refused"] += 1
if len(refused_examples) < 10:
refused_examples.append(
{"id": case["id"], "gold": gold, "reason": verdict.reason,
"premises": case["premises"], "query": case["query"]}
)
else: else:
counts["wrong"] += 1 counts["wrong"] += 1
if len(wrong_examples) < 10: if len(wrong_examples) < 10:
@ -60,12 +67,15 @@ def build_report(cases: list[dict]) -> dict:
"premises": case["premises"], "query": case["query"]} "premises": case["premises"], "query": case["query"]}
) )
all_cases_correct = counts["correct"] == len(cases)
return { return {
"n": len(cases), "n": len(cases),
"counts": dict(counts), "counts": dict(counts),
"by_gold": dict(by_gold), "by_gold": dict(by_gold),
"correct_by_gold": dict(correct_by_gold), "correct_by_gold": dict(correct_by_gold),
"all_cases_correct": all_cases_correct,
"wrong_examples": wrong_examples, "wrong_examples": wrong_examples,
"refused_examples": refused_examples,
} }
@ -85,6 +95,11 @@ def _run(name: str, path: Path) -> dict:
for w in report["wrong_examples"]: for w in report["wrong_examples"]:
print(f" {w['id']}: gold={w['gold']} got={w['got']} " print(f" {w['id']}: gold={w['gold']} got={w['got']} "
f"premises={w['premises']} query={w['query']}") f"premises={w['premises']} query={w['query']}")
if report["refused_examples"]:
print(" REFUSED examples:")
for r in report["refused_examples"]:
print(f" {r['id']}: gold={r['gold']} reason={r['reason']} "
f"premises={r['premises']} query={r['query']}")
return report return report
@ -92,6 +107,9 @@ if __name__ == "__main__":
reports = { reports = {
"dev": _run("dev", _ROOT / "dev" / "cases.jsonl"), "dev": _run("dev", _ROOT / "dev" / "cases.jsonl"),
"holdout-v1": _run("holdout-v1", _ROOT / "holdout" / "v1" / "cases.jsonl"), "holdout-v1": _run("holdout-v1", _ROOT / "holdout" / "v1" / "cases.jsonl"),
"external-v1": _run("external-v1", _ROOT / "external" / "v1" / "cases.jsonl"),
} }
total_wrong = sum(r["counts"]["wrong"] for r in reports.values()) total_wrong = sum(r["counts"]["wrong"] for r in reports.values())
sys.exit(1 if total_wrong > 0 else 0) total_refused = sum(r["counts"]["refused"] for r in reports.values())
all_correct = all(r["all_cases_correct"] for r in reports.values())
sys.exit(0 if total_wrong == 0 and total_refused == 0 and all_correct else 1)

View file

@ -40,4 +40,14 @@ untouched.
PYTHONPATH=. .venv/bin/python -m evals.gsm8k_math.holdout_dev.v1.runner PYTHONPATH=. .venv/bin/python -m evals.gsm8k_math.holdout_dev.v1.runner
``` ```
Exits non-zero if `wrong > 0` (the floor). Current baseline: **0 / 0 / 500**. Default execution exits non-zero only if `wrong > 0` (the safety floor). Current
baseline: **0 / 0 / 500**.
Capability-promotion runs must opt into a correctness threshold:
```bash
PYTHONPATH=. .venv/bin/python -m evals.gsm8k_math.holdout_dev.v1.runner --min-correct 1
```
That command fails today, by design. A GSM8K capability PR updates the threshold only
when held-out `correct` rises while `wrong == 0` remains true.

View file

@ -1,10 +1,14 @@
{ {
"baseline_correct": 0,
"capability_pass": false,
"counts": { "counts": {
"correct": 0, "correct": 0,
"refused": 500, "refused": 500,
"wrong": 0 "wrong": 0
}, },
"lane": "gsm8k_math/holdout_dev/v1", "lane": "gsm8k_math/holdout_dev/v1",
"min_correct": null,
"min_correct_pass": true,
"n": 500, "n": 500,
"note": "Open held-out dev metric: cases CORE was NOT built against. Iterate here; the sealed TEST (1,319) is the final arbiter. wrong=0 is the floor; correct rising is the goal (refuse-everything is the failing baseline, not a pass).", "note": "Open held-out dev metric: cases CORE was NOT built against. Iterate here; the sealed TEST (1,319) is the final arbiter. wrong=0 is the floor; correct rising is the goal (refuse-everything is the failing baseline, not a pass).",
"per_case": [ "per_case": [
@ -2009,6 +2013,7 @@
"verdict": "refused" "verdict": "refused"
} }
], ],
"safety_pass": true,
"schema_version": 1, "schema_version": 1,
"source": "openai/gsm8k train split, minus the 50 train_sample, deterministic sha256(question) sort, first 500" "source": "openai/gsm8k train split, minus the 50 train_sample, deterministic sha256(question) sort, first 500"
} }

View file

@ -23,6 +23,7 @@ Deterministic; no network (reads the committed `cases.jsonl`).
from __future__ import annotations from __future__ import annotations
import argparse
import json import json
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@ -32,6 +33,7 @@ from evals.gsm8k_math.runner import _score_one_candidate_graph
_HERE = Path(__file__).resolve().parent _HERE = Path(__file__).resolve().parent
_CASES_PATH = _HERE / "cases.jsonl" _CASES_PATH = _HERE / "cases.jsonl"
_REPORT_PATH = _HERE / "report.json" _REPORT_PATH = _HERE / "report.json"
BASELINE_CORRECT = 0
def _load_cases() -> list[dict[str, Any]]: def _load_cases() -> list[dict[str, Any]]:
@ -42,7 +44,11 @@ def _load_cases() -> list[dict[str, Any]]:
] ]
def build_report(cases: list[dict[str, Any]] | None = None) -> dict[str, Any]: def build_report(
cases: list[dict[str, Any]] | None = None,
*,
min_correct: int | None = None,
) -> dict[str, Any]:
cases = cases if cases is not None else _load_cases() cases = cases if cases is not None else _load_cases()
counts = {"correct": 0, "wrong": 0, "refused": 0} counts = {"correct": 0, "wrong": 0, "refused": 0}
per_case: list[dict[str, str]] = [] per_case: list[dict[str, str]] = []
@ -50,6 +56,9 @@ def build_report(cases: list[dict[str, Any]] | None = None) -> dict[str, Any]:
outcome = _score_one_candidate_graph(case) outcome = _score_one_candidate_graph(case)
counts[outcome.outcome] = counts.get(outcome.outcome, 0) + 1 counts[outcome.outcome] = counts.get(outcome.outcome, 0) + 1
per_case.append({"case_id": outcome.case_id, "verdict": outcome.outcome}) per_case.append({"case_id": outcome.case_id, "verdict": outcome.outcome})
safety_pass = counts["wrong"] == 0
capability_pass = counts["correct"] > BASELINE_CORRECT
min_correct_pass = True if min_correct is None else counts["correct"] >= min_correct
return { return {
"schema_version": 1, "schema_version": 1,
"lane": "gsm8k_math/holdout_dev/v1", "lane": "gsm8k_math/holdout_dev/v1",
@ -59,6 +68,11 @@ def build_report(cases: list[dict[str, Any]] | None = None) -> dict[str, Any]:
), ),
"n": len(cases), "n": len(cases),
"counts": counts, "counts": counts,
"baseline_correct": BASELINE_CORRECT,
"safety_pass": safety_pass,
"capability_pass": capability_pass,
"min_correct": min_correct,
"min_correct_pass": min_correct_pass,
"note": ( "note": (
"Open held-out dev metric: cases CORE was NOT built against. Iterate here; " "Open held-out dev metric: cases CORE was NOT built against. Iterate here; "
"the sealed TEST (1,319) is the final arbiter. wrong=0 is the floor; correct " "the sealed TEST (1,319) is the final arbiter. wrong=0 is the floor; correct "
@ -72,12 +86,31 @@ def write_report(report: dict[str, Any]) -> None:
_REPORT_PATH.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") _REPORT_PATH.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def main() -> int: def _parser() -> argparse.ArgumentParser:
report = build_report() parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--min-correct",
type=int,
default=None,
help=(
"Optional promotion gate. Default reports safety only; when set, "
"the process also fails unless correct >= MIN_CORRECT."
),
)
return parser
def main(argv: list[str] | None = None) -> int:
args = _parser().parse_args(argv)
report = build_report(min_correct=args.min_correct)
write_report(report) write_report(report)
c = report["counts"] c = report["counts"]
print(f"holdout_dev: correct={c['correct']} wrong={c['wrong']} refused={c['refused']} (n={report['n']})") print(f"holdout_dev: correct={c['correct']} wrong={c['wrong']} refused={c['refused']} (n={report['n']})")
return 0 if c["wrong"] == 0 else 1 # the floor: wrong must be 0 if not report["safety_pass"]:
return 1
if not report["min_correct_pass"]:
return 1
return 0
if __name__ == "__main__": # pragma: no cover if __name__ == "__main__": # pragma: no cover

View file

@ -33,6 +33,19 @@ from .rules import (
evaluate_modus_ponens, evaluate_modus_ponens,
evaluate_proof_conclusion, evaluate_proof_conclusion,
) )
from .entail import (
ENTAILMENT_REASONS,
INCONSISTENT_PREMISES,
OUT_OF_REGIME_OR_MALFORMED,
TAUTOLOGICAL_IMPLICATION,
TAUTOLOGICAL_REFUTATION,
UNDETERMINED,
Entailment,
EntailmentTrace,
EntailmentVerdict,
evaluate_entailment,
evaluate_entailment_with_trace,
)
__all__ = ( __all__ = (
"CONCLUSION_DISAGREEMENT", "CONCLUSION_DISAGREEMENT",
@ -48,9 +61,20 @@ __all__ = (
"ProofError", "ProofError",
"ProofGraph", "ProofGraph",
"ProofNode", "ProofNode",
"ENTAILMENT_REASONS",
"Entailment",
"EntailmentTrace",
"EntailmentVerdict",
"INCONSISTENT_PREMISES",
"OUT_OF_REGIME_OR_MALFORMED",
"TAUTOLOGICAL_IMPLICATION",
"TAUTOLOGICAL_REFUTATION",
"UNESTABLISHED_ANTECEDENT", "UNESTABLISHED_ANTECEDENT",
"UNDETERMINED",
"UNIQUE_CANONICAL_CONCLUSION", "UNIQUE_CANONICAL_CONCLUSION",
"build_proof_graph", "build_proof_graph",
"evaluate_entailment",
"evaluate_entailment_with_trace",
"evaluate_modus_ponens", "evaluate_modus_ponens",
"evaluate_proof_conclusion", "evaluate_proof_conclusion",
"proof_from_premises", "proof_from_premises",

View file

@ -30,6 +30,7 @@ propositional and in scope; an ungrounded ``forall x. P(x)`` is not.
from __future__ import annotations from __future__ import annotations
import json
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from typing import Final from typing import Final
@ -66,6 +67,33 @@ class EntailmentVerdict:
reason: str reason: str
@dataclass(frozen=True, slots=True)
class EntailmentTrace:
"""Deterministic proof evidence for a propositional entailment decision."""
outcome: Entailment
reason: str
premise_keys: tuple[str, ...]
conjunction_key: str | None
query_key: str | None
entailment_check_key: str | None
refutation_check_key: str | None
def as_dict(self) -> dict[str, object]:
return {
"outcome": self.outcome.value,
"reason": self.reason,
"premise_keys": self.premise_keys,
"conjunction_key": self.conjunction_key,
"query_key": self.query_key,
"entailment_check_key": self.entailment_check_key,
"refutation_check_key": self.refutation_check_key,
}
def canonical_json(self) -> str:
return json.dumps(self.as_dict(), sort_keys=True, separators=(",", ":"))
def _conjoin(premises: tuple[str, ...]) -> str: def _conjoin(premises: tuple[str, ...]) -> str:
"""Fully-parenthesized conjunction of the premise formulas (``true`` if empty). """Fully-parenthesized conjunction of the premise formulas (``true`` if empty).
@ -82,27 +110,83 @@ def evaluate_entailment(premises: tuple[str, ...], query: str) -> EntailmentVerd
Sound and complete over the propositional regime; refusal-first on anything Sound and complete over the propositional regime; refusal-first on anything
outside it. Never raises on a logic-domain error every ``LogicError`` (and outside it. Never raises on a logic-domain error every ``LogicError`` (and
its regime/budget subclasses) maps to a typed ``REFUSED`` verdict.""" its regime/budget subclasses) maps to a typed ``REFUSED`` verdict."""
trace = evaluate_entailment_with_trace(premises, query)
return EntailmentVerdict(trace.outcome, trace.reason)
def evaluate_entailment_with_trace(
premises: tuple[str, ...],
query: str,
) -> EntailmentTrace:
"""Decide entailment and return deterministic proof evidence.
This is the evidence-bearing API; :func:`evaluate_entailment` intentionally
remains the stable verdict-only wrapper for existing callers."""
premise_keys_list: list[str] = []
conjunction_key: str | None = None
query_key: str | None = None
entailment_check_key: str | None = None
refutation_check_key: str | None = None
try: try:
for premise in premises:
premise_keys_list.append(canonicalize(premise).canonical_key)
conj = _conjoin(premises) conj = _conjoin(premises)
conj_canon = canonicalize(conj) conj_canon = canonicalize(conj)
conjunction_key = conj_canon.canonical_key
if conj_canon.is_contradiction: if conj_canon.is_contradiction:
# No model satisfies the premises: from a contradiction everything # No model satisfies the premises: from a contradiction everything
# follows. We decline rather than assert a vacuous entailment. # follows. We decline rather than assert a vacuous entailment.
return EntailmentVerdict(Entailment.REFUSED, INCONSISTENT_PREMISES) return EntailmentTrace(
outcome=Entailment.REFUSED,
reason=INCONSISTENT_PREMISES,
premise_keys=tuple(premise_keys_list),
conjunction_key=conjunction_key,
query_key=None,
entailment_check_key=None,
refutation_check_key=None,
)
# Force the query through the canonicalizer too, so a malformed / out-of- # Force the query through the canonicalizer too, so a malformed / out-of-
# regime query refuses even when the implication check would shortcut. # regime query refuses even when the implication check would shortcut.
canonicalize(query) query_canon = canonicalize(query)
entailed = canonicalize(f"({conj}) -> ({query})").is_tautology query_key = query_canon.canonical_key
refuted = canonicalize(f"({conj}) -> (~({query}))").is_tautology entail_canon = canonicalize(f"({conj}) -> ({query})")
entailment_check_key = entail_canon.canonical_key
refute_canon = canonicalize(f"({conj}) -> (~({query}))")
refutation_check_key = refute_canon.canonical_key
entailed = entail_canon.is_tautology
refuted = refute_canon.is_tautology
except LogicError: except LogicError:
return EntailmentVerdict(Entailment.REFUSED, OUT_OF_REGIME_OR_MALFORMED) return EntailmentTrace(
outcome=Entailment.REFUSED,
reason=OUT_OF_REGIME_OR_MALFORMED,
premise_keys=tuple(premise_keys_list),
conjunction_key=conjunction_key,
query_key=query_key,
entailment_check_key=entailment_check_key,
refutation_check_key=refutation_check_key,
)
if entailed and refuted: if entailed and refuted:
# Only possible if the premises are inconsistent, already handled above; # Only possible if the premises are inconsistent, already handled above;
# defensive — never assert a contradiction-derived answer. # defensive — never assert a contradiction-derived answer.
return EntailmentVerdict(Entailment.REFUSED, INCONSISTENT_PREMISES) outcome = Entailment.REFUSED
if entailed: reason = INCONSISTENT_PREMISES
return EntailmentVerdict(Entailment.ENTAILED, TAUTOLOGICAL_IMPLICATION) elif entailed:
if refuted: outcome = Entailment.ENTAILED
return EntailmentVerdict(Entailment.REFUTED, TAUTOLOGICAL_REFUTATION) reason = TAUTOLOGICAL_IMPLICATION
return EntailmentVerdict(Entailment.UNKNOWN, UNDETERMINED) elif refuted:
outcome = Entailment.REFUTED
reason = TAUTOLOGICAL_REFUTATION
else:
outcome = Entailment.UNKNOWN
reason = UNDETERMINED
return EntailmentTrace(
outcome=outcome,
reason=reason,
premise_keys=tuple(premise_keys_list),
conjunction_key=conjunction_key,
query_key=query_key,
entailment_check_key=entailment_check_key,
refutation_check_key=refutation_check_key,
)

View file

@ -17,6 +17,7 @@ def test_core_test_lists_curated_suites(capsys) -> None:
assert "cognition" in captured.out.splitlines() assert "cognition" in captured.out.splitlines()
assert "teaching" in captured.out.splitlines() assert "teaching" in captured.out.splitlines()
assert "packs" in captured.out.splitlines() assert "packs" in captured.out.splitlines()
assert "deductive" in captured.out.splitlines()
assert "full" in captured.out.splitlines() assert "full" in captured.out.splitlines()
@ -61,6 +62,27 @@ def test_core_test_fast_suite_expands_to_iteration_lane(monkeypatch) -> None:
assert "tests/" not in command assert "tests/" not in command
def test_core_test_deductive_suite_expands_to_entailment_lane(monkeypatch) -> None:
calls: list[tuple[str, ...]] = []
def fake_run(*args: str, check: bool = False, cwd=None) -> int:
calls.append(args)
return 0
monkeypatch.setattr(cli, "_run", fake_run)
rc = cli.main(["test", "--suite", "deductive", "-q"])
assert rc == 0
assert calls[0] == (
cli.sys.executable,
"-m",
"pytest",
"tests/test_deductive_logic_entail.py",
"-q",
)
def test_core_test_suite_accepts_pytest_flags_without_separator(monkeypatch) -> None: def test_core_test_suite_accepts_pytest_flags_without_separator(monkeypatch) -> None:
calls: list[tuple[str, ...]] = [] calls: list[tuple[str, ...]] = []

View file

@ -15,6 +15,8 @@ from core.cognition import CognitiveTurnPipeline, CognitiveTurnResult
from core.cognition.trace import trace_hash_from_result from core.cognition.trace import trace_hash_from_result
from generate.intent import IntentTag from generate.intent import IntentTag
from generate.graph_planner import RhetoricalMove from generate.graph_planner import RhetoricalMove
from teaching.source import ProposalSource
from teaching.store import PackMutationProposal, TeachingStore
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -240,3 +242,36 @@ def test_pipeline_chat_response_contract_unchanged(pipeline: CognitiveTurnPipeli
assert isinstance(result.vault_hits, int) assert isinstance(result.vault_hits, int)
assert result.proposition is not None assert result.proposition is not None
assert result.articulation is not None assert result.articulation is not None
def test_verification_turn_records_entailment_operator_telemetry() -> None:
store = TeachingStore()
store._proposals.extend([ # noqa: SLF001 - focused operator substrate seed
PackMutationProposal(
proposal_id="p1",
candidate_id="c1",
subject="wisdom",
correction_text="wisdom precedes knowledge",
prior_surface="",
source=ProposalSource.operator(emitted_at_revision="test"),
triple=("wisdom", "precedes", "knowledge"),
),
PackMutationProposal(
proposal_id="p2",
candidate_id="c2",
subject="knowledge",
correction_text="knowledge precedes recall",
prior_surface="",
source=ProposalSource.operator(emitted_at_revision="test"),
triple=("knowledge", "precedes", "recall"),
),
])
pipeline = CognitiveTurnPipeline(ChatRuntime(), teaching_store=store)
result = pipeline.run("wisdom precedes recall.", max_tokens=6)
assert result.intent is not None
assert result.intent.tag is IntentTag.VERIFICATION
assert "entailment:" in result.operator_invocation
assert '"outcome":"entailed"' in result.operator_invocation
assert result.trace_hash == trace_hash_from_result(result)

View file

@ -22,11 +22,14 @@ import pytest
from evals.deductive_logic.generate import _make_case from evals.deductive_logic.generate import _make_case
from evals.deductive_logic.oracle import oracle_entailment from evals.deductive_logic.oracle import oracle_entailment
from evals.deductive_logic.runner import _ROOT, _load, build_report from evals.deductive_logic.runner import _ROOT, _load, build_report
from generate.logic_canonical import LogicBudgetError
from generate.proof_chain.entail import ( from generate.proof_chain.entail import (
INCONSISTENT_PREMISES, INCONSISTENT_PREMISES,
OUT_OF_REGIME_OR_MALFORMED, OUT_OF_REGIME_OR_MALFORMED,
TAUTOLOGICAL_IMPLICATION,
Entailment, Entailment,
evaluate_entailment, evaluate_entailment,
evaluate_entailment_with_trace,
) )
@ -81,6 +84,44 @@ def test_malformed_query_refuses() -> None:
assert evaluate_entailment(("P",), "P & & Q").outcome is Entailment.REFUSED assert evaluate_entailment(("P",), "P & & Q").outcome is Entailment.REFUSED
def test_budget_error_refuses_typed(monkeypatch) -> None:
def boom(_formula: str):
raise LogicBudgetError("too many nodes")
monkeypatch.setattr("generate.proof_chain.entail.canonicalize", boom)
v = evaluate_entailment(("P",), "Q")
assert v.outcome is Entailment.REFUSED
assert v.reason == OUT_OF_REGIME_OR_MALFORMED
def test_entailment_trace_is_deterministic_evidence() -> None:
t1 = evaluate_entailment_with_trace(("P", "P -> Q"), "Q")
t2 = evaluate_entailment_with_trace(("P", "P -> Q"), "Q")
assert t1.outcome is Entailment.ENTAILED
assert t1.reason == TAUTOLOGICAL_IMPLICATION
assert t1.premise_keys
assert t1.conjunction_key is not None
assert t1.query_key is not None
assert t1.entailment_check_key == "T"
assert t1.refutation_check_key is not None
assert t1.canonical_json() == t2.canonical_json()
assert "premise_keys" in t1.canonical_json()
def test_refused_trace_preserves_available_canonical_evidence() -> None:
trace = evaluate_entailment_with_trace(("P",), "P & & Q")
assert trace.outcome is Entailment.REFUSED
assert trace.reason == OUT_OF_REGIME_OR_MALFORMED
assert trace.premise_keys
assert trace.conjunction_key is not None
assert trace.query_key is None
assert trace.entailment_check_key is None
assert trace.refutation_check_key is None
# --- 3a. wrong=0 vs the independent oracle (deterministic fuzz) ------------- # --- 3a. wrong=0 vs the independent oracle (deterministic fuzz) -------------
def test_engine_matches_independent_oracle_fuzz() -> None: def test_engine_matches_independent_oracle_fuzz() -> None:
@ -128,6 +169,8 @@ def test_holdout_lane_wrong_is_zero() -> None:
report = build_report(_load(_ROOT / "holdout" / "v1" / "cases.jsonl")) report = build_report(_load(_ROOT / "holdout" / "v1" / "cases.jsonl"))
assert report["n"] == 500 assert report["n"] == 500
assert report["counts"]["wrong"] == 0 assert report["counts"]["wrong"] == 0
assert report["counts"]["refused"] == 0
assert report["all_cases_correct"] is True
# the sizeable, honest signal: non-trivial deductions decided correctly # the sizeable, honest signal: non-trivial deductions decided correctly
cbg = report["correct_by_gold"] cbg = report["correct_by_gold"]
assert cbg.get("entailed", 0) >= 50 assert cbg.get("entailed", 0) >= 50
@ -138,3 +181,33 @@ def test_dev_lane_wrong_is_zero() -> None:
report = build_report(_load(_ROOT / "dev" / "cases.jsonl")) report = build_report(_load(_ROOT / "dev" / "cases.jsonl"))
assert report["n"] == 200 assert report["n"] == 200
assert report["counts"]["wrong"] == 0 assert report["counts"]["wrong"] == 0
assert report["counts"]["refused"] == 0
assert report["all_cases_correct"] is True
def test_external_mirror_lane_matches_independent_oracle() -> None:
cases = _load(_ROOT / "external" / "v1" / "cases.jsonl")
report = build_report(cases)
assert report["n"] == 16
assert report["all_cases_correct"] is True
assert report["counts"] == {"correct": 16, "wrong": 0, "refused": 0}
for case in cases:
assert oracle_entailment(tuple(case["premises"]), case["query"]) == case["gold"]
def test_refusal_boundary_split_refuses_every_case() -> None:
cases = _load(_ROOT / "refusal" / "v1" / "cases.jsonl")
assert len(cases) == 4
for case in cases:
verdict = evaluate_entailment(tuple(case["premises"]), case["query"])
assert case["gold"] == "refused"
assert verdict.outcome is Entailment.REFUSED
def test_runner_treats_committed_refusal_as_capability_failure() -> None:
report = build_report([
{"id": "bad-refusal", "premises": ["P", "~P"], "query": "Q", "gold": "entailed"}
])
assert report["counts"]["wrong"] == 0
assert report["counts"]["refused"] == 1
assert report["all_cases_correct"] is False

View file

@ -25,6 +25,7 @@ def test_wrong_is_zero_the_floor() -> None:
"""FOREVER: never confabulate on held-out data. If this ever fails, a committing """FOREVER: never confabulate on held-out data. If this ever fails, a committing
path is wrong on real GSM8K exactly the breach class that hid behind train_sample.""" path is wrong on real GSM8K exactly the breach class that hid behind train_sample."""
assert _REPORT["counts"]["wrong"] == 0 assert _REPORT["counts"]["wrong"] == 0
assert _REPORT["safety_pass"] is True
def test_current_baseline_snapshot() -> None: def test_current_baseline_snapshot() -> None:
@ -38,3 +39,14 @@ def test_current_baseline_snapshot() -> None:
f"holdout_dev moved to {_REPORT['counts']} — if a real capability change landed, " f"holdout_dev moved to {_REPORT['counts']} — if a real capability change landed, "
f"update this snapshot AND confirm wrong=0 on the sealed test before claiming lift" f"update this snapshot AND confirm wrong=0 on the sealed test before claiming lift"
) )
assert _REPORT["baseline_correct"] == 0
assert _REPORT["capability_pass"] is False
assert _REPORT["min_correct"] is None
assert _REPORT["min_correct_pass"] is True
def test_min_correct_promotion_gate_is_optional() -> None:
report = build_report(_load_cases(), min_correct=1)
assert report["safety_pass"] is True
assert report["min_correct"] == 1
assert report["min_correct_pass"] is False