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",
    "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())