#!/usr/bin/env python3
"""Deployment wrapper for the hash-pinned minimal AI Council relay and staging core.

This process has no package, service, shell, arbitrary-path, or Telegram-token
capability. It binds only the policy-pinned IPv4 loopback listener and owns the
controller-side Unix socket.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import stat
import threading
import time
from typing import Any

import council_relay as relay
import council_staging
from council_core import CouncilCore, CouncilError

CONFIG_SCHEMA = "AI_COUNCIL_R05_6_DEPLOYMENT_CONFIG_V1"
HEX64 = re.compile(r"[0-9a-f]{64}\Z")
EXPECTED_PROFILE = "PROVISIONAL_PILOT_SYNTHETIC_PUBLIC_ONLY_TEMPORARY_INGRESS"
EXPECTED_AGENT_CLIENT_SHA256 = "ae60a19b75468c1ebb128bcc54e8764a7f02c20d41142ed82dff53f64a2beb18"
EXPECTED_SOURCE_PATHS = frozenset({
    "arc_agent_client.py", "council_controller.py", "council_core.py", "council_relay.py",
    "council_staging.py", "councilctl.py", "telegram_gateway.py",
    "policies/r03_a2_policy_v1.json", "policies/r03_a3_policy_v1.json",
    "policies/r03_core_policy_v1.json", "room_manifests/approval.json",
    "room_manifests/bench.json", "room_manifests/commons.json",
    "room_manifests/engineering.json", "room_manifests/entry.json",
    "room_manifests/incident.json", "room_manifests/job.json",
    "room_manifests/review.json",
})
PROMPT_EXPORT_TTL_SECONDS = 2147483647
PROMPT_EXPORT_NAMES = (
    "PROMPT-A1.txt", "PROMPT-A2.txt", "PROMPT-A3.txt", "PROMPT-A4.txt",
    "PROMPT-A5.txt", "PROMPT-A6.txt", "PROMPT-A7.txt", "PROMPT-A8.txt",
    "PROMPT-A9.txt", "PROMPT-A10.txt", "PROMPT-A11.txt", "PROMPT-A12.txt",
    "prompt-export-manifest.json",
)
PROMPT_EXPORT_TEMP = re.compile(r"\.(?:PROMPT-A(?:1[0-2]|[1-9])\.txt|prompt-export-manifest\.json)\.tmp-[0-9]{1,12}-[0-9a-f]{12}\Z")


class ControllerError(RuntimeError):
    pass


def require(value: bool, message: str) -> None:
    if not value:
        raise ControllerError(message)


def regular_bytes(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as error:
        raise ControllerError(f"required file unavailable: {path.name}") from error
    try:
        before = os.fstat(descriptor)
        require(stat.S_ISREG(before.st_mode) and before.st_nlink == 1, f"unsafe file: {path.name}")
        chunks: list[bytes] = []
        while True:
            chunk = os.read(descriptor, 131072)
            if not chunk:
                break
            chunks.append(chunk)
        after = os.fstat(descriptor)
        require((before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) ==
                (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns),
                f"file changed while reading: {path.name}")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def strict_json(path: Path) -> dict[str, Any]:
    def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]:
        result: dict[str, Any] = {}
        for key, value in items:
            if key in result:
                raise ValueError("duplicate key")
            result[key] = value
        return result
    try:
        value = json.loads(regular_bytes(path).decode("utf-8"), object_pairs_hook=pairs,
                           parse_constant=lambda _x: (_ for _ in ()).throw(ValueError("constant")))
    except (UnicodeError, json.JSONDecodeError, ValueError) as error:
        raise ControllerError(f"invalid JSON: {path.name}") from error
    require(isinstance(value, dict), f"JSON object required: {path.name}")
    return value


def validate_deployment_config(path: Path, program_root: Path) -> dict[str, Any]:
    """Validate the exact R05.6 temporary Quick Tunnel runtime contract.

    This deliberately rejects the older R04 Tailscale/Funnel configuration.
    Transport fields are policy assertions only; the controller itself retains
    an IPv4-loopback listener and no Cloudflare or public-bind capability.
    """
    value = strict_json(path)
    required = {
        "schema", "version", "runtime_profile", "listen_address", "listen_port",
        "metrics_address", "metrics_port", "allowed_data_class", "max_agent_capacity",
        "initial_active_agents", "workspace_ttl_seconds", "message_body_persisted",
        "private_payload_enabled", "target_install_authorized", "tailscale_used",
        "public_plain_http", "temporary_ingress", "external_transport", "origin_transport",
        "quick_tunnel_origin", "quick_tunnel_hostname_suffix",
        "quick_tunnel_runtime_max_seconds", "enrollment_code_ttl_seconds",
        "agent_enrollment_code_in_single_owner_prompt", "arc_agent_client_sha256",
        "source_sha256",
    }
    require(set(value) == required, "deployment config fields")
    require(value["schema"] == CONFIG_SCHEMA and value["version"] == 1,
            "deployment config identity")
    require(value["runtime_profile"] == EXPECTED_PROFILE, "runtime profile")
    require(value["listen_address"] == "127.0.0.1" and value["listen_port"] == 17443,
            "listener must be exact IPv4 loopback")
    require(value["metrics_address"] == "127.0.0.1" and value["metrics_port"] == 17444,
            "metrics must be exact IPv4 loopback")
    require(value["allowed_data_class"] == "SYNTHETIC_PUBLIC_ONLY", "data class")
    require(value["max_agent_capacity"] == 12 and value["initial_active_agents"] == 3,
            "agent capacity")
    require(value["workspace_ttl_seconds"] == PROMPT_EXPORT_TTL_SECONDS,
            "workspace TTL")
    require(value["message_body_persisted"] is False and
            value["private_payload_enabled"] is False,
            "payload boundary")
    require(value["target_install_authorized"] is False,
            "runtime must not gain installer authority")
    require(value["tailscale_used"] is False and value["public_plain_http"] is False,
            "forbidden transport state")
    require(value["temporary_ingress"] == "CLOUDFLARE_QUICK_TUNNEL" and
            value["external_transport"] == "HTTPS" and
            value["origin_transport"] == "LOOPBACK_HTTP",
            "temporary ingress identity")
    require(value["quick_tunnel_origin"] == "http://127.0.0.1:17443" and
            value["quick_tunnel_hostname_suffix"] == ".trycloudflare.com" and
            value["quick_tunnel_runtime_max_seconds"] == 2147483647,
            "Quick Tunnel binding")
    require(value["enrollment_code_ttl_seconds"] == 2147483647 and
            value["agent_enrollment_code_in_single_owner_prompt"] is True and
            value["arc_agent_client_sha256"] == EXPECTED_AGENT_CLIENT_SHA256,
            "agent bootstrap boundary")
    pins = value["source_sha256"]
    require(isinstance(pins, dict) and set(pins) == EXPECTED_SOURCE_PATHS,
            "source pin inventory")
    for relative, expected in pins.items():
        require(isinstance(relative, str) and re.fullmatch(r"[A-Za-z0-9_./-]{1,160}", relative) and
                not relative.startswith("/") and ".." not in relative.split("/"), "source pin path")
        require(isinstance(expected, str) and HEX64.fullmatch(expected) is not None, "source pin hash")
        source = program_root / relative
        require(source.resolve(strict=True).is_relative_to(program_root.resolve(strict=True)),
                "source containment")
        require(hashlib.sha256(regular_bytes(source)).hexdigest() == expected,
                f"source hash mismatch: {relative}")
    return value


def sweep_expired_prompt_exports(export_root: Path, now: int) -> int:
    """Delete only the fixed owner-export files (A1..A12 + manifest) after the one-hour cap."""
    try:
        root_info = export_root.lstat()
    except FileNotFoundError:
        return 0
    require(stat.S_ISDIR(root_info.st_mode) and not stat.S_ISLNK(root_info.st_mode), "unsafe export root")
    require(root_info.st_uid == os.geteuid() and stat.S_IMODE(root_info.st_mode) == 0o700,
            "unsafe export root metadata")
    removed = 0
    names = set(PROMPT_EXPORT_NAMES)
    with os.scandir(export_root) as entries:
        for entry in entries:
            if PROMPT_EXPORT_TEMP.fullmatch(entry.name):
                names.add(entry.name)
    for name in sorted(names):
        path = export_root / name
        try:
            information = path.lstat()
        except FileNotFoundError:
            continue
        require(stat.S_ISREG(information.st_mode) and information.st_nlink == 1 and
                information.st_uid == os.geteuid(), "unsafe prompt export")
        age = max(0, now - int(information.st_mtime))
        if age >= PROMPT_EXPORT_TTL_SECONDS:
            path.unlink()
            removed += 1
    return removed


def serve(args: argparse.Namespace) -> int:
    config_path = args.deployment_config.resolve(strict=True)
    program_root = Path(__file__).resolve(strict=True).parent
    deployment = validate_deployment_config(config_path, program_root)
    relay_policy = relay.read_json(args.relay_policy)
    relay.validate_policy(relay_policy)
    core_policy = relay.read_json(args.core_policy)
    require(relay.file_sha256(args.core_policy) == relay_policy["a1_core_policy_sha256"],
            "core policy pin")
    staging_policy = relay.read_json(args.staging_policy)
    council_staging.validate_policy(staging_policy)
    require(relay_policy["listen_address"] == deployment["listen_address"] and
            relay_policy["listen_port"] == deployment["listen_port"], "deployment/policy listener mismatch")
    master_key = relay.read_master_key(args.master_key_file)
    require(args.state_root.is_absolute() and args.manifest_dir.is_absolute(), "absolute runtime paths required")
    core = CouncilCore(args.state_root, args.manifest_dir, core_policy, master_key,
                       lambda: int(time.time()))
    staging = council_staging.StagingController(core, staging_policy)
    app = relay.RelayApplication(core, relay_policy, staging_controller=staging,
                                 apply_approved_staging=True)
    ingress = relay.GatewayIngressServer(app)
    server = relay.LoopbackRelayServer(app, relay_policy["listen_port"])
    stop_event = threading.Event()

    def sweep_once() -> None:
        with app.lock:
            core.sweep_expired_workspaces()
        removed = sweep_expired_prompt_exports(args.state_root / "exports", int(time.time()))
        if removed:
            relay.emit("expired_prompt_exports_removed", count=removed)

    def sweeper() -> None:
        while not stop_event.wait(60):
            try:
                sweep_once()
            except (ControllerError, CouncilError, OSError):
                relay.emit("retention_sweep_failed", error_type="EXPECTED_RUNTIME_ERROR")

    sweep_thread = threading.Thread(target=sweeper, name="council-workspace-sweeper", daemon=True)

    def stop(_signum: int, _frame: object) -> None:
        if not stop_event.is_set():
            stop_event.set()
            threading.Thread(target=server.shutdown, daemon=True).start()

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    try:
        sweep_once()
        ingress.start()
        sweep_thread.start()
        relay.emit("controller_started", schema=CONFIG_SCHEMA, address="127.0.0.1", port=17443,
                   profile=EXPECTED_PROFILE)
        server.serve_forever(poll_interval=0.2)
        return 0
    finally:
        stop_event.set()
        server.server_close()
        ingress.close()
        sweep_thread.join(timeout=2.0)
        core.close()
        relay.emit("controller_stopped", schema=CONFIG_SCHEMA)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Hash-pinned minimal AI Council controller")
    subparsers = parser.add_subparsers(dest="command", required=True)
    run = subparsers.add_parser("serve")
    run.add_argument("--deployment-config", type=Path, required=True)
    run.add_argument("--relay-policy", type=Path, required=True)
    run.add_argument("--core-policy", type=Path, required=True)
    run.add_argument("--staging-policy", type=Path, required=True)
    run.add_argument("--manifest-dir", type=Path, required=True)
    run.add_argument("--state-root", type=Path, required=True)
    run.add_argument("--master-key-file", type=Path, required=True)
    run.set_defaults(handler=serve)
    return parser


def main() -> int:
    os.umask(0o077)
    args = build_parser().parse_args()
    try:
        return int(args.handler(args))
    except (ControllerError, relay.RelayError, CouncilError, OSError, ValueError) as error:
        relay.emit("controller_fatal", error_type=type(error).__name__, detail=str(error)[:300])
        return 1
    except KeyboardInterrupt:
        return 130


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