from __future__ import annotations import hashlib import json import sys import tempfile import unittest from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from second_brain_runtime import PublicSecondBrain # noqa: E402 QUANT_DOMAINS = [ "algebra_number_theory", "coding_error_control", "dynamics_consensus", "energy_entropy_physics", "governance_receipts", "information_geometry", "narrative_lineage", "topology_geometry", "trust_aggregation", ] def canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def frontier_row( stable: str, *, title: str, content: str, kind: str, repository: str = "szl-holdings/szl-formulas", quant_domain: str | None = None, admission: str = "DISCOVERED_REVIEW_REQUIRED", ) -> dict[str, Any]: row: dict[str, Any] = { "schema": "szl.second-brain.frontier-candidate/v1", "id": "frontier:" + hashlib.sha256(stable.encode()).hexdigest()[:32], "title": title, "content": content, "content_sha256": hashlib.sha256(content.encode()).hexdigest(), "source_repository": repository, "source_revision": hashlib.sha1(repository.encode()).hexdigest(), "source_path": "fixture/source.md", "source_kind": kind, "admission": admission, "candidate_state": "DISCOVERED_REVIEW_REQUIRED", "content_access": "CONTROLLER_ONLY", } if quant_domain: row["quant_domain"] = quant_domain return row class PublicSecondBrainTest(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.snapshot = Path(self.temporary.name) sources = ( ["doc"] * 152 + ["formula"] * 269 + ["ingest"] * 143 + ["invariant"] * 11 ) rows = [] for index, source in enumerate(sources): text = ( f"governed receipt living anatomy second brain {index} {source}" if index < 8 else f"public knowledge chunk {index} {source}" ) rows.append( { "id": f"{source}:fixture:{index:04d}", "source": source, "sourceId": f"{source}-{index}", "title": f"Fixture {source} {index}", "text": text, "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), } ) corpus_raw = b"".join(canonical_bytes(row) + b"\n" for row in rows) manifest = { "datasetName": "test public projection", "publicChunkCount": 575, "bySource": { "doc": 152, "formula": 269, "ingest": 143, "invariant": 11, }, "projectionSha256": "test-only", "secretScan": "PASS", } manifest_raw = (json.dumps(manifest, indent=2) + "\n").encode("utf-8") formula_authority = { "executable_registry_repository": "szl-holdings/szl-formulas", "executable_registry_count": 21, "locked_proven_count": 8, "locked_proven_ids": [ "F1", "F11", "F12", "F18", "F19", "F22", "F4", "F7", ], "lambda_status": "CONJECTURE_1_OPEN_ADVISORY_ONLY", "f_number_to_executable_registry_mapping": "UNKNOWN_NOT_INFERRED", } frontier_rows = [ frontier_row( "authority", title="SZL formula authority and proof boundary", content=json.dumps(formula_authority, sort_keys=True), kind="formula-authority", admission="REFERENCE_AND_CONSTRAINT_INPUT_ONLY", ) ] for index in range(30): domain = QUANT_DOMAINS[index % len(QUANT_DOMAINS)] frontier_rows.append( frontier_row( f"attributed-{index}", title=f"Attributed formula F{index + 1}", content=( f"Formula F{index + 1}; quant domain {domain}; " "reported status retained; review required." ), kind="attributed-formula", quant_domain=domain, admission="REFERENCE_AND_CONSTRAINT_INPUT_ONLY", ) ) for index in range(21): frontier_rows.append( frontier_row( f"executable-{index}", title=f"Executable formula kernel_{index}", content=( f"Executable formula kernel_{index}; per-obligation status; " "no inferred F-number mapping." ), kind="executable-formula", admission="EXECUTABLE_CONSTRAINT_REVIEW_REQUIRED", ) ) for domain in QUANT_DOMAINS: frontier_rows.append( frontier_row( f"domain-{domain}", title=f"Quant domain {domain}", content=f"Quant domain {domain}; reference and constraint input only.", kind="quant-domain", quant_domain=domain, admission="REFERENCE_AND_CONSTRAINT_INPUT_ONLY", ) ) for index in range(10): repository = ( "szl-holdings/ouroboros" if index < 4 else "szl-holdings/anatomy" ) frontier_rows.append( frontier_row( f"source-{index}", title=f"Source document {index}", content=( "Ouroboros bounded loop convergence termination receipt closure Codex" if index < 4 else "Living Anatomy holographic brain frontier observation" ), kind="source-document", repository=repository, ) ) frontier_rows.sort(key=lambda row: row["id"]) frontier_candidates_raw = b"".join( canonical_bytes(row) + b"\n" for row in frontier_rows ) candidate_set_sha256 = hashlib.sha256(frontier_candidates_raw).hexdigest() source_kind_counts = { "formula-authority": 1, "attributed-formula": 30, "executable-formula": 21, "quant-domain": 9, "source-document": 10, } quant_domain_counts = { domain: sum( 1 for row in frontier_rows if row.get("quant_domain") == domain ) for domain in QUANT_DOMAINS } frontier_state = { "schema": "szl.second-brain.frontier-state/v1", "state": "REVIEW_REQUIRED", "candidate_count": len(frontier_rows), "candidate_set_sha256": candidate_set_sha256, "source_count": 7, "sources": [ { "source_id": "ouroboros_runtime" if index == 0 else f"source_{index}", "repository": ( "szl-holdings/ouroboros" if index == 0 else f"szl-holdings/source-{index}" ), "revision": f"{index + 1:x}" * 40, "path": "README.md", "parser": "markdown", "content_sha256": f"{index + 1:x}" * 64, "candidate_count": 1, } for index in range(6) ], "source_kind_counts": source_kind_counts, "quant_domain_counts": quant_domain_counts, "public_content_access": "HANDLES_ONLY", "controller_content_access": "AUTHORIZED_CONTROLLER_ONLY", "private_graph_nodes_loaded": 0, "raw_graph_nodes_admitted_to_gradients": 0, "training_authority": "NONE", "promotion_authority": "NONE", "execution_authority": "NONE", "merge_authority": "NONE", "lambda": "CONJECTURE_1", "learning_definition": ( "Content-addressed public-source candidates are proposed for human review." ), "state_sha256": "c" * 64, } frontier_state_raw = ( json.dumps(frontier_state, sort_keys=True, indent=2) + "\n" ).encode("utf-8") source = { "schema": "szl.second-brain.snapshot/v1", "source_repository": "szl-holdings/szl-second-brain", "source_revision": "a" * 40, "manifest_sha256": hashlib.sha256(manifest_raw).hexdigest(), "corpus_sha256": hashlib.sha256(corpus_raw).hexdigest(), "public_chunk_count": 575, "frontier_state_sha256": hashlib.sha256(frontier_state_raw).hexdigest(), "frontier_candidates_sha256": hashlib.sha256( frontier_candidates_raw ).hexdigest(), "frontier_candidate_count": len(frontier_rows), "frontier_candidate_set_sha256": candidate_set_sha256, "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", } (self.snapshot / "manifest.json").write_bytes(manifest_raw) (self.snapshot / "brain-corpus.public.jsonl").write_bytes(corpus_raw) (self.snapshot / "frontier-state.v1.json").write_bytes(frontier_state_raw) (self.snapshot / "frontier-candidates.public.jsonl").write_bytes( frontier_candidates_raw ) (self.snapshot / "source.json").write_text( json.dumps(source, indent=2) + "\n", encoding="utf-8", ) def tearDown(self) -> None: self.temporary.cleanup() def test_source_bound_snapshot_loads_retrieval_and_frontier(self) -> None: brain = PublicSecondBrain(self.snapshot) health = brain.health() self.assertTrue(health["ready"]) self.assertEqual(575, health["chunk_count"]) self.assertEqual("a" * 40, health["source_revision"]) self.assertEqual("HANDLES_ONLY", health["content_access"]) self.assertEqual(0, health["private_graph_nodes_loaded"]) self.assertEqual(71, health["frontier"]["candidate_count"]) self.assertEqual(7, health["frontier"]["source_count"]) self.assertEqual(30, health["formula_atlas"]["attributed_formula_count"]) self.assertEqual(21, health["formula_atlas"]["executable_formula_count"]) self.assertEqual(8, health["formula_atlas"]["locked_proven_count"]) self.assertEqual(9, health["quant_domain_count"]) self.assertEqual( {"doc": 152, "formula": 269, "ingest": 143, "invariant": 11}, health["by_source"], ) def test_search_returns_only_receipted_handles(self) -> None: brain = PublicSecondBrain(self.snapshot) result = brain.search("governed receipt anatomy", k=4) self.assertTrue(result["ready"]) self.assertEqual(4, len(result["handles"])) self.assertEqual(4, len(result["scores"])) self.assertEqual(64, len(result["result_sha256"])) for handle in result["handles"]: self.assertEqual( { "nodeId", "nodeKind", "label", "note", "source", "sourceId", "sha256", }, set(handle), ) self.assertNotIn("text", handle) self.assertEqual(64, len(handle["sha256"])) def test_context_has_no_training_or_write_authority(self) -> None: brain = PublicSecondBrain(self.snapshot) context = brain.context("living anatomy", k=3) self.assertTrue(context["ready"]) self.assertEqual("NONE", context["training_authority"]) self.assertEqual("NONE", context["write_authority"]) self.assertEqual(0, context["private_graph_nodes_loaded"]) self.assertEqual(3, len(context["model_handles"])) self.assertEqual(3, len(context["evidence"])) def test_frontier_formula_quant_and_ouroboros_views_are_handles_only(self) -> None: brain = PublicSecondBrain(self.snapshot) frontier = brain.frontier_search("formula quant anatomy ouroboros", k=12) self.assertTrue(frontier["ready"]) self.assertEqual("REVIEW_REQUIRED", frontier["state"]) self.assertTrue(frontier["handles"]) self.assertEqual("NONE", frontier["training_authority"]) self.assertEqual("NONE", frontier["execution_authority"]) self.assertTrue(all("content" not in handle for handle in frontier["handles"])) formulas = brain.formula_view(k=24) self.assertEqual(30, formulas["attributed_formula_count"]) self.assertEqual(21, formulas["executable_formula_count"]) self.assertEqual(8, formulas["locked_proven_count"]) self.assertEqual("UNKNOWN_NOT_INFERRED", formulas["f_number_mapping"]) self.assertEqual( "CONJECTURE_1_OPEN_ADVISORY_ONLY", formulas["lambda_status"], ) quant = brain.quant_view(k=24) self.assertEqual(9, quant["quant_domain_count"]) self.assertEqual(9, len(quant["domains"])) ouroboros = brain.ouroboros_view(k=12) self.assertTrue(ouroboros["loop_contract"]["bounded"]) self.assertTrue(ouroboros["loop_contract"]["terminating"]) self.assertTrue(ouroboros["loop_contract"]["receipt_closed"]) self.assertFalse( ouroboros["loop_contract"]["recommendations_executed"] ) self.assertTrue(ouroboros["handles"]) def test_neural_quant_v7_is_receipted_and_non_mutating(self) -> None: brain = PublicSecondBrain(self.snapshot) payload = brain.neural_quant_view(k=12) self.assertTrue(payload["ready"]) self.assertEqual("7.0.0", payload["version"]) self.assertEqual(575, payload["brain"]["chunk_count"]) self.assertEqual(71, payload["brain"]["frontier_candidate_count"]) self.assertEqual(30, payload["formulas"]["attributed"]) self.assertEqual(21, payload["formulas"]["executable"]) self.assertEqual(8, payload["formulas"]["locked_proven"]) self.assertEqual(9, payload["quant"]["domain_count"]) self.assertEqual("HANDLES_ONLY", payload["content_access"]) self.assertEqual( { "training": "NONE", "promotion": "NONE", "execution": "NONE", "merge": "NONE", "provider_mutation": "NONE", }, payload["authority"], ) self.assertEqual(64, len(payload["view_sha256"])) for group in ("formulas", "quant", "ouroboros"): for handle in payload[group]["handles"]: self.assertNotIn("content", handle) self.assertNotIn("text", handle) def test_corrupt_frontier_candidate_fails_closed(self) -> None: path = self.snapshot / "frontier-candidates.public.jsonl" rows = path.read_text(encoding="utf-8").splitlines() candidate = json.loads(rows[0]) candidate["content"] = "tampered" rows[0] = json.dumps(candidate, sort_keys=True, separators=(",", ":")) path.write_text("\n".join(rows) + "\n", encoding="utf-8") brain = PublicSecondBrain(self.snapshot) health = brain.health() self.assertFalse(health["ready"]) self.assertEqual("UNAVAILABLE", health["state"]) self.assertIn("mismatch", str(health["load_error"]).lower()) def test_corrupt_corpus_fails_closed(self) -> None: path = self.snapshot / "brain-corpus.public.jsonl" path.write_bytes(path.read_bytes() + b"{}\n") brain = PublicSecondBrain(self.snapshot) health = brain.health() self.assertFalse(health["ready"]) self.assertEqual("UNAVAILABLE", health["state"]) self.assertIn("mismatch", str(health["load_error"]).lower()) if __name__ == "__main__": unittest.main()