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

feat(ADR-0172/W2): audit-corpus decomposer (#383)

Browse source 
Add decompose_audit(audit_path) to teaching/math_contemplation.py.
Groups audit_brief_11.json refusal rows by
(refusal_reason, missing_operator), emits one
MathReaderRefusalShapeProposal per group of >=2 rows, each carrying a
4-step ReasoningTrace (observation -> grouping -> hypothesis ->
conclusion).

Determinism:
- Group iteration sorted by (refusal_reason, missing_operator).
- Evidence per group sorted by case_id.
- Output tuple sorted by proposal_id.
- 10x rerun -> byte-identical proposals + trace_ids.

Pure read-only: audit file is not mutated, no proposals written to
disk, no chat/field/generate/algebra imports.

Tests (tests/test_adr_0172_w2_decomposer.py): real-audit emission,
determinism (10x), evidence floor, change-kind dispatch over all four
heuristic branches, four-step trace, case_id sort, proposal_id sort,
empty input -> empty tuple, unmapped operator skip, missing file ->
FileNotFoundError, no-mutation contract.

Added to core test --suite teaching.
This commit is contained in:
Shay 2026-05-27 12:39:53 -07:00 committed by GitHub
parent 87790ad60b
commit af3821f0ed
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 602 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
core/cli.py
View file

@ -65,6 +65,7 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = {
"tests/test_reviewed_teaching_loop.py",
"tests/test_pipeline_teaching_integration.py",
"tests/test_epistemic_invariants.py",
"tests/test_adr_0172_w2_decomposer.py",
),
"packs": (
"tests/test_core_semantic_seed_pack.py",

294
teaching/math_contemplation.py
View file

@ -1,13 +1,22 @@
"""ADR-0167 W2-A / W2-B — Audit-to-evidence adapter + lexical claim signature.
"""ADR-0167 W2-A / W2-B + ADR-0172 W2 — Audit-corpus decomposition.
W2-A deliverable: :func:`audit_to_evidence` and
ADR-0167 W2-A deliverable: :func:`audit_to_evidence` and
:func:`audit_problem_to_evidence` convert :class:`AuditRow` sequences into
typed :class:`MathReaderRefusalEvidence` teaching-corridor records.
W2-B deliverable: For ``sub_type == "lexical"`` evidence, ``claim_signature``
is computed via :func:`lexical_claim_signature` and the ``evidence_hash`` is
recomputed to incorporate the signature. All other sub_types leave
``claim_signature == ""``.
ADR-0167 W2-B deliverable: For ``sub_type == "lexical"`` evidence,
``claim_signature`` is computed via :func:`lexical_claim_signature` and the
``evidence_hash`` is recomputed to incorporate the signature. All other
sub_types leave ``claim_signature == ""``.
ADR-0172 W2 deliverable: :func:`decompose_audit` reads
``audit_brief_11.json``, groups refusal rows by
``(refusal_reason, missing_operator)``, and emits one
:class:`MathReaderRefusalShapeProposal` per group of ≥2 rows. Each
proposal carries a 4-step :class:`ReasoningTrace`
(observation → grouping → hypothesis → conclusion). Pure read-only:
the audit file is not mutated, no proposal is written to disk, and no
teaching-store hook fires.
Pure module. No filesystem writes, no network calls, no global mutation.
Deterministic: same inputs → byte-identical output across all reruns.
@ -15,15 +24,28 @@ Deterministic: same inputs → byte-identical output across all reruns.
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Iterable
from evals.refusal_taxonomy.shape_categories import ShapeCategory
from generate.comprehension.audit import AuditRow, audit_problem
from teaching.math_claim_signature import lexical_claim_signature
from teaching.math_contemplation_proposal import (
MathReaderRefusalShapeProposal,
build_proposal,
)
from teaching.math_evidence import (
MathReaderRefusalEvidence,
SUB_TYPE_FOR_OPERATOR,
from_audit_row,
)
from teaching.math_reasoning_trace import (
ReasoningStep,
ReasoningTrace,
build_trace,
)
def audit_to_evidence(
@ -98,7 +120,267 @@ def audit_problem_to_evidence(
return audit_to_evidence(rows)
# ---------------------------------------------------------------------------
# ADR-0172 W2 — Audit-corpus decomposer
# ---------------------------------------------------------------------------
_CHANGE_KIND_BY_REFUSAL_REASON: dict[str, str] = {
"lexicon_entry": "vocabulary_addition",
"narrowness_violation": "matcher_extension",
"frame_unrecognized": "frame_reclassification",
}
def _audit_row_from_case(case: dict) -> AuditRow:
"""Reconstruct an :class:`AuditRow` from one ``per_case`` JSON entry.
Defensive about missing fields: the audit JSON only carries the
columns it had to populate. Empty/missing fields default to the
natural zero for their declared type.
"""
return AuditRow(
case_id=str(case["case_id"]),
sentence_index=int(case.get("sentence_index", 0)),
token_index=int(case.get("token_index", 0)),
token_text=str(case.get("token_text", "")),
recognized_terms=tuple(case.get("recognized_terms", ())),
skipped_frame=case.get("skipped_frame"),
missing_operator=case.get("missing_operator"),
refusal_reason=str(case.get("refusal_reason", "")),
refusal_detail=str(case.get("refusal_detail", "")),
)
def _change_kind_for_group(refusal_reason: str) -> str:
"""Heuristic per ADR-0172 §"Six open questions" #1."""
return _CHANGE_KIND_BY_REFUSAL_REASON.get(
refusal_reason, "injector_sub_shape"
)
def _modal_anchor_payload(
*,
refusal_reason: str,
missing_operator: str,
evidence: tuple[MathReaderRefusalEvidence, ...],
) -> dict:
"""Build a deterministic, JSON-safe placeholder payload for the group."""
return {
"evidence_count": len(evidence),
"group_key": {
"missing_operator": missing_operator,
"refusal_reason": refusal_reason,
},
"modal_sub_type": evidence[0].sub_type,
}
def _build_reasoning_trace(
*,
refusal_reason: str,
missing_operator: str,
evidence: tuple[MathReaderRefusalEvidence, ...],
change_kind: str,
) -> ReasoningTrace:
"""Construct the 4-step contemplation trace for one group."""
case_ids = tuple(ev.case_id for ev in evidence)
group_payload = {
"missing_operator": missing_operator,
"refusal_reason": refusal_reason,
}
observation = ReasoningStep(
step_index=0,
step_kind="observation",
input_pointers=case_ids,
claim=(
f"{len(evidence)} refusal rows share "
f"(refusal_reason={refusal_reason!r}, "
f"missing_operator={missing_operator!r})"
),
justification=(
"Decomposer iterated audit_brief_11.json per_case rows and "
"found a group whose shared key meets the ≥2-evidence floor."
),
output_payload={
"case_ids": list(case_ids),
"evidence_count": len(evidence),
},
)
grouping = ReasoningStep(
step_index=1,
step_kind="grouping",
input_pointers=case_ids,
claim=(
"Group key encodes the shared (refusal_reason, missing_operator) "
"tuple under which these rows refused."
),
justification=(
"Per ADR-0172 §'Six open questions' #1, the naive grouping is "
"exact equality on the refusal_reason × missing_operator pair."
),
output_payload=group_payload,
)
hypothesis = ReasoningStep(
step_index=2,
step_kind="hypothesis",
input_pointers=case_ids,
claim=(
f"The structural change kind for this group is {change_kind!r}."
),
justification=(
"Dispatched via the refusal_reason heuristic: lexicon_entry → "
"vocabulary_addition; narrowness_violation → matcher_extension; "
"frame_unrecognized → frame_reclassification; else → "
"injector_sub_shape."
),
output_payload={"proposed_change_kind": change_kind},
)
conclusion = ReasoningStep(
step_index=3,
step_kind="conclusion",
input_pointers=case_ids,
claim=(
f"Propose a {change_kind!r} structural change covering "
f"{len(evidence)} evidence rows."
),
justification=(
"Evidence-only proposal; the wrong=0 surface ratification "
"handler decides whether to apply, not this decomposer."
),
output_payload={
"proposed_change_kind": change_kind,
"evidence_count": len(evidence),
},
)
return build_trace(
(observation, grouping, hypothesis, conclusion)
)
def _build_proposal_for_group(
*,
refusal_reason: str,
missing_operator: str,
evidence: tuple[MathReaderRefusalEvidence, ...],
) -> MathReaderRefusalShapeProposal:
"""Assemble one :class:`MathReaderRefusalShapeProposal` for a group."""
change_kind = _change_kind_for_group(refusal_reason)
payload = _modal_anchor_payload(
refusal_reason=refusal_reason,
missing_operator=missing_operator,
evidence=evidence,
)
trace = _build_reasoning_trace(
refusal_reason=refusal_reason,
missing_operator=missing_operator,
evidence=evidence,
change_kind=change_kind,
)
replay_seed = json.dumps(
{
"evidence_hashes": sorted(ev.evidence_hash for ev in evidence),
"missing_operator": missing_operator,
"refusal_reason": refusal_reason,
},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
replay_equivalence_hash = hashlib.sha256(replay_seed).hexdigest()
structural_commonality = (
f"{len(evidence)} refusals share "
f"refusal_reason={refusal_reason!r} ∧ "
f"missing_operator={missing_operator!r}"
)
wrong_zero_assertion = (
"Proposal is evidence-only; ratification handler is the wrong=0 "
"surface, not this proposal."
)
return build_proposal(
shape_category=ShapeCategory.UNCATEGORIZED,
structural_commonality=structural_commonality,
evidence_pointers=evidence,
proposed_change_kind=change_kind,
proposed_change_payload=payload,
wrong_zero_assertion=wrong_zero_assertion,
replay_equivalence_hash=replay_equivalence_hash,
reasoning_trace=trace,
)
def decompose_audit(
audit_path: Path,
) -> tuple[MathReaderRefusalShapeProposal, ...]:
"""Decompose an audit brief into refusal-shape proposals.
Read ``audit_path`` (expected schema: ``audit_brief_11.json``), group
``per_case`` refusal rows by ``(refusal_reason, missing_operator)``,
and emit one :class:`MathReaderRefusalShapeProposal` per group with
≥2 evidence rows. Each proposal carries a 4-step
:class:`ReasoningTrace` (observation → grouping → hypothesis →
conclusion).
Determinism contract
--------------------
- Group iteration order is sorted by ``(refusal_reason,
missing_operator)``.
- Evidence per group is sorted by ``case_id``.
- Output tuple is sorted by ``proposal_id``.
- The same input file produces a byte-identical proposal stream
across every rerun.
Trust boundary
--------------
Pure read-only. ``audit_path`` is read once; no file is written.
Decomposer is teaching-layer code only — does not import from
``chat``/``field``/``generate.stream``/``algebra``.
"""
raw = audit_path.read_text(encoding="utf-8")
data = json.loads(raw)
per_case = data.get("per_case", []) or []
rows: list[AuditRow] = []
for case in per_case:
if not isinstance(case, dict):
continue
if not case.get("case_id"):
continue
rows.append(_audit_row_from_case(case))
evidence_records = audit_to_evidence(rows)
groups: dict[tuple[str, str], list[MathReaderRefusalEvidence]] = {}
for ev in evidence_records:
if ev.missing_operator is None:
continue
key = (ev.refusal_reason, ev.missing_operator)
groups.setdefault(key, []).append(ev)
proposals: list[MathReaderRefusalShapeProposal] = []
for key in sorted(groups.keys()):
refusal_reason, missing_operator = key
group_evs = tuple(sorted(groups[key], key=lambda e: e.case_id))
if len(group_evs) < 2:
continue
proposals.append(
_build_proposal_for_group(
refusal_reason=refusal_reason,
missing_operator=missing_operator,
evidence=group_evs,
)
)
return tuple(sorted(proposals, key=lambda p: p.proposal_id))
__all__ = [
"audit_to_evidence",
"audit_problem_to_evidence",
"decompose_audit",
]

313
tests/test_adr_0172_w2_decomposer.py Normal file
View file

@ -0,0 +1,313 @@
"""ADR-0172 W2 — Audit-corpus decomposer tests.
Pins :func:`teaching.math_contemplation.decompose_audit` against:
- the real ``evals/gsm8k_math/train_sample/v1/audit_brief_11.json`` audit;
- a synthetic mini-audit covering the four change-kind branches;
- determinism (10x rerun identity);
- evidence-floor (≥2-row threshold);
- sort contracts on evidence and proposal output;
- empty-input → empty-output;
- pure read-only behavior (no filesystem mutation).
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from teaching.math_contemplation import decompose_audit
from teaching.math_contemplation_proposal import (
MathReaderRefusalShapeProposal,
)
REAL_AUDIT_PATH = (
Path(__file__).resolve().parent.parent
/ "evals"
/ "gsm8k_math"
/ "train_sample"
/ "v1"
/ "audit_brief_11.json"
)
# ---------------------------------------------------------------------------
# Helpers — synthetic audit fixtures
# ---------------------------------------------------------------------------
def _case(
*,
case_id: str,
refusal_reason: str,
missing_operator: str,
sentence_index: int = 0,
token_index: int = 0,
token_text: str = "x",
refusal_detail: str = "",
) -> dict:
return {
"case_id": case_id,
"outcome": "refused",
"refusal_reason": refusal_reason,
"missing_operator": missing_operator,
"refusal_detail": refusal_detail,
"sentence_index": sentence_index,
"token_index": token_index,
"token_text": token_text,
"recognized_terms": [],
"skipped_frame": None,
}
def _write_audit(tmp_path: Path, per_case: list[dict]) -> Path:
audit = {
"schema_version": 1,
"brief": "synthetic-test",
"case_count": len(per_case),
"per_case": per_case,
}
target = tmp_path / "audit.json"
target.write_text(json.dumps(audit), encoding="utf-8")
return target
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_decompose_audit_emits_at_least_one_proposal() -> None:
"""Real audit_brief_11.json contains ≥1 multi-row group."""
assert REAL_AUDIT_PATH.exists(), REAL_AUDIT_PATH
proposals = decompose_audit(REAL_AUDIT_PATH)
assert len(proposals) >= 1
for p in proposals:
assert isinstance(p, MathReaderRefusalShapeProposal)
assert len(p.evidence_pointers) >= 2
def test_decompose_audit_deterministic_across_reruns() -> None:
"""Same input → byte-identical proposal stream across 10 reruns."""
first = decompose_audit(REAL_AUDIT_PATH)
first_ids = tuple(p.proposal_id for p in first)
for _ in range(9):
rerun = decompose_audit(REAL_AUDIT_PATH)
assert tuple(p.proposal_id for p in rerun) == first_ids
for left, right in zip(first, rerun):
assert left == right
assert left.reasoning_trace.trace_id == right.reasoning_trace.trace_id
def test_decompose_audit_minimum_evidence_threshold(tmp_path: Path) -> None:
"""Singleton groups MUST NOT emit a proposal."""
audit = _write_audit(
tmp_path,
[
_case(
case_id="solo-0001",
refusal_reason="unknown_word",
missing_operator="lexicon_entry",
),
],
)
proposals = decompose_audit(audit)
assert proposals == ()
def test_decompose_audit_change_kind_dispatch(tmp_path: Path) -> None:
"""Each refusal_reason branch yields the expected change_kind.
The dispatch heuristic keys on ``refusal_reason``. We use the literal
strings from the brief plus an ``other`` row that should fall through
to ``injector_sub_shape``. Each branch gets 2 rows so it actually
emits a proposal.
"""
branches = [
("lexicon_entry", "lexicon_entry", "vocabulary_addition"),
("narrowness_violation", "multi_quantity_composition", "matcher_extension"),
("frame_unrecognized", "pre_frame_filler_sentence", "frame_reclassification"),
("incomplete_operation", "quantity_extraction", "injector_sub_shape"),
]
per_case: list[dict] = []
for refusal_reason, missing_operator, _kind in branches:
per_case.append(
_case(
case_id=f"{refusal_reason}-a",
refusal_reason=refusal_reason,
missing_operator=missing_operator,
)
)
per_case.append(
_case(
case_id=f"{refusal_reason}-b",
refusal_reason=refusal_reason,
missing_operator=missing_operator,
)
)
audit = _write_audit(tmp_path, per_case)
proposals = decompose_audit(audit)
by_reason = {
p.evidence_pointers[0].refusal_reason: p for p in proposals
}
for refusal_reason, _missing, expected_kind in branches:
assert refusal_reason in by_reason, refusal_reason
assert by_reason[refusal_reason].proposed_change_kind == expected_kind
def test_decompose_audit_reasoning_trace_has_four_steps(tmp_path: Path) -> None:
"""Every emitted proposal carries an exactly-4-step trace."""
audit = _write_audit(
tmp_path,
[
_case(
case_id="t-0001",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
_case(
case_id="t-0002",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
],
)
proposals = decompose_audit(audit)
assert len(proposals) == 1
trace = proposals[0].reasoning_trace
assert len(trace.steps) == 4
assert [s.step_kind for s in trace.steps] == [
"observation",
"grouping",
"hypothesis",
"conclusion",
]
def test_decompose_audit_evidence_sorted_by_case_id(tmp_path: Path) -> None:
"""Evidence ordering inside a proposal is case_id-sorted."""
audit = _write_audit(
tmp_path,
[
_case(
case_id="zzz-last",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
_case(
case_id="aaa-first",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
_case(
case_id="mmm-mid",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
],
)
proposals = decompose_audit(audit)
assert len(proposals) == 1
ids = [ev.case_id for ev in proposals[0].evidence_pointers]
assert ids == sorted(ids)
assert ids == ["aaa-first", "mmm-mid", "zzz-last"]
def test_decompose_audit_proposal_ids_sorted(tmp_path: Path) -> None:
"""Output tuple is sorted by proposal_id."""
per_case: list[dict] = []
for group in ("lexicon_entry", "narrowness_violation", "frame_unrecognized"):
per_case.append(
_case(
case_id=f"{group}-0001",
refusal_reason=group,
missing_operator="lexicon_entry"
if group == "lexicon_entry"
else (
"multi_quantity_composition"
if group == "narrowness_violation"
else "pre_frame_filler_sentence"
),
)
)
per_case.append(
_case(
case_id=f"{group}-0002",
refusal_reason=group,
missing_operator="lexicon_entry"
if group == "lexicon_entry"
else (
"multi_quantity_composition"
if group == "narrowness_violation"
else "pre_frame_filler_sentence"
),
)
)
audit = _write_audit(tmp_path, per_case)
proposals = decompose_audit(audit)
ids = [p.proposal_id for p in proposals]
assert len(ids) >= 2
assert ids == sorted(ids)
def test_decompose_audit_empty_file_returns_empty_tuple(tmp_path: Path) -> None:
"""Audit with ``per_case == []`` yields ``()``."""
audit = _write_audit(tmp_path, [])
assert decompose_audit(audit) == ()
def test_decompose_audit_no_runtime_mutation(tmp_path: Path) -> None:
"""Decomposer is read-only: audit file bytes unchanged, no sibling files."""
audit = _write_audit(
tmp_path,
[
_case(
case_id="ro-0001",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
_case(
case_id="ro-0002",
refusal_reason="lexicon_entry",
missing_operator="lexicon_entry",
),
],
)
before_bytes = audit.read_bytes()
before_listing = sorted(p.name for p in tmp_path.iterdir())
_ = decompose_audit(audit)
assert audit.read_bytes() == before_bytes
after_listing = sorted(p.name for p in tmp_path.iterdir())
assert after_listing == before_listing
def test_decompose_audit_skips_rows_without_known_operator(tmp_path: Path) -> None:
"""Rows whose missing_operator is unmapped (or null) are dropped silently."""
audit = _write_audit(
tmp_path,
[
_case(
case_id="skip-0001",
refusal_reason="lexicon_entry",
missing_operator="not_a_real_operator",
),
_case(
case_id="skip-0002",
refusal_reason="lexicon_entry",
missing_operator="not_a_real_operator",
),
],
)
proposals = decompose_audit(audit)
assert proposals == ()
def test_decompose_audit_missing_file_raises(tmp_path: Path) -> None:
"""A non-existent audit path raises (no silent empty-tuple swallow)."""
with pytest.raises(FileNotFoundError):
decompose_audit(tmp_path / "nope.json")