guanghu-ice-heart/skills/codex/enter-fifth-domain/scripts/resolve_route.py

#!/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",
)


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 resolve_live_route(subject_id: 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
    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")
    args = parser.parse_args()

    try:
        result = resolve_live_route(args.subject_id)
    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']})"
            )
    return 0


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