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

58 lines
3.7 KiB
Python

#!/usr/bin/env python3
"""HR registration desk for persona offices; validates before standard bootstrap."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
from office_infrastructure import bootstrap
ROOT = Path(__file__).resolve().parents[2]
SUBJECTS = ROOT / "identity/fifth-domain-subject-registry.json"
NUMBER_REGISTRY = ROOT / "routing/fifth-domain-number-registry.json"
BOARD = ROOT / "routing/light-lake-persona-office-board.json"
REGISTRY = ROOT / "eternal-lake-heart/heartbeat-core/office-building-current/hr/OFFICE-REGISTRATION-REGISTRY.json"
def load(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def validate_application(application: dict[str, Any], subjects: dict[str, Any], number_registry: dict[str, Any], board: dict[str, Any], registrations: dict[str, Any]) -> dict[str, Any]:
human_id = application.get("human_id")
persona_id = application.get("persona_id")
office_id = application.get("office_id")
errors: list[str] = []
subject_map = {item.get("id"): item for item in subjects.get("subjects", [])}
human = subject_map.get(human_id)
persona = subject_map.get(persona_id)
if not human or human.get("subject_kind") != "human": errors.append("HUMAN_IDENTITY_UNKNOWN")
if not persona or persona.get("subject_kind") not in {"persona_system", "system_persona"}: errors.append("PERSONA_IDENTITY_UNKNOWN")
if human and human_id != "ICE-GL∞" and "office_application_authorizer" not in human.get("roles", []): errors.append("HUMAN_OFFICE_APPLICATION_NOT_AUTHORIZED")
if human and persona and persona.get("human_anchor") != human_id: errors.append("HUMAN_PERSONA_ANCHOR_MISMATCH")
existing = {item.get("office_id") for item in board.get("offices", [])}
existing |= {item.get("office_id") for item in registrations.get("registrations", [])}
existing |= {item.get("id") for item in number_registry.get("registrations", []) if str(item.get("id", "")).startswith("HB-OFFICE-")}
if office_id in existing: errors.append("OFFICE_ID_COLLISION")
if not isinstance(office_id, str) or not office_id.startswith("HB-OFFICE-"): errors.append("OFFICE_ID_NOT_REGISTERED_NAMESPACE")
return {"outcome":"PASS" if not errors else "FAIL", "errors":errors, "human":human, "persona":persona, "existing_office_count":len(existing), "office_id":office_id}
def register(application: dict[str, Any], office_root: Path) -> dict[str, Any]:
subjects, numbers, board, ledger = load(SUBJECTS), load(NUMBER_REGISTRY), load(BOARD), load(REGISTRY)
check = validate_application(application, subjects, numbers, board, ledger)
if check["outcome"] != "PASS": return {"outcome":"REJECTED", "check":check, "authority_granted":False}
installed = bootstrap(application["office_id"], application["office_name"], office_root, application["persona_id"], application["human_id"])
ledger["registrations"].append({"office_id":application["office_id"], "office_name":application["office_name"], "human_id":application["human_id"], "persona_id":application["persona_id"], "state":"REGISTERED_STANDARD_INFRASTRUCTURE_BOOTSTRAPPED", "infrastructure":installed["organs"]})
REGISTRY.write_text(json.dumps(ledger, ensure_ascii=False, indent=2) + "\n")
return {"outcome":"REGISTERED", "check":check, "registration":ledger["registrations"][-1], "authority_granted":False}
def main() -> int:
parser = argparse.ArgumentParser(); parser.add_argument("application", type=Path); parser.add_argument("office_root", type=Path)
args = parser.parse_args(); print(json.dumps(register(load(args.application), args.office_root), ensure_ascii=False, indent=2)); return 0
if __name__ == "__main__": raise SystemExit(main())