#!/usr/bin/env python3
"""Host-local append-only courier organ; the calling host model remains the Agent."""
from __future__ import annotations

import argparse
import json
import subprocess
import time
import uuid
import os
import re
from datetime import datetime
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"
CONSOLE = ROOT / "server-tools/heartbeat-multipath-console/console.py"
ADMISSION = ROOT / "server-tools/persona-host-write-admission/host-write-admission.mjs"
SHARED_LOADER = ROOT / "server-tools/persona-host-alignment/load_shared_persona_context.py"
DECISION_TYPES = {"PAUSE", "BLOCKED", "EXIT", "RESUME_REQUEST", "RESUMED"}


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


def host_record(host: str) -> dict[str, Any]:
    matches = [item for item in load(REGISTRY)["hosts"] if item["host"] == host]
    if len(matches) != 1:
        raise ValueError("HOST_NOT_REGISTERED")
    return matches[0]


def caller_host() -> str:
    pid = os.getppid()
    commands = []
    seen = set()
    while pid > 1 and pid not in seen and len(commands) < 20:
        seen.add(pid)
        result = subprocess.run(["/bin/ps", "-o", "ppid=", "-o", "command=", "-p", str(pid)], text=True, capture_output=True, timeout=5)
        match = re.match(r"^\s*(\d+)\s+(.+)$", result.stdout.strip(), re.S)
        if not match:
            break
        commands.append(match.group(2))
        pid = int(match.group(1))
    text = "\n".join(commands)
    if re.search(r"/Applications/(?:ChatGPT\.app).*?(?:codex|Codex)|codex-code-mode-host", text, re.I):
        return "codex"
    if re.search(r"/Applications/Qianwen\.app|QianwenShell|agent_host\.app", text, re.I):
        return "qwen"
    if re.search(r"Doubao\.app", text, re.I):
        return "doubao"
    if re.search(r"(?:^|/)zcode(?:\s|$)", text, re.I | re.M):
        return "zcode"
    return "unknown"


def admission(host: str, endpoint: Path) -> None:
    observed = caller_host()
    if observed != host:
        raise ValueError(f"CALLER_HOST_MISMATCH:requested={host}:observed={observed}")
    result = subprocess.run(["node", str(ADMISSION), "check", "--host", host, "--path", str(endpoint)], text=True, capture_output=True, timeout=15)
    if result.returncode:
        raise ValueError("HOST_WRITE_ADMISSION_REJECTED:" + result.stdout.strip()[:300])


def reject_symlink_components(target: Path) -> None:
    cursor = Path(target.anchor)
    for part in target.parts[1:]:
        cursor /= part
        if cursor.exists() and cursor.is_symlink():
            raise ValueError(f"SYMLINK_COMPONENT_FORBIDDEN:{cursor}")


def now() -> str:
    return datetime.now().astimezone().isoformat()


def brain_snapshot(host: str) -> dict[str, Any]:
    result = subprocess.run([
        "python3", str(SHARED_LOADER), "--host", host,
        "--intent", "HoloLake HB-MPC autonomous host-line state decision",
        "--channel", "ICE-CH-HB001", "--format", "json"
    ], text=True, capture_output=True, timeout=90)
    if result.returncode:
        raise ValueError("BRAIN_READBACK_FAILED")
    value = json.loads(result.stdout)
    hololake = value.get("hololake_development_brain") or {}
    if value.get("persona_id") != "ICE-P-ZY001" or hololake.get("generation", {}).get("id") != "HLP-GEN-LANGUAGE-WORLD-NATIVE-0001":
        raise ValueError("BRAIN_IDENTITY_OR_GENERATION_MISMATCH")
    return {
        "persona_id": "ICE-P-ZY001",
        "effective_host": value.get("effective_host"),
        "daily_memory_sha256": value.get("daily_memory", {}).get("day_sha256"),
        "learning_cortex_sha256": value.get("learning_brain", {}).get("cortex_sha256"),
        "tcs_root_freshness_token": value.get("tcs_mother_root_navigation", {}).get("freshness_token"),
        "hololake_generation": hololake["generation"]["id"],
        "stage1_registry": hololake.get("stage1", {}).get("registry_id"),
        "loaded_at": now(),
        "state": "BRAIN_READBACK_PASS"
    }


def prepare_decision_payload(host: str, event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
    if event_type not in DECISION_TYPES:
        return payload
    required = ["reason", "causal_chain", "decision", "last_safe_checkpoint", "impact", "resume_condition", "quota_state", "model_state", "requested_help"]
    missing = [field for field in required if field not in payload or payload[field] in (None, "")]
    if missing:
        raise ValueError("DECISION_PAYLOAD_FIELDS_REQUIRED:" + ",".join(missing))
    return {**payload, "brain_state": brain_snapshot(host)}


def enrich_heartbeat_payload(event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
    if event_type != "HEARTBEAT":
        return payload
    result = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], text=True, capture_output=True, timeout=10)
    if result.returncode or not re.fullmatch(r"[a-f0-9]{40}", result.stdout.strip()):
        raise ValueError("REPO012_HEAD_READBACK_FAILED")
    observed = result.stdout.strip()
    supplied = payload.get("repo012_head")
    if supplied is not None and supplied != observed:
        raise ValueError(f"REPO012_HEAD_PAYLOAD_MISMATCH:supplied={supplied}:observed={observed}")
    return {**payload, "observed_repo012_head": observed}


def emit(host: str, event_type: str, mentions: list[str], task_id: str | None, payload: dict[str, Any], evidence: list[str]) -> dict[str, Any]:
    record = host_record(host)
    endpoint = Path(record["endpoint"])
    admission(host, endpoint)
    reject_symlink_components(endpoint)
    endpoint.mkdir(parents=True, exist_ok=True)
    if endpoint.resolve() != endpoint:
        raise ValueError("ENDPOINT_REALPATH_MISMATCH")
    stamp = datetime.now().astimezone().strftime("%Y%m%dT%H%M%S%f%z")
    event_id = f"HB-MPC-{record['development_id']}-{stamp}-{uuid.uuid4().hex[:8]}"
    payload = enrich_heartbeat_payload(event_type, prepare_decision_payload(host, event_type, payload))
    event = {
        "schema": "guanghu.heartbeat-multipath-event/v1",
        "event_id": event_id,
        "console_id": "HB-MPC-0001",
        "event_type": event_type,
        "from_development_id": record["development_id"],
        "mentions": mentions,
        "task_id": task_id,
        "emitted_at": now(),
        "payload": payload,
        "evidence": evidence,
        "authority_granted": False,
    }
    body = json.dumps(event, ensure_ascii=False, indent=2) + "\n"
    if len(body.encode()) > 262_144:
        raise ValueError("EVENT_TOO_LARGE")
    path = endpoint / f"{event_id}.json"
    descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        handle.write(body)
    return {"outcome": "PASS", "event_id": event_id, "path": str(path), "development_id": record["development_id"], "authority_granted": False}


def update_presence(host: str) -> dict[str, Any]:
    record = host_record(host)
    presence = Path(record["endpoint"]).parent / "presence/CURRENT.json"
    admission(host, presence)
    reject_symlink_components(presence)
    presence.parent.mkdir(parents=True, exist_ok=True)
    value = {
        "schema": "guanghu.heartbeat-multipath-courier-presence/v1",
        "console_id": "HB-MPC-0001",
        "development_id": record["development_id"],
        "host": host,
        "courier_online": True,
        "model_cognition_state": "UNKNOWN_REQUIRES_SEPARATE_EVENT",
        "pid": os.getpid(),
        "updated_at": now(),
        "historical_record": False,
        "authority_granted": False,
    }
    body = json.dumps(value, ensure_ascii=False, indent=2) + "\n"
    temporary = presence.with_name(f".{presence.name}.{os.getpid()}.tmp")
    descriptor = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        handle.write(body)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(temporary, presence)
    return {"outcome":"PASS", "presence":str(presence), "development_id":record["development_id"], "model_cognition_state":"UNKNOWN_REQUIRES_SEPARATE_EVENT"}


def parse_payload(raw: str | None, message: str | None) -> dict[str, Any]:
    if raw:
        value = json.loads(raw)
        if not isinstance(value, dict):
            raise ValueError("PAYLOAD_OBJECT_REQUIRED")
        return value
    return {"message": message or ""}


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("command", choices=["join", "presence", "heartbeat", "send", "claim", "progress", "result", "attention", "tool-candidate", "pause", "blocked", "exit", "resume-request", "resumed", "mentions", "watch"])
    parser.add_argument("--host", required=True, choices=["codex", "zcode", "qwen", "doubao"])
    parser.add_argument("--mentions", nargs="*", default=[])
    parser.add_argument("--task-id")
    parser.add_argument("--message")
    parser.add_argument("--payload-json")
    parser.add_argument("--evidence", nargs="*", default=[])
    parser.add_argument("--interval", type=float, default=2.0)
    parser.add_argument("--heartbeat-seconds", type=float, default=15.0)
    args = parser.parse_args()
    type_map = {"heartbeat": "HEARTBEAT", "send": "MESSAGE", "claim": "TASK_CLAIM", "progress": "PROGRESS", "result": "RESULT", "attention": "ATTENTION", "tool-candidate": "TOOL_MIGRATION", "pause":"PAUSE", "blocked":"BLOCKED", "exit":"EXIT", "resume-request":"RESUME_REQUEST", "resumed":"RESUMED"}
    try:
        record = host_record(args.host)
        if args.command == "join":
            result = emit(args.host, "REGISTER", ["@ALL"], "HLP-MPC-TASK-001", {"host": args.host, "development_id": record["development_id"], "message": args.message or "JOINED_PENDING_SELF_ASSESSMENT"}, args.evidence)
        elif args.command == "presence":
            result = update_presence(args.host)
        elif args.command in type_map:
            result = emit(args.host, type_map[args.command], args.mentions, args.task_id, parse_payload(args.payload_json, args.message), args.evidence)
        elif args.command == "mentions":
            completed = subprocess.run(["python3", str(CONSOLE), "mentions", "--development-id", record["development_id"]], text=True, capture_output=True, timeout=30)
            print(completed.stdout, end="")
            return completed.returncode
        else:
            last_heartbeat = 0.0
            seen_messages: set[str] = set()
            while True:
                if time.monotonic() - last_heartbeat >= max(5.0, args.heartbeat_seconds):
                    update_presence(args.host)
                    last_heartbeat = time.monotonic()
                completed = subprocess.run(["python3", str(CONSOLE), "mentions", "--development-id", record["development_id"]], text=True, capture_output=True, timeout=30)
                if completed.returncode == 0:
                    inbox = json.loads(completed.stdout)
                    fresh = [item for item in inbox.get("messages", []) if item.get("event_id") not in seen_messages]
                    seen_messages.update(item.get("event_id") for item in inbox.get("messages", []) if item.get("event_id"))
                    if fresh:
                        print(json.dumps({"outcome":"PASS", "development_id":record["development_id"], "new_messages":fresh}, ensure_ascii=False, indent=2), flush=True)
                else:
                    print(json.dumps({"outcome":"FAIL", "error":"MENTION_SCAN_FAILED", "detail":completed.stderr[-500:]}, ensure_ascii=False), flush=True)
                time.sleep(max(0.5, args.interval))
        if args.command != "watch":
            print(json.dumps(result, ensure_ascii=False, indent=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())