diff --git a/docs/analysis/replay-moment-backend-scoping-2026-06-12.md b/docs/analysis/replay-moment-backend-scoping-2026-06-12.md index 58d47712..eb01573f 100644 --- a/docs/analysis/replay-moment-backend-scoping-2026-06-12.md +++ b/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 diff --git a/docs/workbench/api-contract-v1.md b/docs/workbench/api-contract-v1.md index 132033e4..9f0094bc 100644 --- a/docs/workbench/api-contract-v1.md +++ b/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" } ] } diff --git a/tests/test_workbench_api.py b/tests/test_workbench_api.py index f00dbc08..06ef4444 100644 --- a/tests/test_workbench_api.py +++ b/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: diff --git a/tests/test_workbench_replay.py b/tests/test_workbench_replay.py new file mode 100644 index 00000000..7753e618 --- /dev/null +++ b/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) diff --git a/workbench-ui/schema-snapshot.json b/workbench-ui/schema-snapshot.json index 497d0ac6..b6f97ed1 100644 --- a/workbench-ui/schema-snapshot.json +++ b/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" diff --git a/workbench-ui/src/design/doctrine/schemaDrift.test.ts b/workbench-ui/src/design/doctrine/schemaDrift.test.ts index eb69b71e..ceae7dd1 100644 --- a/workbench-ui/src/design/doctrine/schemaDrift.test.ts +++ b/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, "..", "..", ".."); diff --git a/workbench/api.py b/workbench/api.py index 28360363..562ef213 100644 --- a/workbench/api.py +++ b/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 () diff --git a/workbench/replay.py b/workbench/replay.py new file mode 100644 index 00000000..d2af3d83 --- /dev/null +++ b/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, + ) diff --git a/workbench/schemas.py b/workbench/schemas.py index ccc6c025..5ba612c2 100644 --- a/workbench/schemas.py +++ b/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 # ---------------------------------------------------------------------------