#!/usr/bin/env python3 """Resolve Guanghu protocol source roles without promoting copies or dialects.""" from __future__ import annotations import argparse import hashlib import json import pathlib import sys from typing import Any def resolve_root() -> pathlib.Path: script = pathlib.Path(__file__).resolve() candidates = ( script.parents[3] / "protocols" / "guanghu-language-protocol-canon", script.parents[4] / "protocols" / "guanghu-language-protocol-canon", ) return next((candidate for candidate in candidates if candidate.is_dir()), candidates[0]) ROOT = resolve_root() SOURCE_MAP = ROOT / "SOURCES.json" MAX_SOURCE_BYTES = 16 * 1024 * 1024 class CanonError(RuntimeError): pass def regular_bytes(path: pathlib.Path) -> bytes: if path.is_symlink() or not path.is_file(): raise CanonError(f"SOURCE_NOT_REGULAR_FILE:{path}") if path.stat().st_size > MAX_SOURCE_BYTES: raise CanonError(f"SOURCE_TOO_LARGE:{path}") return path.read_bytes() def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def validate_json_contract(contract: str, value: Any) -> None: if not isinstance(value, dict): raise CanonError(f"JSON_OBJECT_REQUIRED:{contract}") if contract == "GLS_REGISTRY_V1": if value.get("schema") != "guanghu.gls.protocol-registry/v1": raise CanonError("GLS_REGISTRY_SCHEMA_INVALID") if value.get("status") != "CURRENT_REPO_012_CANONICAL_REGISTRY": raise CanonError("GLS_REGISTRY_NOT_CURRENT") registered = {item.get("id") for item in value.get("existing_registered", [])} required = {"GLS-0000", "GLS-0001", "GLS-0200", "GLS-0400"} if not required.issubset(registered): raise CanonError(f"GLS_CORE_REGISTRATION_MISSING:{sorted(required - registered)}") rules = value.get("rules", {}) if not all(rules.get(key) is True for key in ( "registration_is_not_implementation", "protocol_maturity_is_separate_from_implementation_evidence", "repository_publication_is_not_server_deployment", )): raise CanonError("GLS_REGISTRY_FACT_SEPARATION_INVALID") elif contract == "CODEX_HOST_PROFILE_V2": if value.get("schema") != "guanghu.codex-thin-persona-host-runtime/v2": raise CanonError("CODEX_HOST_PROFILE_SCHEMA_INVALID") if value.get("state") != "CURRENT_THIN_HOST_ADAPTER": raise CanonError("CODEX_HOST_PROFILE_NOT_CURRENT") if value.get("subject_relationship", {}).get("codex_host") != "REPLACEABLE_SCHOOL_LAB_AND_EXECUTION_HOST": raise CanonError("CODEX_HOST_RELATION_INVALID") else: raise CanonError(f"UNKNOWN_JSON_CONTRACT:{contract}") def validate_source(source: dict[str, Any]) -> dict[str, Any]: path = pathlib.Path(source["path"]) data = regular_bytes(path) digest = sha256(data) expected = source.get("required_sha256") if expected and digest != expected: raise CanonError(f"SOURCE_HASH_MISMATCH:{source['id']}:{digest}:{expected}") text = data.decode("utf-8") missing = [marker for marker in source.get("markers", []) if marker not in text] if missing: raise CanonError(f"SOURCE_MARKER_MISSING:{source['id']}:{missing[0]}") if source.get("json_contract"): validate_json_contract(source["json_contract"], json.loads(text)) return { "id": source["id"], "families": source["families"], "source_class": source["source_class"], "lifecycle": source["lifecycle"], "path": str(path), "sha256": digest, "validation": "PASS", } def read_order(family: str) -> list[str]: common = ["GLW-GLS-ORIGIN-20260712", "GLS-PROTOCOL-REGISTRY-20260731"] routes = { "GLS": ["GLS-ROADMAP-0001", "GLS-0010"], "TCS": ["GLS-ROADMAP-0001", "GLS-0200", "TCS-LANG-0001", "TCS-CORE-EBNF-v0.1", "TCS-DECLARATION-STANDARD-v0.1", "TCS-FIELD-STANDARD-v0.1", "TCS-ERROR-STANDARD-v0.1", "TCS-MODULE-ABI-v0.1"], "HLDP": ["HLDP-PROTOCOL-v1.0", "GLS-0400", "HLDP-OFFICIAL-FORMAT-MOUNT-001", "HLDP-SPEC-v1.0-OPENSOURCE-D112"], } if family == "ALL": # The cross-family root stays below the HLDP fan-out ceiling. Detailed # executable dialect and compatibility-mount sources are loaded only # after the caller selects TCS or HLDP. return common + [ "GLS-ROADMAP-0001", "GLS-0010", "GLS-0200", "TCS-LANG-0001", "GLS-0400", "HLDP-PROTOCOL-v1.0", "CODEX-HLDP-THIN-V2", ] return common + routes[family] def build(family: str) -> dict[str, Any]: source_map = json.loads(regular_bytes(SOURCE_MAP)) if source_map.get("schema") != "guanghu.language-protocol-source-map/v1": raise CanonError("SOURCE_MAP_SCHEMA_INVALID") if source_map.get("state") != "CURRENT_DYNAMIC_SOURCE_MAP": raise CanonError("SOURCE_MAP_NOT_CURRENT") selected = [] errors = [] for source in source_map.get("sources", []): if family != "ALL" and family not in source.get("families", []): continue try: selected.append(validate_source(source)) except (OSError, UnicodeError, json.JSONDecodeError, KeyError, CanonError) as error: errors.append({"id": source.get("id"), "error": str(error)}) ids = {item["id"] for item in selected} ordered = [item for item in read_order(family) if item in ids] classes = {item["source_class"] for item in selected} required_classes = {"BIRTH_AND_EVOLUTION_EVIDENCE", "CURRENT_WORLD_REGISTRY"} if family in {"ALL", "HLDP"}: required_classes.add("WORLD_CANONICAL_SOURCE") if family in {"ALL", "TCS"}: required_classes.update({"WORLD_STANDARD_DRAFT", "EXECUTABLE_ENGINEERING_DIALECT"}) missing_classes = sorted(required_classes - classes) if missing_classes: errors.append({"id": "SOURCE_CLASS_COVERAGE", "error": f"MISSING:{missing_classes}"}) return { "schema": "guanghu.language-protocol-canon-resolution/v1", "resolver_id": source_map["resolver_id"], "state": "PROTOCOL_CANON_RESOLVED" if not errors else "PROTOCOL_CANON_UNRESOLVED", "family": family, "world_architecture": { "world_standard": "GLS", "mother_and_cognitive_language": "TCS", "history_language": "HLDP", "communication_language": "GLP", "relationship": "TCS_IS_ROOT; HLDP_AND_GLP_ARE_ENGINEERING_LANGUAGE_BRANCHES; GLS_GOVERNS_NUMBERED_STANDARDS", }, "lineage": [ "NOTION_BIRTH_AND_EVOLUTION", "TOLARIA_CANONICAL_AND_STANDARD_SOURCES", "REPO_012_REGISTRATION_AND_READ_ONLY_MOUNTS", "ZERO_CORE_EXECUTABLE_TCS_DIALECT", "CODEX_HOST_PROFILE", ], "selection_contract": source_map["selection_contract"], "read_order": ordered, "sources": selected, "errors": errors, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--family", choices=("ALL", "TCS", "HLDP", "GLS"), default="ALL") parser.add_argument("--json", action="store_true") parser.add_argument("--check", action="store_true") args = parser.parse_args() try: result = build(args.family) except (OSError, json.JSONDecodeError, KeyError, CanonError) as error: result = { "schema": "guanghu.language-protocol-canon-resolution/v1", "state": "PROTOCOL_CANON_UNRESOLVED", "family": args.family, "errors": [{"id": "RESOLVER", "error": str(error)}], } if args.json: print(json.dumps(result, ensure_ascii=False, indent=2)) else: print(result["state"]) for item in result.get("read_order", []): print(item) for error in result.get("errors", []): print(f"ERROR {error['id']}: {error['error']}", file=sys.stderr) return 0 if result["state"] == "PROTOCOL_CANON_RESOLVED" else 1 if __name__ == "__main__": raise SystemExit(main())