Spaces:
Running
Running
| # SPDX-License-Identifier: Apache-2.0 | |
| """Fail-closed handles-only contract for the refinement observatory.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import re | |
| from pathlib import Path | |
| from typing import Any, Mapping | |
| STATE_SCHEMA = "szl.second-brain.refinement-memory/v1" | |
| HANDLE_SCHEMA = "szl.second-brain.refinement-pattern-handle/v1" | |
| SOURCE_SCHEMA = "szl.anatomy.refinement-source/v1" | |
| HEX_40 = re.compile(r"^[0-9a-f]{40}$") | |
| HEX_64 = re.compile(r"^[0-9a-f]{64}$") | |
| ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$") | |
| FORBIDDEN_KEYS = frozenset( | |
| { | |
| "chain_of_thought", | |
| "hidden_reasoning", | |
| "private_reasoning", | |
| "raw_prompt", | |
| "system_prompt", | |
| "raw_completion", | |
| "transcript", | |
| "credentials", | |
| "credential", | |
| "token", | |
| "secret", | |
| "private_graph", | |
| "content", | |
| "text", | |
| } | |
| ) | |
| class RefinementProjectionError(ValueError): | |
| """The source-bound projection violated the public observatory contract.""" | |
| def canonical_bytes(value: Any) -> bytes: | |
| try: | |
| return json.dumps( | |
| value, | |
| ensure_ascii=False, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| allow_nan=False, | |
| ).encode("utf-8") | |
| except (TypeError, ValueError) as exc: | |
| raise RefinementProjectionError("projection is not strict JSON") from exc | |
| def sha256_hex(value: bytes | str) -> str: | |
| data = value.encode("utf-8") if isinstance(value, str) else value | |
| return hashlib.sha256(data).hexdigest() | |
| def _scan(value: Any, path: str = "$") -> None: | |
| if isinstance(value, Mapping): | |
| for key, item in value.items(): | |
| if str(key).strip().casefold() in FORBIDDEN_KEYS: | |
| raise RefinementProjectionError( | |
| f"forbidden public field at {path}.{key}" | |
| ) | |
| _scan(item, f"{path}.{key}") | |
| elif isinstance(value, list): | |
| for index, item in enumerate(value): | |
| _scan(item, f"{path}[{index}]") | |
| def validate_projection( | |
| state: Mapping[str, Any], source: Mapping[str, Any] | |
| ) -> tuple[dict[str, Any], dict[str, Any]]: | |
| state_payload = dict(state) | |
| source_payload = dict(source) | |
| _scan(state_payload) | |
| _scan(source_payload) | |
| if state_payload.get("schema") != STATE_SCHEMA: | |
| raise RefinementProjectionError("unsupported refinement state schema") | |
| state_digest = str(state_payload.get("state_sha256") or "") | |
| if not HEX_64.fullmatch(state_digest): | |
| raise RefinementProjectionError("state digest is missing") | |
| state_body = dict(state_payload) | |
| state_body.pop("state_sha256", None) | |
| if sha256_hex(canonical_bytes(state_body)) != state_digest: | |
| raise RefinementProjectionError("state digest mismatch") | |
| expected_state = { | |
| "ready": True, | |
| "content_access": "HANDLES_ONLY", | |
| "private_graph_present": False, | |
| "raw_reasoning_present": False, | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| } | |
| for key, expected in expected_state.items(): | |
| if state_payload.get(key) != expected: | |
| raise RefinementProjectionError(f"state boundary mismatch: {key}") | |
| handles = state_payload.get("handles") | |
| if not isinstance(handles, list): | |
| raise RefinementProjectionError("handles are missing") | |
| if state_payload.get("pattern_count") != len(handles): | |
| raise RefinementProjectionError("pattern count mismatch") | |
| seen: set[str] = set() | |
| for handle in handles: | |
| if not isinstance(handle, dict) or handle.get("schema") != HANDLE_SCHEMA: | |
| raise RefinementProjectionError("invalid pattern handle") | |
| handle_id = str(handle.get("id") or "") | |
| digest = str(handle.get("sha256") or "") | |
| pair = str(handle.get("model_pair_sha256") or "") | |
| code = str(handle.get("error_code") or "") | |
| if not handle_id.startswith("refinement:") or handle_id in seen: | |
| raise RefinementProjectionError("pattern handle identity is invalid") | |
| if not HEX_64.fullmatch(digest) or not HEX_64.fullmatch(pair): | |
| raise RefinementProjectionError("pattern handle digest is invalid") | |
| if not ERROR_CODE.fullmatch(code): | |
| raise RefinementProjectionError("pattern error code is invalid") | |
| for key in ("verified_repair_rate", "regression_rate", "mean_audit_confidence"): | |
| number = float(handle.get(key, -1.0)) | |
| if not 0.0 <= number <= 1.0: | |
| raise RefinementProjectionError(f"pattern metric is invalid: {key}") | |
| seen.add(handle_id) | |
| if source_payload.get("schema") != SOURCE_SCHEMA: | |
| raise RefinementProjectionError("unsupported source receipt schema") | |
| receipt_digest = str(source_payload.get("receipt_sha256") or "") | |
| if not HEX_64.fullmatch(receipt_digest): | |
| raise RefinementProjectionError("source receipt digest is missing") | |
| source_body = dict(source_payload) | |
| source_body.pop("receipt_sha256", None) | |
| if sha256_hex(canonical_bytes(source_body)) != receipt_digest: | |
| raise RefinementProjectionError("source receipt digest mismatch") | |
| if source_payload.get("source_repository") != "szl-holdings/szl-second-brain": | |
| raise RefinementProjectionError("source repository drifted") | |
| if source_payload.get("source_path") != "data/refinement-patterns.public.json": | |
| raise RefinementProjectionError("source path drifted") | |
| if source_payload.get("state_sha256") != state_digest: | |
| raise RefinementProjectionError("source/state digest mismatch") | |
| available = source_payload.get("available") | |
| revision = source_payload.get("source_revision") | |
| if available is True: | |
| if not HEX_40.fullmatch(str(revision or "")): | |
| raise RefinementProjectionError("available source revision is not exact") | |
| elif available is False: | |
| if revision not in (None, "") and not HEX_40.fullmatch(str(revision)): | |
| raise RefinementProjectionError("unavailable source revision is invalid") | |
| else: | |
| raise RefinementProjectionError("source availability is invalid") | |
| for key in ("training_authority", "promotion_authority", "execution_authority"): | |
| if source_payload.get(key) != "NONE": | |
| raise RefinementProjectionError(f"source carries forbidden {key}") | |
| return state_payload, source_payload | |
| def load_projection( | |
| state_path: str | Path, source_path: str | Path | |
| ) -> tuple[dict[str, Any], dict[str, Any]]: | |
| state = json.loads(Path(state_path).read_text(encoding="utf-8")) | |
| source = json.loads(Path(source_path).read_text(encoding="utf-8")) | |
| if not isinstance(state, dict) or not isinstance(source, dict): | |
| raise RefinementProjectionError("projection files must contain objects") | |
| return validate_projection(state, source) | |
| def filter_handles( | |
| state: Mapping[str, Any], *, error_code: str | None = None, limit: int = 50 | |
| ) -> list[dict[str, Any]]: | |
| if not 1 <= int(limit) <= 200: | |
| raise RefinementProjectionError("limit must be in [1, 200]") | |
| normalized = None | |
| if error_code is not None and str(error_code).strip(): | |
| normalized = str(error_code).strip().upper() | |
| if not ERROR_CODE.fullmatch(normalized): | |
| raise RefinementProjectionError("error_code filter is invalid") | |
| handles = state.get("handles") | |
| if not isinstance(handles, list): | |
| raise RefinementProjectionError("handles are missing") | |
| return [ | |
| dict(handle) | |
| for handle in handles | |
| if normalized is None or handle.get("error_code") == normalized | |
| ][: int(limit)] | |