feat(lighthouse): add canonical three-host skill navigation
This commit is contained in:
parent
8423a96776
commit
fb5f931e78
14 changed files with 891 additions and 22 deletions
|
|
@ -0,0 +1,134 @@
|
|||
#!/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()
|
||||
REPO_ROOT = SCRIPT_PATH.parents[4]
|
||||
SHARED_ROOT = SCRIPT_PATH.parents[3]
|
||||
if (REPO_ROOT / "routing" / "lighthouse-path-registry.json").is_file():
|
||||
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 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": "BLOCK", "error": "HOST_UNKNOWN_NO_GUESS", "requested_host": host_name}
|
||||
intent = resolve_intent(host_map, intent_text)
|
||||
if not intent:
|
||||
return {"decision": "BLOCK", "error": "INTENT_UNKNOWN_NO_GUESS", "requested_intent": intent_text}
|
||||
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())
|
||||
Loading…
Reference in a new issue