guanghu-ice-heart/server-tools/heartbeat-office-building/office_infrastructure.py

#!/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())