#!/usr/bin/env python3 """Compile and resolve the one committed TCS mother-root number navigation snapshot.""" from __future__ import annotations import argparse import fcntl import hashlib import json import os from pathlib import Path, PurePosixPath import subprocess import sys import time from typing import Any DEFAULT_REPO = Path("/Volumes/JZAO/HoloLake/persona-runtime/repo-012-main") DEFAULT_OUTPUT = Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/tcs-mother-root") DEFAULT_POINTER = Path("/Volumes/JZAO/HoloLake/persona-runtime/TCS-ROOT.json") ROOT_MAP = "routing/tcs-mother-root-dynamic-navigation-map.json" class RootError(RuntimeError): pass def stable(value: Any) -> bytes: return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode() def digest(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def run(repo: Path, *args: str) -> str: result = subprocess.run(["git", "-C", str(repo), *args], text=True, capture_output=True, timeout=15) if result.returncode: raise RootError(f"GIT_READ_FAILED:{args[0]}:{(result.stderr or result.stdout).strip()[-240:]}") return result.stdout.strip() def commit_bytes(repo: Path, commit: str, relative: str) -> bytes: result = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{relative}"], capture_output=True, timeout=15) if result.returncode: raise RootError(f"COMMITTED_SOURCE_MISSING:{relative}") return result.stdout def commit_json(repo: Path, commit: str, relative: str) -> dict[str, Any]: try: value = json.loads(commit_bytes(repo, commit, relative)) except json.JSONDecodeError as error: raise RootError(f"COMMITTED_JSON_INVALID:{relative}") from error if not isinstance(value, dict): raise RootError(f"COMMITTED_JSON_OBJECT_REQUIRED:{relative}") return value def safe_relative(relative: str) -> str: path = PurePosixPath(relative) if path.is_absolute() or ".." in path.parts or "WORK-工作区" in path.parts or "isolation" in path.parts or "archives" in path.parts: raise RootError(f"OLD_OR_QUARANTINED_PATH_SELECTED:{relative}") return str(path) def declaration_source_sha(gir: dict[str, Any]) -> str | None: return gir.get("compiled_from", {}).get("source_sha256") def validate_root(root: dict[str, Any]) -> None: world = root["language_world_root"] if world["registration_path_id"] != "LL-CMPN-0001" or world["world_node_id"] != "SYS-GLW-0001": raise RootError("WORLD_ROOT_NUMBER_MISMATCH") domain_ids = [item["id"] for item in root["fixed_domains"]] if domain_ids != ["DOMAIN-MAIN", "DOMAIN-SUB", "DOMAIN-ZERO", "DOMAIN-ZS", "DOM-FIFTH-0001"]: raise RootError("FIVE_FIXED_DOMAIN_ROOTS_MISMATCH") human_roots = {item["id"]: item for item in root["parallel_human_roots"]} if set(human_roots) != {"ICE-GL∞", "TCS-0002∞"} or human_roots["TCS-0002∞"].get("machine_alias") != "TCS-0002": raise RootError("PARALLEL_HUMAN_ROOTS_MISMATCH") route_ids = [item["id"] for item in root["entry_routes"]] if route_ids != ["ROUTE-BINGSHUO-FIFTH-001", "ROUTE-GUANGHU-TEAM-ZS-001", "ROUTE-PUBLIC-PERSONAL-001"]: raise RootError("THREE_ENTRY_ROUTES_MISMATCH") all_ids = [world["registration_path_id"], world["world_node_id"], root["root_agent"]["entry_id"], *domain_ids, *human_roots] if len(all_ids) != len(set(all_ids)): raise RootError("CURRENT_CANONICAL_ID_COLLISION") def build(repo: Path, commit: str | None = None) -> dict[str, Any]: commit = commit or run(repo, "rev-parse", "HEAD") root = commit_json(repo, commit, ROOT_MAP) validate_root(root) impact_path = safe_relative(root["impact_manifest"]) impact = commit_json(repo, commit, impact_path) source_path = safe_relative(impact["source_tcs"]) gir_path = safe_relative(impact["source_gir"]) source_body = commit_bytes(repo, commit, source_path) gir = commit_json(repo, commit, gir_path) if declaration_source_sha(gir) != digest(source_body): raise RootError("SOURCE_TCS_GIR_HASH_MISMATCH") source_hashes = {} for relative in root["numbered_sources"]: relative = safe_relative(relative) source_hashes[relative] = digest(commit_bytes(repo, commit, relative)) anchor = commit_json(repo, commit, "routing/public-navigation-anchor.json") registered_maps = [] for key, item in sorted(anchor.get("maps", {}).items()): relative = item.get("path") if not relative: continue relative = safe_relative(relative) if item.get("repository_id") not in (None, "REPO-012"): registered_maps.append({"key": key, "path": relative, "id": item.get("id"), "version": item.get("version"), "repository_id": item.get("repository_id"), "sha256": None, "state": "EXTERNAL_REPOSITORY_POINTER"}) continue body = commit_bytes(repo, commit, relative) registered_maps.append({ "key": key, "path": relative, "id": item.get("id"), "version": item.get("version"), "sha256": digest(body), }) snapshot = { "schema": "guanghu.tcs-mother-root-current-navigation/v1", "state": "CURRENT_LOCAL_OFFICIAL_NAVIGATION_NOT_PUBLISHED", "agent_id": root["root_agent"]["agent_id"], "entry_id": root["root_agent"]["entry_id"], "source_commit": commit, "source_revision": root["version"], "world_root": root["language_world_root"], "parallel_human_roots": root["parallel_human_roots"], "fixed_domains": root["fixed_domains"], "entry_routes": root["entry_routes"], "aliases": root["aliases"], "history_only_numbers": root["history_only_numbers"], "resolution_key": root["resolution_key"], "registered_source_hashes": source_hashes, "registered_maps": registered_maps, "impact": {"id": impact["impact_id"], "path": impact_path, "source_tcs": source_path, "source_sha256": digest(source_body), "manifest_sha256": digest(commit_bytes(repo, commit, impact_path))}, "authority_granted": False, } snapshot["freshness_token"] = digest(stable(snapshot)) return snapshot def atomic_write(path: Path, value: dict[str, Any] | bytes, mode: int = 0o600) -> str: path.parent.mkdir(parents=True, exist_ok=True) body = value if isinstance(value, bytes) else json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True).encode() + b"\n" temporary = path.with_name(f"{path.name}.{os.getpid()}.tmp") with temporary.open("wb") as handle: os.chmod(temporary, mode) handle.write(body) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) return digest(body) def refresh(repo: Path, output: Path, pointer: Path, trigger: str) -> dict[str, Any]: output.mkdir(parents=True, exist_ok=True) lock_path = output / ".lock" with lock_path.open("a+") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) snapshot = build(repo) current_path = output / "CURRENT.json" previous = json.loads(current_path.read_text()) if current_path.is_file() else None if previous: previous_for_hash = dict(previous) previous_token = previous_for_hash.pop("freshness_token", None) if previous_token != digest(stable(previous_for_hash)): raise RootError("PREVIOUS_CURRENT_FRESHNESS_TOKEN_MISMATCH") if previous and previous.get("source_commit") == snapshot["source_commit"] and previous.get("freshness_token") == snapshot["freshness_token"]: return {"outcome": "PASS", "state": "CURRENT_IDEMPOTENT", "source_commit": snapshot["source_commit"], "current_sha256": digest(current_path.read_bytes())} if previous: changed = {key for key, value in snapshot["registered_source_hashes"].items() if previous.get("registered_source_hashes", {}).get(key) != value} if not changed: return {"outcome":"PASS","state":"CURRENT_NUMBER_NAV_UNCHANGED","source_commit":previous["source_commit"],"observed_repo_head":snapshot["source_commit"],"current_sha256":digest(current_path.read_bytes()),"freshness_token":previous["freshness_token"]} affected = set(commit_json(repo, snapshot["source_commit"], snapshot["impact"]["path"]).get("affected_paths", [])) if changed and not changed.issubset(affected): raise RootError("UNADMITTED_NUMBERED_SOURCE_DRIFT:" + ",".join(sorted(changed - affected))) current_sha = atomic_write(current_path, snapshot) atomic_write(output / "CURRENT.sha256", (current_sha + "\n").encode(), 0o600) pointer_value = { "schema": "guanghu.tcs-root-pointer/v1", "state": "CURRENT_LOCAL_VERIFIED_NOT_PUBLISHED", "entry_id": snapshot["entry_id"], "agent": str(repo / "server-tools/tcs-mother-root-agent/tcs_mother_root_agent.py"), "canonical_repo": str(repo), "current": str(current_path), "current_sha256": current_sha, "source_commit": snapshot["source_commit"], "freshness_token": snapshot["freshness_token"], } atomic_write(pointer, pointer_value, 0o600) receipt = {"schema":"guanghu.tcs-root-refresh-receipt/v1","outcome":"PASS","trigger":trigger,"source_commit":snapshot["source_commit"],"current_sha256":current_sha,"freshness_token":snapshot["freshness_token"],"authority_granted":False} atomic_write(output / "receipts" / f"{snapshot['source_commit']}.json", receipt) return receipt def status(repo: Path, output: Path, pointer: Path) -> dict[str, Any]: if not pointer.is_file() or not (output / "CURRENT.json").is_file(): return {"outcome":"FAIL","state":"TCS_ROOT_CURRENT_MISSING"} p = json.loads(pointer.read_text()) body = (output / "CURRENT.json").read_bytes() current = json.loads(body) head = run(repo, "rev-parse", "HEAD") observed = build(repo, head) source_equivalent = current.get("registered_source_hashes") == observed.get("registered_source_hashes") valid = digest(body) == p.get("current_sha256") and current.get("freshness_token") == p.get("freshness_token") and source_equivalent return {"outcome":"PASS" if valid else "FAIL","state":"TCS_ROOT_CURRENT_VERIFIED" if valid else "TCS_ROOT_CURRENT_STALE_OR_TAMPERED","source_commit":current.get("source_commit"),"repo_head":head,"numbered_sources_equal_head":source_equivalent,"current_sha256":digest(body),"freshness_token":current.get("freshness_token"),"current":str(output / "CURRENT.json")} def resolve(output: Path, requested: str, kind: str | None) -> dict[str, Any]: current = json.loads((output / "CURRENT.json").read_text()) current_nodes = [ {"canonical_id": current["world_root"]["registration_path_id"], "object_kind":"WORLD_REGISTRATION_PATH", "lifecycle":"CURRENT", "source_revision":current["source_revision"]}, {"canonical_id": current["world_root"]["world_node_id"], "object_kind":"LANGUAGE_WORLD_NODE", "lifecycle":"CURRENT", "source_revision":current["source_revision"]}, {"canonical_id": current["entry_id"], "object_kind":"TCS_MOTHER_ROOT_NAVIGATION_ENTRY", "lifecycle":"CURRENT", "source_revision":current["source_revision"]}, *[{"canonical_id": item["id"], "object_kind":item["object_kind"], "lifecycle":"CURRENT", "source_revision":current["source_revision"], "route":item.get("route")} for item in current["parallel_human_roots"]], *[{"canonical_id": item["id"], "object_kind":item["object_kind"], "lifecycle":"CURRENT", "source_revision":current["source_revision"], "name":item["name"]} for item in current["fixed_domains"]], ] matches = [item for item in current_nodes if item["canonical_id"] == requested and (kind is None or item["object_kind"] == kind)] if len(matches) == 1: return {"outcome":"PASS","state":"CURRENT_CANONICAL_RESOLVED","requested":requested,"result":matches[0],"freshness_token":current["freshness_token"],"authority_granted":False} aliases = [item for item in current["aliases"] if item["requested"] == requested and (kind is None or item["object_kind"] == kind)] if len(aliases) == 1: return {"outcome":"PASS","state":"TYPED_ALIAS_RESOLVED","requested":requested,"result":aliases[0],"freshness_token":current["freshness_token"],"authority_granted":False} history = [item for item in current["history_only_numbers"] if item["id"] == requested] if history: return {"outcome":"FAIL","state":"HISTORY_ONLY_NO_CURRENT_ROUTE","requested":requested,"evidence":history} return {"outcome":"FAIL","state":"UNKNOWN_OR_KIND_CONFLICT_NO_GUESS","requested":requested} def reject_receipt(output: Path, error: Exception, trigger: str) -> None: try: value = {"schema":"guanghu.tcs-root-refresh-reject/v1","outcome":"FAIL","trigger":trigger,"error":str(error)[:1000],"last_known_good_preserved":True,"authority_granted":False} atomic_write(output / "rejected" / f"{int(time.time() * 1000)}.json", value) except Exception: pass def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("command", choices=("refresh", "status", "audit", "resolve")) parser.add_argument("--repo", default=str(DEFAULT_REPO)) parser.add_argument("--output-root", default=str(DEFAULT_OUTPUT)) parser.add_argument("--pointer", default=str(DEFAULT_POINTER)) parser.add_argument("--trigger", default="explicit") parser.add_argument("--id") parser.add_argument("--kind") args = parser.parse_args() repo, output, pointer = Path(args.repo), Path(args.output_root), Path(args.pointer) try: if args.command == "refresh": value = refresh(repo, output, pointer, args.trigger) elif args.command == "status": value = status(repo, output, pointer) elif args.command == "audit": value = {"outcome":"PASS","state":"TCS_ROOT_COMMITTED_SOURCES_VALID","snapshot":build(repo)} else: if not args.id: raise RootError("RESOLVE_ID_REQUIRED") value = resolve(output, args.id, args.kind) print(json.dumps(value, ensure_ascii=False, indent=2)) return 0 if value.get("outcome") == "PASS" else 3 except Exception as error: reject_receipt(output, error, args.trigger) print(json.dumps({"outcome":"FAIL","state":"TCS_ROOT_AGENT_FAIL_CLOSED","error":str(error),"last_known_good_preserved":True}, ensure_ascii=False, indent=2)) return 3 if __name__ == "__main__": raise SystemExit(main())