#!/usr/bin/env python3
"""Task24 synthetic explicit-send observer; the human alone invokes Send."""
from __future__ import annotations

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

PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest")
OWNER = PROFILE / "data" / "owner-actions"
EVIDENCE = Path("/home/cube/projects/richard/traning coach/.omo/evidence")
PRE_SEND = EVIDENCE / "task24-live-window-approval-pre-send-observation.json"
CHECKPOINT = EVIDENCE / "task24-live-window-checkpoint.json"
TERMINAL = EVIDENCE / "task24-live-window-send-terminal-observation.json"
TOKEN = "c8eab7b6685c3c65"
MESSAGE = "156"
SYNTHETIC = "task22_dm_rehearsal"
APPROVAL = "TASK24_OWNER_SYNTHETIC_SEND_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-send.", 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]:
    result: dict[str, int] = {}
    for path in sorted((PROFILE / "data" / "customers").rglob("events.jsonl")):
        for line in path.read_text(encoding="utf-8").splitlines():
            try:
                value = json.loads(line)
            except ValueError:
                continue
            kind = value.get("event_type") if isinstance(value, dict) else None
            if isinstance(kind, str):
                result[kind] = result.get(kind, 0) + 1
    return result


def state() -> dict[str, Any]:
    drafts = load(OWNER / "drafts.json", {})
    generations = load(OWNER / "draft-generations.json", {})
    cards = load(OWNER / "draft-generation-cards.json", {})
    deliveries = load(OWNER / "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,
        "deliveries": deliveries,
        "delivery_count": len(deliveries),
        "provider_count": provider_count(generations),
        "events": event_counts(),
        "gateway": gateway.get("gateway_state") if isinstance(gateway, dict) else None,
        "telegram": telegram.get("state") if isinstance(telegram, dict) else None,
        "drafts_sha": hashlib.sha256((OWNER / "drafts.json").read_bytes()).hexdigest(),
        "generations_sha": hashlib.sha256((OWNER / "draft-generations.json").read_bytes()).hexdigest(),
    }


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


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


def synthetic_scope() -> bool:
    import yaml
    rows = []
    for profile in sorted(path for path in Path("/home/cube/.hermes/profiles").iterdir() if path.is_dir() and not path.is_symlink()):
        registry = load(profile / "customers" / "registry.json", {})
        enabled = [row.get("customer_key") for row in registry.get("customers", []) if isinstance(row, dict) and row.get("enabled") is True]
        config = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8"))
        adaptive = config.get("platforms", {}).get("telegram", {}).get("extra", {}).get("adaptive_nutrition", {})
        rows.append({"profile": profile.name, "enabled": enabled, "delivery_enabled": adaptive.get("delivery_enabled")})
    return rows == [{"profile": "dualcoachtest", "enabled": [SYNTHETIC], "delivery_enabled": True}, {"profile": "physique-coach", "enabled": [], "delivery_enabled": False}]


def baseline() -> tuple[dict[str, Any], str]:
    pre = load(PRE_SEND, {})
    if pre.get("status") != "PASS" or pre.get("token") != TOKEN or str(pre.get("message_id")) != MESSAGE:
        raise SystemExit("BLOCK: sealed pre-send restart evidence is unavailable")
    current = state()
    draft = current["drafts"].get(TOKEN)
    history = current["generations"].get(TOKEN)
    card = current["cards"].get(TOKEN)
    if not isinstance(draft, dict) or not isinstance(history, list) or not history or not isinstance(card, dict):
        fail("approved card durable projection is unavailable", current)
    if (
        draft.get("status") != "approved"
        or history[-1].get("state") != "approved"
        or card.get("state") != "published"
        or str(card.get("message_id")) != MESSAGE
        or current["delivery_count"] != 0
        or current["provider_count"] != 2
        or current["gateway"] != "running"
        or current["telegram"] != "connected"
        or not synthetic_scope()
    ):
        fail("approved/send baseline drifted", current)
    if subprocess.run(["systemctl", "--user", "is-active", "--quiet", SERVICE], check=False).returncode != 0:
        fail("gateway is not active", current)
    baseline = {"draft_tokens": set(current["drafts"]), "card_tokens": set(current["cards"]), "provider_count": current["provider_count"], "events": current["events"], "drafts_sha": current["drafts_sha"], "generations_sha": current["generations_sha"]}
    seal = hashlib.sha256(json.dumps({"pre_send_sha256": hashlib.sha256(PRE_SEND.read_bytes()).hexdigest(), "token": TOKEN, "message": MESSAGE, "draft_tokens": sorted(baseline["draft_tokens"]), "card_tokens": sorted(baseline["card_tokens"]), "provider_count": baseline["provider_count"]}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return baseline, seal


def terminal_delivery(baseline: dict[str, Any], current: dict[str, Any]) -> bool:
    if current["provider_count"] != baseline["provider_count"]:
        fail("explicit Send changed generation provider count", current)
    if current["delivery_count"] > 1:
        fail("explicit Send created duplicate delivery receipts", current)
    if set(current["drafts"]) != baseline["draft_tokens"] or set(current["cards"]) != baseline["card_tokens"]:
        fail("explicit Send created or removed immutable drafts/cards", current)
    if current["delivery_count"] == 0:
        return False
    delivery = next(iter(current["deliveries"].values()))
    history = current["generations"].get(TOKEN)
    draft = current["drafts"].get(TOKEN)
    if not isinstance(delivery, dict) or not isinstance(history, list) or not history or not isinstance(draft, dict):
        return False
    if (
        delivery.get("draft_id") != TOKEN
        or delivery.get("customer_key") != SYNTHETIC
        or delivery.get("status") != "sent_audited"
        or not isinstance(delivery.get("message_id"), str)
        or not delivery.get("message_id")
        or draft.get("status") != "sent"
        or history[-1].get("state") != "sent_audited"
        or not isinstance(history[-1].get("delivery_provider_receipt"), str)
        or current["events"].get("draft_sent", 0) != baseline["events"].get("draft_sent", 0) + 1
    ):
        return False
    return True


def restart_after_send(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:
        if libc.inotify_add_watch(fd, os.fsencode(str(PROFILE)), 0x00000008 | 0x00000080 | 0x01000000) < 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 post-send restart", state())
        deadline = time.monotonic() + 180
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                fail("gateway did not publish connected state after post-send restart", state())
            ready, _, _ = select.select([fd], [], [], remaining)
            if not ready:
                fail("gateway did not publish connected state after post-send restart", state())
            data = os.read(fd, 65536)
            offset = 0
            changed = False
            while offset + 16 <= len(data):
                _, _, _, length = 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
            current = state()
            if current["gateway"] != "running" or current["telegram"] != "connected":
                continue
            if not terminal_delivery(baseline, current):
                fail("sent receipt drifted during post-send restart", current)
            if current["drafts_sha"] != baseline["sent_drafts_sha"] or current["generations_sha"] != baseline["sent_generations_sha"]:
                fail("restart changed durable draft/generation state", current)
            return current
    finally:
        os.close(fd)


def run() -> int:
    before, expected = baseline()
    if os.environ.get("TASK24_SEND_APPROVAL") != APPROVAL:
        raise SystemExit("BLOCK: missing exact synthetic Send permission")
    if os.environ.get("TASK24_SEND_SEAL") != expected:
        raise SystemExit("BLOCK: supplied synthetic Send 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:
        if libc.inotify_add_watch(fd, os.fsencode(str(OWNER)), 0x00000008 | 0x00000080 | 0x01000000) < 0:
            raise OSError(ctypes.get_errno(), "inotify_add_watch")
        current = state()
        if current["delivery_count"] != 0 or current["provider_count"] != before["provider_count"]:
            fail("send state drifted while observer armed", current)
        checkpoint("WAITING_OWNER_SEND_156", current)
        print("WAITING_OWNER_SEND_156", flush=True)
        deadline = time.monotonic() + 1800
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                fail("timed out waiting for Owner explicit Send", state())
            ready, _, _ = select.select([fd], [], [], remaining)
            if not ready:
                fail("timed out waiting for Owner explicit Send", state())
            os.read(fd, 65536)
            current = state()
            if not terminal_delivery(before, current):
                continue
            before["sent_drafts_sha"] = current["drafts_sha"]
            before["sent_generations_sha"] = current["generations_sha"]
            checkpoint("RESTARTING_AFTER_SEND", current)
            after = restart_after_send(before)
            checkpoint("COMPLETE_TASK24_SEND", after)
            delivery = next(iter(after["deliveries"].values()))
            atomic(TERMINAL, {"schema": "task24-send-terminal-observation-v1", "status": "PASS", "message_id": MESSAGE, "token": TOKEN, "synthetic_customer": SYNTHETIC, "delivery_count": 1, "delivery_status": "sent_audited", "provider_message_id_present": bool(delivery.get("message_id")), "delivery_receipt_present": True, "provider_count_delta": 0, "new_draft_count": 0, "new_card_count": 0, "post_send_restart_count": 1, "gateway": "running/connected", "negative_callback_proof": "sealed isolated Task24 matrix passed stale/repeat/wrong-role paths", "timestamp": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")})
            print("TASK24_SEND_TERMINAL_PASS", 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()
    _state, seal = baseline()
    if args.seal:
        print(seal)
        return 0
    return run()


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