#!/usr/bin/env python3
"""Sealed Task24 approval observer with exactly one post-approval restart."""
from __future__ import annotations

import argparse
import ctypes
import datetime as dt
import hashlib
import json
import os
import select
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any

PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest")
OWNER_ACTIONS = PROFILE / "data" / "owner-actions"
EVIDENCE = Path("/home/cube/projects/richard/traning coach/.omo/evidence")
REPAIR = EVIDENCE / "dualcoach-task-24-regenerate-lineage-repair.json"
EDIT = EVIDENCE / "task24-live-window-edit-recovery-observation.json"
CHECKPOINT = EVIDENCE / "task24-live-window-checkpoint.json"
OBSERVATION = EVIDENCE / "task24-live-window-approval-pre-send-observation.json"
SOURCE_ROOT = Path("/home/cube/projects/richard/hermes-agent")
PARENT_TOKEN = "7fa2b209eb5de478827d649ee81c3f0f"
TOKEN = "c8eab7b6685c3c65"
MESSAGE_ID = "156"
APPROVAL = "TASK24_OWNER_APPROVAL_APPROVED"
SERVICE = "hermes-gateway-dualcoachtest.service"


def load(path: Path, default: Any) -> Any:
    return json.loads(path.read_text(encoding="utf-8")) if path.exists() else default


def atomic(path: Path, value: dict[str, Any]) -> None:
    fd, temporary = tempfile.mkstemp(prefix=".task24-approval.", dir=str(path.parent))
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(value, handle, sort_keys=True, separators=(",", ":"))
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        directory = os.open(path.parent, os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
    except BaseException:
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


def provider_count(generations: dict[str, Any]) -> int:
    return len(
        {
            row["generation_provider_receipt"]
            for history in generations.values()
            if isinstance(history, list)
            for row in history
            if isinstance(row, dict)
            and isinstance(row.get("generation_provider_receipt"), str)
        }
    )


def event_counts() -> dict[str, int]:
    counts: dict[str, int] = {}
    for path in sorted((PROFILE / "data" / "customers").rglob("events.jsonl")):
        for line in path.read_text(encoding="utf-8").splitlines():
            try:
                payload = json.loads(line)
            except ValueError:
                continue
            kind = payload.get("event_type") if isinstance(payload, dict) else None
            if isinstance(kind, str):
                counts[kind] = counts.get(kind, 0) + 1
    return counts


def current_state() -> dict[str, Any]:
    drafts = load(OWNER_ACTIONS / "drafts.json", {})
    generations = load(OWNER_ACTIONS / "draft-generations.json", {})
    cards = load(OWNER_ACTIONS / "draft-generation-cards.json", {})
    deliveries = load(OWNER_ACTIONS / "draft-deliveries.json", {})
    gateway = load(PROFILE / "gateway_state.json", {})
    platforms = gateway.get("platforms") if isinstance(gateway, dict) else {}
    telegram = platforms.get("telegram") if isinstance(platforms, dict) else {}
    return {
        "drafts": drafts,
        "generations": generations,
        "cards": cards,
        "delivery_count": len(deliveries),
        "provider_count": provider_count(generations),
        "events": event_counts(),
        "gateway_state": gateway.get("gateway_state") if isinstance(gateway, dict) else None,
        "telegram_state": telegram.get("state") if isinstance(telegram, dict) else None,
    }


def write_checkpoint(stage: str, state: dict[str, Any]) -> None:
    atomic(
        CHECKPOINT,
        {
            "state": stage,
            "current_message_id": MESSAGE_ID,
            "current_token": TOKEN,
            "delivery_count": state["delivery_count"],
            "provider_count": state["provider_count"],
            "timestamp": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        },
    )


def fail(reason: str, state: dict[str, Any]) -> None:
    write_checkpoint("BLOCKED_APPROVAL_PRE_SEND", state)
    atomic(
        OBSERVATION,
        {
            "schema": "task24-approval-pre-send-observation-v1",
            "status": "BLOCKED",
            "reason": reason,
            "delivery_count": state["delivery_count"],
            "provider_count": state["provider_count"],
            "timestamp": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        },
    )
    raise SystemExit(f"BLOCK: {reason}")


def active_current_card(state: dict[str, Any]) -> list[str]:
    candidates: list[str] = []
    for token, draft in state["drafts"].items():
        history = state["generations"].get(token, [])
        card = state["cards"].get(token, {})
        if (
            isinstance(draft, dict)
            and isinstance(history, list)
            and history
            and isinstance(card, dict)
            and draft.get("status") in {"created", "edited", "approved"}
            and history[-1].get("state") in {"draft_created", "approved"}
            and card.get("state") == "published"
        ):
            candidates.append(token)
    return sorted(candidates)


def seal_baseline() -> tuple[dict[str, Any], str]:
    repair = load(REPAIR, {})
    edit = load(EDIT, {})
    if repair.get("status") != "READY_FOR_TASK24_REGENERATE_RECOVERY" or edit.get("status") != "PASS":
        raise SystemExit("BLOCK: sealed lineage/edit recovery evidence is unavailable")
    if edit.get("parent_token") != PARENT_TOKEN or edit.get("child_token") != TOKEN or str(edit.get("child_message_id")) != MESSAGE_ID:
        raise SystemExit("BLOCK: sealed edit recovery evidence does not bind card 156")
    hashes = repair.get("source_sha256")
    if not isinstance(hashes, dict):
        raise SystemExit("BLOCK: lineage repair source seal is malformed")
    for relative, expected in hashes.items():
        if not isinstance(relative, str) or not isinstance(expected, str) or hashlib.sha256((SOURCE_ROOT / relative).read_bytes()).hexdigest() != expected:
            raise SystemExit("BLOCK: lineage repair source seal drifted")
    state = current_state()
    draft = state["drafts"].get(TOKEN)
    history = state["generations"].get(TOKEN)
    card = state["cards"].get(TOKEN)
    if not isinstance(draft, dict) or not isinstance(history, list) or not history or not isinstance(card, dict):
        fail("card 156 durable state is unavailable", state)
    if (
        draft.get("status") != "edited"
        or draft.get("parent_draft_id") != PARENT_TOKEN
        or history[-1].get("state") != "draft_created"
        or history[0].get("lineage_parent_token") != PARENT_TOKEN
        or card.get("state") != "published"
        or str(card.get("message_id")) != MESSAGE_ID
        or state["delivery_count"] != 0
        or state["provider_count"] != 2
        or state["gateway_state"] != "running"
        or state["telegram_state"] != "connected"
        or active_current_card(state) != [TOKEN]
    ):
        fail("card 156 is not the sealed sole current approval card", state)
    if subprocess.run(["systemctl", "--user", "is-active", "--quiet", SERVICE], check=False).returncode != 0:
        fail("gateway service is inactive", state)
    baseline = {
        "draft_tokens": set(state["drafts"]),
        "card_tokens": set(state["cards"]),
        "historical_cards": {token: state["cards"].get(token) for token in state["cards"] if token != TOKEN},
        "provider_count": state["provider_count"],
        "delivery_count": state["delivery_count"],
        "events": state["events"],
        "current_card": card,
    }
    bound = {
        "repair_sha256": hashlib.sha256(REPAIR.read_bytes()).hexdigest(),
        "edit_sha256": hashlib.sha256(EDIT.read_bytes()).hexdigest(),
        "token": TOKEN,
        "message_id": MESSAGE_ID,
        "draft_tokens": sorted(baseline["draft_tokens"]),
        "card_tokens": sorted(baseline["card_tokens"]),
        "provider_count": baseline["provider_count"],
        "delivery_count": baseline["delivery_count"],
    }
    return baseline, hashlib.sha256(json.dumps(bound, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def approved(baseline: dict[str, Any], state: dict[str, Any]) -> bool:
    if state["delivery_count"] != baseline["delivery_count"]:
        fail("approval created a delivery", state)
    if state["provider_count"] != baseline["provider_count"]:
        fail("approval changed provider receipt count", state)
    if set(state["drafts"]) != baseline["draft_tokens"] or set(state["cards"]) != baseline["card_tokens"]:
        fail("approval created or removed immutable drafts/cards", state)
    for token, card in baseline["historical_cards"].items():
        if state["cards"].get(token) != card:
            fail("approval did not preserve historical cards", state)
    draft = state["drafts"].get(TOKEN)
    history = state["generations"].get(TOKEN)
    card = state["cards"].get(TOKEN)
    if not isinstance(draft, dict) or not isinstance(history, list) or not history or not isinstance(card, dict):
        return False
    if draft.get("status") != "approved" or history[-1].get("state") != "approved":
        return False
    if (
        card.get("state") != "published"
        or str(card.get("message_id")) != MESSAGE_ID
        or active_current_card(state) != [TOKEN]
    ):
        fail("approved card identity or current-card uniqueness drifted", state)
    if state["events"].get("draft_approved", 0) != baseline["events"].get("draft_approved", 0) + 1:
        fail("approval ingress is not exactly one Owner transition", state)
    return True


def restart_after_approval(baseline: dict[str, Any]) -> dict[str, Any]:
    libc = ctypes.CDLL(None, use_errno=True)
    fd = libc.inotify_init1(os.O_CLOEXEC)
    if fd < 0:
        raise OSError(ctypes.get_errno(), "inotify_init1")
    try:
        watch = libc.inotify_add_watch(
            fd, os.fsencode(str(PROFILE)), 0x00000008 | 0x00000080 | 0x01000000
        )
        if watch < 0:
            raise OSError(ctypes.get_errno(), "inotify_add_watch")
        subprocess.run(["systemctl", "--user", "restart", SERVICE], check=True)
        if subprocess.run(["systemctl", "--user", "is-active", "--quiet", SERVICE], check=False).returncode != 0:
            fail("gateway did not return active after controlled pre-send restart", current_state())
        deadline = time.monotonic() + 180
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                fail("gateway did not publish connected state after controlled restart", current_state())
            ready, _, _ = select.select([fd], [], [], remaining)
            if not ready:
                fail("gateway did not publish connected state after controlled restart", current_state())
            data = os.read(fd, 65536)
            offset = 0
            changed = False
            while offset + 16 <= len(data):
                _, _, _, length = __import__("struct").unpack_from("iIII", data, offset)
                name = data[offset + 16:offset + 16 + length].split(b"\0", 1)[0].decode("utf-8", "strict")
                offset += 16 + length
                changed = changed or name == "gateway_state.json"
            if not changed:
                continue
            state = current_state()
            if state["gateway_state"] != "running" or state["telegram_state"] != "connected":
                continue
            if not approved(baseline, state):
                fail("approval drifted during controlled pre-send restart", state)
            return state
    finally:
        os.close(fd)


def run() -> int:
    baseline, expected = seal_baseline()
    if os.environ.get("TASK24_APPROVAL_RECOVERY_APPROVAL") != APPROVAL:
        raise SystemExit("BLOCK: missing exact approval observer permission")
    if os.environ.get("TASK24_APPROVAL_RECOVERY_SEAL") != expected:
        raise SystemExit("BLOCK: supplied approval observer seal is stale")
    libc = ctypes.CDLL(None, use_errno=True)
    fd = libc.inotify_init1(os.O_CLOEXEC)
    if fd < 0:
        raise OSError(ctypes.get_errno(), "inotify_init1")
    try:
        watch = libc.inotify_add_watch(
            fd, os.fsencode(str(OWNER_ACTIONS)), 0x00000008 | 0x00000080 | 0x01000000
        )
        if watch < 0:
            raise OSError(ctypes.get_errno(), "inotify_add_watch")
        state = current_state()
        if state["delivery_count"] != 0 or state["provider_count"] != baseline["provider_count"]:
            fail("approval baseline drifted while arming observer", state)
        write_checkpoint("WAITING_OWNER_APPROVE_156", state)
        print("WAITING_OWNER_APPROVE_156", flush=True)
        deadline = time.monotonic() + 1800
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                fail("timed out waiting for Owner approval", current_state())
            ready, _, _ = select.select([fd], [], [], remaining)
            if not ready:
                fail("timed out waiting for Owner approval", current_state())
            os.read(fd, 65536)
            state = current_state()
            if not approved(baseline, state):
                continue
            write_checkpoint("RESTARTING_BEFORE_SEND", state)
            restarted = restart_after_approval(baseline)
            write_checkpoint("WAITING_OWNER_SEND_156", restarted)
            atomic(
                OBSERVATION,
                {
                    "schema": "task24-approval-pre-send-observation-v1",
                    "status": "PASS",
                    "message_id": MESSAGE_ID,
                    "token": TOKEN,
                    "controls": "customer_send|revoke",
                    "provider_count_delta": 0,
                    "delivery_count": restarted["delivery_count"],
                    "new_draft_count": 0,
                    "new_card_count": 0,
                    "restart_count": 1,
                    "timestamp": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
                },
            )
            print(f"WAITING_OWNER_SEND {MESSAGE_ID} {TOKEN}", flush=True)
            return 0
    finally:
        os.close(fd)


def main() -> int:
    parser = argparse.ArgumentParser()
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--seal", action="store_true")
    modes.add_argument("--run", action="store_true")
    args = parser.parse_args()
    _baseline, seal = seal_baseline()
    if args.seal:
        print(seal)
        return 0
    return run()


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