feat(teaching/W0): ReasoningTrace substrate for ADR-0172 Tier 1 (#379)

Schema-only module defining ReasoningStep / ReasoningTrace with
byte-identical canonical serialization and sha256 trace_id derivation.
Replay-equivalence is enforced by:

- sorted-key JSON, no whitespace, ensure_ascii=False, allow_nan=False
- recursive rejection of float values in payloads (replay hazard)
- step_index monotonicity from 0
- empty trace rejected
- Literal-checked step_kind across all eight Tier 1+2 kinds

No runtime hook. No import from chat/field/generate/algebra.
Downstream (W1 ShapeProposal, W2 decomposer) consume this schema.

Tests: 12 new, full teaching suite green (17 passed).
This commit is contained in:
Shay 2026-05-27 12:21:59 -07:00 committed by GitHub
parent 3aea5a1fa8
commit f16ac96fb7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 294 additions and 0 deletions

View file

@ -0,0 +1,169 @@
"""ADR-0172 W0 — ``ReasoningTrace`` substrate.
Schema-only module. Defines the byte-identical, replay-equivalent
reasoning trace carried by ``MathReaderRefusalShapeProposal`` (W1)
and emitted by the audit-corpus decomposer (W2).
Determinism contract:
- Canonical bytes are stable across processes and dict insertion order.
- ``trace_id`` is ``sha256(canonical_bytes)`` of the full step sequence.
- Floating-point values are forbidden in payloads (would break replay).
No runtime hook. No import from ``chat``/``field``/``generate``/``algebra``.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Literal, get_args
__all__ = (
"ReasoningStep",
"ReasoningTrace",
"StepKind",
"build_trace",
"canonical_bytes",
"compute_trace_id",
)
StepKind = Literal[
"observation",
"grouping",
"abstraction",
"hypothesis",
"test_design",
"test_application",
"test_result",
"conclusion",
]
_STEP_KINDS: frozenset[str] = frozenset(get_args(StepKind))
@dataclass(frozen=True)
class ReasoningStep:
step_index: int
step_kind: StepKind
input_pointers: tuple[str, ...]
claim: str
justification: str
output_payload: object
@dataclass(frozen=True)
class ReasoningTrace:
trace_id: str
steps: tuple[ReasoningStep, ...]
def _reject_floats(value: object, *, path: str) -> None:
"""Recursively reject any ``float`` instance in a JSON-shaped value.
Booleans are a ``bool`` subclass of ``int`` and remain permitted.
"""
if isinstance(value, float):
raise ValueError(
f"floating-point values are forbidden in canonical payloads (at {path})"
)
if isinstance(value, dict):
for key, sub in value.items():
if not isinstance(key, str):
raise ValueError(
f"payload dict keys must be strings (at {path}, got {type(key).__name__})"
)
_reject_floats(sub, path=f"{path}.{key}")
elif isinstance(value, (list, tuple)):
for idx, sub in enumerate(value):
_reject_floats(sub, path=f"{path}[{idx}]")
def _validate_payload(payload: object) -> None:
"""Ensure payload is JSON-serializable and float-free."""
try:
json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False)
except (TypeError, ValueError) as exc:
raise ValueError(f"output_payload is not JSON-serializable: {exc}") from exc
_reject_floats(payload, path="output_payload")
def _step_to_canonical(step: ReasoningStep) -> dict[str, object]:
return {
"step_index": step.step_index,
"step_kind": step.step_kind,
"input_pointers": list(step.input_pointers),
"claim": step.claim,
"justification": step.justification,
"output_payload": step.output_payload,
}
def _steps_canonical_bytes(steps: tuple[ReasoningStep, ...]) -> bytes:
payload = [_step_to_canonical(step) for step in steps]
return json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def canonical_bytes(trace: ReasoningTrace) -> bytes:
"""Return the canonical byte serialization of a trace.
The ``trace_id`` field itself is excluded (it is derived from these
very bytes). Stable across processes and dict insertion order.
"""
return _steps_canonical_bytes(trace.steps)
def compute_trace_id(steps: tuple[ReasoningStep, ...]) -> str:
"""Return ``sha256(canonical_bytes(steps))`` as hex digest."""
return hashlib.sha256(_steps_canonical_bytes(steps)).hexdigest()
def build_trace(steps: list[ReasoningStep] | tuple[ReasoningStep, ...]) -> ReasoningTrace:
"""Validate, freeze, and hash a reasoning trace.
Raises ``ValueError`` on:
- empty step list
- ``step_kind`` outside the ``StepKind`` Literal
- ``step_index`` not starting at 0 with monotonic +1 increments
- non-JSON-serializable payload
- any floating-point value inside a payload (replay hazard)
"""
frozen = tuple(steps)
if not frozen:
raise ValueError("reasoning trace must contain at least one step")
for expected_index, step in enumerate(frozen):
if step.step_kind not in _STEP_KINDS:
raise ValueError(
f"step {expected_index}: unknown step_kind {step.step_kind!r}"
)
if step.step_index != expected_index:
raise ValueError(
"step_index must start at 0 and increase monotonically by 1 "
f"(step {expected_index} has step_index={step.step_index})"
)
if not isinstance(step.input_pointers, tuple):
raise ValueError(
f"step {expected_index}: input_pointers must be a tuple"
)
for pointer in step.input_pointers:
if not isinstance(pointer, str):
raise ValueError(
f"step {expected_index}: input_pointers must contain strings only"
)
_validate_payload(step.output_payload)
trace_id = compute_trace_id(frozen)
return ReasoningTrace(trace_id=trace_id, steps=frozen)

View file

@ -0,0 +1,125 @@
"""ADR-0172 W0 — ``ReasoningTrace`` substrate tests."""
from __future__ import annotations
import pytest
from teaching.math_reasoning_trace import (
ReasoningStep,
build_trace,
canonical_bytes,
compute_trace_id,
)
def _step(
*,
index: int = 0,
kind: str = "observation",
pointers: tuple[str, ...] = (),
claim: str = "claim",
justification: str = "because",
payload: object | None = None,
) -> ReasoningStep:
return ReasoningStep(
step_index=index,
step_kind=kind, # type: ignore[arg-type]
input_pointers=pointers,
claim=claim,
justification=justification,
output_payload={} if payload is None else payload,
)
def test_step_index_must_start_at_zero() -> None:
with pytest.raises(ValueError, match="step_index"):
build_trace([_step(index=1)])
def test_step_index_must_be_monotonic() -> None:
with pytest.raises(ValueError, match="step_index"):
build_trace([_step(index=0), _step(index=2, kind="grouping")])
def test_canonical_bytes_stable_across_runs() -> None:
steps_a = [
_step(index=0, kind="observation", payload={"n": 3}),
_step(index=1, kind="grouping", payload={"key": ["a", "b"]}),
]
steps_b = [
_step(index=0, kind="observation", payload={"n": 3}),
_step(index=1, kind="grouping", payload={"key": ["a", "b"]}),
]
trace_a = build_trace(steps_a)
trace_b = build_trace(steps_b)
assert canonical_bytes(trace_a) == canonical_bytes(trace_b)
assert trace_a.trace_id == trace_b.trace_id
def test_trace_id_changes_when_claim_changes() -> None:
base = build_trace([_step(claim="alpha")])
other = build_trace([_step(claim="beta")])
assert base.trace_id != other.trace_id
def test_trace_id_invariant_to_dict_insertion_order() -> None:
payload_in_order = {"alpha": 1, "beta": 2, "gamma": 3}
payload_out_of_order = {"gamma": 3, "alpha": 1, "beta": 2}
trace_a = build_trace([_step(payload=payload_in_order)])
trace_b = build_trace([_step(payload=payload_out_of_order)])
assert trace_a.trace_id == trace_b.trace_id
def test_non_json_serializable_payload_rejected() -> None:
with pytest.raises(ValueError, match="JSON-serializable"):
build_trace([_step(payload={"bad": {1, 2, 3}})])
def test_all_eight_step_kinds_accepted() -> None:
kinds = (
"observation",
"grouping",
"abstraction",
"hypothesis",
"test_design",
"test_application",
"test_result",
"conclusion",
)
steps = [_step(index=i, kind=kind) for i, kind in enumerate(kinds)]
trace = build_trace(steps)
assert tuple(s.step_kind for s in trace.steps) == kinds
def test_empty_trace_rejected() -> None:
with pytest.raises(ValueError, match="at least one step"):
build_trace([])
def test_unknown_step_kind_rejected() -> None:
bad = ReasoningStep(
step_index=0,
step_kind="speculation", # type: ignore[arg-type]
input_pointers=(),
claim="c",
justification="j",
output_payload={},
)
with pytest.raises(ValueError, match="unknown step_kind"):
build_trace([bad])
def test_float_payload_rejected() -> None:
with pytest.raises(ValueError, match="floating-point"):
build_trace([_step(payload={"weight": 0.5})])
def test_compute_trace_id_matches_build_trace() -> None:
steps = (_step(index=0, claim="x"),)
assert compute_trace_id(steps) == build_trace(list(steps)).trace_id
def test_trace_is_frozen() -> None:
trace = build_trace([_step()])
with pytest.raises(Exception): # FrozenInstanceError subclass of AttributeError
trace.trace_id = "mutated" # type: ignore[misc]