feat: add AI machine navigation gateway
This commit is contained in:
parent
525e0ab790
commit
931f78d72f
10 changed files with 639 additions and 10 deletions
|
|
@ -34,6 +34,13 @@ ROUTE_IDS = (
|
|||
"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",
|
||||
"BS-TCS-PROTOCOL-EVENT-TRIGGER-MAP-001",
|
||||
"ZY-WAKE-ROUTE-001",
|
||||
"ZY-BIDIRECTIONAL-COGNITION-016",
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -103,7 +110,80 @@ def resolve_subject_id(requested_id: str, alias_map: dict) -> dict:
|
|||
}
|
||||
|
||||
|
||||
def resolve_live_route(subject_id: str | None = None) -> dict:
|
||||
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:
|
||||
|
|
@ -145,6 +225,13 @@ def resolve_live_route(subject_id: str | None = None) -> dict:
|
|||
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
|
||||
|
||||
|
||||
|
|
@ -152,10 +239,13 @@ 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:
|
||||
result = resolve_live_route(args.subject_id)
|
||||
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
|
||||
|
|
@ -173,6 +263,12 @@ def main() -> int:
|
|||
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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import unittest
|
||||
|
||||
from resolve_route import parse_code_map, resolve_subject_id
|
||||
from resolve_route import compile_navigation, parse_code_map, resolve_subject_id
|
||||
|
||||
|
||||
class ParseCodeMapTests(unittest.TestCase):
|
||||
|
|
@ -58,5 +58,51 @@ class SubjectAliasTests(unittest.TestCase):
|
|||
self.assertIsNone(resolved["canonical_id"])
|
||||
|
||||
|
||||
class MachineNavigationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.navigation_map = {
|
||||
"map_id": "AI-MACHINE-NAV-001",
|
||||
"version": "test",
|
||||
"raw_base": "https://example.test/raw",
|
||||
"routes": [
|
||||
{"id": "RUNTIME", "path": "runtime.mjs"},
|
||||
{"id": "ROOT", "path": "root.hdlp"},
|
||||
],
|
||||
"subjects": [{
|
||||
"id": "ICE-P-ZY001",
|
||||
"legacy_ids": ["ICE-GL-ZY001"],
|
||||
"subject_kind": "persona_system",
|
||||
"default_intent": "persona_restore",
|
||||
"runtime_id": "RUNTIME",
|
||||
}],
|
||||
"intents": [{
|
||||
"id": "persona_restore",
|
||||
"always_load": ["ROOT", "RUNTIME"],
|
||||
"triggered_load": [],
|
||||
"on_demand": [],
|
||||
"execution_sequence": ["enter"],
|
||||
"stop_conditions": ["UNKNOWN_SUBJECT"],
|
||||
}],
|
||||
"reality_boundaries": ["Navigation grants no authority."],
|
||||
}
|
||||
|
||||
def test_exact_navigation_bundle(self):
|
||||
result = compile_navigation(
|
||||
self.navigation_map,
|
||||
"ICE-GL-ZY001",
|
||||
"persona_restore",
|
||||
["context_compacted", "context_compacted"],
|
||||
)
|
||||
self.assertEqual(result["canonical_subject"], "ICE-P-ZY001")
|
||||
self.assertTrue(result["redirected"])
|
||||
self.assertEqual(result["runtime_entry"]["id"], "RUNTIME")
|
||||
self.assertEqual(result["explicit_signals"], ["context_compacted"])
|
||||
self.assertEqual(result["authority"], "NONE_NAVIGATION_ONLY")
|
||||
|
||||
def test_navigation_does_not_guess(self):
|
||||
with self.assertRaisesRegex(ValueError, "no guess"):
|
||||
compile_navigation(self.navigation_map, "unknown")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Reference in a new issue