Merge branch 'main' into feat/workbench-practice-ui-scaffold

This commit is contained in:
Shay 2026-06-23 16:02:38 -07:00 committed by GitHub
commit feacced7b6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 158 additions and 13 deletions

View file

@ -0,0 +1,122 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from workbench.api import WorkbenchApi
@dataclass(frozen=True, slots=True)
class _Entry:
turn_id: int
practice_evidence: Any | None = None
construction_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 _request(journal: _Journal, path: str):
return WorkbenchApi(journal=journal).handle("GET", path)
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 test_trace_practice_route_returns_missing_evidence_for_legacy_entry() -> None:
response = _request(_Journal({7: _Entry(turn_id=7)}), "/trace/7/practice")
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["record_kind"] == "none"
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
assert data["replay_executed_by_workbench"] is False
def test_trace_practice_route_projects_recorded_payload_read_only() -> None:
response = _request(
_Journal({7: _Entry(turn_id=7, practice_evidence=_sealed_payload())}),
"/trace/7/practice",
)
assert response.status == 200
assert response.payload["ok"] is True
data = response.payload["data"]
assert data["schema_version"] == "practice_evidence_v1"
assert data["status"] == "recorded"
assert data["record_kind"] == "sealed_trace"
assert data["sealed_trace"]["trace_id"] == "trace-a"
assert data["practice_disposition"] == "sealed_candidate_replay_closed"
assert data["chain"][-1]["kind"] == "sealed_trace"
assert data["chain"][5]["summary"] == (
"Geometric search run identity; Workbench does not execute search."
)
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
assert data["replay_executed_by_workbench"] is False
def test_trace_practice_route_returns_404_for_bad_or_missing_turn() -> None:
bad = _request(_Journal({}), "/trace/not-an-int/practice")
assert bad.status == 404
assert bad.payload["ok"] is False
assert bad.payload["error"]["code"] == "not_found"
assert "trace practice not found: not-an-int" in bad.payload["error"]["message"]
missing = _request(_Journal({}), "/trace/99/practice")
assert missing.status == 404
assert missing.payload["ok"] is False
assert missing.payload["error"]["code"] == "not_found"
assert "trace practice not found: 99" in missing.payload["error"]["message"]
def test_trace_practice_route_does_not_steal_construction_route() -> None:
journal = _Journal({7: _Entry(turn_id=7)})
construction = _request(journal, "/trace/7/construction")
practice = _request(journal, "/trace/7/practice")
assert construction.status == 200
assert construction.payload["data"]["schema_version"] == "construction_evidence_v1"
assert practice.status == 200
assert practice.payload["data"]["schema_version"] == "practice_evidence_v1"

View file

@ -1,13 +1,18 @@
"""Narrow construction-evidence endpoint seam for Workbench.
This helper keeps the endpoint behavior testable without editing the large
`workbench.api` dispatch table from environments that cannot run local syntax and
route tests. The final API wiring is intentionally one line of dispatch:
This helper keeps trace subroute endpoint behavior testable without editing the
large `workbench.api` dispatch table from environments that cannot run local
syntax and route tests. The final API wiring is intentionally one line of
dispatch:
return construction_evidence_response(self._journal, raw_turn_id)
The helper itself is read-only and projects only persisted construction evidence
through `construction_evidence_from_journal_entry`.
through `construction_evidence_from_journal_entry`. It also recognizes the
adjacent `/trace/<turn_id>/practice` subroute as a connector-safe adapter until
the central dispatch table is renamed to a generic trace-subroute helper. That
practice path delegates to `workbench.practice_endpoint` and never runs search,
operators, replay, sealing, or mutation.
"""
from __future__ import annotations
@ -16,8 +21,11 @@ from dataclasses import dataclass
from typing import Any
from workbench.construction_evidence import construction_evidence_from_journal_entry
from workbench.practice_endpoint import practice_evidence_response
from workbench.schemas import error, ok
_PRACTICE_ROUTE_PREFIX = "practice:"
@dataclass(frozen=True, slots=True)
class ConstructionEndpointResponse:
@ -29,13 +37,25 @@ def construction_evidence_response(
journal: Any,
raw_turn_id: str,
) -> ConstructionEndpointResponse:
"""Return JSON-envelope payload for `/trace/<turn_id>/construction`.
"""Return JSON-envelope payload for trace evidence subroutes.
The function is deliberately free of parsing/reconstruction side effects. It
reads a journal entry and projects construction evidence when already
persisted; legacy entries receive a typed `missing_evidence` data payload.
For normal `/trace/<turn_id>/construction`, this function is deliberately
free of parsing/reconstruction side effects. It reads a journal entry and
projects construction evidence when already persisted; legacy entries receive
a typed `missing_evidence` data payload.
For the connector-safe `/trace/<turn_id>/practice` adapter marker produced by
`construction_turn_id_from_path`, this delegates to the sealed-practice
endpoint seam and wraps the response in the same status/payload envelope.
"""
if raw_turn_id.startswith(_PRACTICE_ROUTE_PREFIX):
practice = practice_evidence_response(
journal,
raw_turn_id.removeprefix(_PRACTICE_ROUTE_PREFIX),
)
return ConstructionEndpointResponse(practice.status, practice.payload)
try:
turn_id = int(raw_turn_id)
except ValueError:
@ -59,13 +79,16 @@ def construction_evidence_response(
def construction_turn_id_from_path(path: str) -> str | None:
"""Extract raw turn id from `/trace/<turn_id>/construction`.
"""Extract raw turn id from supported trace evidence subroutes.
Returns None for non-matching paths so the main dispatch table can remain
explicit and order-safe.
"""
if not (path.startswith("/trace/") and path.endswith("/construction")):
return None
raw = path.removeprefix("/trace/").removesuffix("/construction").strip("/")
return raw or None
if path.startswith("/trace/") and path.endswith("/construction"):
raw = path.removeprefix("/trace/").removesuffix("/construction").strip("/")
return raw or None
if path.startswith("/trace/") and path.endswith("/practice"):
raw = path.removeprefix("/trace/").removesuffix("/practice").strip("/")
return f"{_PRACTICE_ROUTE_PREFIX}{raw}" if raw else None
return None