From f177efc837aa5b33d02c57966c6c43111da51ad0 Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 12 Jun 2026 12:49:41 -0700 Subject: [PATCH] feat(workbench): add R2 backend read substrate --- docs/workbench/api-contract-v1.md | 173 +++++++ docs/workbench/data-shapes-v1.md | 105 ++++ tests/test_workbench_r2_read_substrate.py | 261 ++++++++++ workbench/api.py | 85 +++- workbench/journal.py | 8 + workbench/readers.py | 568 +++++++++++++++++++++- workbench/schemas.py | 96 ++++ 7 files changed, 1291 insertions(+), 5 deletions(-) create mode 100644 tests/test_workbench_r2_read_substrate.py diff --git a/docs/workbench/api-contract-v1.md b/docs/workbench/api-contract-v1.md index aea423dc..132033e4 100644 --- a/docs/workbench/api-contract-v1.md +++ b/docs/workbench/api-contract-v1.md @@ -52,6 +52,7 @@ Errors: W-026 error codes: - `bad_request` +- `evidence_unavailable` - `not_found` - `unsupported` - `read_error` @@ -68,6 +69,7 @@ Exception mapping: | malformed request, invalid path, unsafe traversal | 400 | `bad_request` | | well-formed missing artifact/proposal/lane/trace | 404 | `not_found` | | artifact exceeds the W-026 read size limit | 413 | `read_error` | +| deterministic evidence source absent | 501 | `evidence_unavailable` | | deferred W-027+ route | 501 | `unsupported` | | filesystem read failure | 500 | `read_error` | | unexpected runtime failure | 500 | `runtime_unavailable` | @@ -378,6 +380,177 @@ Forbidden: --- +# R2 Read Projections + +These endpoints are read-only projections over existing artifacts. They do not +create event stores, run stores, pack stores, or vault persistence. + +## GET /packs + +Purpose: + +List readable pack manifests from `language_packs/data/*/manifest.json` and +cleanly readable JSON manifests under `packs/*/*/manifest.json`. + +Query: + +- `limit`: non-negative integer, default `100` +- `offset`: non-negative integer, default `0` + +Trust boundary: + +Pack list reads only fixed repository roots. `GET /packs/{pack_id}` validates +`pack_id` against the safe pattern `[A-Za-z0-9][A-Za-z0-9_.-]{0,127}` before +any filesystem access. Path traversal and encoded slashes return +`400 bad_request`. + +Response: + +```json +{ + "ok": true, + "data": { + "items": [ + { + "pack_id": "en_core_cognition_v1", + "source": "language_pack", + "manifest_path": "language_packs/data/en_core_cognition_v1/manifest.json", + "version": "1.2.0", + "language": "en", + "modality": null, + "determinism_class": "D0", + "checksum": "82d5...", + "checksums": { + "checksum": "82d5..." + } + } + ], + "limit": 100, + "offset": 0 + } +} +``` + +Manifest checksum fields are surfaced verbatim from the manifest. The API also +returns a separate `manifest_digest` on detail, computed from the manifest file +bytes, so byte integrity and manifest-authored checksums remain distinct. + +## GET /packs/{pack_id} + +Purpose: + +Return one pack manifest projection. Unknown safe ids return `404 not_found`; +unsafe ids return `400 bad_request`. + +Response includes the summary fields plus: + +```json +{ + "manifest_digest": "sha256:...", + "manifest": {} +} +``` + +--- + +## GET /audit/events + +Purpose: + +Return a merged audit timeline over existing deterministic artifacts: + +- `engine_state/manifest.json` when present, +- `teaching/proposals/proposals.jsonl`, +- `teaching/math_proposals/proposals.jsonl`, +- persisted Workbench telemetry JSONL files under `workbench_data/`. + +The route does not invent or append an event store. Events are sorted by +`timestamp`, then source/path/type/id tiebreakers. + +Query: + +- `limit`: non-negative integer, default `100` +- `offset`: non-negative integer, default `0` + +Each event includes `source` and `mutation_boundary`. Mutation-boundary events +identify review transitions, accepted corpus appends, operator telemetry, or +engine checkpoint boundaries. + +--- + +## GET /runs + +Purpose: + +List deterministic run/session projections that can be derived from existing +artifacts. + +Current evidence sources: + +- `workbench_data/turn_journal.jsonl` as `workbench_turn_journal` when turn + entries exist. +- `engine_state/manifest.json` as `engine_state_checkpoint` when a checkpoint + manifest exists. + +Known gap: + +The current persisted artifacts do not record a separate durable user-facing +session id. The API exposes the artifact boundary and includes `evidence_gap` +instead of synthesizing a false session identity. + +--- + +## GET /runs/{session_id} + +Purpose: + +Return detail for a deterministic run projection. Unknown ids return +`404 not_found`. Turn journal details include paginated `turns`, where each turn +links to `/trace/{turn_id}`. + +Query: + +- `limit`: non-negative integer for turn refs, default `100` +- `offset`: non-negative integer for turn refs, default `0` + +--- + +## GET /vault/summary + +Purpose: + +Return a cold summary of persisted vault evidence only when +`engine_state/session_state.json` contains a Shape B+ `vault` snapshot. + +If the persisted snapshot is absent, the route returns: + +```json +{ + "ok": false, + "error": { + "code": "evidence_unavailable", + "message": "vault evidence unavailable: engine_state/session_state.json is absent" + } +} +``` + +The route never reaches into live runtime memory. + +## GET /vault/entries + +Purpose: + +Return metadata for persisted vault entries. The endpoint surfaces metadata and +a digest of each persisted versor encoding; it does not emit raw versor +coordinates or approximate recall scores. + +Query: + +- `limit`: non-negative integer, default `100` +- `offset`: non-negative integer, default `0` + +--- + # Evals ## GET /evals diff --git a/docs/workbench/data-shapes-v1.md b/docs/workbench/data-shapes-v1.md index cf4add27..e1f691ca 100644 --- a/docs/workbench/data-shapes-v1.md +++ b/docs/workbench/data-shapes-v1.md @@ -36,6 +36,7 @@ export type WorkbenchResponse = export type WorkbenchError = { code: | "bad_request" + | "evidence_unavailable" | "not_found" | "unsupported" | "read_error" @@ -225,6 +226,110 @@ arbitrary user-supplied filesystem path. --- +# R2 Read Projections + +```ts +export type PackSummary = { + pack_id: string; + source: "language_pack" | "runtime_pack"; + manifest_path: string; + version: string | null; + language: string | null; + modality: string | null; + determinism_class: string | null; + checksum: string | null; + checksums: Record; +}; + +export type PackDetail = PackSummary & { + manifest_digest: string; + manifest: Record; +}; +``` + +`checksum` and `checksums` are manifest-authored values and must remain +verbatim. `manifest_digest` is the backend-computed digest of the manifest file +bytes. + +```ts +export type AuditEvent = { + event_id: string; + source: + | "engine_state_manifest" + | "math_proposal_log" + | "operator_telemetry" + | "reboot_telemetry" + | "teaching_proposal_log"; + source_path: string; + timestamp: string | null; + event_type: string; + mutation_boundary: boolean; + summary: string; + ref_id: string | null; + payload_digest: string; + payload: unknown; +}; +``` + +Audit events are projections over existing artifacts. `event_id` and +`payload_digest` are deterministic backend digests; they are not new stored +identifiers. + +```ts +export type RunSummary = { + session_id: string; + source: "engine_state_manifest" | "turn_journal"; + turn_count: number; + started_at: string | null; + updated_at: string | null; + checkpoint_present: boolean; + checkpoint_revision: string | null; + artifact_refs: ArtifactRef[]; + evidence_gap: string | null; +}; + +export type RunTurnRef = { + turn_id: number; + trace_hash: string | null; + timestamp: string; + trace_path: string; + surface_excerpt: string; +}; + +export type RunDetail = RunSummary & { + turns: RunTurnRef[]; + manifest: Record | null; +}; +``` + +When no durable per-session id exists, `session_id` names the artifact boundary +(`workbench_turn_journal` or `engine_state_checkpoint`) and `evidence_gap` +states the missing persisted fact. + +```ts +export type VaultSummary = { + source_path: string; + entry_count: number; + store_count: number; + reproject_interval: number; + max_entries: number | null; + persisted: boolean; +}; + +export type VaultEntry = { + entry_index: number; + epistemic_status: string; + epistemic_state: string; + metadata: Record; + versor_digest: string | null; +}; +``` + +Vault shapes are available only from persisted `engine_state/session_state.json` +evidence. If absent, the API returns `501 evidence_unavailable`. + +--- + # UI state tags These are display-only semantic tags. diff --git a/tests/test_workbench_r2_read_substrate.py b/tests/test_workbench_r2_read_substrate.py new file mode 100644 index 00000000..269db6e5 --- /dev/null +++ b/tests/test_workbench_r2_read_substrate.py @@ -0,0 +1,261 @@ +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from workbench import readers +from workbench.api import WorkbenchApi +from workbench.journal import TurnJournal, TurnJournalEntry +from workbench.schemas import ChatTurnResult, TurnVerdict + + +def _request(api: WorkbenchApi, method: str, path: str): + return api.handle(method, path, b"") + + +def _snapshot(root: Path) -> dict[str, bytes]: + snap: dict[str, bytes] = {} + if not root.exists(): + return snap + for path in sorted(root.rglob("*")): + if not path.is_file(): + continue + rel = path.relative_to(root) + if "__pycache__" in rel.parts or rel.suffix in {".pyc", ".pyo"}: + continue + snap[rel.as_posix()] = path.read_bytes() + return snap + + +def _chat_result(prompt: str = "What is truth?") -> ChatTurnResult: + return ChatTurnResult( + prompt=prompt, + surface="Truth is coherent structure.", + articulation_surface="Truth is coherent structure.", + walk_surface="truth -> coherence", + grounding_source="pack", + epistemic_state="decoded", + normative_clearance="cleared", + normative_detail="", + trace_hash="sha256:trace", + refusal_emitted=False, + hedge_injected=False, + mutation_mode="runtime_turn", + identity_verdict=TurnVerdict(outcome="cleared", runtime_detail=""), + safety_verdict=TurnVerdict(outcome="cleared", runtime_detail=""), + ethics_verdict=TurnVerdict(outcome="cleared", runtime_detail=""), + proposal_candidates=[], + turn_cost_ms=7, + checkpoint_emitted=False, + ) + + +def _entry(turn_id: int, timestamp: str) -> TurnJournalEntry: + result = replace(_chat_result(f"prompt {turn_id}"), turn_id=turn_id) + return TurnJournalEntry.from_chat_turn(result, turn_id=turn_id, timestamp=timestamp) + + +def test_packs_list_get_pagination_and_checksum_verbatim() -> None: + api = WorkbenchApi() + + first = _request(api, "GET", "/packs?limit=1&offset=0") + second = _request(api, "GET", "/packs?limit=1&offset=1") + detail = _request(api, "GET", "/packs/en_core_cognition_v1") + + assert first.status == 200 + assert first.payload["ok"] is True + assert first.payload["generated_at"] + assert len(first.payload["data"]["items"]) == 1 + assert len(second.payload["data"]["items"]) == 1 + assert first.payload["data"]["items"][0]["pack_id"] <= second.payload["data"]["items"][0]["pack_id"] + assert detail.status == 200 + assert detail.payload["data"]["pack_id"] == "en_core_cognition_v1" + assert detail.payload["data"]["checksum"] == detail.payload["data"]["manifest"]["checksum"] + assert detail.payload["data"]["checksums"]["checksum"] == detail.payload["data"]["manifest"]["checksum"] + + +def test_pack_id_path_traversal_rejected_before_filesystem_access() -> None: + api = WorkbenchApi() + + response = _request(api, "GET", "/packs/..%2F..%2Fpyproject.toml") + + assert response.status == 400 + assert response.payload["ok"] is False + assert response.payload["error"]["code"] == "bad_request" + + +def test_unknown_pack_returns_404_not_synthetic_data() -> None: + response = _request(WorkbenchApi(), "GET", "/packs/not_a_real_pack_v1") + + assert response.status == 404 + assert response.payload["ok"] is False + assert response.payload["error"]["code"] == "not_found" + + +def test_audit_events_merge_existing_artifacts_with_stable_order( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + proposal_log = tmp_path / "proposal_events.jsonl" + proposal_log.write_text( + json.dumps( + { + "event": "transition", + "proposal_id": "proposal-1", + "to": "accepted", + "timestamp": "2026-06-12T02:00:00+00:00", + }, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + math_log = tmp_path / "math_proposals.jsonl" + math_log.write_text( + json.dumps( + { + "proposal_id": "math-1", + "domain": "math", + "timestamp": "2026-06-12T01:00:00+00:00", + }, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + telemetry_root = tmp_path / "workbench_data" + telemetry_root.mkdir() + (telemetry_root / "operator_telemetry.jsonl").write_text( + json.dumps( + { + "event": "operator_ratify", + "proposal_id": "math-1", + "timestamp": "2026-06-12T03:00:00+00:00", + }, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + + monkeypatch.setattr(readers, "DEFAULT_PROPOSAL_LOG_PATH", proposal_log) + monkeypatch.setattr(readers, "MATH_PROPOSALS_JSONL", math_log) + monkeypatch.setattr(readers, "WORKBENCH_TELEMETRY_ROOT", telemetry_root) + monkeypatch.setattr(readers, "ENGINE_STATE_ROOT", tmp_path / "engine_state") + + api = WorkbenchApi() + page = _request(api, "GET", "/audit/events?limit=2&offset=0") + next_page = _request(api, "GET", "/audit/events?limit=2&offset=2") + again = _request(api, "GET", "/audit/events?limit=2&offset=0") + + assert page.status == 200 + items = page.payload["data"]["items"] + assert [item["source"] for item in items] == ["math_proposal_log", "teaching_proposal_log"] + assert [item["timestamp"] for item in items] == sorted(item["timestamp"] for item in items) + assert items[1]["mutation_boundary"] is True + assert next_page.payload["data"]["items"][0]["source"] == "operator_telemetry" + assert page.payload["data"] == again.payload["data"] + + +def test_runs_project_turn_journal_without_synthesizing_unknown_sessions( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(readers, "ENGINE_STATE_ROOT", tmp_path / "engine_state") + journal = TurnJournal(tmp_path / "workbench_data") + journal.append(_entry(1, "2026-06-12T00:00:00+00:00")) + journal.append(_entry(2, "2026-06-12T00:01:00+00:00")) + api = WorkbenchApi(journal=journal) + + runs = _request(api, "GET", "/runs?limit=1&offset=0") + detail = _request(api, "GET", f"/runs/{readers.JOURNAL_RUN_ID}?limit=1&offset=1") + missing = _request(api, "GET", "/runs/not-a-real-session") + + assert runs.status == 200 + assert runs.payload["data"]["items"][0]["session_id"] == readers.JOURNAL_RUN_ID + assert runs.payload["data"]["items"][0]["turn_count"] == 2 + assert detail.status == 200 + assert detail.payload["data"]["turns"][0]["turn_id"] == 2 + assert detail.payload["data"]["turns"][0]["trace_path"] == "/trace/2" + assert missing.status == 404 + + +def test_vault_absent_returns_typed_evidence_unavailable( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(readers, "ENGINE_STATE_ROOT", tmp_path / "engine_state") + + response = _request(WorkbenchApi(), "GET", "/vault/summary") + + assert response.status == 501 + assert response.payload["ok"] is False + assert response.payload["error"]["code"] == "evidence_unavailable" + + +def test_vault_summary_and_entries_read_persisted_session_state_only( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + engine_state = tmp_path / "engine_state" + engine_state.mkdir() + (engine_state / "session_state.json").write_text( + json.dumps( + { + "vault": { + "versors": [{"dtype": "float32", "shape": [1], "data": "AAAAAA=="}], + "metadata": [ + { + "id": "entry-1", + "epistemic_status": "coherent", + "epistemic_state": "decoded", + } + ], + "store_count": 1, + "reproject_interval": 20, + "max_entries": None, + } + }, + sort_keys=True, + ), + encoding="utf-8", + ) + monkeypatch.setattr(readers, "ENGINE_STATE_ROOT", engine_state) + api = WorkbenchApi() + + summary = _request(api, "GET", "/vault/summary") + entries = _request(api, "GET", "/vault/entries?limit=1&offset=0") + + assert summary.status == 200 + assert summary.payload["data"]["entry_count"] == 1 + assert summary.payload["data"]["persisted"] is True + assert entries.status == 200 + assert entries.payload["data"]["items"][0]["metadata"]["id"] == "entry-1" + assert entries.payload["data"]["items"][0]["versor_digest"].startswith("sha256:") + + +def test_r2_read_routes_do_not_mutate_guarded_roots() -> None: + repo_root = Path(__file__).resolve().parent.parent + guarded = { + "teaching": repo_root / "teaching", + "packs": repo_root / "packs", + "language_packs/data": repo_root / "language_packs" / "data", + "engine_state": repo_root / "engine_state", + } + before = {name: _snapshot(path) for name, path in guarded.items()} + api = WorkbenchApi() + + responses = [ + _request(api, "GET", "/packs?limit=2"), + _request(api, "GET", "/packs/en_core_cognition_v1"), + _request(api, "GET", "/audit/events?limit=2"), + _request(api, "GET", "/runs?limit=2"), + _request(api, "GET", "/vault/summary"), + _request(api, "GET", "/vault/entries?limit=2"), + ] + + assert [response.status for response in responses] == [200, 200, 200, 200, 501, 501] + assert {name: _snapshot(path) for name, path in guarded.items()} == before diff --git a/workbench/api.py b/workbench/api.py index 761ce0b6..28360363 100644 --- a/workbench/api.py +++ b/workbench/api.py @@ -19,7 +19,7 @@ from core.epistemic_state import ( ) from workbench import readers from workbench.journal import DEFAULT_JOURNAL_DIR, TurnJournal, TurnJournalEntry -from workbench.readers import ArtifactTooLargeError +from workbench.readers import ArtifactTooLargeError, EvidenceUnavailableError from workbench.schemas import ChatTurnResult, MathRatifyResult, ProposalRef, TurnVerdict, error, ok @@ -28,6 +28,20 @@ MAX_CHAT_PROMPT_CHARS = 4096 _CHAT_TURN_LOCK = threading.Lock() +def _pagination( + query: dict[str, list[str]], + *, + default_limit: int = 100, +) -> tuple[int, int]: + limit = int(query.get("limit", [str(default_limit)])[0]) + offset = int(query.get("offset", ["0"])[0]) + if limit < 0: + raise ValueError("limit must be non-negative") + if offset < 0: + raise ValueError("offset must be non-negative") + return limit, offset + + @dataclass(frozen=True, slots=True) class ApiResponse: status: int @@ -91,6 +105,8 @@ class WorkbenchApi: except FileNotFoundError as exc: missing = str(exc) or "resource" return ApiResponse(404, error("not_found", f"not found: {missing}")) + except EvidenceUnavailableError as exc: + return ApiResponse(501, error("evidence_unavailable", str(exc))) except ArtifactTooLargeError as exc: return ApiResponse(413, error("read_error", str(exc))) except OSError as exc: @@ -134,6 +150,73 @@ class WorkbenchApi: if method == "GET" and path.startswith("/math-proposals/"): proposal_id = unquote(path.removeprefix("/math-proposals/")) return ApiResponse(200, ok(readers.read_math_proposal(proposal_id))) + if method == "GET" and path == "/packs": + limit, offset = _pagination(query) + return ApiResponse( + 200, + ok( + { + "items": readers.list_packs(limit=limit, offset=offset), + "limit": limit, + "offset": offset, + } + ), + ) + if method == "GET" and path.startswith("/packs/"): + pack_id = unquote(path.removeprefix("/packs/")) + return ApiResponse(200, ok(readers.read_pack(pack_id))) + if method == "GET" and path == "/audit/events": + limit, offset = _pagination(query) + return ApiResponse( + 200, + ok( + { + "items": readers.list_audit_events(limit=limit, offset=offset), + "limit": limit, + "offset": offset, + } + ), + ) + if method == "GET" and path == "/runs": + limit, offset = _pagination(query) + return ApiResponse( + 200, + ok( + { + "items": readers.list_runs(self._journal, limit=limit, offset=offset), + "limit": limit, + "offset": offset, + } + ), + ) + if method == "GET" and path.startswith("/runs/"): + session_id = unquote(path.removeprefix("/runs/")) + turn_limit, turn_offset = _pagination(query) + return ApiResponse( + 200, + ok( + readers.read_run( + session_id, + self._journal, + turn_limit=turn_limit, + turn_offset=turn_offset, + ) + ), + ) + if method == "GET" and path == "/vault/summary": + return ApiResponse(200, ok(readers.read_vault_summary())) + if method == "GET" and path == "/vault/entries": + limit, offset = _pagination(query) + return ApiResponse( + 200, + ok( + { + "items": readers.list_vault_entries(limit=limit, offset=offset), + "limit": limit, + "offset": offset, + } + ), + ) if method == "GET" and path == "/evals": return ApiResponse(200, ok({"lanes": readers.list_eval_lanes()})) if method == "GET" and path.startswith("/evals/"): diff --git a/workbench/journal.py b/workbench/journal.py index 2f97a805..d5eef714 100644 --- a/workbench/journal.py +++ b/workbench/journal.py @@ -137,6 +137,14 @@ class TurnJournal: entries = self._read_entries() return [entry.summary() for entry in entries[offset : offset + limit]] + def list_entries(self, *, limit: int = 50, offset: int = 0) -> list[TurnJournalEntry]: + if limit < 0: + raise ValueError("limit must be non-negative") + if offset < 0: + raise ValueError("offset must be non-negative") + entries = self._read_entries() + return entries[offset : offset + limit] + def get_entry(self, turn_id: int) -> TurnJournalEntry: for entry in self._read_entries(): if entry.turn_id == turn_id: diff --git a/workbench/readers.py b/workbench/readers.py index d735793c..abeadd89 100644 --- a/workbench/readers.py +++ b/workbench/readers.py @@ -5,6 +5,7 @@ from __future__ import annotations import hashlib import json import os +import re import threading from pathlib import Path, PurePosixPath from typing import Any, get_args @@ -13,6 +14,7 @@ from engine_state import EngineStateStore, get_git_revision from evals.framework import discover_lanes, get_lane, run_lane from teaching.proposals import DEFAULT_PROPOSAL_LOG_PATH, ProposalLog, ReviewState from workbench.schemas import ( + AuditEvent, ArtifactDetail, ArtifactRef, EvalLaneSummary, @@ -21,15 +23,25 @@ from workbench.schemas import ( MathProposalSummary, MathRatifyResult, MathReasoningStep, + PackDetail, + PackSummary, ProposalDetail, ProposalSummary, + RunDetail, + RunSummary, + RunTurnRef, RuntimeStatus, + VaultEntry, + VaultSummary, ) REPO_ROOT = Path(__file__).resolve().parents[1] SAFE_EVAL_LANES = frozenset({"contemplation_quality"}) MAX_ARTIFACT_BYTES = 16 * 1024 * 1024 READ_CHUNK_BYTES = 64 * 1024 +SAFE_PACK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") +JOURNAL_RUN_ID = "workbench_turn_journal" +ENGINE_STATE_RUN_ID = "engine_state_checkpoint" _EVAL_RUN_LOCK = threading.Lock() _REVIEW_STATES = frozenset(get_args(ReviewState)) ALLOWED_ARTIFACT_ROOTS = ( @@ -41,6 +53,10 @@ ALLOWED_ARTIFACT_ROOTS = ( ) MATH_PROPOSALS_JSONL = REPO_ROOT / "teaching" / "math_proposals" / "proposals.jsonl" +LANGUAGE_PACK_ROOT = REPO_ROOT / "language_packs" / "data" +RUNTIME_PACK_ROOT = REPO_ROOT / "packs" +WORKBENCH_TELEMETRY_ROOT = REPO_ROOT / "workbench_data" +ENGINE_STATE_ROOT = REPO_ROOT / "engine_state" _DEFAULT_MATH_AUDIT_PATH = ( REPO_ROOT / "evals" @@ -63,6 +79,10 @@ class ArtifactTooLargeError(OSError): """Raised when an artifact is too large for direct Workbench reads.""" +class EvidenceUnavailableError(OSError): + """Raised when a read route has no persisted evidence source to project.""" + + def _sha256_bytes(content: bytes) -> str: return "sha256:" + hashlib.sha256(content).hexdigest() @@ -79,6 +99,61 @@ def _relative(path: Path) -> str: return path.resolve().relative_to(REPO_ROOT.resolve()).as_posix() +def _display_path(path: Path) -> str: + try: + return _relative(path) + except ValueError: + return path.resolve().as_posix() + + +def _check_read_size(path: Path, artifact_id: str | None = None) -> None: + if path.stat().st_size > MAX_ARTIFACT_BYTES: + label = artifact_id or _display_path(path) + raise ArtifactTooLargeError( + f"artifact exceeds {MAX_ARTIFACT_BYTES} byte read limit: {label}" + ) + + +def _canonical_json_bytes(payload: Any) -> bytes: + return json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def _read_json_object(path: Path) -> dict[str, Any]: + _check_read_size(path) + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"expected JSON object in {_display_path(path)}") + return payload + + +def _read_jsonl_records(path: Path) -> list[tuple[int, dict[str, Any]]]: + if not path.exists(): + return [] + _check_read_size(path) + records: list[tuple[int, dict[str, Any]]] = [] + for line_no, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + stripped = raw_line.strip() + if not stripped: + continue + payload = json.loads(stripped) + if isinstance(payload, dict): + records.append((line_no, payload)) + return records + + +def _page(items: list[Any], *, limit: int, offset: int) -> list[Any]: + if limit < 0: + raise ValueError("limit must be non-negative") + if offset < 0: + raise ValueError("offset must be non-negative") + return items[offset : offset + limit] + + def _validate_artifact_id(artifact_id: str) -> PurePosixPath: if not artifact_id or artifact_id.startswith("/"): raise ValueError("artifact id must be a repo-relative path") @@ -174,10 +249,7 @@ def read_artifact(artifact_id: str) -> ArtifactDetail: path = _resolve_artifact(artifact_id) if not path.exists() or not path.is_file(): raise FileNotFoundError(artifact_id) - if path.stat().st_size > MAX_ARTIFACT_BYTES: - raise ArtifactTooLargeError( - f"artifact exceeds {MAX_ARTIFACT_BYTES} byte read limit: {artifact_id}" - ) + _check_read_size(path, artifact_id) raw = path.read_bytes() text = raw.decode("utf-8") content_type = "text" @@ -204,6 +276,103 @@ def read_artifact(artifact_id: str) -> ArtifactDetail: ) +def _validate_pack_id(pack_id: str) -> str: + if not SAFE_PACK_ID_RE.fullmatch(pack_id): + raise ValueError("pack id contains unsafe characters") + return pack_id + + +def _manifest_checksum_fields(manifest: dict[str, Any]) -> dict[str, str]: + checksums: dict[str, str] = {} + for key in sorted(manifest): + value = manifest[key] + if not isinstance(value, str): + continue + lowered = key.lower() + if "checksum" in lowered or lowered.endswith("_sha256") or lowered == "sha256": + checksums[key] = value + return checksums + + +def _pack_source(path: Path) -> str: + resolved = path.resolve() + language_root = LANGUAGE_PACK_ROOT.resolve() + runtime_root = RUNTIME_PACK_ROOT.resolve() + if language_root == resolved or language_root in resolved.parents: + return "language_pack" + if runtime_root == resolved or runtime_root in resolved.parents: + return "runtime_pack" + return "runtime_pack" + + +def _pack_manifest_paths() -> list[Path]: + paths: list[Path] = [] + if LANGUAGE_PACK_ROOT.exists(): + paths.extend(sorted(LANGUAGE_PACK_ROOT.glob("*/manifest.json"))) + if RUNTIME_PACK_ROOT.exists(): + paths.extend(sorted(RUNTIME_PACK_ROOT.glob("*/*/manifest.json"))) + return paths + + +def _pack_detail_from_manifest(path: Path) -> PackDetail | None: + manifest = _read_json_object(path) + pack_id = str(manifest.get("pack_id") or manifest.get("register_id") or path.parent.name) + if not SAFE_PACK_ID_RE.fullmatch(pack_id): + return None + checksums = _manifest_checksum_fields(manifest) + return PackDetail( + pack_id=pack_id, + source=_pack_source(path), # type: ignore[arg-type] + manifest_path=_display_path(path), + version=(str(manifest["version"]) if "version" in manifest else None), + language=(str(manifest["language"]) if "language" in manifest else None), + modality=(str(manifest["modality"]) if "modality" in manifest else None), + determinism_class=( + str(manifest["determinism_class"]) + if "determinism_class" in manifest + else None + ), + checksum=(str(manifest["checksum"]) if "checksum" in manifest else None), + checksums=checksums, + manifest_digest=_sha256_file(path), + manifest=manifest, + ) + + +def _all_pack_details() -> list[PackDetail]: + details: list[PackDetail] = [] + for path in _pack_manifest_paths(): + detail = _pack_detail_from_manifest(path) + if detail is not None: + details.append(detail) + return sorted(details, key=lambda item: (item.pack_id, item.source, item.manifest_path)) + + +def list_packs(*, limit: int = 100, offset: int = 0) -> list[PackSummary]: + return [ + PackSummary( + pack_id=detail.pack_id, + source=detail.source, + manifest_path=detail.manifest_path, + version=detail.version, + language=detail.language, + modality=detail.modality, + determinism_class=detail.determinism_class, + checksum=detail.checksum, + checksums=detail.checksums, + ) + for detail in _page(_all_pack_details(), limit=limit, offset=offset) + ] + + +def read_pack(pack_id: str) -> PackDetail: + safe_id = _validate_pack_id(pack_id) + for detail in _all_pack_details(): + if detail.pack_id == safe_id: + return detail + raise FileNotFoundError(pack_id) + + def _state_value(value: Any) -> str: text = str(value or "unknown") return text if text in _REVIEW_STATES else "unknown" @@ -580,6 +749,397 @@ def ratify_math_proposal( raise NotImplementedError(f"handler {handler_name} application not implemented") +def _payload_digest(payload: Any) -> str: + return _sha256_bytes(_canonical_json_bytes(payload)) + + +def _event_timestamp(payload: dict[str, Any]) -> str | None: + for key in ("timestamp", "created_at", "emitted_at", "reviewed_at", "review_date"): + value = payload.get(key) + if isinstance(value, str) and value: + return value + proposal = payload.get("proposal") + if isinstance(proposal, dict): + source = proposal.get("source") + if isinstance(source, dict): + value = source.get("emitted_at") + if isinstance(value, str) and value: + return value + return None + + +def _proposal_ref_id(payload: dict[str, Any]) -> str | None: + value = payload.get("proposal_id") + if isinstance(value, str) and value: + return value + proposal = payload.get("proposal") + if isinstance(proposal, dict): + value = proposal.get("proposal_id") + if isinstance(value, str) and value: + return value + return None + + +def _audit_event( + *, + source: str, + source_path: Path, + line_no: int, + event_type: str, + payload: dict[str, Any], + mutation_boundary: bool, + ref_id: str | None = None, + summary: str | None = None, +) -> AuditEvent: + display_path = _display_path(source_path) + digest = _payload_digest(payload) + event_id = _sha256_bytes( + _canonical_json_bytes( + { + "digest": digest, + "event_type": event_type, + "line_no": line_no, + "source": source, + "source_path": display_path, + } + ) + ) + label = summary or event_type + if ref_id: + label = f"{label}: {ref_id}" + return AuditEvent( + event_id=event_id, + source=source, # type: ignore[arg-type] + source_path=display_path, + timestamp=_event_timestamp(payload), + event_type=event_type, + mutation_boundary=mutation_boundary, + summary=label, + ref_id=ref_id, + payload_digest=digest, + payload=payload, + ) + + +def _teaching_proposal_audit_events() -> list[AuditEvent]: + path = DEFAULT_PROPOSAL_LOG_PATH + events: list[AuditEvent] = [] + for line_no, payload in _read_jsonl_records(path): + event_type = str(payload.get("event") or "proposal_event") + ref_id = _proposal_ref_id(payload) + events.append( + _audit_event( + source="teaching_proposal_log", + source_path=path, + line_no=line_no, + event_type=event_type, + payload=payload, + mutation_boundary=event_type in {"transition", "accepted_corpus_append"}, + ref_id=ref_id, + summary="teaching proposal event", + ) + ) + return events + + +def _math_proposal_audit_events() -> list[AuditEvent]: + path = MATH_PROPOSALS_JSONL + events: list[AuditEvent] = [] + for line_no, payload in _read_jsonl_records(path): + ref_id = ( + str(payload.get("proposal_id")) + if isinstance(payload.get("proposal_id"), str) + else None + ) + events.append( + _audit_event( + source="math_proposal_log", + source_path=path, + line_no=line_no, + event_type="math_proposal_record", + payload=payload, + mutation_boundary=False, + ref_id=ref_id, + summary="math proposal record", + ) + ) + return events + + +def _telemetry_audit_events() -> list[AuditEvent]: + if not WORKBENCH_TELEMETRY_ROOT.exists(): + return [] + events: list[AuditEvent] = [] + for path in sorted(WORKBENCH_TELEMETRY_ROOT.rglob("*.jsonl")): + if not path.is_file(): + continue + for line_no, payload in _read_jsonl_records(path): + event_name = str(payload.get("event") or "") + event_type = str(payload.get("type") or event_name or "telemetry_event") + if event_type == "reboot": + source = "reboot_telemetry" + mutation_boundary = False + summary = "reboot telemetry" + elif event_name.startswith("operator_"): + source = "operator_telemetry" + mutation_boundary = True + summary = "workbench operator telemetry" + else: + continue + ref_id = ( + str(payload.get("proposal_id")) + if isinstance(payload.get("proposal_id"), str) + else None + ) + events.append( + _audit_event( + source=source, + source_path=path, + line_no=line_no, + event_type=event_type, + payload=payload, + mutation_boundary=mutation_boundary, + ref_id=ref_id, + summary=summary, + ) + ) + return events + + +def _engine_state_manifest_audit_event() -> list[AuditEvent]: + path = ENGINE_STATE_ROOT / "manifest.json" + if not path.exists(): + return [] + manifest = _read_json_object(path) + return [ + _audit_event( + source="engine_state_manifest", + source_path=path, + line_no=1, + event_type="engine_state_checkpoint", + payload=manifest, + mutation_boundary=True, + ref_id=str(manifest.get("written_at_revision") or "unknown"), + summary="engine state checkpoint", + ) + ] + + +def list_audit_events(*, limit: int = 100, offset: int = 0) -> list[AuditEvent]: + events: list[AuditEvent] = [] + events.extend(_engine_state_manifest_audit_event()) + events.extend(_teaching_proposal_audit_events()) + events.extend(_math_proposal_audit_events()) + events.extend(_telemetry_audit_events()) + events.sort( + key=lambda event: ( + event.timestamp or "", + event.source, + event.source_path, + event.event_type, + event.event_id, + ) + ) + return _page(events, limit=limit, offset=offset) + + +def _artifact_ref_for_path(path: Path, kind: str) -> ArtifactRef | None: + if not path.exists() or not path.is_file(): + return None + _check_read_size(path) + return ArtifactRef( + artifact_id=_display_path(path), + kind=kind, # type: ignore[arg-type] + path=_display_path(path), + digest=_sha256_file(path), + created_at=None, + ) + + +def _load_engine_manifest() -> tuple[Path, dict[str, Any]] | None: + path = ENGINE_STATE_ROOT / "manifest.json" + if not path.exists(): + return None + return path, _read_json_object(path) + + +def _journal_entries(journal: Any) -> list[Any]: + path = getattr(journal, "path", None) + if isinstance(path, Path) and path.exists(): + _check_read_size(path) + return list(journal.list_entries(limit=1_000_000, offset=0)) + + +def _engine_run_summary(manifest_item: tuple[Path, dict[str, Any]] | None) -> RunSummary | None: + if manifest_item is None: + return None + path, manifest = manifest_item + artifact = _artifact_ref_for_path(path, "engine_state_manifest") + return RunSummary( + session_id=ENGINE_STATE_RUN_ID, + source="engine_state_manifest", + turn_count=int(manifest.get("turn_count") or 0), + started_at=None, + updated_at=None, + checkpoint_present=True, + checkpoint_revision=str(manifest.get("written_at_revision") or "unknown"), + artifact_refs=[artifact] if artifact is not None else [], + evidence_gap="engine_state manifest has no durable per-session id", + ) + + +def _journal_run_summary(journal: Any, manifest: dict[str, Any] | None) -> RunSummary | None: + entries = _journal_entries(journal) + if not entries: + return None + path = getattr(journal, "path", None) + artifact = _artifact_ref_for_path(path, "trace") if isinstance(path, Path) else None + return RunSummary( + session_id=JOURNAL_RUN_ID, + source="turn_journal", + turn_count=len(entries), + started_at=str(entries[0].timestamp), + updated_at=str(entries[-1].timestamp), + checkpoint_present=manifest is not None, + checkpoint_revision=( + str(manifest.get("written_at_revision") or "unknown") + if manifest is not None + else None + ), + artifact_refs=[artifact] if artifact is not None else [], + evidence_gap="turn journal is one local grouping; no separate durable session id is recorded", + ) + + +def list_runs(journal: Any, *, limit: int = 100, offset: int = 0) -> list[RunSummary]: + manifest_item = _load_engine_manifest() + manifest = manifest_item[1] if manifest_item is not None else None + runs = [ + item + for item in ( + _journal_run_summary(journal, manifest), + _engine_run_summary(manifest_item), + ) + if item is not None + ] + runs.sort(key=lambda run: (run.updated_at or "", run.session_id)) + return _page(runs, limit=limit, offset=offset) + + +def _run_detail_from_summary( + summary: RunSummary, + *, + turns: list[RunTurnRef], + manifest: dict[str, Any] | None, +) -> RunDetail: + return RunDetail( + session_id=summary.session_id, + source=summary.source, + turn_count=summary.turn_count, + started_at=summary.started_at, + updated_at=summary.updated_at, + checkpoint_present=summary.checkpoint_present, + checkpoint_revision=summary.checkpoint_revision, + artifact_refs=summary.artifact_refs, + evidence_gap=summary.evidence_gap, + turns=turns, + manifest=manifest, + ) + + +def read_run( + session_id: str, + journal: Any, + *, + turn_limit: int = 100, + turn_offset: int = 0, +) -> RunDetail: + manifest_item = _load_engine_manifest() + manifest = manifest_item[1] if manifest_item is not None else None + if session_id == JOURNAL_RUN_ID: + summary = _journal_run_summary(journal, manifest) + if summary is None: + raise FileNotFoundError(session_id) + entries = _journal_entries(journal) + turns = [ + RunTurnRef( + turn_id=int(entry.turn_id), + trace_hash=entry.trace_hash, + timestamp=str(entry.timestamp), + trace_path=f"/trace/{entry.turn_id}", + surface_excerpt=str(entry.surface)[:120], + ) + for entry in _page(entries, limit=turn_limit, offset=turn_offset) + ] + return _run_detail_from_summary(summary, turns=turns, manifest=manifest) + if session_id == ENGINE_STATE_RUN_ID: + summary = _engine_run_summary(manifest_item) + if summary is None: + raise FileNotFoundError(session_id) + return _run_detail_from_summary(summary, turns=[], manifest=manifest) + raise FileNotFoundError(session_id) + + +def _load_vault_snapshot() -> tuple[Path, dict[str, Any]]: + path = ENGINE_STATE_ROOT / "session_state.json" + if not path.exists(): + raise EvidenceUnavailableError( + "vault evidence unavailable: engine_state/session_state.json is absent" + ) + payload = _read_json_object(path) + vault = payload.get("vault") + if not isinstance(vault, dict): + raise EvidenceUnavailableError( + "vault evidence unavailable: engine_state/session_state.json has no vault snapshot" + ) + return path, vault + + +def read_vault_summary() -> VaultSummary: + path, vault = _load_vault_snapshot() + metadata = vault.get("metadata") + entries = metadata if isinstance(metadata, list) else [] + return VaultSummary( + source_path=_display_path(path), + entry_count=len(entries), + store_count=int(vault.get("store_count") or 0), + reproject_interval=int(vault.get("reproject_interval") or 0), + max_entries=( + int(vault["max_entries"]) + if isinstance(vault.get("max_entries"), int) + else None + ), + persisted=True, + ) + + +def list_vault_entries(*, limit: int = 100, offset: int = 0) -> list[VaultEntry]: + _path, vault = _load_vault_snapshot() + metadata_raw = vault.get("metadata") + versors_raw = vault.get("versors") + if not isinstance(metadata_raw, list): + raise ValueError("vault snapshot metadata must be a list") + versors = versors_raw if isinstance(versors_raw, list) else [] + entries: list[VaultEntry] = [] + for index, metadata_item in enumerate(metadata_raw): + metadata = metadata_item if isinstance(metadata_item, dict) else {} + versor = versors[index] if index < len(versors) else None + entries.append( + VaultEntry( + entry_index=index, + epistemic_status=str(metadata.get("epistemic_status") or "unknown"), + epistemic_state=str(metadata.get("epistemic_state") or "unknown"), + metadata=metadata, + versor_digest=( + _sha256_bytes(_canonical_json_bytes(versor)) + if versor is not None + else None + ), + ) + ) + return _page(entries, limit=limit, offset=offset) + + def run_safe_eval_lane( lane_name: str, *, diff --git a/workbench/schemas.py b/workbench/schemas.py index 48537640..ccc6c025 100644 --- a/workbench/schemas.py +++ b/workbench/schemas.py @@ -9,6 +9,7 @@ from typing import Any, Literal ErrorCode = Literal[ "bad_request", + "evidence_unavailable", "not_found", "unsupported", "read_error", @@ -278,3 +279,98 @@ class MathRatifyResult: applied: bool = False target_path: str | None = None evidence_hash: str | None = None + + +PackSource = Literal["language_pack", "runtime_pack"] + + +@dataclass(frozen=True, slots=True) +class PackSummary: + pack_id: str + source: PackSource + manifest_path: str + version: str | None + language: str | None + modality: str | None + determinism_class: str | None + checksum: str | None + checksums: dict[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PackDetail(PackSummary): + manifest_digest: str = "" + manifest: dict[str, Any] = field(default_factory=dict) + + +AuditSource = Literal[ + "engine_state_manifest", + "math_proposal_log", + "operator_telemetry", + "reboot_telemetry", + "teaching_proposal_log", +] + + +@dataclass(frozen=True, slots=True) +class AuditEvent: + event_id: str + source: AuditSource + source_path: str + timestamp: str | None + event_type: str + mutation_boundary: bool + summary: str + ref_id: str | None + payload_digest: str + payload: Any + + +RunSource = Literal["engine_state_manifest", "turn_journal"] + + +@dataclass(frozen=True, slots=True) +class RunSummary: + session_id: str + source: RunSource + turn_count: int + started_at: str | None + updated_at: str | None + checkpoint_present: bool + checkpoint_revision: str | None + artifact_refs: list[ArtifactRef] = field(default_factory=list) + evidence_gap: str | None = None + + +@dataclass(frozen=True, slots=True) +class RunTurnRef: + turn_id: int + trace_hash: str | None + timestamp: str + trace_path: str + surface_excerpt: str + + +@dataclass(frozen=True, slots=True) +class RunDetail(RunSummary): + turns: list[RunTurnRef] = field(default_factory=list) + manifest: dict[str, Any] | None = None + + +@dataclass(frozen=True, slots=True) +class VaultSummary: + source_path: str + entry_count: int + store_count: int + reproject_interval: int + max_entries: int | None + persisted: bool + + +@dataclass(frozen=True, slots=True) +class VaultEntry: + entry_index: int + epistemic_status: str + epistemic_state: str + metadata: dict[str, Any] + versor_digest: str | None