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

feat(ADR-0167/W3-A): e2e determinism + cognition regression — LexicalClaim slice closed (#357)

Browse source 
Wave 3, closes the LexicalClaim slice of ADR-0167.  After this PR the
math reader's refusal taxonomy is evidence, not terminus: lexical
refusals flow through audit row → typed evidence → dedup signature →
HITL ratification (W2-D) → pack write → next-audit-pass-resolves.

Deliverables
------------
- tests/test_math_evidence_e2e.py (new, 7 tests):
  * test_full_pipeline_from_audit_to_evidence
  * test_e2e_replay_equivalence
  * test_lexical_ratification_advances_unknown_word_row (case 0040 'sees')
  * test_e2e_determinism_across_processes
  * test_cognition_teaching_corridor_unaffected
  * test_evidence_dedup_via_claim_signature
  * test_audit_artifact_round_trip_with_signatures
- evals/gsm8k_math/train_sample/v1/audit_brief_11.md: Post-W2 baseline
  table + cognition regression line + case 0050 hazard status + pointer
  to the new e2e regression module.
- tests/test_candidate_domain_partition.py: minimal allowlist patch to
  test_existing_cognition_tests_untouched so that future ADR-0167 PRs
  can add their own evidence test files without tripping a structurally
  brittle hard-coded whitelist (W2-C partition risk; recorded in PR body).

Hard constraints held
---------------------
- wrong == 0: case 0050 hazard still refuses at sentence_index 0
  after the tmpdir-pack 'sees' ratification; no admission introduced.
- Cognition regression: zero modifications to cognition test bodies;
  only the W2-C whitelist assertion was loosened.
- Determinism: in-process and cross-process evidence_hash byte-identical.
- No real-pack mutation: a per-test digest fixture asserts
  language_packs/data/en_core_math_v1/ is byte-identical before and
  after each test.

Out of scope
------------
- Frame/Composition/Reference/Slot ratification handlers (follow-up ADRs).
- Workbench v1 wiring of math candidates (ADR-0167 §Q4).
- Auto-ratification — HITL only, forever.
- The two partition risks Gemini flagged in W2-C (cognition pack indexing,
  replay-gate default) remain follow-up.

With this PR merged the engine can ratify math-domain lexical claims
from its own refusal evidence through the existing HITL teaching
corridor — the thesis claim of ADR-0167 becomes a concrete green test.
This commit is contained in:
Shay 2026-05-27 07:27:24 -07:00 committed by GitHub
parent 94fca8ea5d
commit cc6f13a939
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 578 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
evals/gsm8k_math/train_sample/v1/audit_brief_11.md
View file

@ -147,3 +147,60 @@ with open('evals/gsm8k_math/train_sample/v1/cases.jsonl') as f:
```
The artifact `audit_brief_11.json` is pinned by `tests/test_brief_11b_audit_artifact.py`.
## Post-W2 baseline (ADR-0167 LexicalClaim-first)
Measured by running the comprehension audit over the 50 cases on the W3-A
branch (`feat/adr-0167-w3a-e2e-determinism`, base = `origin/main` after W1-A,
W2-A, W2-B, W2-C, and W2-D merged). No pack mutation occurred during the
measurement — counts reflect the same real `en_core_math_v1` pack the test
suite runs against.
| dimension | count |
| --- | --- |
| admitted | 0 |
| refused | 50 |
Refusal reasons:
| reason | count |
| --- | --- |
| incomplete_operation | 20 |
| unexpected_category | 17 |
| unknown_word | 5 |
| unattached_quantity | 4 |
| unresolved_pronoun | 3 |
| no_question_target | 1 |
Missing operators (first refusal per case):
| missing_operator | count |
| --- | --- |
| quantity_extraction | 11 |
| pre_frame_filler_sentence | 9 |
| multi_quantity_composition | 8 |
| fraction_percentage_literal | 4 |
| unit_binding | 4 |
| lexicon_entry | 3 |
| pronoun_resolution | 3 |
| descriptive_frame_question | 2 |
| multi_subject_sentence | 2 |
| compound_numeric_literal | 1 |
| compound_time_literal | 1 |
| question_frame_slot | 1 |
| question_target_slot | 1 |
Adapter, signature, ratification, and e2e tests all green; cognition
regression untouched.
Case 0050 hazard verification: refused at sentence_index 0 (verified by both
`tests/test_math_lexical_ratification.py::test_hazard_case_0050_remains_refused`
and `tests/test_math_evidence_e2e.py::test_lexical_ratification_advances_unknown_word_row`
step 8).
Regression net: any future PR that touches the math reader → evidence wire
must keep `tests/test_math_evidence_e2e.py` green. It covers the full path
from `AuditRow` → `MathReaderRefusalEvidence` → `claim_signature` →
`apply_lexical_claim` → re-audit, plus a cross-process determinism check and
the cognition-domain partition guard from W2-C.

15
tests/test_candidate_domain_partition.py
View file

@ -83,10 +83,15 @@ def test_existing_cognition_tests_untouched():
check=True,
)
lines = [line.strip() for line in result.stdout.splitlines() if line.strip()]
# Allowlist of new/modified test files contributed by ADR-0167 PRs.
# Future ADR-0167 PRs append their own file here so this regression net
# stays meaningful as the wave progresses.
allowed = {
"test_candidate_domain_partition.py", # W2-C
"test_math_evidence_e2e.py", # W3-A
}
for line in lines:
path = line.split()[-1]
# We only expect test_candidate_domain_partition.py as new/modified file.
# Note: test_math_evidence_schema.py was added by W1-A and merged, so
# depending on if we are checking against HEAD, we should verify it is untouched by us.
# git status --porcelain shows changes compared to HEAD.
assert Path(path).name == "test_candidate_domain_partition.py"
assert Path(path).name in allowed, (
f"unexpected new/modified test file: {path}"
)

511
tests/test_math_evidence_e2e.py Normal file
View file

@ -0,0 +1,511 @@
"""ADR-0167 W3-A — End-to-end determinism + cognition regression.
Closes the LexicalClaim-first slice: refusal → adapter → signature →
ratification → re-audit → row movement.
Pure tests. Every ratification uses a tmpdir pack copy; the real
``language_packs/data/en_core_math_v1/`` is byte-identical before and
after the suite runs.
"""
from __future__ import annotations
import functools
import hashlib
import json
import shutil
import subprocess
import sys
from dataclasses import asdict
from pathlib import Path
import pytest
from core.protocol import canonical_bytes
from generate.comprehension import lexicon as comprehension_lexicon
from generate.comprehension import lifecycle
from generate.comprehension.audit import AuditRow, audit_problem
from generate.comprehension.state import ReaderRefusal
from teaching.discovery import DiscoveryCandidate
from teaching.math_claim_signature import lexical_claim_signature
from teaching.math_contemplation import audit_to_evidence
from teaching.math_evidence import (
SUB_TYPE_FOR_OPERATOR,
MathReaderRefusalEvidence,
from_audit_row,
)
from teaching.math_lexical_ratification import apply_lexical_claim
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
REPO_ROOT = Path(__file__).resolve().parents[1]
AUDIT_ARTIFACT_PATH = (
REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "audit_brief_11.json"
)
CASES_PATH = (
REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "cases.jsonl"
)
PACK_ROOT = REPO_ROOT / "language_packs" / "data" / "en_core_math_v1"
PACK_HAZARD_CASE_ID = "gsm8k-train-sample-v1-0050"
RATIFICATION_TARGET_CASE_ID = "gsm8k-train-sample-v1-0040"
RATIFICATION_TARGET_SURFACE = "sees"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _row_from_artifact_case(case: dict[str, object]) -> AuditRow:
return AuditRow(
case_id=str(case["case_id"]),
sentence_index=int(case["sentence_index"]),
token_index=int(case["token_index"]),
token_text=str(case["token_text"]),
recognized_terms=tuple(str(t) for t in case["recognized_terms"]), # type: ignore[arg-type]
skipped_frame=(
None if case["skipped_frame"] is None else str(case["skipped_frame"])
),
missing_operator=(
None
if case["missing_operator"] is None
else str(case["missing_operator"])
),
refusal_reason=str(case["refusal_reason"]),
refusal_detail=str(case["refusal_detail"]),
)
def _load_artifact_rows() -> list[AuditRow]:
artifact = json.loads(AUDIT_ARTIFACT_PATH.read_text(encoding="utf-8"))
return [_row_from_artifact_case(case) for case in artifact["per_case"]]
def _load_cases() -> list[dict]:
return [
json.loads(line)
for line in CASES_PATH.read_text(encoding="utf-8").splitlines()
if line.strip()
]
def _real_pack_digest() -> str:
"""Recursive sha256 of all bytes in the real math pack (sorted paths)."""
digest = hashlib.sha256()
for path in sorted(PACK_ROOT.rglob("*")):
if path.is_file():
digest.update(path.relative_to(PACK_ROOT).as_posix().encode("utf-8"))
digest.update(b"\x00")
digest.update(path.read_bytes())
digest.update(b"\x00")
return digest.hexdigest()
@pytest.fixture()
def pack_copy(tmp_path: Path):
"""Copy the real math pack to tmpdir; clear lexicon caches around use."""
target = tmp_path / "en_core_math_v1"
shutil.copytree(PACK_ROOT, target)
comprehension_lexicon._CACHE.clear()
lifecycle._get_lexicon.cache_clear()
yield target
comprehension_lexicon._CACHE.clear()
lifecycle._get_lexicon.cache_clear()
@pytest.fixture(autouse=True)
def _real_pack_untouched():
"""Hard assertion: the real math pack must be byte-identical after each test."""
before = _real_pack_digest()
yield
after = _real_pack_digest()
assert after == before, "real en_core_math_v1 pack was mutated by a test"
def _use_pack_for_reader(monkeypatch: pytest.MonkeyPatch, pack_root: Path) -> None:
comprehension_lexicon._CACHE.clear()
@functools.cache
def _tmp_lexicon():
return comprehension_lexicon.load_lexicon(pack_root)
monkeypatch.setattr(lifecycle, "_get_lexicon", _tmp_lexicon)
# ---------------------------------------------------------------------------
# Test 1 — full pipeline from audit to evidence
# ---------------------------------------------------------------------------
def test_full_pipeline_from_audit_to_evidence() -> None:
rows = _load_artifact_rows()
evidence = audit_to_evidence(rows)
expected = sum(
1
for row in rows
if row.missing_operator is not None
and row.missing_operator in SUB_TYPE_FOR_OPERATOR
)
assert len(evidence) == expected
for record in evidence:
assert isinstance(record, MathReaderRefusalEvidence)
assert len(record.evidence_hash) == 64
assert all(c in "0123456789abcdef" for c in record.evidence_hash)
if record.sub_type == "lexical":
assert len(record.claim_signature) == 64
assert all(c in "0123456789abcdef" for c in record.claim_signature)
else:
assert record.claim_signature == ""
# audit_row round-trips intact
rebuilt = from_audit_row(
record.audit_row,
record.sub_type,
claim_signature=record.claim_signature,
)
assert rebuilt.evidence_hash == record.evidence_hash
# ---------------------------------------------------------------------------
# Test 2 — replay equivalence (in-process)
# ---------------------------------------------------------------------------
def test_e2e_replay_equivalence() -> None:
rows = _load_artifact_rows()
first = audit_to_evidence(rows)
second = audit_to_evidence(rows)
assert len(first) == len(second)
for a, b in zip(first, second):
assert a == b
assert a.evidence_hash == b.evidence_hash
assert a.to_canonical_bytes() == b.to_canonical_bytes()
# ---------------------------------------------------------------------------
# Test 3 — load-bearing integration test: lexical ratification advances a row
# ---------------------------------------------------------------------------
def test_lexical_ratification_advances_unknown_word_row(
pack_copy: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The ADR-0167 thesis-claim test: engine teaches itself in math domain.
Steps (per W3-A brief):
1. Run audit on case 0040 with tmpdir pack → refusal at 'sees'
2. Wrap refusal in MathReaderRefusalEvidence via audit_to_evidence
3. apply_lexical_claim with category='drain_token', tmpdir pack
4. Re-audit case 0040 with tmpdir pack → refusal still happens
5. Assert previously-blocking token resolves (no unknown_word at 'sees')
6. Assert refusal moved away from 'lexicon_entry'
7. Assert wrong == 0 holds (no admission)
8. Assert case 0050 hazard still pinned at sentence_index 0
"""
cases = _load_cases()
case_0040 = next(
c for c in cases if c["case_id"] == RATIFICATION_TARGET_CASE_ID
)
case_0050 = next(c for c in cases if c["case_id"] == PACK_HAZARD_CASE_ID)
# Step 1: initial audit produces a lexicon_entry refusal at 'sees'.
_use_pack_for_reader(monkeypatch, pack_copy)
result_before, rows_before = audit_problem(
case_0040["question"], case_id=case_0040["case_id"]
)
assert isinstance(result_before, ReaderRefusal)
assert rows_before, "expected at least one audit row from case 0040"
first_refusal = rows_before[0]
assert first_refusal.missing_operator == "lexicon_entry"
assert first_refusal.token_text == RATIFICATION_TARGET_SURFACE
# Step 2: wrap as evidence.
evidence = audit_to_evidence([first_refusal])
assert len(evidence) == 1
claim = evidence[0]
assert claim.sub_type == "lexical"
assert claim.claim_signature # non-empty for lexical sub_type
# Step 3: ratify into tmpdir drain_token.jsonl.
receipt = apply_lexical_claim(
claim=claim,
category="drain_token",
reviewer="w3a_test",
pack_root=pack_copy,
)
assert receipt.lemma == RATIFICATION_TARGET_SURFACE
assert receipt.category == "drain_token"
assert receipt.file_sha256_before != receipt.file_sha256_after
# Step 4 + 5 + 6: re-audit; previously-blocking token resolves; refusal moved.
comprehension_lexicon._CACHE.clear()
lifecycle._get_lexicon.cache_clear()
_use_pack_for_reader(monkeypatch, pack_copy)
result_after, rows_after = audit_problem(
case_0040["question"], case_id=case_0040["case_id"]
)
# Lexicon resolves 'sees' as drain_token in the tmpdir pack.
lex = comprehension_lexicon.load_lexicon(pack_copy)
resolved = comprehension_lexicon.lookup(lex, RATIFICATION_TARGET_SURFACE)
assert resolved is not None
assert resolved.category == "drain_token"
# Refusal still happens (case 0040 has many barriers) but not on 'sees'
# and not as a lexicon_entry miss on that token.
assert isinstance(result_after, ReaderRefusal)
if rows_after:
new_refusal = rows_after[0]
# Either we advanced past sentence 0, OR we hit a different token,
# OR a different missing_operator class entirely.
progressed = (
new_refusal.token_text != RATIFICATION_TARGET_SURFACE
or new_refusal.missing_operator != "lexicon_entry"
or new_refusal.sentence_index != first_refusal.sentence_index
)
assert progressed, (
"ratification of 'sees' did not move the refusal class for case 0040: "
f"before={first_refusal.missing_operator!r}@s{first_refusal.sentence_index}"
f"/{first_refusal.token_text!r}, "
f"after={new_refusal.missing_operator!r}@s{new_refusal.sentence_index}"
f"/{new_refusal.token_text!r}"
)
# Step 7: wrong == 0 — case 0040 still refuses (not admitted).
assert not _is_admission(result_after)
# Step 8: case 0050 hazard still pinned.
result_hazard, _hazard_rows = audit_problem(
case_0050["question"], case_id=case_0050["case_id"]
)
assert isinstance(result_hazard, ReaderRefusal)
assert result_hazard.sentence_index == 0
def _is_admission(result: object) -> bool:
"""Return True iff the reader admitted (produced a MathProblemGraph)."""
from generate.math_problem_graph import MathProblemGraph
return isinstance(result, MathProblemGraph)
# ---------------------------------------------------------------------------
# Test 4 — cross-process determinism
# ---------------------------------------------------------------------------
def test_e2e_determinism_across_processes() -> None:
rows = _load_artifact_rows()
in_process = audit_to_evidence(rows)
assert in_process, "expected at least one evidence record"
expected_hash = in_process[0].evidence_hash
expected_case_id = in_process[0].case_id
script = (
"import json, sys\n"
"from pathlib import Path\n"
"from generate.comprehension.audit import AuditRow\n"
"from teaching.math_contemplation import audit_to_evidence\n"
f"artifact = json.loads(Path({str(AUDIT_ARTIFACT_PATH)!r}).read_text())\n"
"def to_row(c):\n"
" return AuditRow(\n"
" case_id=str(c['case_id']),\n"
" sentence_index=int(c['sentence_index']),\n"
" token_index=int(c['token_index']),\n"
" token_text=str(c['token_text']),\n"
" recognized_terms=tuple(str(t) for t in c['recognized_terms']),\n"
" skipped_frame=None if c['skipped_frame'] is None else str(c['skipped_frame']),\n"
" missing_operator=None if c['missing_operator'] is None else str(c['missing_operator']),\n"
" refusal_reason=str(c['refusal_reason']),\n"
" refusal_detail=str(c['refusal_detail']),\n"
" )\n"
"rows = [to_row(c) for c in artifact['per_case']]\n"
"ev = audit_to_evidence(rows)\n"
"print(ev[0].case_id)\n"
"print(ev[0].evidence_hash)\n"
)
proc = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
check=False,
cwd=str(REPO_ROOT),
)
assert proc.returncode == 0, f"subprocess failed: {proc.stderr}"
out_lines = [line for line in proc.stdout.splitlines() if line.strip()]
assert out_lines[0] == expected_case_id
assert out_lines[1] == expected_hash
# ---------------------------------------------------------------------------
# Test 5 — cognition teaching corridor unaffected
# ---------------------------------------------------------------------------
def _cognition_candidate() -> DiscoveryCandidate:
return DiscoveryCandidate(
candidate_id="w3a_cognition_probe",
proposed_chain={
"subject": "wisdom",
"intent": "cause",
"connective": None,
"object": None,
},
trigger="would_have_grounded",
source_turn_trace="t_w3a",
pack_consistent=True,
boundary_clean=True,
)
def test_cognition_teaching_corridor_unaffected() -> None:
cand = _cognition_candidate()
assert cand.domain == "cognition"
# Canonical-bytes invariant (per W2-C contract: domain is serialised in
# the canonical bytes; only as_dict() omits the key for cognition).
cb = canonical_bytes(cand)
assert b'"domain":"cognition"' in cb
assert b'"domain":"math"' not in cb
# Replay-equivalent across reruns.
assert canonical_bytes(_cognition_candidate()) == cb
# Round-trip through as_dict/from_dict without raising.
payload = cand.as_dict()
assert "domain" not in payload # cognition omits per W2-C
restored = DiscoveryCandidate.from_dict(payload)
assert restored.domain == "cognition"
assert canonical_bytes(restored) == cb
# Defensive: math candidate canonical bytes differ.
math_cand = DiscoveryCandidate(
candidate_id="w3a_math_probe",
proposed_chain={
"subject": "crayons",
"intent": "lexical_claim",
"connective": None,
"object": None,
},
trigger="would_have_grounded",
source_turn_trace="t_w3a_math",
pack_consistent=True,
boundary_clean=True,
domain="math",
)
assert canonical_bytes(math_cand) != cb
assert b'"domain":"math"' in canonical_bytes(math_cand)
# ---------------------------------------------------------------------------
# Test 6 — evidence dedup via claim_signature
# ---------------------------------------------------------------------------
def test_evidence_dedup_via_claim_signature() -> None:
"""Two cases refusing on the same surface collapse under signature dedup."""
shared_surface = "crayons"
refusal_detail = f"no primitive or lexicon match for '{shared_surface}'"
row_a = AuditRow(
case_id="case-A",
sentence_index=0,
token_index=4,
token_text=shared_surface,
recognized_terms=("Ava", "has", "5"),
skipped_frame=None,
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail=refusal_detail,
)
row_b = AuditRow(
case_id="case-B",
sentence_index=0,
token_index=4,
token_text=shared_surface,
recognized_terms=("Ava", "has", "5"),
skipped_frame=None,
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail=refusal_detail,
)
evidence = audit_to_evidence([row_a, row_b])
assert len(evidence) == 2
e_a, e_b = evidence
# Signatures match (dedup key)
assert e_a.claim_signature == e_b.claim_signature
assert (
e_a.claim_signature
== lexical_claim_signature(surface=shared_surface, refusal_detail=refusal_detail)
)
# Per-record hashes differ (case_id is part of canonical bytes)
assert e_a.evidence_hash != e_b.evidence_hash
# Future Workbench dedup logic — pin the contract via a minimal snippet.
deduped: dict[str, MathReaderRefusalEvidence] = {}
for record in evidence:
deduped.setdefault(record.claim_signature, record)
assert len(deduped) == 1
# ---------------------------------------------------------------------------
# Test 7 — audit-artifact round-trip with signatures
# ---------------------------------------------------------------------------
def test_audit_artifact_round_trip_with_signatures(tmp_path: Path) -> None:
rows = _load_artifact_rows()
evidence = audit_to_evidence(rows)
out_path = tmp_path / "evidence.jsonl"
with out_path.open("w", encoding="utf-8") as fh:
for record in evidence:
payload = {
"case_id": record.case_id,
"sentence_index": record.sentence_index,
"token_index": record.token_index,
"refusal_reason": record.refusal_reason,
"missing_operator": record.missing_operator,
"claim_signature": record.claim_signature,
"evidence_hash": record.evidence_hash,
"sub_type": record.sub_type,
"audit_row": asdict(record.audit_row),
}
fh.write(
json.dumps(payload, sort_keys=True, ensure_ascii=False) + "\n"
)
# Reload, rebuild evidence from the audit_row, and check byte-equality.
rebuilt: list[MathReaderRefusalEvidence] = []
with out_path.open("r", encoding="utf-8") as fh:
for line in fh:
payload = json.loads(line)
audit_row = AuditRow(
case_id=payload["audit_row"]["case_id"],
sentence_index=payload["audit_row"]["sentence_index"],
token_index=payload["audit_row"]["token_index"],
token_text=payload["audit_row"]["token_text"],
recognized_terms=tuple(payload["audit_row"]["recognized_terms"]),
skipped_frame=payload["audit_row"]["skipped_frame"],
missing_operator=payload["audit_row"]["missing_operator"],
refusal_reason=payload["audit_row"]["refusal_reason"],
refusal_detail=payload["audit_row"]["refusal_detail"],
)
rebuilt.append(
from_audit_row(
audit_row,
payload["sub_type"],
claim_signature=payload["claim_signature"],
)
)
assert len(rebuilt) == len(evidence)
for original, restored in zip(evidence, rebuilt):
assert original.evidence_hash == restored.evidence_hash
assert original.to_canonical_bytes() == restored.to_canonical_bytes()