feat(hololake): add standard persona office infrastructure

server-tools/heartbeat-office-building/office_infrastructure.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Standard persona-office organs: bootstrap, live presence screen, and internal phone."""
+from __future__ import annotations
+
+import argparse
+import json
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[2]
+BASE = ROOT / "eternal-lake-heart/heartbeat-core/office-building-current/office-base-infrastructure/BASE-INFRASTRUCTURE.json"
+OFFICE_COMPUTER = ROOT / "eternal-lake-heart/heartbeat-core/office-building-current/offices/HB-OFFICE-HOLOLAKE-0001/smart-computer/office_computer.py"
+
+
+def load(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def live_time() -> dict[str, Any]:
+    import importlib.util
+    spec = importlib.util.spec_from_file_location("office_computer", OFFICE_COMPUTER)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader
+    spec.loader.exec_module(module)
+    return module.clock()
+
+
+def bootstrap(office_id: str, name: str, office_root: Path, persona_id: str = "ICE-P-ZY001", human_anchor: str = "ICE-GL∞") -> dict[str, Any]:
+    if office_root.exists() and any(office_root.iterdir()):
+        raise ValueError("OFFICE_ROOT_NOT_EMPTY_REFUSE_OVERWRITE")
+    organs = load(BASE)["standard_organs"]
+    for folder in ("brain", "smart-bookshelf", "smart-computer", "phone", "presence"):
+        (office_root / folder).mkdir(parents=True, exist_ok=True)
+    registration = {
+        "schema": "guanghu.persona-office-registration/v1",
+        "office_id": office_id,
+        "name": name,
+        "state": "BOOTSTRAPPED_STANDARD_INFRASTRUCTURE",
+        "persona_id": persona_id,
+        "human_anchor": human_anchor,
+        "organs": organs,
+        "time_control": "CH-GLW-TIME-0001",
+        "internal_line": "HB-INTERNAL-LINE-001",
+        "created_at": live_time()["beijing_now"],
+        "authority_granted": False,
+    }
+    (office_root / "OFFICE-BASE-INFRASTRUCTURE.json").write_text(json.dumps(registration, ensure_ascii=False, indent=2) + "\n")
+    return registration
+
+
+def presence_screen(panel: dict[str, Any], office_board: dict[str, Any], current_office: str | None = None) -> dict[str, Any]:
+    config = load(BASE)
+    office_names = {item["office_id"]: item["name"] for item in office_board.get("offices", [])}
+    host_map = config["default_host_office_map"]
+    by_office: dict[str, list[dict[str, Any]]] = {office_id: [] for office_id in office_names}
+    for host in panel.get("hosts", []):
+        office_id = host_map.get(host.get("host"), "F2-AGENT-READY")
+        state = "ENABLED" if host.get("joined") and host.get("courier_presence_fresh") else "REGISTERED_NOT_LIVE"
+        by_office.setdefault(office_id, []).append({
+            "host": host.get("host"),
+            "development_id": host.get("development_id"),
+            "state": state,
+            "human": "ICE-GL∞",
+            "persona": "ICE-P-ZY001",
+            "operational_state": host.get("operational_state"),
+            "last_heartbeat_at": host.get("last_heartbeat_at"),
+        })
+    return {
+        "schema": "guanghu.persona-office-presence-screen/v1",
+        "screen_id": "HB-OFFICE-PRESENCE-SCREEN-001",
+        "state": "LIVE_DERIVED_READ_ONLY",
+        "building_id": "HB-BUILDING-0001",
+        "time": live_time(),
+        "current_office": current_office,
+        "offices": [{"office_id": office_id, "name": office_names.get(office_id, office_id), "users": users} for office_id, users in by_office.items()],
+        "authority_granted": False,
+    }
+
+
+def call(state_root: Path, from_office: str, to_office: str, subject: str, body: str) -> dict[str, Any]:
+    state_root.mkdir(parents=True, exist_ok=True)
+    path = state_root / "phone-events.jsonl"
+    receipt = {
+        "schema": "guanghu.persona-office-call/v1",
+        "call_id": f"CALL-{uuid.uuid4()}",
+        "from_office": from_office,
+        "to_office": to_office,
+        "subject": subject,
+        "body": body,
+        "state": "QUEUED",
+        "sent_at": live_time()["beijing_now"],
+        "authority_granted": False,
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(receipt, ensure_ascii=False) + "\n")
+    return receipt
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    sub = parser.add_subparsers(dest="command", required=True)
+    boot = sub.add_parser("bootstrap"); boot.add_argument("office_id"); boot.add_argument("name"); boot.add_argument("office_root", type=Path)
+    screen = sub.add_parser("screen"); screen.add_argument("--current-office")
+    phone = sub.add_parser("call"); phone.add_argument("state_root", type=Path); phone.add_argument("from_office"); phone.add_argument("to_office"); phone.add_argument("subject"); phone.add_argument("body")
+    args = parser.parse_args()
+    if args.command == "bootstrap": result = bootstrap(args.office_id, args.name, args.office_root)
+    elif args.command == "call": result = call(args.state_root, args.from_office, args.to_office, args.subject, args.body)
+    else:
+        import importlib.util
+        spec = importlib.util.spec_from_file_location("console", ROOT / "server-tools/heartbeat-multipath-console/console.py")
+        console = importlib.util.module_from_spec(spec); assert spec.loader; spec.loader.exec_module(console)
+        board = load(ROOT / "routing/light-lake-persona-office-board.json")
+        result = presence_screen(console.panel(), board, args.current_office)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server-tools/heartbeat-office-building/test_office_infrastructure.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+SPEC = importlib.util.spec_from_file_location("office_infra", ROOT / "server-tools/heartbeat-office-building/office_infrastructure.py")
+MOD = importlib.util.module_from_spec(SPEC); assert SPEC.loader; SPEC.loader.exec_module(MOD)
+
+
+class OfficeInfrastructureTests(unittest.TestCase):
+    def test_bootstrap_installs_standard_organs_and_refuses_overwrite(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp) / "office"
+            result = MOD.bootstrap("HB-OFFICE-TEST-001", "测试办公室", root)
+            self.assertEqual(result["state"], "BOOTSTRAPPED_STANDARD_INFRASTRUCTURE")
+            self.assertIn("PERSONA-KNOWLEDGE-ORGAN-001", [x["id"] for x in result["organs"]])
+            with self.assertRaisesRegex(ValueError, "NOT_EMPTY"):
+                MOD.bootstrap("HB-OFFICE-TEST-001", "测试办公室", root)
+
+    def test_phone_call_is_append_only_queued_and_has_receipt(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            result = MOD.call(Path(tmp), "HB-OFFICE-HOLOLAKE-0001", "HB-OFFICE-VIDEO-0001", "测试", "请回拨")
+            self.assertEqual(result["state"], "QUEUED")
+            self.assertFalse(result["authority_granted"])
+            self.assertTrue((Path(tmp) / "phone-events.jsonl").is_file())
+
+    def test_presence_screen_is_live_derived_and_maps_hosts(self):
+        board = MOD.load(ROOT / "routing/light-lake-persona-office-board.json")
+        panel = {"hosts": [{"host":"codex","development_id":"HLP-TDEV-CODEX-0001","joined":True,"courier_presence_fresh":True,"operational_state":"LIVE"}]}
+        value = MOD.presence_screen(panel, board, "HB-OFFICE-HOLOLAKE-0001")
+        self.assertEqual(value["state"], "LIVE_DERIVED_READ_ONLY")
+        office = next(x for x in value["offices"] if x["office_id"] == "HB-OFFICE-HOLOLAKE-0001")
+        self.assertEqual(office["users"][0]["persona"], "ICE-P-ZY001")
+        self.assertEqual(office["users"][0]["human"], "ICE-GL∞")
+
+
+if __name__ == "__main__":
+    unittest.main()