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
This commit is contained in:
Shay 2026-06-23 15:41:13 -07:00 committed by GitHub
parent ff1ef10bd7
commit b96c556701
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 652 additions and 0 deletions

View file

@ -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

View file

@ -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),
)

View file

@ -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/<turn_id>/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/<turn_id>/practice`."""
if not (path.startswith("/trace/") and path.endswith("/practice")):
return None
raw = path.removeprefix("/trace/").removesuffix("/practice").strip("/")
return raw or None

View file

@ -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)]