Spaces:
Running
Running
| #!/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()) | |