From b96c5567010bb04f72185d81ea24bae1e4aa45a4 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 23 Jun 2026 15:41:13 -0700 Subject: [PATCH] feat(workbench): add sealed practice evidence read model (#899) * feat(workbench): add practice evidence read model * feat(workbench): add practice evidence endpoint seam * feat(workbench): persist practice evidence field when present * test(workbench): cover practice evidence read model seam --- tests/test_workbench_practice_evidence.py | 188 ++++++++++ workbench/journal.py | 2 + workbench/practice_endpoint.py | 57 +++ workbench/practice_evidence.py | 405 ++++++++++++++++++++++ 4 files changed, 652 insertions(+) create mode 100644 tests/test_workbench_practice_evidence.py create mode 100644 workbench/practice_endpoint.py create mode 100644 workbench/practice_evidence.py diff --git a/tests/test_workbench_practice_evidence.py b/tests/test_workbench_practice_evidence.py new file mode 100644 index 00000000..7dc425f9 --- /dev/null +++ b/tests/test_workbench_practice_evidence.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from workbench.practice_endpoint import ( + practice_evidence_response, + practice_turn_id_from_path, +) +from workbench.practice_evidence import ( + PRACTICE_EVIDENCE_ABSENT, + practice_evidence_from_journal_entry, +) + + +@dataclass(frozen=True, slots=True) +class _Entry: + turn_id: int + practice_evidence: Any | None = None + + +class _Journal: + def __init__(self, entries: dict[int, Any]) -> None: + self._entries = entries + + def get_entry(self, turn_id: int) -> Any: + try: + return self._entries[turn_id] + except KeyError: + raise FileNotFoundError(str(turn_id)) from None + + +def _sealed_payload() -> dict[str, Any]: + return { + "trace_id": "trace-a", + "trace_policy_version": "sealed_practice_trace.v1", + "input_digest": "i" * 64, + "problem_frame_digest": "p" * 64, + "original_contract_assessment_id": "assessment-a", + "residual_ids": ["residual-a"], + "search_gate_decision_id": "g" * 64, + "compute_budget_id": "b" * 64, + "geometric_search_run_id": "r" * 64, + "candidate_attempt_ids": ["attempt-a"], + "candidate_attempt_binding_ids": ["binding-a"], + "replay_result_ids": ["replay-a"], + "replay_refusal_ids": [], + "upstream_identity_chain": ["p" * 64, "assessment-a", "residual-a"], + "practice_disposition": "sealed_candidate_replay_closed", + "trace_records": ["p" * 64, "assessment-a", "residual-a"], + "evidence_spans": [ + {"text": "delta", "start": 0, "end": 5, "sentence_index": 0} + ], + "created_by_policy": "sealed_practice_trace.v1", + "explanation": "sealed practice trace prose", + } + + +def _trace_refusal_payload() -> dict[str, Any]: + return { + "trace_refusal_id": "refusal-a", + "trace_policy_version": "sealed_practice_trace.v1", + "input_digest": None, + "practice_disposition": "trace_invalid_input", + "reason_codes": ["invalid_run_type"], + "explanation": "Practice trace refused.", + } + + +def test_practice_evidence_from_journal_entry_returns_missing_for_legacy_turn() -> None: + evidence = practice_evidence_from_journal_entry(_Entry(turn_id=7)) + + assert evidence.schema_version == "practice_evidence_v1" + assert evidence.turn_id == 7 + assert evidence.status == "missing_evidence" + assert evidence.missing_reason == PRACTICE_EVIDENCE_ABSENT + assert evidence.chain == [] + assert evidence.diagnostic_only is True + assert evidence.serving_allowed is False + assert evidence.mutation_allowed is False + assert evidence.replay_execution_allowed is False + assert evidence.replay_executed_by_workbench is False + + +def test_practice_evidence_projects_raw_sealed_trace_payload_read_only() -> None: + evidence = practice_evidence_from_journal_entry( + _Entry(turn_id=7, practice_evidence=_sealed_payload()) + ) + + assert evidence.status == "recorded" + assert evidence.record_kind == "sealed_trace" + assert evidence.practice_disposition == "sealed_candidate_replay_closed" + assert evidence.sealed_trace is not None + assert evidence.sealed_trace.trace_id == "trace-a" + assert evidence.sealed_trace.evidence_spans[0].text == "delta" + assert evidence.trace_refusal is None + assert evidence.serving_allowed is False + assert evidence.mutation_allowed is False + assert evidence.replay_execution_allowed is False + assert [card.kind for card in evidence.chain] == [ + "problem_frame", + "contract_assessment", + "residuals", + "search_gate", + "compute_budget", + "geometric_search_run", + "candidate_attempts", + "attempt_bindings", + "replay_results", + "replay_refusals", + "sealed_trace", + ] + assert evidence.chain[5].summary == ( + "Geometric search run identity; Workbench does not execute search." + ) + assert evidence.chain[8].refs == ["replay-a"] + assert evidence.chain[9].status == "missing_evidence" + + +def test_practice_evidence_projects_raw_trace_refusal_payload_read_only() -> None: + evidence = practice_evidence_from_journal_entry( + _Entry(turn_id=8, practice_evidence=_trace_refusal_payload()) + ) + + assert evidence.status == "recorded" + assert evidence.record_kind == "trace_refusal" + assert evidence.practice_disposition == "trace_invalid_input" + assert evidence.sealed_trace is None + assert evidence.trace_refusal is not None + assert evidence.trace_refusal.reason_codes == ["invalid_run_type"] + assert evidence.chain[0].kind == "trace_refusal" + assert evidence.serving_allowed is False + assert evidence.mutation_allowed is False + + +def test_practice_evidence_response_returns_missing_evidence_for_legacy_turn() -> None: + response = practice_evidence_response(_Journal({7: _Entry(turn_id=7)}), "7") + + assert response.status == 200 + assert response.payload["ok"] is True + data = response.payload["data"] + assert data["schema_version"] == "practice_evidence_v1" + assert data["turn_id"] == 7 + assert data["status"] == "missing_evidence" + assert data["diagnostic_only"] is True + assert data["serving_allowed"] is False + assert data["mutation_allowed"] is False + assert data["replay_execution_allowed"] is False + + +def test_practice_evidence_response_projects_recorded_payload() -> None: + response = practice_evidence_response( + _Journal({7: _Entry(turn_id=7, practice_evidence=_sealed_payload())}), + "7", + ) + + assert response.status == 200 + data = response.payload["data"] + assert data["status"] == "recorded" + assert data["record_kind"] == "sealed_trace" + assert data["sealed_trace"]["trace_id"] == "trace-a" + assert data["chain"][-1]["kind"] == "sealed_trace" + + +def test_practice_evidence_response_returns_404_for_bad_turn_id() -> None: + response = practice_evidence_response(_Journal({}), "not-an-int") + + assert response.status == 404 + assert response.payload["ok"] is False + assert response.payload["error"]["code"] == "not_found" + assert "not-an-int" in response.payload["error"]["message"] + + +def test_practice_evidence_response_returns_404_for_missing_turn() -> None: + response = practice_evidence_response(_Journal({}), "8") + + assert response.status == 404 + assert response.payload["ok"] is False + assert response.payload["error"]["code"] == "not_found" + assert "8" in response.payload["error"]["message"] + + +def test_practice_turn_id_from_path_matches_only_practice_route() -> None: + assert practice_turn_id_from_path("/trace/7/practice") == "7" + assert practice_turn_id_from_path("/trace/7/construction") is None + assert practice_turn_id_from_path("/trace/7/pipeline") is None + assert practice_turn_id_from_path("/trace/7") is None + assert practice_turn_id_from_path("/trace//practice") is None diff --git a/workbench/journal.py b/workbench/journal.py index 42212d01..677dc01a 100644 --- a/workbench/journal.py +++ b/workbench/journal.py @@ -60,6 +60,7 @@ class TurnJournalEntry: pipeline_record: CognitivePipelineRecord | dict[str, Any] | None = None field_evidence: FieldEvidence | dict[str, Any] | None = None construction_evidence: ConstructionEvidence | dict[str, Any] | None = None + practice_evidence: dict[str, Any] | None = None trace_integrity: TraceIntegrity | None = None journal_digest: str = "" @@ -102,6 +103,7 @@ class TurnJournalEntry: pipeline_record=to_data(result.pipeline_record), field_evidence=to_data(result.field_evidence), construction_evidence=to_data(result.construction_evidence), + practice_evidence=to_data(getattr(result, "practice_evidence", None)), trace_integrity=_trace_integrity_for_hash(result.trace_hash), ) diff --git a/workbench/practice_endpoint.py b/workbench/practice_endpoint.py new file mode 100644 index 00000000..a8d16d51 --- /dev/null +++ b/workbench/practice_endpoint.py @@ -0,0 +1,57 @@ +"""Narrow sealed-practice evidence endpoint seam for Workbench. + +The helper is read-only and projects only persisted practice evidence through +`practice_evidence_from_journal_entry`. It never runs geometric search, replay, +operator execution, or trace sealing. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from workbench.practice_evidence import practice_evidence_from_journal_entry +from workbench.schemas import error, ok + + +@dataclass(frozen=True, slots=True) +class PracticeEndpointResponse: + status: int + payload: dict[str, Any] + + +def practice_evidence_response( + journal: Any, + raw_turn_id: str, +) -> PracticeEndpointResponse: + """Return JSON-envelope payload for `/trace//practice`.""" + + try: + turn_id = int(raw_turn_id) + except ValueError: + return PracticeEndpointResponse( + 404, + error("not_found", f"trace practice not found: {raw_turn_id}"), + ) + + try: + entry = journal.get_entry(turn_id) + except FileNotFoundError: + return PracticeEndpointResponse( + 404, + error("not_found", f"trace practice not found: {turn_id}"), + ) + + return PracticeEndpointResponse( + 200, + ok(practice_evidence_from_journal_entry(entry)), + ) + + +def practice_turn_id_from_path(path: str) -> str | None: + """Extract raw turn id from `/trace//practice`.""" + + if not (path.startswith("/trace/") and path.endswith("/practice")): + return None + raw = path.removeprefix("/trace/").removesuffix("/practice").strip("/") + return raw or None diff --git a/workbench/practice_evidence.py b/workbench/practice_evidence.py new file mode 100644 index 00000000..25421df6 --- /dev/null +++ b/workbench/practice_evidence.py @@ -0,0 +1,405 @@ +"""Read-only sealed-practice evidence projection helpers for Workbench. + +This module is intentionally inert. It projects already-persisted practice evidence +for a turn into a Workbench-facing read model. It does not run geometric search, +execute candidate operators, run replay, seal traces, mutate journals, or grant +serving authority. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Literal + +PipelineEvidenceStatus = Literal["recorded", "missing_evidence"] +PracticeRecordKind = Literal["sealed_trace", "trace_refusal"] +PracticeCardKind = Literal[ + "problem_frame", + "contract_assessment", + "residuals", + "search_gate", + "compute_budget", + "geometric_search_run", + "candidate_attempts", + "attempt_bindings", + "replay_results", + "replay_refusals", + "sealed_trace", + "trace_refusal", +] + +PRACTICE_EVIDENCE_ABSENT = "sealed practice evidence was not persisted for this turn" + + +@dataclass(frozen=True, slots=True) +class PracticeSourceSpanView: + text: str + start: int + end: int + sentence_index: int | None = None + + +@dataclass(frozen=True, slots=True) +class PracticeEvidenceCard: + """A read-only upstream identity card. + + `refs` are identity references only. They are not executable handles, mutation + handles, or proof of serving authority. + """ + + kind: PracticeCardKind + status: PipelineEvidenceStatus + refs: list[str] = field(default_factory=list) + summary: str = "" + + +@dataclass(frozen=True, slots=True) +class SealedPracticeTraceView: + trace_id: str + trace_policy_version: str + input_digest: str + problem_frame_digest: str + original_contract_assessment_id: str + residual_ids: list[str] + search_gate_decision_id: str + compute_budget_id: str + geometric_search_run_id: str + candidate_attempt_ids: list[str] + candidate_attempt_binding_ids: list[str] + replay_result_ids: list[str] + replay_refusal_ids: list[str] + upstream_identity_chain: list[str] + practice_disposition: str + trace_records: list[str] + evidence_spans: list[PracticeSourceSpanView] = field(default_factory=list) + created_by_policy: str = "" + explanation: str = "" + + +@dataclass(frozen=True, slots=True) +class PracticeTraceRefusalView: + trace_refusal_id: str + trace_policy_version: str + input_digest: str | None + practice_disposition: str + reason_codes: list[str] + explanation: str = "" + + +@dataclass(frozen=True, slots=True) +class PracticeEvidence: + """Workbench-facing sealed-practice evidence for one turn. + + This record is a projection only. The authority fields are deliberately + load-bearing: Workbench may inspect this evidence, but this endpoint never + runs replay, mutates journals, executes operators, or allows serving. + """ + + schema_version: Literal["practice_evidence_v1"] + turn_id: int + status: PipelineEvidenceStatus + missing_reason: str | None + record_kind: PracticeRecordKind | None + practice_disposition: str | None + chain: list[PracticeEvidenceCard] = field(default_factory=list) + sealed_trace: SealedPracticeTraceView | None = None + trace_refusal: PracticeTraceRefusalView | None = None + diagnostic_only: bool = True + serving_allowed: bool = False + mutation_allowed: bool = False + replay_execution_allowed: bool = False + replay_executed_by_workbench: bool = False + + +def missing_practice_evidence(turn_id: int, reason: str) -> PracticeEvidence: + """Return the honest absence state for legacy turns. + + Absence is not an error and not a failed proof. It means the selected turn has + no persisted sealed-practice evidence to project. + """ + + return PracticeEvidence( + schema_version="practice_evidence_v1", + turn_id=turn_id, + status="missing_evidence", + missing_reason=reason, + record_kind=None, + practice_disposition=None, + chain=[], + sealed_trace=None, + trace_refusal=None, + diagnostic_only=True, + serving_allowed=False, + mutation_allowed=False, + replay_execution_allowed=False, + replay_executed_by_workbench=False, + ) + + +def practice_evidence_from_journal_entry(entry: Any) -> PracticeEvidence: + """Project persisted sealed-practice evidence from a journal entry if present.""" + + turn_id = int(getattr(entry, "turn_id")) + payload = getattr(entry, "practice_evidence", None) + if payload is None: + return missing_practice_evidence(turn_id, PRACTICE_EVIDENCE_ABSENT) + + if isinstance(payload, PracticeEvidence): + return payload + + try: + return _practice_evidence_from_payload(turn_id, payload) + except (KeyError, TypeError, ValueError) as exc: + return missing_practice_evidence( + turn_id, + f"practice evidence payload projection failed: {exc}", + ) + + +def _practice_evidence_from_payload(turn_id: int, payload: Any) -> PracticeEvidence: + schema_version = _field(payload, "schema_version") + if schema_version == "practice_evidence_v1": + status = _field(payload, "status") + if status == "missing_evidence": + return missing_practice_evidence( + turn_id, + _field(payload, "missing_reason") or PRACTICE_EVIDENCE_ABSENT, + ) + if status != "recorded": + return missing_practice_evidence(turn_id, f"unsupported status: {status}") + sealed_payload = _field(payload, "sealed_trace") + refusal_payload = _field(payload, "trace_refusal") + elif _field(payload, "trace_id") is not None: + sealed_payload = payload + refusal_payload = None + elif _field(payload, "trace_refusal_id") is not None: + sealed_payload = None + refusal_payload = payload + else: + return missing_practice_evidence( + turn_id, + "practice evidence payload has unsupported shape", + ) + + sealed_trace = _sealed_trace_view(sealed_payload) if sealed_payload is not None else None + trace_refusal = _trace_refusal_view(refusal_payload) if refusal_payload is not None else None + if sealed_trace is not None and trace_refusal is not None: + return missing_practice_evidence( + turn_id, + "practice evidence payload cannot contain both sealed trace and trace refusal", + ) + if sealed_trace is None and trace_refusal is None: + return missing_practice_evidence( + turn_id, + "recorded practice evidence missing sealed trace/refusal payload", + ) + + return _recorded_practice_evidence( + turn_id=turn_id, + sealed_trace=sealed_trace, + trace_refusal=trace_refusal, + ) + + +def _recorded_practice_evidence( + *, + turn_id: int, + sealed_trace: SealedPracticeTraceView | None, + trace_refusal: PracticeTraceRefusalView | None, +) -> PracticeEvidence: + if sealed_trace is not None: + return PracticeEvidence( + schema_version="practice_evidence_v1", + turn_id=turn_id, + status="recorded", + missing_reason=None, + record_kind="sealed_trace", + practice_disposition=sealed_trace.practice_disposition, + chain=_sealed_trace_chain(sealed_trace), + sealed_trace=sealed_trace, + trace_refusal=None, + diagnostic_only=True, + serving_allowed=False, + mutation_allowed=False, + replay_execution_allowed=False, + replay_executed_by_workbench=False, + ) + assert trace_refusal is not None + return PracticeEvidence( + schema_version="practice_evidence_v1", + turn_id=turn_id, + status="recorded", + missing_reason=None, + record_kind="trace_refusal", + practice_disposition=trace_refusal.practice_disposition, + chain=_trace_refusal_chain(trace_refusal), + sealed_trace=None, + trace_refusal=trace_refusal, + diagnostic_only=True, + serving_allowed=False, + mutation_allowed=False, + replay_execution_allowed=False, + replay_executed_by_workbench=False, + ) + + +def _sealed_trace_view(payload: Any) -> SealedPracticeTraceView: + return SealedPracticeTraceView( + trace_id=str(_required(payload, "trace_id")), + trace_policy_version=str(_required(payload, "trace_policy_version")), + input_digest=str(_required(payload, "input_digest")), + problem_frame_digest=str(_required(payload, "problem_frame_digest")), + original_contract_assessment_id=str( + _required(payload, "original_contract_assessment_id") + ), + residual_ids=_str_list(_field(payload, "residual_ids", [])), + search_gate_decision_id=str(_required(payload, "search_gate_decision_id")), + compute_budget_id=str(_required(payload, "compute_budget_id")), + geometric_search_run_id=str(_required(payload, "geometric_search_run_id")), + candidate_attempt_ids=_str_list(_field(payload, "candidate_attempt_ids", [])), + candidate_attempt_binding_ids=_str_list( + _field(payload, "candidate_attempt_binding_ids", []) + ), + replay_result_ids=_str_list(_field(payload, "replay_result_ids", [])), + replay_refusal_ids=_str_list(_field(payload, "replay_refusal_ids", [])), + upstream_identity_chain=_str_list(_field(payload, "upstream_identity_chain", [])), + practice_disposition=str(_enum_value(_required(payload, "practice_disposition"))), + trace_records=_str_list(_field(payload, "trace_records", [])), + evidence_spans=[_span_view(span) for span in _as_list(_field(payload, "evidence_spans", []))], + created_by_policy=str(_field(payload, "created_by_policy", "")), + explanation=str(_field(payload, "explanation", "")), + ) + + +def _trace_refusal_view(payload: Any) -> PracticeTraceRefusalView: + input_digest = _field(payload, "input_digest") + return PracticeTraceRefusalView( + trace_refusal_id=str(_required(payload, "trace_refusal_id")), + trace_policy_version=str(_required(payload, "trace_policy_version")), + input_digest=None if input_digest is None else str(input_digest), + practice_disposition=str(_enum_value(_required(payload, "practice_disposition"))), + reason_codes=_str_list(_field(payload, "reason_codes", [])), + explanation=str(_field(payload, "explanation", "")), + ) + + +def _sealed_trace_chain(trace: SealedPracticeTraceView) -> list[PracticeEvidenceCard]: + return [ + PracticeEvidenceCard( + kind="problem_frame", + status="recorded", + refs=[trace.problem_frame_digest], + summary="Problem frame digest bound into the sealed practice trace.", + ), + PracticeEvidenceCard( + kind="contract_assessment", + status="recorded", + refs=[trace.original_contract_assessment_id], + summary="Original contract assessment identity recorded before practice.", + ), + PracticeEvidenceCard( + kind="residuals", + status="recorded" if trace.residual_ids else "missing_evidence", + refs=trace.residual_ids, + summary="Residual identities admitted as the practice target.", + ), + PracticeEvidenceCard( + kind="search_gate", + status="recorded", + refs=[trace.search_gate_decision_id], + summary="Search gate decision identity; Workbench does not rerun the gate.", + ), + PracticeEvidenceCard( + kind="compute_budget", + status="recorded", + refs=[trace.compute_budget_id], + summary="Compute budget decision identity; Workbench does not allocate budget.", + ), + PracticeEvidenceCard( + kind="geometric_search_run", + status="recorded", + refs=[trace.geometric_search_run_id], + summary="Geometric search run identity; Workbench does not execute search.", + ), + PracticeEvidenceCard( + kind="candidate_attempts", + status="recorded" if trace.candidate_attempt_ids else "missing_evidence", + refs=trace.candidate_attempt_ids, + summary="Candidate attempt identities recorded upstream.", + ), + PracticeEvidenceCard( + kind="attempt_bindings", + status="recorded" if trace.candidate_attempt_binding_ids else "missing_evidence", + refs=trace.candidate_attempt_binding_ids, + summary="Candidate attempt/run binding identities recorded upstream.", + ), + PracticeEvidenceCard( + kind="replay_results", + status="recorded" if trace.replay_result_ids else "missing_evidence", + refs=trace.replay_result_ids, + summary="Replay adapter result identities; Workbench does not run replay here.", + ), + PracticeEvidenceCard( + kind="replay_refusals", + status="recorded" if trace.replay_refusal_ids else "missing_evidence", + refs=trace.replay_refusal_ids, + summary="Replay adapter refusal identities; Workbench does not retry replay here.", + ), + PracticeEvidenceCard( + kind="sealed_trace", + status="recorded", + refs=[trace.trace_id], + summary="Sealed practice trace identity and disposition.", + ), + ] + + +def _trace_refusal_chain(refusal: PracticeTraceRefusalView) -> list[PracticeEvidenceCard]: + refs = [refusal.trace_refusal_id] + return [ + PracticeEvidenceCard( + kind="trace_refusal", + status="recorded", + refs=refs, + summary="Practice trace refused before a sealed practice trace could be projected.", + ) + ] + + +def _span_view(payload: Any) -> PracticeSourceSpanView: + sentence_index = _field(payload, "sentence_index") + return PracticeSourceSpanView( + text=str(_required(payload, "text")), + start=int(_required(payload, "start")), + end=int(_required(payload, "end")), + sentence_index=None if sentence_index is None else int(sentence_index), + ) + + +def _field(payload: Any, name: str, default: Any = None) -> Any: + if isinstance(payload, dict): + return payload.get(name, default) + return getattr(payload, name, default) + + +def _required(payload: Any, name: str) -> Any: + value = _field(payload, name) + if value is None: + raise KeyError(name) + return value + + +def _enum_value(value: Any) -> Any: + return getattr(value, "value", value) + + +def _as_list(value: Any) -> list[Any]: + if isinstance(value, list): + return value + if isinstance(value, tuple): + return list(value) + return [] + + +def _str_list(value: Any) -> list[str]: + return [str(_enum_value(item)) for item in _as_list(value)]