Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| # SPDX-License-Identifier: Apache-2.0 | |
| """Source-bound Second Brain + frontier runtime for SZL Living Anatomy v7. | |
| The runtime loads two immutable public-memory planes bundled by the deployment | |
| workflow: | |
| 1. the 575-chunk governed retrieval projection; and | |
| 2. the review-gated frontier candidate set bound to its public source manifest. | |
| Every source revision, row digest, set digest, formula count, quant-domain count, | |
| and authority boundary is replayed before readiness. Public methods return handles, | |
| counts, and digests only. Candidate and corpus content remain internal to this | |
| read-only process. The owner's private graph is never loaded, model weights are not | |
| trained, and no promotion, execution, merge, or provider authority is granted. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| import re | |
| import threading | |
| from collections import Counter | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any | |
| from frontier_source_contract import ( | |
| FrontierSourceBindings, | |
| metadata_handle_identity, | |
| validate_source_revision, | |
| ) | |
| from pipeline_observation import load_pipeline_snapshot | |
| SCHEMA_HEALTH = "szl.living-anatomy.second-brain.health/v2" | |
| SCHEMA_SEARCH = "szl.living-anatomy.second-brain.search/v1" | |
| SCHEMA_CONTEXT = "szl.living-anatomy.second-brain.context/v1" | |
| SCHEMA_MANIFEST = "szl.living-anatomy.second-brain.manifest/v2" | |
| SCHEMA_FRONTIER = "szl.living-anatomy.frontier-search/v1" | |
| SCHEMA_FORMULAS = "szl.living-anatomy.formula-atlas/v1" | |
| SCHEMA_QUANT = "szl.living-anatomy.quant-atlas/v1" | |
| SCHEMA_OUROBOROS = "szl.living-anatomy.ouroboros-observation/v2" | |
| SCHEMA_NEURAL_QUANT = "szl.living-anatomy.neural-quant-v7/v1" | |
| SCHEMA_PIPELINE = "szl.living-anatomy.rag-dag-pipeline/v1" | |
| SOURCE_REPOSITORY = "szl-holdings/szl-second-brain" | |
| CANONICAL_DATASET = "SZLHOLDINGS/szl-second-brain-inrepo" | |
| PUBLIC_CHUNK_COUNT = 575 | |
| MIN_FRONTIER_CANDIDATES = 70 | |
| MIN_FRONTIER_SOURCES = 7 | |
| EXPECTED_ATTRIBUTED_FORMULAS = 30 | |
| EXPECTED_EXECUTABLE_FORMULAS = 21 | |
| EXPECTED_QUANT_DOMAINS = 9 | |
| EXPECTED_LOCKED_PROVEN = 8 | |
| PRIVATE_GRAPH_NODES_DISCLOSED = 9464 | |
| MAX_QUERY_CHARS = 500 | |
| MAX_K = 24 | |
| HEX_40 = re.compile(r"^[0-9a-f]{40}$") | |
| HEX_64 = re.compile(r"^[0-9a-f]{64}$") | |
| FRONTIER_ID = re.compile(r"^frontier:[0-9a-f]{32}$") | |
| TOKEN = re.compile(r"[a-z0-9λ_.:/+-]+", re.I) | |
| STOP = { | |
| "the", | |
| "is", | |
| "a", | |
| "an", | |
| "of", | |
| "and", | |
| "or", | |
| "to", | |
| "in", | |
| "for", | |
| "on", | |
| "at", | |
| "by", | |
| "as", | |
| "what", | |
| "which", | |
| "who", | |
| "how", | |
| "why", | |
| "does", | |
| "did", | |
| "are", | |
| "was", | |
| "be", | |
| "it", | |
| "this", | |
| "that", | |
| "with", | |
| "from", | |
| "into", | |
| "over", | |
| "not", | |
| } | |
| ROOT = Path(__file__).resolve().parent | |
| SNAPSHOT_ROOT = Path( | |
| os.environ.get( | |
| "SECOND_BRAIN_SNAPSHOT_ROOT", | |
| str(ROOT / ".runtime" / "second-brain"), | |
| ) | |
| ).resolve() | |
| def _utc_now() -> str: | |
| return ( | |
| datetime.now(timezone.utc) | |
| .replace(microsecond=0) | |
| .isoformat() | |
| .replace("+00:00", "Z") | |
| ) | |
| def _sha256_bytes(value: bytes) -> str: | |
| return hashlib.sha256(value).hexdigest() | |
| def _canonical_bytes(value: Any) -> bytes: | |
| return json.dumps( | |
| value, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| ensure_ascii=False, | |
| ).encode("utf-8") | |
| def _canonical_sha256(value: Any) -> str: | |
| return _sha256_bytes(_canonical_bytes(value)) | |
| def _tokenize(value: str) -> list[str]: | |
| return [ | |
| token.lower() | |
| for token in TOKEN.findall(value or "") | |
| if len(token) > 1 and token.lower() not in STOP | |
| ] | |
| def _read_json(path: Path) -> dict[str, Any]: | |
| payload = json.loads(path.read_text(encoding="utf-8")) | |
| if not isinstance(payload, dict): | |
| raise ValueError(f"{path.name} must contain a JSON object") | |
| return payload | |
| def _valid_revision(value: Any) -> bool: | |
| return isinstance(value, str) and bool(HEX_40.fullmatch(value.lower())) | |
| def _valid_digest(value: Any) -> bool: | |
| return isinstance(value, str) and bool(HEX_64.fullmatch(value.lower())) | |
| def _bounded_k(value: Any, default: int = 6) -> int: | |
| try: | |
| parsed = int(value) | |
| except (TypeError, ValueError): | |
| parsed = default | |
| return max(1, min(parsed, MAX_K)) | |
| class PublicSecondBrain: | |
| """Validated read-only index over retrieval and frontier public memory.""" | |
| def __init__(self, snapshot_root: Path | None = None) -> None: | |
| self.snapshot_root = Path(snapshot_root or SNAPSHOT_ROOT).resolve() | |
| self.manifest_path = self.snapshot_root / "manifest.json" | |
| self.corpus_path = self.snapshot_root / "brain-corpus.public.jsonl" | |
| self.frontier_state_path = self.snapshot_root / "frontier-state.v1.json" | |
| self.frontier_candidates_path = ( | |
| self.snapshot_root / "frontier-candidates.public.jsonl" | |
| ) | |
| self.source_path = self.snapshot_root / "source.json" | |
| self._lock = threading.RLock() | |
| self._rows: list[dict[str, Any]] = [] | |
| self._df: Counter[str] = Counter() | |
| self._frontier_rows: list[dict[str, Any]] = [] | |
| self._frontier_df: Counter[str] = Counter() | |
| self._manifest: dict[str, Any] = {} | |
| self._frontier_state: dict[str, Any] = {} | |
| self._formula_authority: dict[str, Any] = {} | |
| self._source: dict[str, Any] = {} | |
| self._load_error: str | None = None | |
| self._loaded_at: str | None = None | |
| self.reload() | |
| def reload(self) -> dict[str, Any]: | |
| with self._lock: | |
| self._rows = [] | |
| self._df = Counter() | |
| self._frontier_rows = [] | |
| self._frontier_df = Counter() | |
| self._manifest = {} | |
| self._frontier_state = {} | |
| self._formula_authority = {} | |
| self._source = {} | |
| self._load_error = None | |
| self._loaded_at = None | |
| try: | |
| manifest_raw = self.manifest_path.read_bytes() | |
| corpus_raw = self.corpus_path.read_bytes() | |
| frontier_state_raw = self.frontier_state_path.read_bytes() | |
| frontier_candidates_raw = self.frontier_candidates_path.read_bytes() | |
| source = _read_json(self.source_path) | |
| manifest = json.loads(manifest_raw.decode("utf-8")) | |
| frontier_state = json.loads(frontier_state_raw.decode("utf-8")) | |
| if not isinstance(manifest, dict): | |
| raise ValueError("manifest.json must contain a JSON object") | |
| if not isinstance(frontier_state, dict): | |
| raise ValueError("frontier-state.v1.json must contain a JSON object") | |
| self._validate_source_receipt( | |
| source, | |
| manifest_raw=manifest_raw, | |
| corpus_raw=corpus_raw, | |
| frontier_state_raw=frontier_state_raw, | |
| frontier_candidates_raw=frontier_candidates_raw, | |
| ) | |
| rows, document_frequency = self._load_retrieval_rows( | |
| manifest, | |
| corpus_raw, | |
| source, | |
| ) | |
| ( | |
| frontier_rows, | |
| frontier_frequency, | |
| formula_authority, | |
| ) = self._load_frontier_rows( | |
| frontier_state, | |
| frontier_candidates_raw, | |
| source, | |
| ) | |
| self._rows = rows | |
| self._df = document_frequency | |
| self._frontier_rows = frontier_rows | |
| self._frontier_df = frontier_frequency | |
| self._manifest = manifest | |
| self._frontier_state = frontier_state | |
| self._formula_authority = formula_authority | |
| self._source = source | |
| self._loaded_at = _utc_now() | |
| except Exception as exc: | |
| self._load_error = f"{type(exc).__name__}: {exc}" | |
| return self.health() | |
| def _validate_source_receipt( | |
| source: dict[str, Any], | |
| *, | |
| manifest_raw: bytes, | |
| corpus_raw: bytes, | |
| frontier_state_raw: bytes, | |
| frontier_candidates_raw: bytes, | |
| ) -> None: | |
| if source.get("schema") != "szl.second-brain.snapshot/v1": | |
| raise ValueError("unsupported source receipt schema") | |
| if source.get("source_repository") != SOURCE_REPOSITORY: | |
| raise ValueError("unexpected Second Brain source repository") | |
| if not _valid_revision(source.get("source_revision")): | |
| raise ValueError("Second Brain source revision is not an exact Git SHA") | |
| expected_digests = { | |
| "manifest_sha256": _sha256_bytes(manifest_raw), | |
| "corpus_sha256": _sha256_bytes(corpus_raw), | |
| "frontier_state_sha256": _sha256_bytes(frontier_state_raw), | |
| "frontier_candidates_sha256": _sha256_bytes(frontier_candidates_raw), | |
| } | |
| for key, measured in expected_digests.items(): | |
| if source.get(key) != measured: | |
| raise ValueError(f"Second Brain {key} mismatch") | |
| expected_authority = { | |
| "private_graph_nodes_materialized": 0, | |
| "raw_graph_nodes_admitted_to_gradients": 0, | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| "lambda_state": "CONJECTURE_1", | |
| } | |
| for key, value in expected_authority.items(): | |
| if source.get(key) != value: | |
| raise ValueError(f"Second Brain source authority drift: {key}") | |
| def _load_retrieval_rows( | |
| manifest: dict[str, Any], | |
| corpus_raw: bytes, | |
| source: dict[str, Any], | |
| ) -> tuple[list[dict[str, Any]], Counter[str]]: | |
| rows: list[dict[str, Any]] = [] | |
| document_frequency: Counter[str] = Counter() | |
| ids: set[str] = set() | |
| for line_number, line in enumerate( | |
| corpus_raw.decode("utf-8").splitlines(), | |
| start=1, | |
| ): | |
| if not line.strip(): | |
| continue | |
| row = json.loads(line) | |
| if not isinstance(row, dict) or not row.get("id"): | |
| raise ValueError(f"invalid corpus row at line {line_number}") | |
| node_id = str(row["id"]) | |
| if node_id in ids: | |
| raise ValueError(f"duplicate corpus id at line {line_number}") | |
| ids.add(node_id) | |
| text = str(row.get("text") or "") | |
| declared_digest = str(row.get("sha256") or "") | |
| measured_digest = _sha256_bytes(text.encode("utf-8")) | |
| if declared_digest and declared_digest != measured_digest: | |
| raise ValueError(f"row digest mismatch at line {line_number}") | |
| tokens = _tokenize(f"{row.get('title', '')} {text}") | |
| stored = { | |
| "id": node_id, | |
| "title": str(row.get("title") or ""), | |
| "source": str(row.get("source") or "unknown"), | |
| "sourceId": row.get("sourceId"), | |
| "sha256": measured_digest, | |
| "_tf": Counter(tokens), | |
| } | |
| rows.append(stored) | |
| document_frequency.update(set(tokens)) | |
| declared_count = int(manifest.get("publicChunkCount") or 0) | |
| receipt_count = int(source.get("public_chunk_count") or 0) | |
| if declared_count != len(rows) or receipt_count != len(rows): | |
| raise ValueError( | |
| "public chunk count mismatch " | |
| f"(manifest={declared_count}, receipt={receipt_count}, loaded={len(rows)})" | |
| ) | |
| if len(rows) != PUBLIC_CHUNK_COUNT: | |
| raise ValueError( | |
| f"expected {PUBLIC_CHUNK_COUNT} public chunks, loaded {len(rows)}" | |
| ) | |
| if str(manifest.get("secretScan") or "").upper() != "PASS": | |
| raise ValueError("public projection secret scan is not PASS") | |
| return rows, document_frequency | |
| def _load_frontier_rows( | |
| state: dict[str, Any], | |
| candidates_raw: bytes, | |
| source: dict[str, Any], | |
| ) -> tuple[list[dict[str, Any]], Counter[str], dict[str, Any]]: | |
| if state.get("schema") != "szl.second-brain.frontier-state/v1": | |
| raise ValueError("unsupported frontier state schema") | |
| expected_state = { | |
| "state": "REVIEW_REQUIRED", | |
| "public_content_access": "HANDLES_ONLY", | |
| "controller_content_access": "AUTHORIZED_CONTROLLER_ONLY", | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| "lambda": "CONJECTURE_1", | |
| } | |
| for key, value in expected_state.items(): | |
| if state.get(key) != value: | |
| raise ValueError(f"frontier authority drift: {key}") | |
| if int(state.get("private_graph_nodes_loaded") or 0) != 0: | |
| raise ValueError("private graph material entered the frontier index") | |
| if int(state.get("raw_graph_nodes_admitted_to_gradients") or 0) != 0: | |
| raise ValueError("frontier index admitted raw graph nodes to gradients") | |
| source_count = state.get("source_count") | |
| sources = state.get("sources") | |
| if type(source_count) is not int or source_count < MIN_FRONTIER_SOURCES: | |
| raise ValueError("frontier source inventory is incomplete") | |
| if not isinstance(sources, list) or len(sources) != source_count: | |
| raise ValueError("frontier source manifest count drifted") | |
| receipt_source_count = source.get("frontier_source_count") | |
| if type(receipt_source_count) is not int or receipt_source_count != source_count: | |
| raise ValueError("frontier source receipt count mismatch") | |
| source_ids: set[str] = set() | |
| expected_source_counts: dict[tuple[str, str, str], int] = {} | |
| observed_source_counts: dict[tuple[str, str, str], int] = {} | |
| for index, entry in enumerate(sources, start=1): | |
| if not isinstance(entry, dict): | |
| raise ValueError(f"invalid frontier source at index {index}") | |
| source_id = str(entry.get("source_id") or "") | |
| repository = str(entry.get("repository") or "") | |
| revision = str(entry.get("revision") or "") | |
| digest = str(entry.get("content_sha256") or "") | |
| parser = entry.get("parser") | |
| path = entry.get("path") | |
| candidate_count = entry.get("candidate_count") | |
| if not source_id or source_id in source_ids: | |
| raise ValueError(f"frontier source identity failed at index {index}") | |
| if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository): | |
| raise ValueError(f"frontier source repository failed at index {index}") | |
| validate_source_revision(entry) | |
| if not HEX_64.fullmatch(digest): | |
| raise ValueError(f"frontier source binding failed at index {index}") | |
| if ( | |
| not isinstance(parser, str) | |
| or not parser.strip() | |
| or not isinstance(path, str) | |
| or not path.strip() | |
| or type(candidate_count) is not int | |
| or candidate_count < 1 | |
| ): | |
| raise ValueError(f"frontier source metadata failed at index {index}") | |
| binding = (repository, revision, path) | |
| if binding in expected_source_counts: | |
| raise ValueError(f"duplicate frontier source binding at index {index}") | |
| source_ids.add(source_id) | |
| expected_source_counts[binding] = candidate_count | |
| observed_source_counts[binding] = 0 | |
| source_bindings = FrontierSourceBindings(sources) | |
| kind_counts = state.get("source_kind_counts") | |
| domain_counts = state.get("quant_domain_counts") | |
| if not isinstance(kind_counts, dict) or not isinstance(domain_counts, dict): | |
| raise ValueError("frontier formula/quant summaries are unavailable") | |
| if int(kind_counts.get("attributed-formula") or 0) != EXPECTED_ATTRIBUTED_FORMULAS: | |
| raise ValueError("attributed formula count drifted") | |
| if int(kind_counts.get("executable-formula") or 0) != EXPECTED_EXECUTABLE_FORMULAS: | |
| raise ValueError("executable formula count drifted") | |
| if int(kind_counts.get("quant-domain") or 0) != EXPECTED_QUANT_DOMAINS: | |
| raise ValueError("quant-domain record count drifted") | |
| if len(domain_counts) != EXPECTED_QUANT_DOMAINS: | |
| raise ValueError("quant-domain taxonomy drifted") | |
| rows: list[dict[str, Any]] = [] | |
| document_frequency: Counter[str] = Counter() | |
| ids: set[str] = set() | |
| canonical_lines: list[bytes] = [] | |
| formula_authority: dict[str, Any] | None = None | |
| for line_number, line in enumerate( | |
| candidates_raw.decode("utf-8").splitlines(), | |
| start=1, | |
| ): | |
| if not line.strip(): | |
| continue | |
| row = json.loads(line) | |
| if ( | |
| not isinstance(row, dict) | |
| or row.get("schema") | |
| != "szl.second-brain.frontier-candidate/v1" | |
| ): | |
| raise ValueError(f"invalid frontier candidate at line {line_number}") | |
| candidate_id = str(row.get("id") or "") | |
| if not FRONTIER_ID.fullmatch(candidate_id) or candidate_id in ids: | |
| raise ValueError( | |
| f"frontier candidate identity failed at line {line_number}" | |
| ) | |
| ids.add(candidate_id) | |
| revision = str(row.get("source_revision") or "") | |
| digest = str(row.get("content_sha256") or "") | |
| if not _valid_digest(digest): | |
| raise ValueError( | |
| f"frontier source binding failed at line {line_number}" | |
| ) | |
| binding = source_bindings.binding_for(row) | |
| if binding not in expected_source_counts: | |
| raise ValueError(f"frontier source manifest binding failed at line {line_number}") | |
| observed_source_counts[binding] += 1 | |
| content = str(row.get("content") or "") | |
| if _sha256_bytes(content.encode("utf-8")) != digest: | |
| raise ValueError(f"frontier digest mismatch at line {line_number}") | |
| if row.get("candidate_state") != "DISCOVERED_REVIEW_REQUIRED": | |
| raise ValueError( | |
| f"frontier candidate was promoted at line {line_number}" | |
| ) | |
| if row.get("content_access") != "CONTROLLER_ONLY": | |
| raise ValueError( | |
| f"frontier content boundary drifted at line {line_number}" | |
| ) | |
| tokens = _tokenize( | |
| " ".join( | |
| ( | |
| str(row.get("title") or ""), | |
| str(row.get("source_repository") or ""), | |
| str(row.get("source_path") or ""), | |
| str(row.get("source_kind") or ""), | |
| str(row.get("quant_domain") or ""), | |
| content, | |
| ) | |
| ) | |
| ) | |
| stored = { | |
| "id": candidate_id, | |
| "title": str(row.get("title") or ""), | |
| "content": content, | |
| "sha256": digest, | |
| "source_repository": str(row.get("source_repository") or ""), | |
| "source_revision": revision, | |
| "source_revision_kind": row.get("source_revision_kind", "git-sha1"), | |
| "provenance": row.get("provenance"), | |
| "source_path": str(row.get("source_path") or ""), | |
| "source_kind": str(row.get("source_kind") or ""), | |
| "quant_domain": ( | |
| str(row.get("quant_domain")) | |
| if row.get("quant_domain") | |
| else None | |
| ), | |
| "admission": str(row.get("admission") or ""), | |
| "_tf": Counter(tokens), | |
| } | |
| rows.append(stored) | |
| document_frequency.update(set(tokens)) | |
| if stored["source_kind"] == "formula-authority": | |
| formula_authority = json.loads(content) | |
| canonical_lines.append(_canonical_bytes(row) + b"\n") | |
| declared_count = int(state.get("candidate_count") or -1) | |
| receipt_count = int(source.get("frontier_candidate_count") or -1) | |
| if ( | |
| len(rows) != declared_count | |
| or len(rows) != receipt_count | |
| or len(rows) < MIN_FRONTIER_CANDIDATES | |
| ): | |
| raise ValueError( | |
| "frontier candidate count mismatch " | |
| f"(state={declared_count}, receipt={receipt_count}, loaded={len(rows)})" | |
| ) | |
| if observed_source_counts != expected_source_counts: | |
| raise ValueError("frontier per-source candidate counts mismatch") | |
| source_bindings.verify() | |
| measured_set = _sha256_bytes(b"".join(canonical_lines)) | |
| if state.get("candidate_set_sha256") != measured_set: | |
| raise ValueError("frontier state candidate-set digest mismatch") | |
| if source.get("frontier_candidate_set_sha256") != measured_set: | |
| raise ValueError("frontier receipt candidate-set digest mismatch") | |
| if not isinstance(formula_authority, dict): | |
| raise ValueError("frontier formula authority is unavailable") | |
| if int(formula_authority.get("locked_proven_count") or 0) != EXPECTED_LOCKED_PROVEN: | |
| raise ValueError("locked-proven formula count drifted") | |
| if ( | |
| formula_authority.get("lambda_status") | |
| != "CONJECTURE_1_OPEN_ADVISORY_ONLY" | |
| ): | |
| raise ValueError("Lambda formula authority drifted") | |
| if ( | |
| formula_authority.get("f_number_to_executable_registry_mapping") | |
| != "UNKNOWN_NOT_INFERRED" | |
| ): | |
| raise ValueError("unknown formula mapping was silently inferred") | |
| return rows, document_frequency, formula_authority | |
| def ready(self) -> bool: | |
| return ( | |
| self._load_error is None | |
| and len(self._rows) == PUBLIC_CHUNK_COUNT | |
| and len(self._frontier_rows) >= MIN_FRONTIER_CANDIDATES | |
| ) | |
| def source_revision(self) -> str | None: | |
| value = self._source.get("source_revision") | |
| return str(value) if _valid_revision(value) else None | |
| def frontier_candidate_set_sha256(self) -> str | None: | |
| value = self._frontier_state.get("candidate_set_sha256") | |
| return str(value) if _valid_digest(value) else None | |
| def health(self) -> dict[str, Any]: | |
| with self._lock: | |
| by_source: dict[str, int] = {} | |
| for row in self._rows: | |
| source = str(row.get("source") or "unknown") | |
| by_source[source] = by_source.get(source, 0) + 1 | |
| return { | |
| "schema": SCHEMA_HEALTH, | |
| "service": "living-anatomy-yachay-second-brain-v7", | |
| "ready": self.ready, | |
| "state": ( | |
| "SOURCE_BOUND_RETRIEVAL_AND_FRONTIER" | |
| if self.ready | |
| else "UNAVAILABLE" | |
| ), | |
| "transport_state": "REACHABLE", | |
| "evidence_state": "MEASURED" if self.ready else "UNAVAILABLE", | |
| "verification_state": ( | |
| "STRUCTURAL_ONLY" if self.ready else "FAILED" | |
| ), | |
| "authority_state": "READ_ONLY", | |
| "kind": "SOFTWARE", | |
| "content_access": "HANDLES_ONLY", | |
| "source_repository": SOURCE_REPOSITORY, | |
| "source_revision": self.source_revision, | |
| "canonical_dataset": CANONICAL_DATASET, | |
| "chunk_count": len(self._rows), | |
| "declared_public_chunk_count": PUBLIC_CHUNK_COUNT, | |
| "by_source": by_source, | |
| "frontier": { | |
| "ready": self.ready, | |
| "state": self._frontier_state.get("state", "UNAVAILABLE"), | |
| "candidate_count": len(self._frontier_rows), | |
| "source_count": self._frontier_state.get("source_count", 0), | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "public_content_access": "HANDLES_ONLY", | |
| "internal_content_access": "READ_ONLY_PROCESS_INTERNAL", | |
| "candidate_state": "DISCOVERED_REVIEW_REQUIRED", | |
| "learning_definition": self._frontier_state.get( | |
| "learning_definition" | |
| ), | |
| }, | |
| "formula_atlas": { | |
| "attributed_formula_count": int( | |
| self._frontier_state.get("source_kind_counts", {}).get( | |
| "attributed-formula", | |
| 0, | |
| ) | |
| ), | |
| "executable_formula_count": int( | |
| self._frontier_state.get("source_kind_counts", {}).get( | |
| "executable-formula", | |
| 0, | |
| ) | |
| ), | |
| "locked_proven_count": self._formula_authority.get( | |
| "locked_proven_count", | |
| 0, | |
| ), | |
| "locked_proven_ids": self._formula_authority.get( | |
| "locked_proven_ids", | |
| [], | |
| ), | |
| "mapping": self._formula_authority.get( | |
| "f_number_to_executable_registry_mapping", | |
| "UNKNOWN_NOT_INFERRED", | |
| ), | |
| "lambda_status": self._formula_authority.get( | |
| "lambda_status", | |
| "CONJECTURE_1_OPEN_ADVISORY_ONLY", | |
| ), | |
| }, | |
| "quant_domain_count": len( | |
| self._frontier_state.get("quant_domain_counts", {}) | |
| ), | |
| "snapshot_root": str(self.snapshot_root), | |
| "loaded_at": self._loaded_at, | |
| "load_error": self._load_error, | |
| "index_is_model_weights": False, | |
| "private_graph_nodes_loaded": 0, | |
| "private_graph_nodes_disclosed_elsewhere": ( | |
| PRIVATE_GRAPH_NODES_DISCLOSED | |
| ), | |
| "raw_graph_nodes_admitted_to_gradients": 0, | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| "lambda_state": "CONJECTURE_1", | |
| "limits": [ | |
| "Lexical overlap is not correctness.", | |
| "Public APIs return handles and digests, never corpus or candidate content.", | |
| "Frontier candidates require separate human-governed review.", | |
| "The owner's private graph is not bundled, queried, or exposed.", | |
| "This read-only organ cannot train, promote, merge, authorize, or execute an action.", | |
| ], | |
| } | |
| def manifest(self) -> dict[str, Any]: | |
| health = self.health() | |
| return { | |
| "schema": SCHEMA_MANIFEST, | |
| "service": health["service"], | |
| "ready": health["ready"], | |
| "source": { | |
| "repository": SOURCE_REPOSITORY, | |
| "revision": self.source_revision, | |
| "dataset": CANONICAL_DATASET, | |
| "snapshot_receipt": self._source, | |
| }, | |
| "retrieval": { | |
| "chunk_count": len(self._rows), | |
| "by_source": health["by_source"], | |
| "manifest": self._manifest, | |
| }, | |
| "frontier": { | |
| "state": self._frontier_state, | |
| "candidate_count": len(self._frontier_rows), | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "formula_atlas": health["formula_atlas"], | |
| "quant_domain_count": health["quant_domain_count"], | |
| }, | |
| "interfaces": { | |
| "health": "/api/anatomy/v1/brain/health", | |
| "manifest": "/api/anatomy/v1/brain/manifest", | |
| "search": "/api/anatomy/v1/brain/search", | |
| "context": "/api/anatomy/v1/brain/context", | |
| "frontier": "/api/anatomy/v1/brain/frontier", | |
| "formulas": "/api/anatomy/v1/brain/formulas", | |
| "quant": "/api/anatomy/v1/brain/quant", | |
| "ouroboros": "/api/anatomy/v1/brain/ouroboros", | |
| "pipeline": "/api/anatomy/v1/brain/pipeline", | |
| "neural_quant_v7": "/api/anatomy/v1/brain/neural-quant-v7", | |
| }, | |
| "authority_state": "READ_ONLY", | |
| "content_access": "HANDLES_ONLY", | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| "limits": health["limits"], | |
| } | |
| def _retrieval_handle(row: dict[str, Any]) -> dict[str, Any]: | |
| return { | |
| "nodeId": row["id"], | |
| "nodeKind": "INDEX", | |
| "label": "DECLARED", | |
| "note": str(row.get("title") or "")[:160], | |
| "source": row.get("source"), | |
| "sourceId": row.get("sourceId"), | |
| "sha256": row.get("sha256"), | |
| } | |
| def _frontier_handle(row: dict[str, Any]) -> dict[str, Any]: | |
| handle: dict[str, Any] = { | |
| "nodeId": row["id"], | |
| "nodeKind": "FRONTIER_CANDIDATE", | |
| "label": "REVIEW_REQUIRED", | |
| "note": str(row.get("title") or "")[:180], | |
| "sourceRepository": row.get("source_repository"), | |
| "sourceRevision": row.get("source_revision"), | |
| "sourcePath": row.get("source_path"), | |
| "sourceKind": row.get("source_kind"), | |
| "sha256": row.get("sha256"), | |
| "admission": row.get("admission"), | |
| "candidateState": "DISCOVERED_REVIEW_REQUIRED", | |
| "contentAccess": "HANDLES_ONLY", | |
| } | |
| if row.get("quant_domain"): | |
| handle["quantDomain"] = row["quant_domain"] | |
| handle.update(metadata_handle_identity(row)) | |
| return handle | |
| def _score_rows( | |
| query: str, | |
| rows: list[dict[str, Any]], | |
| document_frequency: Counter[str], | |
| *, | |
| k: int, | |
| ) -> list[tuple[float, dict[str, Any]]]: | |
| query_tokens = _tokenize(query) | |
| if not query_tokens: | |
| return [] | |
| query_frequency = Counter(query_tokens) | |
| total = max(1, len(rows)) | |
| scored: list[tuple[float, dict[str, Any]]] = [] | |
| for row in rows: | |
| score = 0.0 | |
| row_frequency: Counter[str] = row["_tf"] | |
| length = max(1, sum(row_frequency.values())) | |
| for term, query_count in query_frequency.items(): | |
| term_frequency = row_frequency.get(term, 0) | |
| if not term_frequency: | |
| continue | |
| inverse_document_frequency = math.log( | |
| 1.0 | |
| + (total - document_frequency.get(term, 0) + 0.5) | |
| / (document_frequency.get(term, 0) + 0.5) | |
| ) | |
| score += ( | |
| inverse_document_frequency | |
| * ((term_frequency * 2.2) / ( | |
| term_frequency + 1.2 + 0.75 * length / 180.0 | |
| )) | |
| * query_count | |
| ) | |
| if score > 0: | |
| scored.append((score, row)) | |
| scored.sort(key=lambda item: (-item[0], str(item[1]["id"]))) | |
| return scored[:k] | |
| def search(self, query: str, k: int = 6) -> dict[str, Any]: | |
| query = str(query or "").strip()[:MAX_QUERY_CHARS] | |
| requested_k = _bounded_k(k, default=6) | |
| with self._lock: | |
| if not self.ready: | |
| return { | |
| "schema": SCHEMA_SEARCH, | |
| "ready": False, | |
| "query": query, | |
| "handles": [], | |
| "scores": [], | |
| "source_revision": self.source_revision, | |
| "error": self._load_error or "Second Brain snapshot unavailable", | |
| "authority_state": "READ_ONLY", | |
| "content_access": "HANDLES_ONLY", | |
| } | |
| top = self._score_rows( | |
| query, | |
| self._rows, | |
| self._df, | |
| k=requested_k, | |
| ) | |
| handles = [self._retrieval_handle(row) for _, row in top] | |
| scores = [round(score, 6) for score, _ in top] | |
| result_body = { | |
| "query": query, | |
| "handles": handles, | |
| "scores": scores, | |
| "source_revision": self.source_revision, | |
| "corpus_chunk_count": len(self._rows), | |
| "ranking": "BM25_LIKE_LEXICAL", | |
| "content_access": "HANDLES_ONLY", | |
| } | |
| return { | |
| "schema": SCHEMA_SEARCH, | |
| "ready": True, | |
| **result_body, | |
| "result_sha256": _canonical_sha256(result_body), | |
| "authority_state": "READ_ONLY", | |
| "index_is_model_weights": False, | |
| "honesty": ( | |
| "Ranked lexical overlap over the public source-bound projection; " | |
| "scores are relevance signals, never correctness or proof." | |
| ), | |
| } | |
| def context(self, query: str, k: int = 6) -> dict[str, Any]: | |
| search = self.search(query, k=k) | |
| handles = search.get("handles") if isinstance(search, dict) else [] | |
| handles = handles if isinstance(handles, list) else [] | |
| model_handles = [ | |
| { | |
| key: handle[key] | |
| for key in ("nodeId", "nodeKind", "label", "note") | |
| if key in handle | |
| } | |
| for handle in handles | |
| if isinstance(handle, dict) | |
| ] | |
| evidence = [ | |
| { | |
| "node_id": handle.get("nodeId"), | |
| "source": handle.get("source"), | |
| "source_id": handle.get("sourceId"), | |
| "sha256": handle.get("sha256"), | |
| } | |
| for handle in handles | |
| if isinstance(handle, dict) | |
| ] | |
| body = { | |
| "query": search.get("query"), | |
| "model_handles": model_handles, | |
| "evidence": evidence, | |
| "source_revision": search.get("source_revision"), | |
| "ready": bool(search.get("ready")), | |
| } | |
| return { | |
| "schema": SCHEMA_CONTEXT, | |
| **body, | |
| "context_sha256": _canonical_sha256(body), | |
| "training_authority": "NONE", | |
| "write_authority": "NONE", | |
| "private_graph_nodes_loaded": 0, | |
| "honesty": search.get("honesty"), | |
| "error": search.get("error"), | |
| } | |
| def frontier_search( | |
| self, | |
| query: str, | |
| k: int = 12, | |
| *, | |
| source_kinds: set[str] | None = None, | |
| quant_domain: str | None = None, | |
| source_repository: str | None = None, | |
| ) -> dict[str, Any]: | |
| query = str(query or "").strip()[:MAX_QUERY_CHARS] | |
| requested_k = _bounded_k(k, default=12) | |
| with self._lock: | |
| if not self.ready: | |
| return { | |
| "schema": SCHEMA_FRONTIER, | |
| "ready": False, | |
| "state": "UNAVAILABLE", | |
| "query": query, | |
| "handles": [], | |
| "scores": [], | |
| "source_revision": self.source_revision, | |
| "error": self._load_error or "frontier snapshot unavailable", | |
| "content_access": "HANDLES_ONLY", | |
| "execution_authority": "NONE", | |
| } | |
| filtered = [ | |
| row | |
| for row in self._frontier_rows | |
| if ( | |
| source_kinds is None | |
| or row.get("source_kind") in source_kinds | |
| ) | |
| and ( | |
| quant_domain is None | |
| or row.get("quant_domain") == quant_domain | |
| ) | |
| and ( | |
| source_repository is None | |
| or row.get("source_repository") == source_repository | |
| ) | |
| ] | |
| local_df: Counter[str] = Counter() | |
| for row in filtered: | |
| local_df.update(row["_tf"].keys()) | |
| top = self._score_rows( | |
| query, | |
| filtered, | |
| local_df, | |
| k=requested_k, | |
| ) | |
| handles = [self._frontier_handle(row) for _, row in top] | |
| scores = [round(score, 6) for score, _ in top] | |
| body = { | |
| "query": query, | |
| "handles": handles, | |
| "scores": scores, | |
| "candidate_count": len(self._frontier_rows), | |
| "filtered_candidate_count": len(filtered), | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "source_revision": self.source_revision, | |
| "ranking": "LEXICAL_RELEVANCE_NOT_CORRECTNESS", | |
| "content_access": "HANDLES_ONLY", | |
| } | |
| return { | |
| "schema": SCHEMA_FRONTIER, | |
| "ready": True, | |
| "state": "REVIEW_REQUIRED", | |
| **body, | |
| "result_sha256": _canonical_sha256(body), | |
| "candidate_state": "DISCOVERED_REVIEW_REQUIRED", | |
| "training_authority": "NONE", | |
| "promotion_authority": "NONE", | |
| "execution_authority": "NONE", | |
| "merge_authority": "NONE", | |
| } | |
| def formula_view(self, query: str = "formula authority", k: int = 24) -> dict[str, Any]: | |
| kinds = { | |
| "formula-authority", | |
| "attributed-formula", | |
| "executable-formula", | |
| } | |
| search = self.frontier_search( | |
| query or "formula authority proof status", | |
| k=k, | |
| source_kinds=kinds, | |
| ) | |
| return { | |
| "schema": SCHEMA_FORMULAS, | |
| "ready": search.get("ready", False), | |
| "state": search.get("state", "UNAVAILABLE"), | |
| "source_revision": self.source_revision, | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "attributed_formula_count": EXPECTED_ATTRIBUTED_FORMULAS, | |
| "executable_formula_count": EXPECTED_EXECUTABLE_FORMULAS, | |
| "locked_proven_count": self._formula_authority.get( | |
| "locked_proven_count", | |
| 0, | |
| ), | |
| "locked_proven_ids": self._formula_authority.get( | |
| "locked_proven_ids", | |
| [], | |
| ), | |
| "f_number_mapping": self._formula_authority.get( | |
| "f_number_to_executable_registry_mapping", | |
| "UNKNOWN_NOT_INFERRED", | |
| ), | |
| "lambda_status": self._formula_authority.get( | |
| "lambda_status", | |
| "CONJECTURE_1_OPEN_ADVISORY_ONLY", | |
| ), | |
| "handles": search.get("handles", []), | |
| "scores": search.get("scores", []), | |
| "content_access": "HANDLES_ONLY", | |
| "proof_boundary": ( | |
| "Per-obligation status, source-reported status, and locked-proven " | |
| "membership are separate. No status string promotes a formula." | |
| ), | |
| "execution_authority": "NONE", | |
| } | |
| def quant_view( | |
| self, | |
| query: str = "quant math information geometry coding trust energy", | |
| k: int = 24, | |
| ) -> dict[str, Any]: | |
| search = self.frontier_search( | |
| query, | |
| k=k, | |
| source_kinds={ | |
| "quant-domain", | |
| "attributed-formula", | |
| "executable-formula", | |
| }, | |
| ) | |
| domain_counts = self._frontier_state.get("quant_domain_counts", {}) | |
| domains = [ | |
| { | |
| "id": domain, | |
| "candidate_count": int(count), | |
| "authority": "REFERENCE_AND_CONSTRAINT_INPUT_ONLY", | |
| } | |
| for domain, count in sorted(domain_counts.items()) | |
| ] | |
| return { | |
| "schema": SCHEMA_QUANT, | |
| "ready": search.get("ready", False), | |
| "state": search.get("state", "UNAVAILABLE"), | |
| "source_revision": self.source_revision, | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "quant_domain_count": len(domains), | |
| "domains": domains, | |
| "handles": search.get("handles", []), | |
| "scores": search.get("scores", []), | |
| "content_access": "HANDLES_ONLY", | |
| "lambda_status": "CONJECTURE_1_OPEN_ADVISORY_ONLY", | |
| "execution_authority": "NONE", | |
| } | |
| def _pipeline_snapshot(self) -> dict[str, Any]: | |
| controller = next((source for source in self._frontier_state.get("sources", []) | |
| if source.get("source_id") == "ouroboros_runtime"), {}) | |
| return load_pipeline_snapshot( | |
| self.snapshot_root / "pipeline-source.json", | |
| { | |
| "controller_repository": "szl-holdings/szl-ouroboros", | |
| "controller_revision": controller.get("revision"), | |
| "workflow": ".github/workflows/codex-continuous-frontier.yml", | |
| "second_brain_repository": SOURCE_REPOSITORY, | |
| "second_brain_revision": self.source_revision, | |
| "state_file_sha256": self._source.get("frontier_state_sha256"), | |
| "candidate_file_sha256": self._source.get("frontier_candidates_sha256"), | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "candidate_count": len(self._frontier_rows), | |
| }, | |
| file_sha256=self._source.get("pipeline_source_sha256") if self.ready else None, | |
| ) | |
| def pipeline_view(self) -> dict[str, Any]: | |
| snapshot = self._pipeline_snapshot() | |
| core = { | |
| "source_revision": self.source_revision, | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "rag": { | |
| "state": "SOURCE_BOUND_RETRIEVAL" if self.ready else "UNAVAILABLE", | |
| "chunk_count": len(self._rows), | |
| "candidate_count": len(self._frontier_rows), | |
| "context_endpoint": "/api/anatomy/v1/brain/context", | |
| }, | |
| "upstream": snapshot, | |
| "content_access": "HANDLES_ONLY", | |
| "authority": { | |
| "training": "NONE", "promotion": "NONE", "execution": "NONE", | |
| "merge": "NONE", "provider_mutation": "NONE", | |
| }, | |
| "execution_authorized": False, | |
| } | |
| return { | |
| "schema": SCHEMA_PIPELINE, | |
| "ready": self.ready, | |
| "state": "SOURCE_BOUND_INSPECTION" if snapshot["ready"] else "PARTIAL", | |
| **core, | |
| "view_sha256": _canonical_sha256(core), | |
| } | |
| def ouroboros_view(self, k: int = 16) -> dict[str, Any]: | |
| source_receipt = next( | |
| ( | |
| source | |
| for source in self._frontier_state.get("sources", []) | |
| if source.get("source_id") == "ouroboros_runtime" | |
| and source.get("repository") in { | |
| "szl-holdings/szl-ouroboros", | |
| "szl-holdings/ouroboros", | |
| } | |
| ), | |
| {}, | |
| ) | |
| # The maintained Python kernel and archived TypeScript lineage are | |
| # different identities. Select handles using the validated source | |
| # receipt, never a fixed repository alias or an unfiltered fallback. | |
| search = self.frontier_search( | |
| "ouroboros bounded loop convergence termination receipt closure codex", | |
| k=k, | |
| source_repository=source_receipt.get("repository", ""), | |
| ) | |
| ready = bool(search.get("ready") and source_receipt) | |
| pipeline = self._pipeline_snapshot() | |
| run_observation = pipeline.get("ouroboros_observation") or {} | |
| measurements = run_observation.get("observation") or {} | |
| observed = run_observation.get("state") == "OBSERVED" | |
| return { | |
| "schema": SCHEMA_OUROBOROS, | |
| "ready": ready, | |
| "state": "REVIEW_REQUIRED" if ready else "UNAVAILABLE", | |
| "observation_state": ("RECORDED_REVIEW_ATTEMPT" if observed | |
| else "SOURCE_METADATA_ONLY" if ready else "UNAVAILABLE"), | |
| "source_revision": self.source_revision, | |
| "source": source_receipt, | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "handles": search.get("handles", []) if ready else [], | |
| "scores": search.get("scores", []) if ready else [], | |
| "contract_requirements": { | |
| "bounded": True, | |
| "terminating": True, | |
| "receipt_closed": True, | |
| }, | |
| "loop_contract": { | |
| "bounded": measurements.get("bounded") if observed else None, | |
| "terminating": measurements.get("terminated") if observed else None, | |
| "receipt_closed": measurements.get("receipt_closed") if observed else None, | |
| "codex_role": "ADVISORY_REVIEW_ONLY", | |
| "recommendations_executed": False, | |
| }, | |
| "runtime_evidence": { | |
| "state": run_observation.get("state", pipeline["state"]), | |
| "run_id": (run_observation.get("run") or {}).get("id"), | |
| "receipt_sha256": (run_observation.get("artifact") or {}).get("receipt_sha256"), | |
| "reason": run_observation.get("reason", pipeline["reason"]), | |
| "scope": "RECORDED_REVIEW_ATTEMPT", | |
| "signature_verified": False, | |
| "durable_external_acknowledgment_verified": False, | |
| }, | |
| "content_access": "HANDLES_ONLY", | |
| "training_authority": "NONE", | |
| "execution_authority": "NONE", | |
| } | |
| def neural_quant_view(self, k: int = 24) -> dict[str, Any]: | |
| health = self.health() | |
| formulas = self.formula_view(k=k) | |
| quant = self.quant_view(k=k) | |
| ouroboros = self.ouroboros_view(k=min(k, 16)) | |
| core = { | |
| "source_revision": self.source_revision, | |
| "candidate_set_sha256": self.frontier_candidate_set_sha256, | |
| "brain": { | |
| "chunk_count": health["chunk_count"], | |
| "frontier_candidate_count": health["frontier"][ | |
| "candidate_count" | |
| ], | |
| "frontier_source_count": health["frontier"]["source_count"], | |
| "candidate_state": "DISCOVERED_REVIEW_REQUIRED", | |
| }, | |
| "formulas": { | |
| "attributed": formulas["attributed_formula_count"], | |
| "executable": formulas["executable_formula_count"], | |
| "locked_proven": formulas["locked_proven_count"], | |
| "locked_proven_ids": formulas["locked_proven_ids"], | |
| "mapping": formulas["f_number_mapping"], | |
| "lambda_status": formulas["lambda_status"], | |
| "handles": formulas["handles"], | |
| }, | |
| "quant": { | |
| "domain_count": quant["quant_domain_count"], | |
| "domains": quant["domains"], | |
| "handles": quant["handles"], | |
| }, | |
| "ouroboros": { | |
| "source": ouroboros["source"], | |
| "observation_state": ouroboros["observation_state"], | |
| "contract_requirements": ouroboros["contract_requirements"], | |
| "loop_contract": ouroboros["loop_contract"], | |
| "runtime_evidence": ouroboros["runtime_evidence"], | |
| "handles": ouroboros["handles"], | |
| }, | |
| "pipeline": self.pipeline_view(), | |
| "content_access": "HANDLES_ONLY", | |
| "authority": { | |
| "training": "NONE", | |
| "promotion": "NONE", | |
| "execution": "NONE", | |
| "merge": "NONE", | |
| "provider_mutation": "NONE", | |
| }, | |
| } | |
| return { | |
| "schema": SCHEMA_NEURAL_QUANT, | |
| "version": "7.0.0", | |
| "ready": self.ready, | |
| "state": "MEASURED" if self.ready else "UNAVAILABLE", | |
| **core, | |
| "view_sha256": _canonical_sha256(core), | |
| "honesty": ( | |
| "A read-only holographic observation of source-bound public memory, " | |
| "formula/quant constraints, and bounded loop metadata. It neither " | |
| "trains nor executes." | |
| ), | |
| } | |