feat(ADR-0167/W2-A): audit-to-evidence adapter (#352)

Wave 2, parallel with W2-B/C/D. Implements the type-A→type-B converter
from AuditRow to MathReaderRefusalEvidence per ADR-0167 W2-A brief.

Deliverables:
- teaching/math_contemplation.py:
  - audit_to_evidence(audit_rows): pure deterministic adapter, uses
    SUB_TYPE_FOR_OPERATOR for subtype assignment, skips rows where
    missing_operator is None, leaves claim_signature="" (W2-B will fill)
  - audit_problem_to_evidence(problem_text, case_id): convenience wrapper
    that runs the reader and adapts the output
- tests/test_math_contemplation_adapter.py: 8 tests covering
  determinism, input-order preservation, sub-type mapping
  exhaustiveness, distinct hashes across cases, empty input handling,
  None-operator skip, and round-trip from problem text

Invariants:
- Deterministic across reruns (verified by determinism rerun)
- No I/O in adapter path
- Input order preserved (no internal sort)
- claim_signature == "" for all W2-A records (W2-B coordination)

Validation:
- tests/test_math_contemplation_adapter.py: 8 passed
- tests/test_math_evidence_schema.py: 11 passed (W1-A regression)
- tests/test_brief_11b_audit_artifact.py + step2_lexicon + brief_11_audit:
  45 passed (regression)
- Determinism rerun: identical results
This commit is contained in:
Shay 2026-05-27 06:44:46 -07:00 committed by GitHub
parent 05aaff224e
commit 9da61b96a0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 195 additions and 0 deletions

View file

@ -0,0 +1,47 @@
"""ADR-0167 W2-A — audit-to-evidence adapter.
Pure deterministic conversion from comprehension audit rows into typed
``MathReaderRefusalEvidence`` records for the teaching corridor.
"""
from __future__ import annotations
from typing import Iterable
from generate.comprehension.audit import AuditRow, audit_problem
from teaching.math_evidence import (
MathReaderRefusalEvidence,
SUB_TYPE_FOR_OPERATOR,
from_audit_row,
)
def audit_to_evidence(
audit_rows: Iterable[AuditRow],
) -> tuple[MathReaderRefusalEvidence, ...]:
"""Convert audit rows into typed teaching-corridor evidence records.
Pure function. Deterministic. No filesystem, no network, no mutation.
Sub-type assignment from ``teaching.math_evidence.SUB_TYPE_FOR_OPERATOR``.
Skips rows with ``missing_operator=None`` (no sub_type no candidate).
"""
out: list[MathReaderRefusalEvidence] = []
for row in audit_rows:
if row.missing_operator is None:
continue
sub_type = SUB_TYPE_FOR_OPERATOR[row.missing_operator]
out.append(from_audit_row(row, sub_type, claim_signature=""))
return tuple(out)
def audit_problem_to_evidence(
problem_text: str,
*,
case_id: str,
) -> tuple[MathReaderRefusalEvidence, ...]:
"""Run reader audit on ``problem_text`` and return mapped evidence rows."""
_result, rows = audit_problem(problem_text, case_id=case_id)
return audit_to_evidence(rows)
__all__ = ["audit_problem_to_evidence", "audit_to_evidence"]

View file

@ -0,0 +1,148 @@
"""Tests for ADR-0167 W2-A — audit-to-evidence adapter."""
from __future__ import annotations
import json
from pathlib import Path
from generate.comprehension.audit import AuditRow
from teaching.math_contemplation import audit_problem_to_evidence, audit_to_evidence
from teaching.math_evidence import SUB_TYPE_FOR_OPERATOR
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"]),
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(
Path("evals/gsm8k_math/train_sample/v1/audit_brief_11.json").read_text()
)
return [_row_from_artifact_case(case) for case in artifact["per_case"]]
def test_adapter_round_trips_full_audit_artifact() -> None:
rows = _load_artifact_rows()
out = audit_to_evidence(rows)
expected = sum(1 for row in rows if row.missing_operator is not None)
assert len(out) == expected
def test_adapter_is_deterministic() -> None:
rows = _load_artifact_rows()
baseline = audit_to_evidence(rows)
for _ in range(10):
assert audit_to_evidence(rows) == baseline
def test_skips_none_operator_rows() -> None:
rows = [
AuditRow(
case_id="a",
sentence_index=0,
token_index=0,
token_text="?",
recognized_terms=(),
skipped_frame=None,
missing_operator=None,
refusal_reason="no_question_target",
refusal_detail="none",
)
]
assert audit_to_evidence(rows) == ()
def test_every_missing_operator_maps_to_a_sub_type() -> None:
rows = _load_artifact_rows()
for row in rows:
if row.missing_operator is not None:
assert row.missing_operator in SUB_TYPE_FOR_OPERATOR
assert len(audit_to_evidence(rows)) > 0
def test_evidence_hashes_distinct_across_cases() -> None:
row1 = AuditRow(
case_id="case-1",
sentence_index=0,
token_index=0,
token_text="crayons",
recognized_terms=("She", "has"),
skipped_frame="descriptive_frame",
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail="no primitive or lexicon match for 'crayons'",
)
row2 = AuditRow(
case_id="case-2",
sentence_index=0,
token_index=0,
token_text="crayons",
recognized_terms=("She", "has"),
skipped_frame="descriptive_frame",
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail="no primitive or lexicon match for 'crayons'",
)
out1 = audit_to_evidence([row1])[0]
out2 = audit_to_evidence([row2])[0]
assert out1.evidence_hash != out2.evidence_hash
def test_empty_audit_returns_empty_tuple() -> None:
assert audit_to_evidence([]) == ()
def test_adapter_preserves_input_order() -> None:
rows = [
AuditRow(
case_id="x-1",
sentence_index=0,
token_index=1,
token_text="a",
recognized_terms=("one",),
skipped_frame="initial_state",
missing_operator="lexicon_entry",
refusal_reason="unknown_word",
refusal_detail="a",
),
AuditRow(
case_id="x-2",
sentence_index=1,
token_index=2,
token_text="b",
recognized_terms=("two",),
skipped_frame="operation_frame",
missing_operator="question_target_slot",
refusal_reason="no_question_target",
refusal_detail="b",
),
]
out = audit_to_evidence(rows)
assert [ev.case_id for ev in out] == ["x-1", "x-2"]
def test_audit_problem_to_evidence_round_trip() -> None:
out = audit_problem_to_evidence(
"Tom opens an amusement park. How many days?",
case_id="probe",
)
assert len(out) >= 1
assert all(ev.case_id == "probe" for ev in out)