277 lines
10 KiB
Python
Executable file
277 lines
10 KiB
Python
Executable file
#!/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())
|