Merge pull request #716 from AssetOverflow/feat/wb-r3-replay-backend

feat(workbench): sealed single-turn replay backend — GET /replay/{turn_id} (Wave R3)

docs/analysis/replay-moment-backend-scoping-2026-06-12.md

@@ -2,7 +2,32 @@
 
 Date: 2026-06-12
 Plan: `docs/workbench/wave-R-mastery-revamp.md` § Wave R3 (first item).
-Status: scoped; implementation is the first R3 PR.
+Status: implemented — with one design correction recorded below.
+
+## Amendment (implementation PR, same day)
+
+The "sealed genesis-prefix replay" decision below was **wrong** and was
+corrected before any code shipped.  This doc was written before reading the
+chat handler; implementation began by reading it, which surfaced the
+decisive fact: `workbench/api.py::_run_chat_turn` constructs a **fresh
+`ChatRuntime()` per turn**.  The journaled turns were never one continuous
+session, so feeding prompts `1..N` through a single runtime would
+accumulate session state the original turns never had — prefix replay
+would manufacture spurious divergence by construction.
+
+What shipped instead: **sealed fresh-runtime single-turn replay**
+(`comparison_basis: "sealed_fresh_runtime_single_turn"`), which matches
+the original per-turn execution shape and is O(1) instead of O(N).  The
+honesty consequence is unchanged in kind but renamed: the unknown is not
+prefix completeness but whether the original turn's fresh runtime loaded
+an engine-state checkpoint present at the time (`origin_state:
+"unrecorded"` — the journal does not record it).  `refused:
+prefix_too_long` is obsolete (no prefix); the contract section in
+`docs/workbench/api-contract-v1.md` § Replay is authoritative for the
+shipped shape.  The five proof obligations below survive, re-aimed at the
+single-turn model, and all execute in `tests/test_workbench_replay.py`.
+
+The original analysis is preserved unedited below as the decision record.
 
 ## What exists today
 

docs/workbench/api-contract-v1.md

@@ -276,18 +276,41 @@ Response:
 
 # Replay
 
-## GET /replay/{artifact_id}
+## GET /replay/{turn_id}
+
+(Wave R3 — supersedes the unwired W-026 `{artifact_id}` placeholder.
+Design record: `docs/analysis/replay-moment-backend-scoping-2026-06-12.md`.)
 
 Purpose:
 
-Return replay comparison information.
+Re-execute a journaled turn's prompt in a **sealed fresh runtime**
+(`ChatRuntime(no_load_state=True)` — no checkpoint load, no checkpoint
+write, no proposal lineage) and compare the resulting envelope leaf-by-leaf
+against the recorded `TurnJournalEntry`.
 
 Important:
 
-The API must not claim `"equivalent": true` unless a real replay/compare path
-has run. Comparing an artifact digest to itself is not replay evidence. Until
-the replay theater is wired, this route should return `unsupported` or a
-non-equivalent comparison with explicit divergence evidence.
+The API must not claim `"equivalent": true` unless a real re-execution has
+run. Comparing a digest to itself is not replay evidence; if the replay
+runtime fails, the route returns `runtime_unavailable` (500) and no
+comparison object exists.
+
+Honesty envelope: the claim demonstrated is *same prompt, same genesis
+substrate → bit-identical envelope*. The original turn ran in its own fresh
+runtime that may have loaded an engine-state checkpoint present at the
+time; the journal does not record whether one existed, so `origin_state` is
+`"unrecorded"` and a divergence means nondeterminism **or** origin-state
+influence — the API never claims to distinguish them, and the frontend must
+not render a divergence as a determinism-failure verdict.
+
+Severity classes: every `TurnJournalEntry` field is classified exactly once
+(enforced by test). `critical` divergences break equivalence; wall-clock
+fields (`timestamp`, `turn_cost_ms`, `journal_digest`) are `informational`
+and never do.
+
+Errors: unknown or malformed `turn_id` → 404 `not_found`; replay runtime
+failure → 500 `runtime_unavailable`. The route is read-only: it appends no
+journal entry and writes no engine state.
 
 Response:
 
@@ -295,16 +318,19 @@ Response:
 {
   "ok": true,
   "data": {
-    "artifact_id": "artifact-001",
-    "original_hash": "sha256:...",
-    "replay_hash": null,
-    "equivalent": false,
+    "turn_id": 7,
+    "comparison_basis": "sealed_fresh_runtime_single_turn",
+    "origin_state": "unrecorded",
+    "original_trace_hash": "sha256:...",
+    "replay_trace_hash": "sha256:...",
+    "equivalent": true,
+    "replay_turn_cost_ms": 412,
     "divergences": [
       {
-        "path": "$",
-        "original": "artifact digest",
-        "replay": null,
-        "severity": "info"
+        "path": "timestamp",
+        "original": "2026-06-12T18:00:00+00:00",
+        "replay": "2026-06-12T23:55:01+00:00",
+        "severity": "informational"
       }
     ]
   }

tests/test_workbench_api.py

@@ -202,11 +202,14 @@ def test_unknown_trace_returns_404_not_placeholder_success() -> None:
     assert response.payload["error"]["code"] == "not_found"
 
 
-def test_replay_is_explicitly_unsupported_in_w026() -> None:
+def test_replay_path_style_ids_refuse_not_found() -> None:
+    # Wave R3 wired GET /replay/{turn_id} (tests/test_workbench_replay.py owns
+    # its obligations); the W-026 path-style artifact id is not a turn id and
+    # must refuse cleanly rather than 500 or fabricate a comparison.
     replay = _request("GET", "/replay/evals/cognition/results/example.json")
 
-    assert replay.status == 501
-    assert replay.payload["error"]["code"] == "unsupported"
+    assert replay.status == 404
+    assert replay.payload["error"]["code"] == "not_found"
 
 
 def test_unexpected_dispatch_error_stays_json(monkeypatch) -> None:

tests/test_workbench_replay.py

@@ -0,0 +1,192 @@
+"""Sealed single-turn replay — GET /replay/{turn_id} (Wave R3).
+
+Proof obligations from docs/analysis/replay-moment-backend-scoping-2026-06-12.md:
+each test here must MEANINGFULLY FAIL under the violation it is written to
+catch (CLAUDE.md schema-defined proof obligations rule).
+"""
+
+from __future__ import annotations
+
+from dataclasses import fields, replace
+from pathlib import Path
+
+import pytest
+
+from chat.runtime import ChatRuntime
+from workbench import api as workbench_api
+from workbench.api import WorkbenchApi, _run_sealed_chat_turn, _with_turn_cost_and_id
+from workbench.journal import TurnJournal, TurnJournalEntry
+from workbench.replay import CRITICAL_FIELDS, INFORMATIONAL_FIELDS, replay_turn
+
+_ENGINE_STATE_DIR = Path("engine_state")
+
+
+def _snapshot(root: Path) -> dict[str, bytes]:
+    snap: dict[str, bytes] = {}
+    if not root.exists():
+        return snap
+    for path in sorted(root.rglob("*")):
+        if path.is_file() and "__pycache__" not in path.relative_to(root).parts:
+            snap[path.relative_to(root).as_posix()] = path.read_bytes()
+    return snap
+
+
+@pytest.fixture(scope="module")
+def recorded() -> tuple[TurnJournalEntry, object]:
+    """One sealed-origin turn, journaled — the genesis-true positive control.
+
+    Module-scoped: the sealed runtime turn is the expensive part; every test
+    derives tampered copies from this single recorded entry.
+    """
+    result = _run_sealed_chat_turn("What is truth?")
+    result = _with_turn_cost_and_id(result, 7, 1)
+    entry = TurnJournalEntry.from_chat_turn(result, turn_id=1)
+    return entry, result
+
+
+def _api_with_entry(tmp_path: Path, entry: TurnJournalEntry) -> WorkbenchApi:
+    journal = TurnJournal(journal_dir=tmp_path / "workbench_data")
+    journal.append(entry)
+    return WorkbenchApi(journal=journal)
+
+
+def test_field_classification_is_exhaustive() -> None:
+    """A future journal field must force an explicit severity decision."""
+    names = {spec.name for spec in fields(TurnJournalEntry)}
+    assert CRITICAL_FIELDS | INFORMATIONAL_FIELDS == names
+    assert not CRITICAL_FIELDS & INFORMATIONAL_FIELDS
+
+
+def test_sealed_origin_turn_replays_equivalent(tmp_path, recorded) -> None:
+    """Positive control: genesis-origin entry replays bit-equivalent."""
+    entry, _ = recorded
+    api = _api_with_entry(tmp_path, entry)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    assert response.status == 200
+    data = response.payload["data"]
+    assert data["equivalent"] is True
+    assert data["comparison_basis"] == "sealed_fresh_runtime_single_turn"
+    assert data["origin_state"] == "unrecorded"
+    assert data["original_trace_hash"] == entry.trace_hash
+    assert data["replay_trace_hash"] == entry.trace_hash
+    assert all(d["severity"] == "informational" for d in data["divergences"])
+
+
+def test_tampered_prompt_breaks_equivalence(tmp_path, recorded) -> None:
+    """Obligation 1: mutating the recorded prompt flips equivalent to false."""
+    entry, _ = recorded
+    tampered = replace(entry, prompt="What is beauty?")
+    api = _api_with_entry(tmp_path, tampered)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    assert response.status == 200
+    data = response.payload["data"]
+    assert data["equivalent"] is False
+    critical_paths = {d["path"] for d in data["divergences"] if d["severity"] == "critical"}
+    assert critical_paths  # the different prompt produced a different envelope
+
+
+def test_tampered_surface_diverges_at_exactly_that_leaf(tmp_path, recorded) -> None:
+    """Obligation 2: comparison reads the RECORDED entry, leaf-precise."""
+    entry, _ = recorded
+    tampered = replace(entry, surface=entry.surface + " [tampered]")
+    api = _api_with_entry(tmp_path, tampered)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    data = response.payload["data"]
+    assert data["equivalent"] is False
+    critical = [d for d in data["divergences"] if d["severity"] == "critical"]
+    assert [d["path"] for d in critical] == ["surface"]
+    assert critical[0]["original"] == entry.surface + " [tampered]"
+    assert critical[0]["replay"] == entry.surface
+
+
+def test_no_comparison_without_real_execution(tmp_path, recorded, monkeypatch) -> None:
+    """Obligation 3: a digest-to-itself shortcut cannot pass.
+
+    If the runtime cannot execute, the endpoint must surface a typed error
+    and never a fabricated comparison — equivalent: true is unreachable
+    without a real re-execution.
+    """
+    entry, _ = recorded
+    api = _api_with_entry(tmp_path, entry)
+
+    def broken(prompt: str):
+        raise RuntimeError("runtime unavailable in test")
+
+    monkeypatch.setattr(workbench_api, "_run_sealed_chat_turn", broken)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    assert response.status == 500
+    assert response.payload["error"]["code"] == "runtime_unavailable"
+    assert "data" not in response.payload or not response.payload.get("data")
+
+
+def test_replay_leaves_no_trace(tmp_path, recorded) -> None:
+    """Obligation 4: GET /replay writes nothing — journal and engine state."""
+    entry, _ = recorded
+    journal = TurnJournal(journal_dir=tmp_path / "workbench_data")
+    journal.append(entry)
+    api = WorkbenchApi(journal=journal)
+    journal_before = _snapshot(tmp_path / "workbench_data")
+    engine_state_before = _snapshot(_ENGINE_STATE_DIR)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    assert response.status == 200
+    assert _snapshot(tmp_path / "workbench_data") == journal_before
+    assert _snapshot(_ENGINE_STATE_DIR) == engine_state_before
+
+
+def test_wall_clock_divergence_does_not_break_equivalence(tmp_path, recorded) -> None:
+    """Obligation 5: timestamp/cost/digest differ freely; equivalence holds."""
+    entry, _ = recorded
+    aged = replace(
+        entry,
+        timestamp="1970-01-01T00:00:00+00:00",
+        turn_cost_ms=999_999,
+        journal_digest="sha256:not-a-real-digest",
+    )
+    api = _api_with_entry(tmp_path, aged)
+
+    response = api.handle("GET", "/replay/1", b"")
+
+    data = response.payload["data"]
+    assert data["equivalent"] is True
+    informational_paths = {d["path"] for d in data["divergences"]}
+    assert "timestamp" in informational_paths
+    assert "turn_cost_ms" in informational_paths
+
+
+def test_unknown_and_malformed_turn_ids_refuse_404(tmp_path, recorded) -> None:
+    entry, _ = recorded
+    api = _api_with_entry(tmp_path, entry)
+
+    for raw in ("999", "not-a-number", ""):
+        response = api.handle("GET", f"/replay/{raw}", b"")
+        assert response.status == 404
+        assert response.payload["error"]["code"] == "not_found"
+
+
+def test_sealed_runtime_construction_is_sealed() -> None:
+    """The replay executor's runtime has no engine-state store at all."""
+    runtime = ChatRuntime(no_load_state=True)
+    assert runtime._engine_state_store is None
+    # checkpoint_engine_state must be a no-op, not an error.
+    runtime.checkpoint_engine_state()
+
+
+def test_replay_turn_propagates_executor_failure(recorded) -> None:
+    """replay_turn itself never swallows execution failure into a result."""
+    entry, _ = recorded
+
+    def broken(prompt: str):
+        raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        replay_turn(entry, execute=broken)

workbench-ui/schema-snapshot.json

@@ -201,6 +201,22 @@
     "trace_hash",
     "grounding_source"
   ],
+  "TurnReplayComparison": [
+    "turn_id",
+    "comparison_basis",
+    "origin_state",
+    "original_trace_hash",
+    "replay_trace_hash",
+    "equivalent",
+    "replay_turn_cost_ms",
+    "divergences"
+  ],
+  "TurnReplayDivergence": [
+    "path",
+    "original",
+    "replay",
+    "severity"
+  ],
   "TurnVerdict": [
     "outcome",
     "runtime_detail"

workbench-ui/src/design/doctrine/schemaDrift.test.ts

@@ -28,6 +28,11 @@ const NOT_YET_MIRRORED = new Set([
   "RunDetail",
   "VaultSummary",
   "VaultEntry",
+  // Wave R3 sealed turn replay backend — TS mirrors land with the frontend
+  // Replay Moment PR (which also retires the W-026 artifact-keyed
+  // ReplayComparison/ReplayDivergence pair on both sides):
+  "TurnReplayComparison",
+  "TurnReplayDivergence",
 ]);
 
 const UI_ROOT = join(__dirname, "..", "..", "..");

workbench/api.py

@@ -20,6 +20,7 @@ from core.epistemic_state import (
 from workbench import readers
 from workbench.journal import DEFAULT_JOURNAL_DIR, TurnJournal, TurnJournalEntry
 from workbench.readers import ArtifactTooLargeError, EvidenceUnavailableError
+from workbench.replay import replay_turn
 from workbench.schemas import ChatTurnResult, MathRatifyResult, ProposalRef, TurnVerdict, error, ok
 
 
@@ -255,7 +256,25 @@ class WorkbenchApi:
             except FileNotFoundError:
                 return ApiResponse(404, error("not_found", f"trace turn not found: {turn_id}"))
         if method == "GET" and path.startswith("/replay/"):
-            return ApiResponse(501, error("unsupported", "route is deferred beyond W-026"))
+            raw_turn_id = unquote(path.removeprefix("/replay/"))
+            try:
+                turn_id = int(raw_turn_id)
+            except ValueError:
+                return ApiResponse(404, error("not_found", f"replay turn not found: {raw_turn_id}"))
+            try:
+                entry = self._journal.get_entry(turn_id)
+            except FileNotFoundError:
+                return ApiResponse(404, error("not_found", f"replay turn not found: {turn_id}"))
+            # Replay executes a real runtime turn; serialize with live chat
+            # turns the same way POST /chat/turn does.
+            with _CHAT_TURN_LOCK:
+                try:
+                    comparison = replay_turn(entry, execute=_run_sealed_chat_turn)
+                except Exception as exc:  # no comparison may be fabricated
+                    return ApiResponse(
+                        500, error("runtime_unavailable", f"replay failed: {exc}")
+                    )
+            return ApiResponse(200, ok(comparison))
         return ApiResponse(404, error("not_found", f"route not found: {method} {path}"))
 
     def _math_ratify(self, proposal_id: str, body: bytes) -> ApiResponse:
@@ -445,8 +464,19 @@ def _proposal_refs(runtime: ChatRuntime, before_ids: set[str]) -> list[ProposalR
     return refs
 
 
-def _run_chat_turn(prompt: str) -> ChatTurnResult:
-    runtime = ChatRuntime()
+def _run_sealed_chat_turn(prompt: str) -> ChatTurnResult:
+    """Replay executor: same envelope assembly, sealed runtime.
+
+    ``no_load_state=True`` nulls the engine-state store by construction —
+    no checkpoint load, ``checkpoint_engine_state`` no-ops, no proposal-log
+    lineage — so a replay can neither read nor leave runtime state.
+    """
+    return _run_chat_turn(prompt, runtime=ChatRuntime(no_load_state=True))
+
+
+def _run_chat_turn(prompt: str, runtime: ChatRuntime | None = None) -> ChatTurnResult:
+    if runtime is None:
+        runtime = ChatRuntime()
     before_candidate_ids = {
         str(getattr(candidate, "candidate_id", "") or "")
         for candidate in getattr(runtime, "_pending_candidates", ()) or ()

workbench/replay.py

@@ -0,0 +1,106 @@
+"""Sealed single-turn replay over the turn journal (Wave R3).
+
+``GET /replay/{turn_id}`` re-executes a journaled prompt in a sealed fresh
+runtime and compares the resulting envelope leaf-by-leaf against the
+recorded :class:`~workbench.journal.TurnJournalEntry`.
+
+The claim demonstrated is exactly the architectural one: same prompt, same
+genesis substrate -> bit-identical envelope.  The original turn ran in its
+own fresh ``ChatRuntime()`` that may have loaded an engine-state checkpoint
+present at the time; the journal does not record whether one existed, so
+``origin_state`` is reported as ``"unrecorded"`` and a divergence means
+nondeterminism OR origin-state influence — never claimed to be one or the
+other.
+
+This module owns classification and comparison only.  Execution is
+injected by the caller (``workbench.api`` passes its own chat-turn
+executor over a sealed runtime), which keeps the comparison pure and the
+no-fabricated-equivalence obligation testable: if the executor raises,
+no comparison object exists.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import fields, replace
+from typing import Callable
+
+from workbench.journal import TurnJournalEntry
+from workbench.schemas import (
+    ChatTurnResult,
+    TurnReplayComparison,
+    TurnReplayDivergence,
+)
+
+# Every TurnJournalEntry field must appear in exactly one of these sets —
+# enforced by tests so a future journal field forces an explicit
+# classification decision instead of silently defaulting.
+CRITICAL_FIELDS = frozenset(
+    {
+        "turn_id",
+        "prompt",
+        "surface",
+        "articulation_surface",
+        "walk_surface",
+        "trace_hash",
+        "grounding_source",
+        "epistemic_state",
+        "normative_clearance",
+        "verdicts",
+        "refusal_emitted",
+        "hedge_injected",
+        "proposal_candidates",
+        "checkpoint_emitted",
+    }
+)
+# Wall-clock by nature, or derived over wall-clock bytes (journal_digest
+# hashes the timestamp): expected to differ on every replay and never
+# evidence against equivalence.
+INFORMATIONAL_FIELDS = frozenset({"timestamp", "turn_cost_ms", "journal_digest"})
+
+
+def replay_turn(
+    entry: TurnJournalEntry,
+    *,
+    execute: Callable[[str], ChatTurnResult],
+) -> TurnReplayComparison:
+    """Re-execute ``entry.prompt`` via ``execute`` and compare envelopes.
+
+    ``execute`` must run the prompt through the same envelope-assembly path
+    that produced the recorded entry, over a sealed runtime.  This function
+    never fabricates a comparison: if ``execute`` raises, the exception
+    propagates and no ``TurnReplayComparison`` exists.
+    """
+    started = time.perf_counter()
+    result = execute(entry.prompt)
+    elapsed_ms = max(0, int(round((time.perf_counter() - started) * 1000)))
+    result = replace(result, turn_cost_ms=elapsed_ms, turn_id=entry.turn_id)
+    replayed = TurnJournalEntry.from_chat_turn(result, turn_id=entry.turn_id)
+
+    divergences: list[TurnReplayDivergence] = []
+    for spec in fields(TurnJournalEntry):
+        original_value = getattr(entry, spec.name)
+        replay_value = getattr(replayed, spec.name)
+        if original_value == replay_value:
+            continue
+        divergences.append(
+            TurnReplayDivergence(
+                path=spec.name,
+                original=original_value,
+                replay=replay_value,
+                severity=(
+                    "critical" if spec.name in CRITICAL_FIELDS else "informational"
+                ),
+            )
+        )
+
+    return TurnReplayComparison(
+        turn_id=entry.turn_id,
+        comparison_basis="sealed_fresh_runtime_single_turn",
+        origin_state="unrecorded",
+        original_trace_hash=entry.trace_hash,
+        replay_trace_hash=replayed.trace_hash,
+        equivalent=not any(d.severity == "critical" for d in divergences),
+        replay_turn_cost_ms=elapsed_ms,
+        divergences=divergences,
+    )

workbench/schemas.py

@@ -231,6 +231,45 @@ class ReplayComparison:
     divergences: list[ReplayDivergence] = field(default_factory=list)
 
 
+# ---------------------------------------------------------------------------
+# Wave R3 — sealed single-turn replay over the turn journal.
+# Scoping: docs/analysis/replay-moment-backend-scoping-2026-06-12.md.
+# The W-026 artifact-keyed pair above has no live consumer and is retired
+# when the frontend Replay Moment re-points to this turn-keyed shape.
+# ---------------------------------------------------------------------------
+
+TurnReplayDivergenceSeverity = Literal["critical", "informational"]
+# The only basis implemented: a fresh ChatRuntime(no_load_state=True) —
+# genesis substrate, no checkpoint load, no checkpoint write, no proposal
+# lineage — re-executes the recorded prompt once.
+TurnReplayBasis = Literal["sealed_fresh_runtime_single_turn"]
+# The journal does not record whether an engine-state checkpoint existed
+# when the original turn ran, so the origin state is honestly unrecorded:
+# a divergence means nondeterminism OR origin-state influence, and the
+# response must never claim to distinguish them.
+TurnReplayOriginState = Literal["unrecorded"]
+
+
+@dataclass(frozen=True, slots=True)
+class TurnReplayDivergence:
+    path: str
+    original: Any
+    replay: Any
+    severity: TurnReplayDivergenceSeverity
+
+
+@dataclass(frozen=True, slots=True)
+class TurnReplayComparison:
+    turn_id: int
+    comparison_basis: TurnReplayBasis
+    origin_state: TurnReplayOriginState
+    original_trace_hash: str | None
+    replay_trace_hash: str | None
+    equivalent: bool
+    replay_turn_cost_ms: int
+    divergences: list[TurnReplayDivergence] = field(default_factory=list)
+
+
 # ---------------------------------------------------------------------------
 # ADR-0172 W4 — Math proposal schemas
 # ---------------------------------------------------------------------------