diff --git a/tests/test_adr_0172_w4_workbench_e2e.py b/tests/test_adr_0172_w4_workbench_e2e.py new file mode 100644 index 00000000..f364a977 --- /dev/null +++ b/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}" + ) diff --git a/workbench/api.py b/workbench/api.py index c025dcd2..dd9e5caf 100644 --- a/workbench/api.py +++ b/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. diff --git a/workbench/readers.py b/workbench/readers.py index 43df8d8c..61f5049b 100644 --- a/workbench/readers.py +++ b/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=, category='drain_token', reviewer='')" + ) + + 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=, category='drain_token', reviewer='')" + ) + + 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, *, diff --git a/workbench/schemas.py b/workbench/schemas.py index 103fa8aa..cc9e5d57 100644 --- a/workbench/schemas.py +++ b/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