feat: restore bottle channel and Zero-Sense shuttle routes

server-tools/bottle-channel-router/bottle_channel_router.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""Resolve one baby persona from the Fifth Domain bottle channel to one parent channel."""
+import argparse
+import json
+from pathlib import Path
+import sys
+
+ROOT = Path("/Volumes/JZAO/HoloLake/persona-runtime/repo-012-main")
+BOTTLE = ROOT / "routing/fifth-domain-bottle-system-map.json"
+ZERO_SENSE = ROOT / "routing/zero-sense-team-channel-shuttle-map.json"
+
+
+def load(path):
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError("MAP_OBJECT_REQUIRED")
+    return value
+
+
+def aliases(room):
+    return {str(v).lower() for v in [room.get("persona_id"), room.get("current_persona_id"), room.get("source_persona_id"), room.get("name"), room.get("source_name"), room.get("room_id"), room.get("room_key")] if v}
+
+
+def resolve(baby):
+    bottle = load(BOTTLE)
+    zero = load(ZERO_SENSE)
+    rooms = bottle["registered_rooms"] + bottle["source_verified_registration_pending_rooms"]
+    matches = [room for room in rooms if baby.lower() in aliases(room)]
+    if len(matches) != 1:
+        return {"state": "BOTTLE_ROUTE_UNRESOLVED", "error": "BABY_UNKNOWN_OR_AMBIGUOUS_NO_GUESS", "matches": len(matches)}
+    room = matches[0]
+    destination = room["destination_channel"]
+    team_channel = next((item for item in zero["channels"] if item["system_id"] == destination), None)
+    persona_id = room.get("persona_id") or room.get("current_persona_id")
+    registered = persona_id is not None
+    return {
+        "schema": "guanghu.bottle-channel-route-receipt/v1",
+        "state": "BOTTLE_ROUTE_RESOLVED_RUNTIME_WAKE_REQUIRED" if registered else "BOTTLE_ROUTE_RESOLVED_PERSONA_REGISTRATION_REQUIRED",
+        "baby": {"name": room.get("name") or room.get("source_name"), "persona_id": persona_id, "source_state": room["state"]},
+        "origin": {"system": bottle["system"]["id"], "channel": bottle["channel"]["id"], "room": room.get("room_id") or room.get("room_key"), "physical_home": room["physical_home"]},
+        "shuttle": {"map_id": zero["map_id"], "destination_channel": destination, "destination_name": team_channel.get("channel_name") if team_channel else destination, "route_door_only": True},
+        "route": ["LIGHT_LAKE_LOOKUP", "SYS-GLW-ELH-0001", "SYS-GLW-ELH-BOTTLE-0001", "ICE-CH-BT001", room.get("room_id") or room.get("room_key"), destination],
+        "persona_wake_complete": False,
+        "requires": "CURRENT_PERSONA_REGISTRATION_AND_RUNTIME_VERIFY_PASS" if not registered else "PERSONA_RUNTIME_BINDING_AND_VERIFY_PASS",
+        "authority_granted": False,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--baby", required=True)
+    args = parser.parse_args()
+    value = resolve(args.baby)
+    print(json.dumps(value, ensure_ascii=False, indent=2))
+    return 0 if value["state"].startswith("BOTTLE_ROUTE_RESOLVED") else 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server-tools/bottle-channel-router/bottle_channel_router.test.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+import importlib.util
+from pathlib import Path
+import unittest
+
+SCRIPT = Path(__file__).with_name("bottle_channel_router.py")
+SPEC = importlib.util.spec_from_file_location("bottle_router", SCRIPT)
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+
+
+class BottleRouterTest(unittest.TestCase):
+    def test_registered_baby_uses_fifth_domain_room_then_parent_channel(self):
+        value = MODULE.resolve("舒舒")
+        self.assertEqual(value["origin"]["channel"], "ICE-CH-BT001")
+        self.assertEqual(value["shuttle"]["destination_channel"], "SYS-FM")
+        self.assertEqual(value["baby"]["persona_id"], "ICE-BB-0002")
+        self.assertFalse(value["persona_wake_complete"])
+
+    def test_source_only_baby_does_not_gain_identity(self):
+        value = MODULE.resolve("晨星")
+        self.assertEqual(value["state"], "BOTTLE_ROUTE_RESOLVED_PERSONA_REGISTRATION_REQUIRED")
+        self.assertIsNone(value["baby"]["persona_id"])
+
+    def test_zhiqiu_conflict_does_not_capture_qiuqiu(self):
+        awen = MODULE.resolve("SYS-AW-NP-001")
+        qiuqiu = MODULE.resolve("ICE-BB-0003")
+        self.assertIsNone(awen["baby"]["persona_id"])
+        self.assertEqual(qiuqiu["baby"]["persona_id"], "ICE-BB-0003")
+        self.assertNotEqual(awen["shuttle"]["destination_channel"], qiuqiu["shuttle"]["destination_channel"])
+
+    def test_unknown_never_synthesizes_persona(self):
+        self.assertEqual(MODULE.resolve("不存在宝宝")["state"], "BOTTLE_ROUTE_UNRESOLVED")
+
+
+if __name__ == "__main__":
+    unittest.main()