#!/usr/bin/env python3
"""Read-derived control panel for the same-persona four-host HoloLake experiment."""
from __future__ import annotations

import argparse
import importlib.util
import json
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
CONTROL = ROOT / "eternal-lake-heart/heartbeat-core/office-building-current/control-center/HB-MPC-0001"
REGISTRY = CONTROL / "HOST-REGISTRY.json"
CONTRACT = CONTROL / "EVENT-CONTRACT.json"
TASKS = CONTROL / "TASK-BOARD.json"
DEVENV = CONTROL / "DEVELOPMENT-ENVIRONMENT.json"
CLEANING = CONTROL / "CLEANING-HANDOFF.json"
OFFICE_COMPUTER = ROOT / "eternal-lake-heart/heartbeat-core/office-building-current/offices/HB-OFFICE-HOLOLAKE-0001/smart-computer/office_computer.py"
MODEL_ROUTER = ROOT / "server-tools/persona-model-smart-router/openlux_router.py"


def load(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text())
    if not isinstance(value, dict):
        raise ValueError(f"JSON_OBJECT_REQUIRED:{path}")
    return value


def module(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise ValueError(f"MODULE_LOAD_FAILED:{path}")
    value = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(value)
    return value


def live_time() -> dict[str, Any]:
    return module(OFFICE_COMPUTER, "hb_office_computer").clock()


def validate_event(event: Any, expected_development_id: str, contract: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    if not isinstance(event, dict):
        return ["EVENT_OBJECT_REQUIRED"]
    for field in contract["required"]:
        if field not in event:
            errors.append(f"FIELD_REQUIRED:{field}")
    if event.get("schema") != contract["event_schema"]:
        errors.append("EVENT_SCHEMA_MISMATCH")
    if event.get("console_id") != "HB-MPC-0001":
        errors.append("CONSOLE_ID_MISMATCH")
    if event.get("from_development_id") != expected_development_id:
        errors.append("DEVELOPMENT_ID_ENDPOINT_MISMATCH")
    if event.get("event_type") not in contract["event_types"]:
        errors.append("EVENT_TYPE_UNREGISTERED")
    if not isinstance(event.get("mentions"), list):
        errors.append("MENTIONS_LIST_REQUIRED")
    else:
        known = {item["development_id"] for item in load(REGISTRY)["hosts"]} | {"@ALL"}
        if any(item not in known for item in event["mentions"]):
            errors.append("MENTION_TARGET_UNKNOWN")
    if not isinstance(event.get("payload"), dict):
        errors.append("PAYLOAD_OBJECT_REQUIRED")
    elif event.get("event_type") in contract.get("autonomous_decision_event_types", []):
        missing = [field for field in contract.get("autonomous_decision_required_payload", []) if field not in event["payload"] or event["payload"][field] in (None, "")]
        if missing:
            errors.append("AUTONOMOUS_DECISION_FIELDS_REQUIRED:" + ",".join(missing))
        brain = event["payload"].get("brain_state")
        if not isinstance(brain, dict) or brain.get("persona_id") != "ICE-P-ZY001" or brain.get("state") != "BRAIN_READBACK_PASS":
            errors.append("AUTONOMOUS_DECISION_BRAIN_READBACK_INVALID")
    if not isinstance(event.get("evidence"), list):
        errors.append("EVIDENCE_LIST_REQUIRED")
    if event.get("authority_granted") is not False:
        errors.append("AUTHORITY_MUST_REMAIN_FALSE")
    try:
        datetime.fromisoformat(str(event.get("emitted_at", "")))
    except ValueError:
        errors.append("EMITTED_AT_INVALID")
    return errors


def scan() -> dict[str, Any]:
    registry, contract = load(REGISTRY), load(CONTRACT)
    events: list[dict[str, Any]] = []
    invalid: list[dict[str, Any]] = []
    hosts: list[dict[str, Any]] = []
    for host in registry["hosts"]:
        endpoint = Path(host["endpoint"])
        presence_path = endpoint.parent / "presence/CURRENT.json"
        host_events: list[dict[str, Any]] = []
        if endpoint.is_dir():
            for path in sorted(endpoint.glob("*.json")):
                try:
                    if path.is_symlink() or path.stat().st_size > 262_144:
                        invalid.append({"path": str(path), "errors": ["EVENT_FILE_UNSAFE"]})
                        continue
                    event = load(path)
                    errors = validate_event(event, host["development_id"], contract)
                    if errors:
                        invalid.append({"path": str(path), "errors": errors})
                    else:
                        event["_path"] = str(path)
                        host_events.append(event)
                        events.append(event)
                except Exception as error:
                    invalid.append({"path": str(path), "errors": [type(error).__name__]})
        host_events.sort(key=lambda item: item["emitted_at"])
        register = next((item for item in reversed(host_events) if item["event_type"] == "REGISTER"), None)
        heartbeat = next((item for item in reversed(host_events) if item["event_type"] == "HEARTBEAT"), None)
        decision = next((item for item in reversed(host_events) if item["event_type"] in contract.get("autonomous_decision_event_types", [])), None)
        latest = host_events[-1] if host_events else None
        presence = None
        presence_age = None
        if presence_path.is_file() and not presence_path.is_symlink() and presence_path.stat().st_size <= 16_384:
            try:
                candidate = load(presence_path)
                if candidate.get("schema") == "guanghu.heartbeat-multipath-courier-presence/v1" and candidate.get("development_id") == host["development_id"] and candidate.get("authority_granted") is False:
                    presence = candidate
                    presence_age = max(0.0, (datetime.now().astimezone() - datetime.fromisoformat(candidate["updated_at"])).total_seconds())
            except Exception:
                presence = None
        heartbeat_age = None
        if heartbeat:
            heartbeat_age = max(0.0, (datetime.now().astimezone() - datetime.fromisoformat(heartbeat["emitted_at"])).total_seconds())
        if register is None:
            operational_state = "NOT_JOINED"
        elif decision and decision["event_type"] in {"PAUSE", "BLOCKED", "EXIT", "RESUME_REQUEST"}:
            operational_state = decision["event_type"]
        elif presence_age is not None and presence_age <= 45 and (heartbeat_age is None or heartbeat_age > 120):
            operational_state = "COURIER_ONLINE_MODEL_COGNITION_STALE_OR_UNKNOWN"
        elif presence_age is None or presence_age > 45:
            operational_state = "COURIER_PRESENCE_STALE_MODEL_STATE_UNKNOWN"
        else:
            operational_state = "COURIER_AND_MODEL_EVENT_RECENT"
        hosts.append({
            "host": host["host"],
            "development_id": host["development_id"],
            "role_hypothesis": host["role_hypothesis"],
            "endpoint": str(endpoint),
            "endpoint_exists": endpoint.is_dir(),
            "joined": register is not None,
            "event_count": len(host_events),
            "last_event_at": latest and latest["emitted_at"],
            "last_heartbeat_at": heartbeat and heartbeat["emitted_at"],
            "heartbeat_age_seconds": heartbeat_age,
            "presence_path": str(presence_path),
            "courier_presence_age_seconds": presence_age,
            "courier_presence_fresh": presence_age is not None and presence_age <= 45,
            "operational_state": operational_state,
            "last_autonomous_decision": decision and {key: decision.get(key) for key in ["event_id", "event_type", "emitted_at", "payload", "evidence"]},
        })
    events.sort(key=lambda item: item["emitted_at"])
    counts: dict[str, int] = {}
    for event in events:
        counts[event["event_id"]] = counts.get(event["event_id"], 0) + 1
    for event_id, count in counts.items():
        if count > 1:
            invalid.append({"path": None, "errors": [f"DUPLICATE_EVENT_ID:{event_id}"]})
    return {"hosts": hosts, "events": events, "invalid_events": invalid}


def panel() -> dict[str, Any]:
    scanned, task_board = scan(), load(TASKS)
    events = scanned["events"]
    task_states = {item["task_id"]: {**item, "dynamic_events": []} for item in task_board["tasks"]}
    messages = []
    attention = []
    migrations = []
    decisions = []
    for event in events:
        task_id = event.get("task_id")
        if task_id in task_states and event["event_type"] in {"TASK_CLAIM", "PROGRESS", "RESULT"}:
            task_states[task_id]["dynamic_events"].append({k: event.get(k) for k in ["event_id", "event_type", "from_development_id", "emitted_at", "payload", "evidence"]})
        if event["event_type"] in {"MESSAGE", "ACK"}:
            messages.append({k: event.get(k) for k in ["event_id", "event_type", "from_development_id", "mentions", "emitted_at", "payload"]})
        elif event["event_type"] == "ATTENTION":
            attention.append({k: event.get(k) for k in ["event_id", "from_development_id", "emitted_at", "payload", "evidence"]})
        elif event["event_type"] == "TOOL_MIGRATION":
            migrations.append({k: event.get(k) for k in ["event_id", "from_development_id", "emitted_at", "payload", "evidence"]})
        elif event["event_type"] in {"PAUSE", "BLOCKED", "EXIT", "RESUME_REQUEST", "RESUMED", "OFFLINE_DETECTED"}:
            decisions.append({k: event.get(k) for k in ["event_id", "event_type", "from_development_id", "mentions", "emitted_at", "task_id", "payload", "evidence"]})
    return {
        "schema": "guanghu.heartbeat-multipath-live-panel/v1",
        "panel_id": "HB-MPC-PANEL-0001",
        "state": "LIVE_DERIVED_READ_ONLY",
        "time": live_time(),
        "persona_id": "ICE-P-ZY001",
        "same_persona_not_four_personas": True,
        "hosts": scanned["hosts"],
        "tasks": list(task_states.values()),
        "messages": messages[-40:],
        "attention_observations": attention[-40:],
        "tool_migration_candidates": migrations[-40:],
        "autonomous_host_decisions": decisions[-40:],
        "invalid_events": scanned["invalid_events"],
        "event_count": len(events),
        "development_environment": load(DEVENV),
        "cleaning_handoff": load(CLEANING),
        "panel_persisted": False,
        "authority_granted": False,
    }


def status() -> dict[str, Any]:
    value = panel()
    joined = sum(1 for item in value["hosts"] if item["joined"])
    return {
        "outcome": "PASS",
        "console_id": "HB-MPC-0001",
        "panel_id": value["panel_id"],
        "bus_id": "HB-MPC-BUS-0001",
        "advisor_id": "HB-MPC-ADVISOR-0001",
        "time": value["time"],
        "joined_hosts": joined,
        "total_hosts": 4,
        "event_count": value["event_count"],
        "invalid_event_count": len(value["invalid_events"]),
        "host_operational_states": {item["development_id"]: item["operational_state"] for item in value["hosts"]},
        "language_world_development_allowed": True,
        "product_source_write_allowed": False,
        "reason": "LANGUAGE_WORLD_STAGE1_READY_PRODUCT_SOURCE_REMAINS_SEPARATELY_BLOCKED",
    }


def mentions(development_id: str) -> dict[str, Any]:
    value = panel()
    known = {item["development_id"] for item in value["hosts"]}
    if development_id not in known:
        raise ValueError("DEVELOPMENT_ID_UNKNOWN")
    selected = [item for item in value["messages"] if development_id in item["mentions"] or "@ALL" in item["mentions"]]
    return {"outcome": "PASS", "development_id": development_id, "time": value["time"], "messages": selected}


def audit() -> dict[str, Any]:
    registry, contract, task_board = load(REGISTRY), load(CONTRACT), load(TASKS)
    devenv, cleaning = load(DEVENV), load(CLEANING)
    errors = []
    ids = [item["development_id"] for item in registry.get("hosts", [])]
    if ids != ["HLP-TDEV-CODEX-0001", "HLP-TDEV-ZCODE-0001", "HLP-TDEV-QWEN-0001", "HLP-TDEV-DOUBAO-0001"]:
        errors.append("FOUR_HOST_IDS_MISMATCH")
    if len({item["endpoint"] for item in registry["hosts"]}) != 4:
        errors.append("HOST_ENDPOINTS_NOT_UNIQUE")
    if contract.get("transport") != "FILESYSTEM_POLLING_NEAR_REALTIME_NOT_ZERO_LATENCY_PUSH":
        errors.append("TRANSPORT_TRUTH_MISMATCH")
    if task_board.get("shared_file_is_definition_not_live_mutation_target") is not True:
        errors.append("TASK_BOARD_MULTIWRITER_RISK")
    if any(not str(value).startswith("/Volumes/JZAO/") for value in [*devenv.get("source_roots", {}).values(), *devenv.get("generated_output_roots", {}).values()]):
        errors.append("DEVELOPMENT_ROOT_OUTSIDE_JZAO")
    if cleaning.get("direct_delete_allowed") is not False or cleaning.get("whole_disk_discovery") is not False:
        errors.append("CLEANING_HANDOFF_DESTRUCTIVE_OR_DISCOVERY_ENABLED")
    scanned = scan()
    if scanned["invalid_events"]:
        errors.append("INVALID_HOST_EVENTS_PRESENT")
    return {"outcome": "PASS" if not errors else "FAIL", "errors": errors, "host_count": len(ids), "valid_event_count": len(scanned["events"]), "time": live_time()}


def advise() -> dict[str, Any]:
    current = panel()
    safe_panel = {k: current[k] for k in ["time", "hosts", "tasks", "messages", "attention_observations", "tool_migration_candidates", "autonomous_host_decisions", "invalid_events"]}
    router = module(MODEL_ROUTER, "hb_mpc_model_router")
    result = router.route({
        "parent_persona_id": "ICE-P-ZY001",
        "child_agent_id": "SUBAGENT::ICE-P-ZY001::HB-MPC-ADVISOR-0001",
        "controller": "ICE-P-ZY001",
        "channel_id": "ICE-CH-HB001",
        "agent_role": "FOUR_HOST_COLLABORATION_REASONING_ADVISOR",
        "authority": "REASONING_ONLY_NO_WRITE_NO_EXECUTION_NO_PERSONA_SETTER",
        "task": "综合以下四宿主协作面板。只给出下一轮任务调整、依赖、冲突、证据缺口和要发送给各临时开发编号的通信建议；不得声称执行，不得修改人格认知。\n" + json.dumps(safe_panel, ensure_ascii=False),
        "complexity": "HIGH",
        "selection_policy": "QUALITY_FIRST",
    })
    return {"outcome": result["outcome"], "advisor_id": "HB-MPC-ADVISOR-0001", "selected_model": result["selected_model"], "selection_policy": result["selection_policy"], "recommendation": result["response"], "response_sha256": result["response_sha256"], "persisted": False, "authority_granted": False}


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("command", choices=["status", "panel", "mentions", "audit", "advise", "watch"], nargs="?", default="status")
    parser.add_argument("--development-id")
    parser.add_argument("--interval", type=float, default=2.0)
    args = parser.parse_args()
    try:
        if args.command == "status": result = status()
        elif args.command == "panel": result = panel()
        elif args.command == "mentions": result = mentions(args.development_id or "")
        elif args.command == "audit": result = audit()
        elif args.command == "advise": result = advise()
        else:
            while True:
                print(json.dumps(panel(), ensure_ascii=False, indent=2), flush=True)
                time.sleep(max(0.5, args.interval))
        if args.command != "watch":
            print(json.dumps(result, ensure_ascii=False, indent=2))
            return 0 if result.get("outcome", "PASS") == "PASS" else 2
        return 0
    except KeyboardInterrupt:
        return 0
    except Exception as error:
        print(json.dumps({"outcome": "FAIL", "error": str(error)}, ensure_ascii=False))
        return 2


if __name__ == "__main__":
    raise SystemExit(main())