feat(ADR-0172/W4): workbench math-proposals integration + e2e tests (#385)

Wires teaching/math_proposals/proposals.jsonl into the CORE Workbench
API (ADR-0160) alongside the existing cognition proposal queue:

workbench/schemas.py
  - MathReasoningStep, MathProposalSummary, MathProposalDetail,
    MathRatifyResult schemas

workbench/readers.py
  - MATH_PROPOSALS_JSONL + _DEFAULT_MATH_AUDIT_PATH constants
  - teaching/math_proposals added to ALLOWED_ARTIFACT_ROOTS
  - _HANDLER_DISPATCH table (vocabulary_addition→LexicalClaim; all
    others not yet implemented)
  - list_math_proposals(), read_math_proposal(), ratify_math_proposal()
  - read_math_proposal() re-runs decompose_audit() to recover full
    4-step reasoning trace (canonical_bytes only carries trace_id)
  - ratify_math_proposal() raises NotImplementedError with clear
    "handler not yet implemented: {change_kind}" for unhandled kinds

workbench/api.py
  - GET /math-proposals, GET /math-proposals/{id}
  - POST /math-proposals/{id}/ratify → _math_ratify()
    (vocabulary_addition→200/routed; unhandled→501 with loud message)

tests/test_adr_0172_w4_workbench_e2e.py — 6 tests:
  1. loads from JSONL
  2. renders domain:math badge (distinct from cognition /proposals)
  3. ratify-vocabulary_addition routes to LexicalClaim (200)
  4. ratify-matcher_extension fails loudly (501 "handler not yet
     implemented")
  5. all 4 trace steps visible in detail response
  6. no cross-contamination between math and cognition queues

teaching + runtime suites green (28 + 20 passed).

Brief-gap note: canonical_bytes() excludes proposal_id and serialises
evidence pointers as hashes only. D1 loader derives proposal_id via
sha256(line_bytes) and re-runs decompose_audit() to recover full trace
for read_math_proposal(). This works but means the JSONL cannot be
loaded without the original audit file. If a future wave needs
standalone JSONL loading, C1 should emit a richer format.

tests/test_adr_0172_w4_workbench_e2e.py

@@ -0,0 +1,360 @@
+"""ADR-0172 W4 — Workbench math-proposals e2e tests.
+
+Six tests:
+
+1. Loads from JSONL: GET /math-proposals returns items when proposals.jsonl exists.
+2. Renders domain badge: items carry domain='math', distinct from cognition /proposals.
+3. ratify-vocabulary_addition routes to LexicalClaim handler (200, routing_status=routed).
+4. ratify-matcher_extension fails loudly (501, not_implemented).
+5. All 4 trace steps visible: GET /math-proposals/{id} includes 4-step trace.
+6. No cross-contamination: cognition /proposals and math /math-proposals are independent.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from workbench.api import WorkbenchApi
+from workbench.readers import (
+    MATH_PROPOSALS_JSONL,
+    _DEFAULT_MATH_AUDIT_PATH,
+    _load_math_proposals_raw,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+REAL_AUDIT_PATH = _DEFAULT_MATH_AUDIT_PATH
+
+
+def _write_jsonl(tmp_path: Path, proposals_jsonl_bytes: bytes) -> Path:
+    p = tmp_path / "proposals.jsonl"
+    p.write_bytes(proposals_jsonl_bytes)
+    return p
+
+
+def _api(jsonl_path: Path | None = None, audit_path: Path | None = None) -> WorkbenchApi:
+    """Return a WorkbenchApi with its readers patched to use tmp paths."""
+    import workbench.readers as _r
+
+    api = WorkbenchApi()
+    _orig_list = _r.list_math_proposals
+    _orig_read = _r.read_math_proposal
+    _orig_ratify = _r.ratify_math_proposal
+
+    if jsonl_path is not None:
+        import functools
+
+        api._list_math_proposals = functools.partial(_r.list_math_proposals, jsonl_path=jsonl_path)
+        api._read_math_proposal = functools.partial(
+            _r.read_math_proposal, jsonl_path=jsonl_path, audit_path=audit_path or REAL_AUDIT_PATH
+        )
+        api._ratify_math_proposal = functools.partial(
+            _r.ratify_math_proposal, jsonl_path=jsonl_path
+        )
+        # Patch the module-level functions temporarily
+        _r.list_math_proposals = api._list_math_proposals  # type: ignore[assignment]
+        _r.read_math_proposal = api._read_math_proposal  # type: ignore[assignment]
+        _r.ratify_math_proposal = api._ratify_math_proposal  # type: ignore[assignment]
+
+    yield api
+
+    # Restore
+    _r.list_math_proposals = _orig_list
+    _r.read_math_proposal = _orig_read
+    _r.ratify_math_proposal = _orig_ratify
+
+
+def _get(api: WorkbenchApi, path: str) -> dict:
+    resp = api.handle("GET", path)
+    return {"status": resp.status, "payload": resp.payload}
+
+
+def _post(api: WorkbenchApi, path: str, body: bytes = b"{}") -> dict:
+    resp = api.handle("POST", path, body)
+    return {"status": resp.status, "payload": resp.payload}
+
+
+def _make_synthetic_proposals_jsonl(tmp_path: Path, change_kinds: list[str]) -> tuple[Path, list[str]]:
+    """Build a synthetic proposals.jsonl with one proposal per change_kind.
+
+    Returns (jsonl_path, list_of_proposal_ids).
+    """
+    from evals.refusal_taxonomy.shape_categories import ShapeCategory
+    from teaching.math_contemplation import decompose_audit
+    from teaching.math_contemplation_proposal import build_proposal, canonical_bytes
+    from teaching.math_evidence import MathReaderRefusalEvidence, SUB_TYPE_FOR_OPERATOR
+    from teaching.math_reasoning_trace import ReasoningStep, build_trace
+    from generate.comprehension.audit import AuditRow
+
+    def _ev(case_id: str, missing_op: str) -> MathReaderRefusalEvidence:
+        from teaching.math_evidence import from_audit_row
+        row = AuditRow(
+            case_id=case_id,
+            sentence_index=0,
+            token_index=0,
+            token_text="weight",
+            recognized_terms=(),
+            skipped_frame=None,
+            missing_operator=missing_op,
+            refusal_reason="lexicon_entry",
+            refusal_detail="",
+        )
+        return from_audit_row(row, "lexical", claim_signature="")
+
+    lines: list[bytes] = []
+    proposal_ids: list[str] = []
+
+    for i, ck in enumerate(change_kinds):
+        ev1 = _ev(f"c{i:02d}a", "drain_token")
+        ev2 = _ev(f"c{i:02d}b", "drain_token")
+
+        obs = ReasoningStep(
+            step_index=0, step_kind="observation", input_pointers=(ev1.case_id, ev2.case_id),
+            claim=f"synthetic observation for {ck}", justification="test",
+            output_payload={"evidence_count": 2},
+        )
+        grp = ReasoningStep(
+            step_index=1, step_kind="grouping", input_pointers=(ev1.case_id, ev2.case_id),
+            claim="group key", justification="test", output_payload={"k": "v"},
+        )
+        hyp = ReasoningStep(
+            step_index=2, step_kind="hypothesis", input_pointers=(ev1.case_id, ev2.case_id),
+            claim=f"change_kind={ck}", justification="test", output_payload={"ck": ck},
+        )
+        con = ReasoningStep(
+            step_index=3, step_kind="conclusion", input_pointers=(ev1.case_id, ev2.case_id),
+            claim="conclude", justification="test", output_payload={"done": 1},
+        )
+        trace = build_trace((obs, grp, hyp, con))
+
+        p = build_proposal(
+            shape_category=ShapeCategory.UNCATEGORIZED,
+            structural_commonality=f"synthetic {ck} test proposal — sufficient length for wrong_zero_assertion",
+            evidence_pointers=(ev1, ev2),
+            proposed_change_kind=ck,  # type: ignore[arg-type]
+            proposed_change_payload={"evidence_count": 2, "group_key": {"k": "v"}, "modal_sub_type": "lexical"},
+            wrong_zero_assertion=(
+                "synthetic proposal for test; ratification handler enforces wrong=0"
+            ),
+            replay_equivalence_hash=hashlib.sha256(f"replay-{ck}-{i}".encode()).hexdigest(),
+            reasoning_trace=trace,
+        )
+        cb = canonical_bytes(p)
+        lines.append(cb + b"\n")
+        proposal_ids.append(p.proposal_id)
+
+    jsonl_path = tmp_path / "proposals.jsonl"
+    jsonl_path.write_bytes(b"".join(lines))
+    return jsonl_path, proposal_ids
+
+
+# ---------------------------------------------------------------------------
+# Test 1: Loads from JSONL
+# ---------------------------------------------------------------------------
+
+
+def test_1_loads_from_jsonl(tmp_path: Path) -> None:
+    """GET /math-proposals returns items from proposals.jsonl when it exists."""
+    from teaching.math_contemplation import decompose_audit
+    from teaching.math_contemplation_proposal import canonical_bytes as cb
+
+    proposals = decompose_audit(REAL_AUDIT_PATH)
+    jsonl_bytes = b"".join(cb(p) + b"\n" for p in proposals)
+    jsonl = _write_jsonl(tmp_path, jsonl_bytes)
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    _r.MATH_PROPOSALS_JSONL = jsonl
+    try:
+        api = WorkbenchApi()
+        resp = api.handle("GET", "/math-proposals")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+
+    assert resp.status == 200
+    data = resp.payload.get("data", {})
+    items = data.get("items", [])
+    assert len(items) == len(proposals)
+
+
+# ---------------------------------------------------------------------------
+# Test 2: Renders domain:math badge, distinct from cognition domain
+# ---------------------------------------------------------------------------
+
+
+def test_2_renders_domain_math_badge(tmp_path: Path) -> None:
+    """Items from /math-proposals have domain='math'; /proposals returns cognition (no domain field)."""
+    from teaching.math_contemplation import decompose_audit
+    from teaching.math_contemplation_proposal import canonical_bytes as cb
+
+    proposals = decompose_audit(REAL_AUDIT_PATH)
+    jsonl_bytes = b"".join(cb(p) + b"\n" for p in proposals)
+    jsonl = _write_jsonl(tmp_path, jsonl_bytes)
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    _r.MATH_PROPOSALS_JSONL = jsonl
+    try:
+        api = WorkbenchApi()
+        resp = api.handle("GET", "/math-proposals")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+
+    items = resp.payload["data"]["items"]
+    assert all(item["domain"] == "math" for item in items), (
+        "all math proposals must carry domain='math'"
+    )
+
+    # Cognition /proposals carries ProposalSummary — no domain field at all
+    cog_resp = WorkbenchApi().handle("GET", "/proposals")
+    cog_items = cog_resp.payload.get("data", {}).get("items", [])
+    for cog_item in cog_items:
+        assert cog_item.get("domain") != "math", (
+            "cognition proposals must not carry domain='math'"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Test 3: ratify-vocabulary_addition routes to LexicalClaim handler
+# ---------------------------------------------------------------------------
+
+
+def test_3_ratify_vocabulary_addition_routes_to_lexical_claim(tmp_path: Path) -> None:
+    """POST /math-proposals/{id}/ratify routes vocabulary_addition to LexicalClaim (200)."""
+    jsonl_path, proposal_ids = _make_synthetic_proposals_jsonl(tmp_path, ["vocabulary_addition"])
+    vocab_pid = proposal_ids[0]
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    _r.MATH_PROPOSALS_JSONL = jsonl_path
+    try:
+        api = WorkbenchApi()
+        resp = api.handle("POST", f"/math-proposals/{vocab_pid}/ratify")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+
+    assert resp.status == 200, f"expected 200, got {resp.status}: {resp.payload}"
+    data = resp.payload.get("data", {})
+    assert data.get("handler_name") == "LexicalClaim", (
+        f"vocabulary_addition must route to LexicalClaim, got: {data.get('handler_name')!r}"
+    )
+    assert data.get("routing_status") == "routed"
+    assert data.get("change_kind") == "vocabulary_addition"
+
+
+# ---------------------------------------------------------------------------
+# Test 4: ratify-matcher_extension fails loudly
+# ---------------------------------------------------------------------------
+
+
+def test_4_ratify_matcher_extension_fails_loudly(tmp_path: Path) -> None:
+    """POST /math-proposals/{id}/ratify returns 501 for matcher_extension with clear message."""
+    jsonl_path, proposal_ids = _make_synthetic_proposals_jsonl(tmp_path, ["matcher_extension"])
+    matcher_pid = proposal_ids[0]
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    _r.MATH_PROPOSALS_JSONL = jsonl_path
+    try:
+        api = WorkbenchApi()
+        resp = api.handle("POST", f"/math-proposals/{matcher_pid}/ratify")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+
+    assert resp.status == 501, f"expected 501 for unimplemented handler, got {resp.status}"
+    err = resp.payload.get("error", {})
+    message = err.get("message", "")
+    assert "handler not yet implemented" in message, (
+        f"expected 'handler not yet implemented' in message, got: {message!r}"
+    )
+    assert "matcher_extension" in message, (
+        f"expected change_kind in error message, got: {message!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Test 5: All 4 trace steps visible
+# ---------------------------------------------------------------------------
+
+
+def test_5_all_4_trace_steps_visible(tmp_path: Path) -> None:
+    """GET /math-proposals/{id} includes reasoning_trace_steps with all 4 steps."""
+    from teaching.math_contemplation import decompose_audit
+    from teaching.math_contemplation_proposal import canonical_bytes as cb
+
+    proposals = decompose_audit(REAL_AUDIT_PATH)
+    assert proposals, "real audit must produce at least one proposal"
+    jsonl_bytes = b"".join(cb(p) + b"\n" for p in proposals)
+    jsonl = _write_jsonl(tmp_path, jsonl_bytes)
+
+    first_id = proposals[0].proposal_id
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    orig_audit = _r._DEFAULT_MATH_AUDIT_PATH
+    _r.MATH_PROPOSALS_JSONL = jsonl
+    _r._DEFAULT_MATH_AUDIT_PATH = REAL_AUDIT_PATH
+    try:
+        api = WorkbenchApi()
+        resp = api.handle("GET", f"/math-proposals/{first_id}")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+        _r._DEFAULT_MATH_AUDIT_PATH = orig_audit
+
+    assert resp.status == 200, f"expected 200, got {resp.status}: {resp.payload}"
+    detail = resp.payload.get("data", {})
+    steps = detail.get("reasoning_trace_steps", [])
+    assert len(steps) == 4, f"expected 4 trace steps, got {len(steps)}"
+    step_kinds = [s["step_kind"] for s in steps]
+    assert "observation" in step_kinds
+    assert "grouping" in step_kinds
+    assert "hypothesis" in step_kinds
+    assert "conclusion" in step_kinds
+
+
+# ---------------------------------------------------------------------------
+# Test 6: No cross-contamination between math and cognition queues
+# ---------------------------------------------------------------------------
+
+
+def test_6_no_cross_contamination(tmp_path: Path) -> None:
+    """Math /math-proposals and cognition /proposals are independent queues."""
+    from teaching.math_contemplation import decompose_audit
+    from teaching.math_contemplation_proposal import canonical_bytes as cb
+
+    proposals = decompose_audit(REAL_AUDIT_PATH)
+    jsonl_bytes = b"".join(cb(p) + b"\n" for p in proposals)
+    jsonl = _write_jsonl(tmp_path, jsonl_bytes)
+
+    import workbench.readers as _r
+    orig = _r.MATH_PROPOSALS_JSONL
+    _r.MATH_PROPOSALS_JSONL = jsonl
+    try:
+        api = WorkbenchApi()
+        math_resp = api.handle("GET", "/math-proposals")
+        cog_resp = api.handle("GET", "/proposals")
+    finally:
+        _r.MATH_PROPOSALS_JSONL = orig
+
+    math_items = math_resp.payload.get("data", {}).get("items", [])
+    cog_items = cog_resp.payload.get("data", {}).get("items", [])
+
+    # Math queue carries domain:math; cognition queue has no domain:math entries
+    assert all(item.get("domain") == "math" for item in math_items)
+    for cog_item in cog_items:
+        assert cog_item.get("domain") != "math", (
+            "cognition proposal leaked into math domain"
+        )
+
+    # proposal_ids from the two queues are disjoint
+    math_ids = {item["proposal_id"] for item in math_items}
+    cog_ids = {item["proposal_id"] for item in cog_items}
+    assert math_ids.isdisjoint(cog_ids), (
+        f"proposal IDs overlap between math and cognition queues: {math_ids & cog_ids}"
+    )

workbench/api.py

@@ -18,7 +18,7 @@ from core.epistemic_state import (
 )
 from workbench import readers
 from workbench.readers import ArtifactTooLargeError
-from workbench.schemas import ChatTurnResult, ProposalRef, TurnVerdict, error, ok
+from workbench.schemas import ChatTurnResult, MathRatifyResult, ProposalRef, TurnVerdict, error, ok
 
 
 MAX_CHAT_BODY_BYTES = 64 * 1024
@@ -75,6 +75,14 @@ class WorkbenchApi:
         if method == "GET" and path.startswith("/proposals/"):
             proposal_id = unquote(path.removeprefix("/proposals/"))
             return ApiResponse(200, ok(readers.read_proposal(proposal_id)))
+        if method == "GET" and path == "/math-proposals":
+            return ApiResponse(200, ok({"items": readers.list_math_proposals()}))
+        if method == "POST" and path.endswith("/ratify") and path.startswith("/math-proposals/"):
+            proposal_id = unquote(path.removeprefix("/math-proposals/").removesuffix("/ratify"))
+            return self._math_ratify(proposal_id)
+        if method == "GET" and path.startswith("/math-proposals/"):
+            proposal_id = unquote(path.removeprefix("/math-proposals/"))
+            return ApiResponse(200, ok(readers.read_math_proposal(proposal_id)))
         if method == "GET" and path == "/evals":
             return ApiResponse(200, ok({"lanes": readers.list_eval_lanes()}))
         if method == "GET" and path.startswith("/evals/"):
@@ -103,6 +111,14 @@ class WorkbenchApi:
             return ApiResponse(501, error("unsupported", "route is deferred beyond W-026"))
         return ApiResponse(404, error("not_found", f"route not found: {method} {path}"))
 
+    def _math_ratify(self, proposal_id: str) -> ApiResponse:
+        """Route ratification by change_kind; 501 for unimplemented handlers."""
+        try:
+            result: MathRatifyResult = readers.ratify_math_proposal(proposal_id)
+        except NotImplementedError as exc:
+            return ApiResponse(501, error("unsupported", str(exc)))
+        return ApiResponse(200, ok(result))
+
     def _chat_turn(self, body: bytes) -> ApiResponse:
         """Execute one live runtime turn.
 

workbench/readers.py

@@ -17,6 +17,10 @@ from workbench.schemas import (
     ArtifactRef,
     EvalLaneSummary,
     EvalRunResult,
+    MathProposalDetail,
+    MathProposalSummary,
+    MathRatifyResult,
+    MathReasoningStep,
     ProposalDetail,
     ProposalSummary,
     RuntimeStatus,
@@ -31,10 +35,27 @@ _REVIEW_STATES = frozenset(get_args(ReviewState))
 ALLOWED_ARTIFACT_ROOTS = (
     REPO_ROOT / "engine_state",
     REPO_ROOT / "teaching" / "proposals",
+    REPO_ROOT / "teaching" / "math_proposals",
     REPO_ROOT / "evals",
     REPO_ROOT / "contemplation" / "runs",
 )
 
+MATH_PROPOSALS_JSONL = REPO_ROOT / "teaching" / "math_proposals" / "proposals.jsonl"
+_DEFAULT_MATH_AUDIT_PATH = (
+    REPO_ROOT
+    / "evals"
+    / "gsm8k_math"
+    / "train_sample"
+    / "v1"
+    / "audit_brief_11.json"
+)
+
+# Dispatch table: proposed_change_kind → handler name.
+# Handlers not listed here are not yet implemented.
+_HANDLER_DISPATCH: dict[str, str] = {
+    "vocabulary_addition": "LexicalClaim",
+}
+
 
 class ArtifactTooLargeError(OSError):
     """Raised when an artifact is too large for direct Workbench reads."""
@@ -268,6 +289,175 @@ def read_eval_lane(lane_name: str) -> EvalLaneSummary:
     )
 
 
+def _load_math_proposals_raw(jsonl_path: Path) -> list[dict[str, Any]]:
+    """Parse proposals.jsonl; derive proposal_id = sha256(canonical_line_bytes)."""
+    if not jsonl_path.exists():
+        return []
+    results: list[dict[str, Any]] = []
+    for raw_line in jsonl_path.read_bytes().splitlines():
+        stripped = raw_line.strip()
+        if not stripped:
+            continue
+        proposal_id = hashlib.sha256(stripped).hexdigest()
+        data: dict[str, Any] = json.loads(stripped)
+        data["proposal_id"] = proposal_id
+        results.append(data)
+    return results
+
+
+def _math_proposal_summary(record: dict[str, Any]) -> MathProposalSummary:
+    payload = record.get("proposed_change_payload") or {}
+    evidence_count = int(payload.get("evidence_count", 0)) if isinstance(payload, dict) else 0
+    return MathProposalSummary(
+        proposal_id=str(record.get("proposal_id", "")),
+        domain="math",
+        shape_category=str(record.get("shape_category", "")),
+        proposed_change_kind=str(record.get("proposed_change_kind", "")),
+        structural_commonality=str(record.get("structural_commonality", "")),
+        evidence_count=evidence_count,
+        replay_equivalence_hash=str(record.get("replay_equivalence_hash", "")),
+    )
+
+
+def list_math_proposals(*, jsonl_path: Path | None = None) -> list[MathProposalSummary]:
+    path = jsonl_path or MATH_PROPOSALS_JSONL
+    records = _load_math_proposals_raw(path)
+    return [_math_proposal_summary(r) for r in records]
+
+
+def _math_trace_steps_from_proposal(proposal: Any) -> list[MathReasoningStep]:
+    """Extract 4 ReasoningStep objects from a MathReaderRefusalShapeProposal."""
+    trace = getattr(proposal, "reasoning_trace", None)
+    if trace is None:
+        return []
+    steps_raw = getattr(trace, "steps", ())
+    steps: list[MathReasoningStep] = []
+    for step in steps_raw:
+        steps.append(
+            MathReasoningStep(
+                step_index=int(getattr(step, "step_index", 0)),
+                step_kind=str(getattr(step, "step_kind", "")),
+                claim=str(getattr(step, "claim", "")),
+                justification=str(getattr(step, "justification", "")),
+                input_pointers=list(getattr(step, "input_pointers", ())),
+                output_payload=getattr(step, "output_payload", {}),
+            )
+        )
+    return steps
+
+
+def read_math_proposal(
+    proposal_id: str,
+    *,
+    jsonl_path: Path | None = None,
+    audit_path: Path | None = None,
+) -> MathProposalDetail:
+    """Return full proposal detail including 4-step reasoning trace.
+
+    Re-runs :func:`teaching.math_contemplation.decompose_audit` to recover
+    the full :class:`MathReaderRefusalShapeProposal` (canonical bytes only
+    carry the trace_id, not the full steps).  Deterministic: same audit →
+    same proposals.
+    """
+    from teaching.math_contemplation import decompose_audit
+
+    # Verify the proposal_id exists in the JSONL first (fast path).
+    path = jsonl_path or MATH_PROPOSALS_JSONL
+    records = _load_math_proposals_raw(path)
+    record = next((r for r in records if r.get("proposal_id") == proposal_id), None)
+    if record is None:
+        raise FileNotFoundError(proposal_id)
+
+    # Re-run decomposer to get the full proposal with trace steps.
+    apath = audit_path or _DEFAULT_MATH_AUDIT_PATH
+    full_proposals = decompose_audit(apath)
+    full = next((p for p in full_proposals if p.proposal_id == proposal_id), None)
+    if full is None:
+        raise FileNotFoundError(f"{proposal_id} (not found in decomposer output)")
+
+    change_kind = str(record.get("proposed_change_kind", ""))
+    handler_name = _HANDLER_DISPATCH.get(change_kind)
+
+    trace_id = str(record.get("reasoning_trace_id", ""))
+    trace_steps = _math_trace_steps_from_proposal(full)
+
+    evidence_pointers = record.get("evidence_pointers", [])
+    evidence_hashes = list(evidence_pointers) if isinstance(evidence_pointers, list) else []
+
+    suggested_cli: str | None = None
+    if handler_name == "LexicalClaim":
+        suggested_cli = (
+            f"# ratify via Python REPL:\n"
+            f"from teaching.math_lexical_ratification import apply_lexical_claim\n"
+            f"# apply_lexical_claim(claim=<evidence>, category='drain_token', reviewer='<you>')"
+        )
+
+    return MathProposalDetail(
+        proposal_id=proposal_id,
+        domain="math",
+        shape_category=str(record.get("shape_category", "")),
+        proposed_change_kind=change_kind,
+        structural_commonality=str(record.get("structural_commonality", "")),
+        evidence_count=len(evidence_hashes),
+        replay_equivalence_hash=str(record.get("replay_equivalence_hash", "")),
+        wrong_zero_assertion=str(record.get("wrong_zero_assertion", "")),
+        proposed_change_payload=record.get("proposed_change_payload"),
+        reasoning_trace_id=trace_id,
+        reasoning_trace_steps=trace_steps,
+        evidence_hashes=evidence_hashes,
+        handler_name=handler_name,
+        suggested_ratify_cli=suggested_cli,
+    )
+
+
+def ratify_math_proposal(
+    proposal_id: str,
+    *,
+    jsonl_path: Path | None = None,
+) -> MathRatifyResult:
+    """Dispatch ratification by change_kind; fail loudly for unimplemented handlers.
+
+    ADR-0160 "Proposal before mutation" doctrine: this function validates
+    routing and returns the handler name + suggested CLI without applying
+    the change.  Mutation requires an explicit operator action outside the
+    workbench (e.g. calling apply_lexical_claim() directly).
+
+    Raises FileNotFoundError if proposal_id not found.
+    Raises NotImplementedError with a clear message for unhandled change_kinds.
+    """
+    path = jsonl_path or MATH_PROPOSALS_JSONL
+    records = _load_math_proposals_raw(path)
+    record = next((r for r in records if r.get("proposal_id") == proposal_id), None)
+    if record is None:
+        raise FileNotFoundError(proposal_id)
+
+    change_kind = str(record.get("proposed_change_kind", ""))
+    handler_name = _HANDLER_DISPATCH.get(change_kind)
+
+    if handler_name is None:
+        raise NotImplementedError(
+            f"handler not yet implemented for change_kind={change_kind!r}; "
+            f"see docs/handoff/ADR-0167-FOLLOWUPS.md §1 for the scoping ADR required "
+            f"before this change_kind can be ratified"
+        )
+
+    suggested_cli: str | None = None
+    if handler_name == "LexicalClaim":
+        suggested_cli = (
+            f"from teaching.math_lexical_ratification import apply_lexical_claim\n"
+            f"# apply_lexical_claim(claim=<evidence>, category='drain_token', reviewer='<you>')"
+        )
+
+    return MathRatifyResult(
+        proposal_id=proposal_id,
+        change_kind=change_kind,
+        handler_name=handler_name,
+        routing_status="routed",
+        message=f"routed to {handler_name} handler",
+        suggested_cli=suggested_cli,
+    )
+
+
 def run_safe_eval_lane(
     lane_name: str,
     *,

workbench/schemas.py

@@ -196,3 +196,50 @@ class ReplayComparison:
     replay_hash: str | None
     equivalent: bool
     divergences: list[ReplayDivergence] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# ADR-0172 W4 — Math proposal schemas
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class MathReasoningStep:
+    step_index: int
+    step_kind: str
+    claim: str
+    justification: str
+    input_pointers: list[str]
+    output_payload: Any
+
+
+@dataclass(frozen=True, slots=True)
+class MathProposalSummary:
+    proposal_id: str
+    domain: Literal["math"]
+    shape_category: str
+    proposed_change_kind: str
+    structural_commonality: str
+    evidence_count: int
+    replay_equivalence_hash: str
+
+
+@dataclass(frozen=True, slots=True)
+class MathProposalDetail(MathProposalSummary):
+    wrong_zero_assertion: str
+    proposed_change_payload: Any
+    reasoning_trace_id: str
+    reasoning_trace_steps: list[MathReasoningStep]
+    evidence_hashes: list[str]
+    handler_name: str | None
+    suggested_ratify_cli: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class MathRatifyResult:
+    proposal_id: str
+    change_kind: str
+    handler_name: str
+    routing_status: Literal["routed", "not_implemented"]
+    message: str
+    suggested_cli: str | None = None