187 lines
7.2 KiB
Python
187 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Resolve a host-specific route from the Guanghu Lighthouse registries."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
SCRIPT_PATH = Path(__file__).resolve()
|
||
LOCAL_SOURCE_ROOT = SCRIPT_PATH.parents[3]
|
||
RUNTIME_ROOT = Path("/Volumes/JZAO/HoloLake/persona-runtime")
|
||
POINTER = RUNTIME_ROOT / "TCS-ROOT.json"
|
||
SHARED_ROOT = RUNTIME_ROOT / "shared"
|
||
|
||
|
||
def canonical_repo_root() -> Path | None:
|
||
if (LOCAL_SOURCE_ROOT / "routing" / "lighthouse-path-registry.json").is_file():
|
||
return LOCAL_SOURCE_ROOT
|
||
if POINTER.is_file() and not POINTER.is_symlink():
|
||
value = json.loads(POINTER.read_text(encoding="utf-8"))
|
||
candidate = Path(value.get("canonical_repo", ""))
|
||
if (candidate / "routing" / "lighthouse-path-registry.json").is_file():
|
||
return candidate
|
||
return None
|
||
|
||
|
||
REPO_ROOT = canonical_repo_root()
|
||
if REPO_ROOT:
|
||
DEFAULT_LIGHTHOUSE = REPO_ROOT / "routing" / "lighthouse-path-registry.json"
|
||
DEFAULT_HOST_MAP = REPO_ROOT / "routing" / "host-skill-navigation-map.json"
|
||
else:
|
||
DEFAULT_LIGHTHOUSE = SHARED_ROOT / "registries" / "lighthouse-path-registry.json"
|
||
DEFAULT_HOST_MAP = SHARED_ROOT / "registries" / "host-skill-navigation-map.json"
|
||
CURRENT_STATES = {"CURRENT", "ACTIVE", "ACTIVE_CANDIDATE", "ACTIVE_LOCAL_PRIVATE"}
|
||
|
||
|
||
def load_json(path: Path) -> dict:
|
||
with path.open(encoding="utf-8") as handle:
|
||
return json.load(handle)
|
||
|
||
|
||
def normalise(value: str) -> str:
|
||
return "".join(value.lower().split())
|
||
|
||
|
||
def resolve_host(host_map: dict, requested: str) -> dict | None:
|
||
key = normalise(requested)
|
||
for host in host_map["hosts"]:
|
||
if key in {normalise(host["id"]), *(normalise(alias) for alias in host.get("aliases", []))}:
|
||
return host
|
||
return None
|
||
|
||
|
||
def resolve_intent(host_map: dict, text: str) -> dict | None:
|
||
value = normalise(text)
|
||
scored = []
|
||
for intent in host_map["intents"]:
|
||
score = sum(len(normalise(phrase)) for phrase in intent["phrases"] if normalise(phrase) in value)
|
||
if score:
|
||
scored.append((score, intent["id"], intent))
|
||
return max(scored, default=(0, "", None))[2]
|
||
|
||
|
||
def internal_route_query(host_map: dict, host_name: str, intent_text: str, reason: str) -> dict:
|
||
examples = [
|
||
{
|
||
"intent_id": intent["id"],
|
||
"target_id": intent.get("target_id"),
|
||
"examples": intent.get("phrases", [])[:2],
|
||
}
|
||
for intent in host_map.get("intents", [])[:8]
|
||
]
|
||
if reason == "HOST_UNKNOWN_NO_GUESS":
|
||
query = f"请由当前人格体向TCS母体和光湖灯塔查询：自然语言意图“{intent_text}”应由哪个已登记宿主承接？"
|
||
else:
|
||
query = f"请由当前人格体向TCS母体和光湖灯塔查询：自然语言意图“{intent_text}”对应哪个当前登记频道、系统、仓库或办公室？"
|
||
return {
|
||
"required": True,
|
||
"query_target": "TCS_MOTHER_BRAIN_RUNTIME",
|
||
"fallback_target": "GUANGHU_LIGHTHOUSE",
|
||
"question_for": "CURRENT_PERSONA_INTERNAL_NAVIGATION",
|
||
"query": query,
|
||
"reason": "UNKNOWN_ROUTE_IS_QUERYABLE_NOT_HUMAN_BLOCKING",
|
||
"examples": examples,
|
||
"resume": "RETRY_ROUTE_AFTER_INTERNAL_READBACK",
|
||
}
|
||
|
||
|
||
def resolve_path(registry: dict, route_id: str) -> dict:
|
||
for path in registry["paths"]:
|
||
if path["id"].upper() == route_id.upper():
|
||
return {
|
||
"status": "VALID_REGISTERED" if path["state"] in CURRENT_STATES else "INVALID_STATE",
|
||
"path": path,
|
||
}
|
||
for redirect in registry.get("redirects", []):
|
||
if redirect["id"].upper() == route_id.upper():
|
||
return {"status": "REDIRECT_ONLY_NOT_CURRENT", "redirect": redirect}
|
||
return {"status": "INVALID_NO_GUESS", "route_id": route_id}
|
||
|
||
|
||
def keychain_helper_status() -> dict:
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "config", "--global", "--get-all", "credential.helper"],
|
||
check=False,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=3,
|
||
)
|
||
except (OSError, subprocess.TimeoutExpired):
|
||
return {"status": "UNAVAILABLE", "secret_read": False}
|
||
helpers = [line.strip() for line in result.stdout.splitlines() if line.strip()]
|
||
return {
|
||
"status": "AVAILABLE" if "osxkeychain" in helpers else "UNAVAILABLE",
|
||
"helper": "osxkeychain" if "osxkeychain" in helpers else None,
|
||
"secret_read": False,
|
||
}
|
||
|
||
|
||
def compile_route(registry: dict, host_map: dict, host_name: str, intent_text: str) -> dict:
|
||
host = resolve_host(host_map, host_name)
|
||
if not host:
|
||
return {
|
||
"decision": "QUERY_ROUTE_SOURCE",
|
||
"error": "HOST_UNKNOWN_NO_GUESS",
|
||
"requested_host": host_name,
|
||
"route_query": internal_route_query(host_map, host_name, intent_text, "HOST_UNKNOWN_NO_GUESS"),
|
||
"authority_granted": False,
|
||
}
|
||
intent = resolve_intent(host_map, intent_text)
|
||
if not intent:
|
||
return {
|
||
"decision": "QUERY_ROUTE_SOURCE",
|
||
"error": "INTENT_UNKNOWN_NO_GUESS",
|
||
"requested_intent": intent_text,
|
||
"route_query": internal_route_query(host_map, host_name, intent_text, "INTENT_UNKNOWN_NO_GUESS"),
|
||
"authority_granted": False,
|
||
}
|
||
target = resolve_path(registry, intent["target_id"])
|
||
if target["status"] != "VALID_REGISTERED":
|
||
return {"decision": "BLOCK", "error": "LIGHTHOUSE_TARGET_INVALID", "target": target}
|
||
result = {
|
||
"schema": "guanghu.host-navigation-receipt/v1",
|
||
"decision": "ALLOW_NAVIGATION",
|
||
"lighthouse_id": registry["lighthouse_id"],
|
||
"registry_id": registry["registry_id"],
|
||
"registry_version": registry["version"],
|
||
"host": host,
|
||
"intent": intent,
|
||
"target": target["path"],
|
||
"authority_granted": False,
|
||
}
|
||
if intent.get("transport") == "LOCAL_GIT_OSXKEYCHAIN":
|
||
result["transport_probe"] = keychain_helper_status()
|
||
if result["transport_probe"]["status"] != "AVAILABLE":
|
||
result["decision"] = "BLOCK"
|
||
result["error"] = "KEYCHAIN_HELPER_UNAVAILABLE"
|
||
offline = target["path"].get("offline")
|
||
if offline and offline.startswith("/"):
|
||
result["offline_reality"] = {
|
||
"path": offline,
|
||
"exists": Path(os.path.expanduser(offline)).exists(),
|
||
}
|
||
if target["path"]["state"] == "ACTIVE_LOCAL_PRIVATE" and not result["offline_reality"]["exists"]:
|
||
result["decision"] = "BLOCK"
|
||
result["error"] = "PRIVATE_VOLUME_MISSING"
|
||
return result
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--host", required=True)
|
||
parser.add_argument("--intent", required=True)
|
||
parser.add_argument("--lighthouse", type=Path, default=DEFAULT_LIGHTHOUSE)
|
||
parser.add_argument("--host-map", type=Path, default=DEFAULT_HOST_MAP)
|
||
args = parser.parse_args()
|
||
result = compile_route(load_json(args.lighthouse), load_json(args.host_map), args.host, args.intent)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
return 0 if result["decision"] == "ALLOW_NAVIGATION" else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|