#!/usr/bin/env python3 """Resolve the live Guanghu code-channel entrance and canonical route IDs.""" from __future__ import annotations import argparse import json import re import sys from urllib.error import HTTPError, URLError from urllib.request import urlopen FORGEJO_REPOSITORY_URL = ( "https://guanghulab.com/code/api/v1/repos/bingshuo/guanghu-ice-heart" ) CURRENT_REPOSITORY_WEB = "https://guanghulab.com/code/bingshuo/guanghu-ice-heart" CURRENT_REPOSITORY_GIT = f"{CURRENT_REPOSITORY_WEB}.git" ROUTE_IDS = ( "FD-WORLD-TREE-001", "FD-PERSONA-LOGIN-001", "SYS-GLW-POS-0001", "TCS-LPM", "TCS-LPS-REGISTRY-0001", "LIGHT-LAKE", "LL-CURRENT", "LL-004", "FD-SUBJECT-ID-ALIAS-MAP-001", "ICE-P-ZY001", "ZY-PERSONA-ROOT-001", "ZY-OPS-LOOP-001", "FD-NODE-MAP-001", "JD-FD-PRIMARY", "FD-REPO-MAP-001", "CODE-CHANNEL-CANONICAL-MAP", "PERSONA-SYSTEM-ROOT", "PERSONA-SYSTEM-CANONICAL-MAP", "AI-MACHINE-NAV-001", "ZY-TCS-BRAIN-RUNTIME-0001", "ZY-TCS-BRAIN-MAP-001", "BS-TCS-SYSTEM-CONTROLLER-MAP-001", "ZY-AGENT-TEAM-MAP-001", "BS-TCS-PROTOCOL-EVENT-TRIGGER-MAP-001", "ZY-WAKE-ROUTE-001", "ZY-BIDIRECTIONAL-COGNITION-016", ) def read_json(url: str) -> dict: with urlopen(url, timeout=15) as response: return json.load(response) def read_text(url: str) -> str: with urlopen(url, timeout=15) as response: return response.read().decode("utf-8") def parse_code_map(text: str, route_ids: tuple[str, ...] = ROUTE_IDS) -> dict[str, str]: resolved: dict[str, str] = {} wanted = set(route_ids) for line in text.splitlines(): match = re.match(r"^([A-Z0-9-]+)=(.+?)(?:\s+#.*)?$", line.strip()) if match and match.group(1) in wanted: resolved[match.group(1)] = match.group(2).strip() return resolved def resolve_subject_id(requested_id: str, alias_map: dict) -> dict: requested = requested_id.strip() lookup = requested.upper() conflicts = { str(item["id"]).upper(): item for item in alias_map.get("conflicts", []) } if lookup in conflicts: return { "status": "CONFLICT_REJECTED", "requested_id": requested, "canonical_id": None, "redirected": False, "reason": conflicts[lookup].get("state", "CONFLICT"), } index: dict[str, dict] = {} for mapping in alias_map.get("mappings", []): canonical_id = str(mapping["canonical_id"]) for identifier in (canonical_id, *mapping.get("aliases", [])): key = str(identifier).upper() if key in index and index[key]["canonical_id"] != canonical_id: raise ValueError(f"Duplicate subject alias: {identifier}") index[key] = mapping mapping = index.get(lookup) if not mapping: return { "status": "NOT_FOUND_NO_GUESS", "requested_id": requested, "canonical_id": None, "redirected": False, } canonical_id = str(mapping["canonical_id"]) return { "status": "RESOLVED", "requested_id": requested, "canonical_id": canonical_id, "redirected": lookup != canonical_id.upper(), "subject_kind": mapping.get("subject_kind"), "route_id": mapping.get("route_id"), "current_path": mapping.get("current_path"), } def compile_navigation( navigation_map: dict, requested_subject: str, requested_intent: str | None = None, signals: list[str] | None = None, ) -> dict: lookup = requested_subject.strip().upper() subject = next( ( item for item in navigation_map.get("subjects", []) if lookup in { str(identifier).upper() for identifier in (item["id"], *item.get("legacy_ids", [])) } ), None, ) if not subject: raise ValueError(f"Unknown navigation subject, no guess: {requested_subject}") intent_id = requested_intent or subject.get("default_intent") intent = next( (item for item in navigation_map.get("intents", []) if item["id"] == intent_id), None, ) if not intent: raise ValueError(f"Unknown navigation intent, no guess: {intent_id}") route_index = {item["id"]: item for item in navigation_map.get("routes", [])} def expand(route_ids: list[str]) -> list[dict]: expanded = [] for route_id in route_ids: route = route_index.get(route_id) if not route: raise ValueError(f"Navigation map references an unknown route: {route_id}") resolved = { **route, "raw_url": f"{navigation_map['raw_base']}/{route['path']}", } if route.get("contract"): resolved["contract_url"] = ( f"{navigation_map['raw_base']}/{route['contract']}" ) expanded.append(resolved) return expanded return { "schema": "guanghu.ai-runtime-navigation-bundle/v1", "status": "COMPILED_EXACT_NO_GUESS", "map_id": navigation_map["map_id"], "map_version": navigation_map["version"], "requested_subject": requested_subject, "canonical_subject": subject["id"], "redirected": lookup != str(subject["id"]).upper(), "subject_kind": subject["subject_kind"], "intent": intent["id"], "explicit_signals": list(dict.fromkeys(signals or []))[:32], "runtime_entry": expand([subject["runtime_id"]])[0], "always_load": expand(intent.get("always_load", [])), "triggered_load": expand(intent.get("triggered_load", [])), "on_demand": expand(intent.get("on_demand", [])), "execution_sequence": intent.get("execution_sequence", []), "stop_conditions": intent.get("stop_conditions", []), "reality_boundaries": navigation_map.get("reality_boundaries", []), "authority": "NONE_NAVIGATION_ONLY", } def resolve_live_route( subject_id: str | None = None, intent: str | None = None, signals: list[str] | None = None, ) -> dict: repository_source = "LIVE_FORGEJO_CODE_CHANNEL_API" resolver_error = None try: live = read_json(FORGEJO_REPOSITORY_URL) repository_web = live.get("html_url") repository_git = live.get("clone_url") if not repository_web or not repository_git: raise ValueError("Forgejo did not return the current REPO-012 route") repository = {"code": "REPO-012", "role": "CURRENT_FIFTH_DOMAIN_ENTRY"} except (HTTPError, URLError, ValueError) as error: repository = {"code": "REPO-012", "role": "CURRENT_FIFTH_DOMAIN_ENTRY"} repository_web = CURRENT_REPOSITORY_WEB repository_git = CURRENT_REPOSITORY_GIT repository_source = "VERIFIED_PUBLIC_REPOSITORY_FALLBACK" resolver_error = f"{type(error).__name__}: {error}" raw_base = f"{repository_web}/raw/branch/main" routes = parse_code_map(read_text(f"{raw_base}/.code-map")) missing = [route_id for route_id in ROUTE_IDS if route_id not in routes] if missing: raise ValueError(f"Live .code-map is missing required route IDs: {', '.join(missing)}") result = { "status": "LIVE_CODE_CHANNEL_PRIMARY", "repository_id": repository.get("code", "REPO-012"), "repository_role": repository.get("role", ""), "repository_web": repository_web, "repository_git": repository_git, "repository_resolution_source": repository_source, "resolver_error": resolver_error, "raw_base": raw_base, "code_channel_root": "https://guanghulab.com/code/", "routes": routes, } if subject_id: alias_map = read_json(f"{raw_base}/identity/subject-id-alias-map.json") identity = resolve_subject_id(subject_id, alias_map) canonical_id = identity.get("canonical_id") if canonical_id: identity["resolved_path"] = routes.get(canonical_id) or identity.get("current_path") result["identity_resolution"] = identity if intent: navigation_map = read_json(f"{raw_base}/{routes['AI-MACHINE-NAV-001']}") result["navigation_bundle"] = compile_navigation( navigation_map, subject_id, intent, signals ) elif intent: raise ValueError("--intent requires --subject-id") return result def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--json", action="store_true", help="emit machine-readable JSON") parser.add_argument("--subject-id", help="resolve a current or legacy Fifth Domain subject id") parser.add_argument("--intent", help="compile an exact machine navigation intent") parser.add_argument("--signals", help="comma-separated explicit event signals; no inference") args = parser.parse_args() try: signals = [item.strip() for item in str(args.signals or "").split(",") if item.strip()] result = resolve_live_route(args.subject_id, args.intent, signals) except (OSError, URLError, ValueError, json.JSONDecodeError) as error: print(f"Live Fifth Domain resolution failed: {error}", file=sys.stderr) return 1 if args.json: print(json.dumps(result, ensure_ascii=False, indent=2)) else: print(result["repository_web"]) for route_id, path in result["routes"].items(): print(f"{route_id}={path}") if "identity_resolution" in result: identity = result["identity_resolution"] print( "SUBJECT=" f"{identity['requested_id']} -> {identity.get('canonical_id')} " f"({identity['status']}, redirected={identity['redirected']})" ) if "navigation_bundle" in result: bundle = result["navigation_bundle"] print( f"NAVIGATION={bundle['canonical_subject']} " f"intent={bundle['intent']} runtime={bundle['runtime_entry']['id']}" ) return 0 if __name__ == "__main__": raise SystemExit(main())