scripts/materialize_second_brain.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""Materialize the public Second Brain projection for Living Anatomy.

The source repository remains authoritative. This operator resolves an exact Git
commit, downloads the public 575-chunk retrieval corpus plus the review-gated
frontier candidate set from that immutable revision, validates every row and
digest, then writes one source receipt beside the snapshot. It never reads or
exports the private Second Brain graph, trains weights, promotes candidates, or
grants action authority.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from frontier_source_contract import FrontierSourceBindings, validate_source_revision
from pipeline_observation import (
    MAX_SNAPSHOT_BYTES, SNAPSHOT_SCHEMA, UPSTREAM_PATH, UPSTREAM_REPOSITORY,
)

DEFAULT_REPOSITORY = "szl-holdings/szl-second-brain"
DEFAULT_REF = "main"
EXPECTED_PUBLIC_CHUNKS = 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
MAX_MANIFEST_BYTES = 256 * 1024
MAX_CORPUS_BYTES = 4 * 1024 * 1024
MAX_FRONTIER_STATE_BYTES = 512 * 1024
MAX_FRONTIER_CANDIDATE_BYTES = 4 * 1024 * 1024
USER_AGENT = "szl-living-anatomy-second-brain-materializer/2.0"
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}$")
SECRET_PATTERNS = (
    re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{24,}\b"),
    re.compile(r"\bgh[pousr]_[A-Za-z0-9]{30,}\b"),
    re.compile(r"\bhf_[A-Za-z0-9]{30,}\b"),
    re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
)


class SnapshotError(RuntimeError):
    """A source-bound public snapshot violated its declared contract."""


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,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def request_bytes(url: str, *, token: str | None = None, limit: int) -> bytes:
    headers = {
        "Accept": "application/vnd.github+json, application/json, text/plain;q=0.9, */*;q=0.8",
        "User-Agent": USER_AGENT,
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    request = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=45) as response:
            body = response.read(limit + 1)
    except (urllib.error.URLError, TimeoutError, ConnectionError) as exc:
        raise SnapshotError(
            f"fetch failed for {url}: {type(exc).__name__}"
        ) from exc
    if len(body) > limit:
        raise SnapshotError(f"response exceeded {limit} bytes: {url}")
    return body


def resolve_revision(
    repository: str,
    ref: str,
    *,
    token: str | None = None,
    api_url: str = "https://api.github.com",
) -> str:
    if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository):
        raise ValueError(f"invalid GitHub repository: {repository!r}")
    encoded_ref = urllib.parse.quote(ref, safe="")
    url = f"{api_url.rstrip('/')}/repos/{repository}/commits/{encoded_ref}"
    try:
        raw = request_bytes(url, token=token, limit=MAX_MANIFEST_BYTES)
    except SnapshotError:
        # A repository-scoped Actions token can lack cross-repo read permission
        # even when the source repository is public. Retry the immutable public
        # read without credentials; never broaden token scope.
        if not token:
            raise
        raw = request_bytes(url, token=None, limit=MAX_MANIFEST_BYTES)
    payload = json.loads(raw)
    revision = (
        str(payload.get("sha") or "").lower()
        if isinstance(payload, dict)
        else ""
    )
    if not HEX_40.fullmatch(revision):
        raise SnapshotError("GitHub did not return an exact source revision")
    return revision


def validate_snapshot(
    manifest_raw: bytes,
    corpus_raw: bytes,
    *,
    expected_chunks: int = EXPECTED_PUBLIC_CHUNKS,
) -> dict[str, Any]:
    manifest = json.loads(manifest_raw.decode("utf-8"))
    if not isinstance(manifest, dict):
        raise ValueError("manifest must be a JSON object")
    if str(manifest.get("secretScan") or "").upper() != "PASS":
        raise ValueError("public projection secretScan must be PASS")

    rows = 0
    by_source: dict[str, int] = {}
    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}: {node_id}")
        ids.add(node_id)
        text = str(row.get("text") or "")
        declared = str(row.get("sha256") or "")
        measured = sha256_bytes(text.encode("utf-8"))
        if declared and declared != measured:
            raise ValueError(f"row digest mismatch at line {line_number}: {node_id}")
        source = str(row.get("source") or "unknown")
        by_source[source] = by_source.get(source, 0) + 1
        rows += 1

    declared_count = int(manifest.get("publicChunkCount") or 0)
    if rows != declared_count or rows != expected_chunks:
        raise ValueError(
            "public chunk count mismatch: "
            f"loaded={rows}, manifest={declared_count}, expected={expected_chunks}"
        )
    declared_by_source = manifest.get("bySource")
    if isinstance(declared_by_source, dict):
        normalized = {
            str(key): int(value) for key, value in declared_by_source.items()
        }
        if normalized != by_source:
            raise ValueError(
                f"source histogram mismatch: loaded={by_source}, manifest={normalized}"
            )
    return {
        "public_chunk_count": rows,
        "by_source": by_source,
        "manifest_sha256": sha256_bytes(manifest_raw),
        "corpus_sha256": sha256_bytes(corpus_raw),
        "manifest_projection_sha256": manifest.get("projectionSha256"),
        "secret_scan": "PASS",
    }


def _reject_secret_like_material(value: str) -> None:
    for pattern in SECRET_PATTERNS:
        if pattern.search(value):
            raise ValueError("frontier candidate set contains secret-like material")


def validate_frontier_snapshot(
    state_raw: bytes,
    candidates_raw: bytes,
) -> dict[str, Any]:
    state = json.loads(state_raw.decode("utf-8"))
    if not isinstance(state, dict):
        raise ValueError("frontier state must be a JSON object")
    if state.get("schema") != "szl.second-brain.frontier-state/v1":
        raise ValueError("unsupported frontier state schema")
    expected = {
        "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.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 snapshot")
    if int(state.get("raw_graph_nodes_admitted_to_gradients") or 0) != 0:
        raise ValueError("frontier snapshot 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")

    source_ids: set[str] = set()
    expected_source_counts: dict[tuple[str, str, str], int] = {}
    observed_source_counts: dict[tuple[str, str, str], int] = {}
    source_candidate_count = 0
    for index, source in enumerate(sources, start=1):
        if not isinstance(source, dict):
            raise ValueError(f"invalid frontier source at index {index}")
        source_id = str(source.get("source_id") or "")
        repository = str(source.get("repository") or "")
        revision = str(source.get("revision") or "")
        digest = str(source.get("content_sha256") or "")
        parser = str(source.get("parser") or "")
        path = str(source.get("path") or "")
        candidate_count = source.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(source)
        if not HEX_64.fullmatch(digest):
            raise ValueError(f"frontier source binding failed at index {index}")
        if (
            not isinstance(source.get("parser"), str)
            or not parser.strip()
            or not isinstance(source.get("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}")
        expected_source_counts[binding] = candidate_count
        observed_source_counts[binding] = 0
        source_ids.add(source_id)
        source_candidate_count += candidate_count

    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 and quant summaries are missing")
    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")

    decoded = candidates_raw.decode("utf-8")
    _reject_secret_like_material(decoded)
    rows = 0
    ids: set[str] = set()
    canonical_lines: list[bytes] = []
    formula_authority: dict[str, Any] | None = None
    for line_number, line in enumerate(decoded.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)
        digest = str(row.get("content_sha256") or "")
        if not HEX_64.fullmatch(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}")
        if row.get("source_kind") == "formula-authority":
            formula_authority = json.loads(content)
        canonical_lines.append(canonical_bytes(row) + b"\n")
        rows += 1

    declared_count = int(state.get("candidate_count") or -1)
    if rows != declared_count or rows < MIN_FRONTIER_CANDIDATES:
        raise ValueError(
            "frontier candidate count mismatch: "
            f"loaded={rows}, declared={declared_count}, minimum={MIN_FRONTIER_CANDIDATES}"
        )
    if source_candidate_count != declared_count:
        raise ValueError(
            "frontier source candidate total mismatch: "
            f"manifest={source_candidate_count}, declared={declared_count}"
        )
    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 candidate-set digest mismatch")
    if not HEX_64.fullmatch(str(state.get("state_sha256") or "")):
        raise ValueError("frontier state digest is malformed")
    if not isinstance(formula_authority, dict):
        raise ValueError("formula authority candidate is missing")
    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 {
        "frontier_candidate_count": rows,
        "frontier_source_count": source_count,
        "frontier_state_sha256": sha256_bytes(state_raw),
        "frontier_candidates_sha256": sha256_bytes(candidates_raw),
        "frontier_candidate_set_sha256": measured_set,
        "formula_counts": {
            "attributed": EXPECTED_ATTRIBUTED_FORMULAS,
            "executable": EXPECTED_EXECUTABLE_FORMULAS,
            "locked_proven": EXPECTED_LOCKED_PROVEN,
        },
        "quant_domain_count": EXPECTED_QUANT_DOMAINS,
        "lambda_state": "CONJECTURE_1",
        "candidate_state": "DISCOVERED_REVIEW_REQUIRED",
    }


def atomic_write(path: Path, payload: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        dir=path.parent,
        prefix=f".{path.name}.",
        delete=False,
    ) as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
        temporary = Path(handle.name)
    os.replace(temporary, path)


def materialize_pipeline_source(*, token: str | None, api_url: str,
                                raw_url: str) -> bytes:
    """Retain the exact admitted Git source; unavailable input stays explicit."""
    revision = None
    snapshot_json = None
    source_digest = None
    try:
        revision = resolve_revision(UPSTREAM_REPOSITORY, "main", token=token, api_url=api_url)
        raw = request_bytes(
            f"{raw_url.rstrip('/')}/{UPSTREAM_REPOSITORY}/{revision}/{UPSTREAM_PATH}",
            limit=MAX_SNAPSHOT_BYTES,
        )
        snapshot_json = raw.decode("utf-8")
        source_digest = sha256_bytes(raw)
    except (SnapshotError, OSError, ValueError, UnicodeError):
        # This optional observation dependency cannot turn the independent,
        # valid public retrieval projection into an invented healthy loop.
        snapshot_json = None
    wrapper = {
        "schema": SNAPSHOT_SCHEMA,
        "source_repository": UPSTREAM_REPOSITORY,
        "source_revision": revision,
        "source_path": UPSTREAM_PATH,
        "source_snapshot_sha256": source_digest,
        "snapshot_json": snapshot_json,
        "captured_at": utc_now(),
    }
    wrapper["receipt_sha256"] = sha256_bytes(canonical_bytes(wrapper))
    return (json.dumps(wrapper, sort_keys=True, indent=2) + "\n").encode("utf-8")


def materialize(
    output: Path,
    *,
    repository: str = DEFAULT_REPOSITORY,
    ref: str = DEFAULT_REF,
    token: str | None = None,
    api_url: str = "https://api.github.com",
    raw_url: str = "https://raw.githubusercontent.com",
) -> dict[str, Any]:
    revision = resolve_revision(repository, ref, token=token, api_url=api_url)
    immutable_base = f"{raw_url.rstrip('/')}/{repository}/{revision}/data"
    manifest_raw = request_bytes(
        f"{immutable_base}/manifest.json",
        limit=MAX_MANIFEST_BYTES,
    )
    corpus_raw = request_bytes(
        f"{immutable_base}/brain-corpus.public.jsonl",
        limit=MAX_CORPUS_BYTES,
    )
    frontier_state_raw = request_bytes(
        f"{immutable_base}/frontier-state.v1.json",
        limit=MAX_FRONTIER_STATE_BYTES,
    )
    frontier_candidates_raw = request_bytes(
        f"{immutable_base}/frontier-candidates.public.jsonl",
        limit=MAX_FRONTIER_CANDIDATE_BYTES,
    )
    retrieval = validate_snapshot(manifest_raw, corpus_raw)
    frontier = validate_frontier_snapshot(
        frontier_state_raw,
        frontier_candidates_raw,
    )
    pipeline_raw = materialize_pipeline_source(token=token, api_url=api_url, raw_url=raw_url)
    pipeline_source = json.loads(pipeline_raw)
    receipt = {
        "schema": "szl.second-brain.snapshot/v1",
        "source_repository": repository,
        "source_ref": ref,
        "source_revision": revision,
        "source_relation": "github-exact-revision-public-projection-and-frontier",
        "canonical_dataset": "SZLHOLDINGS/szl-second-brain-inrepo",
        "manifest_path": "data/manifest.json",
        "corpus_path": "data/brain-corpus.public.jsonl",
        "frontier_state_path": "data/frontier-state.v1.json",
        "frontier_candidates_path": "data/frontier-candidates.public.jsonl",
        "pipeline_source_path": ".runtime/second-brain/pipeline-source.json",
        "pipeline_source_sha256": sha256_bytes(pipeline_raw),
        "pipeline_dependency": {
            key: pipeline_source[key]
            for key in ("source_repository", "source_revision", "source_snapshot_sha256")
        },
        "manifest_sha256": retrieval["manifest_sha256"],
        "corpus_sha256": retrieval["corpus_sha256"],
        "manifest_projection_sha256": retrieval[
            "manifest_projection_sha256"
        ],
        "public_chunk_count": retrieval["public_chunk_count"],
        "by_source": retrieval["by_source"],
        "secret_scan": retrieval["secret_scan"],
        **frontier,
        "materialized_at": utc_now(),
        "authority_state": "READ_ONLY",
        "content_access": "HANDLES_ONLY",
        "frontier_content_access": "HANDLES_ONLY_PUBLIC_CONTROLLER_ONLY_INTERNAL",
        "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",
    }
    receipt["receipt_sha256"] = sha256_bytes(canonical_bytes(receipt))
    atomic_write(output / "manifest.json", manifest_raw)
    atomic_write(output / "pipeline-source.json", pipeline_raw)
    atomic_write(output / "brain-corpus.public.jsonl", corpus_raw)
    atomic_write(output / "frontier-state.v1.json", frontier_state_raw)
    atomic_write(
        output / "frontier-candidates.public.jsonl",
        frontier_candidates_raw,
    )
    atomic_write(
        output / "source.json",
        (json.dumps(receipt, sort_keys=True, indent=2) + "\n").encode("utf-8"),
    )
    return receipt


def token_from_environment() -> str | None:
    for key in ("GH_READ_TOKEN", "GH_ADMIN_TOKEN", "GITHUB_TOKEN"):
        value = os.environ.get(key)
        if value and value.strip():
            return value.strip()
    return None


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--output",
        type=Path,
        default=Path(".runtime/second-brain"),
    )
    parser.add_argument("--repository", default=DEFAULT_REPOSITORY)
    parser.add_argument("--ref", default=DEFAULT_REF)
    parser.add_argument(
        "--api-url",
        default=os.environ.get("GITHUB_API_URL", "https://api.github.com"),
    )
    parser.add_argument("--raw-url", default="https://raw.githubusercontent.com")
    args = parser.parse_args()
    receipt = materialize(
        args.output,
        repository=args.repository,
        ref=args.ref,
        token=token_from_environment(),
        api_url=args.api_url,
        raw_url=args.raw_url,
    )
    print(
        json.dumps(
            {
                "source_repository": receipt["source_repository"],
                "source_revision": receipt["source_revision"],
                "public_chunk_count": receipt["public_chunk_count"],
                "corpus_sha256": receipt["corpus_sha256"],
                "frontier_candidate_count": receipt[
                    "frontier_candidate_count"
                ],
                "frontier_candidate_set_sha256": receipt[
                    "frontier_candidate_set_sha256"
                ],
                "quant_domain_count": receipt["quant_domain_count"],
                "output": str(args.output),
            },
            sort_keys=True,
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())