#!/usr/bin/env python3
"""Sealed, forward-only observer for the Task24 card-154 Edit recovery.

This process never calls Telegram or a provider and never writes the profile. It
only observes durable owner-action files. The human alone presses Edit and sends
one exact Owner-DM message.
"""
from __future__ import annotations

import argparse
import ctypes
import datetime as dt
import hashlib
import json
import os
import select
import subprocess
import sys
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_RECORD = EVIDENCE / "dualcoach-task-24-regenerate-lineage-repair.json"
CHECKPOINT = EVIDENCE / "task24-live-window-checkpoint.json"
OBSERVATION = EVIDENCE / "task24-live-window-edit-recovery-observation.json"
SOURCE_ROOT = Path("/home/cube/projects/richard/hermes-agent")
PARENT_TOKEN = "7fa2b209eb5de478827d649ee81c3f0f"
PARENT_MESSAGE_ID = "154"
EDIT_TEXT = "현재 계획을 유지하며 다음 기록을 확인하겠습니다."
APPROVAL = "TASK24_OWNER_EDIT_RECOVERY_APPROVED"


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


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


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:
                row = json.loads(line)
            except ValueError:
                continue
            kind = row.get("event_type") if isinstance(row, dict) else None
            if isinstance(kind, str):
                counts[kind] = counts.get(kind, 0) + 1
    return counts


def state() -> dict[str, Any]:
    drafts = load_json(OWNER_ACTIONS / "drafts.json", {})
    generations = load_json(OWNER_ACTIONS / "draft-generations.json", {})
    cards = load_json(OWNER_ACTIONS / "draft-generation-cards.json", {})
    deliveries = load_json(OWNER_ACTIONS / "draft-deliveries.json", {})
    gateway = load_json(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 atomic_json(path: Path, payload: dict[str, Any]) -> None:
    fd, temporary = tempfile.mkstemp(prefix=".task24-edit-recovery.", dir=str(path.parent))
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(payload, 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 checkpoint(stage: str, current_message_id: str, current_token: str, current: dict[str, Any]) -> None:
    atomic_json(
        CHECKPOINT,
        {
            "state": stage,
            "current_message_id": current_message_id,
            "current_token": current_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], *, message_id: str = PARENT_MESSAGE_ID, token: str = PARENT_TOKEN) -> None:
    checkpoint("BLOCKED_EDIT_RECOVERY", message_id, token, current)
    atomic_json(
        OBSERVATION,
        {
            "schema": "task24-edit-recovery-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 sealed_baseline() -> tuple[dict[str, Any], str]:
    repair = load_json(REPAIR_RECORD, {})
    if repair.get("status") != "READY_FOR_TASK24_REGENERATE_RECOVERY":
        raise SystemExit("BLOCK: sealed lineage-repair record is unavailable")
    source_hashes = repair.get("source_sha256")
    if not isinstance(source_hashes, dict):
        raise SystemExit("BLOCK: sealed lineage-repair source hashes are unavailable")
    for relative, expected in source_hashes.items():
        if not isinstance(relative, str) or not isinstance(expected, str):
            raise SystemExit("BLOCK: sealed lineage-repair source hash is malformed")
        actual = hashlib.sha256((SOURCE_ROOT / relative).read_bytes()).hexdigest()
        if actual != expected:
            raise SystemExit("BLOCK: lineage-repair source seal drifted")
    current = state()
    drafts = current["drafts"]
    generations = current["generations"]
    cards = current["cards"]
    parent = drafts.get(PARENT_TOKEN)
    history = generations.get(PARENT_TOKEN)
    card = cards.get(PARENT_TOKEN)
    if not isinstance(parent, dict) or not isinstance(history, list) or not history or not isinstance(card, dict):
        fail("card-154 durable state is unavailable", current)
    if (
        parent.get("status") != "created"
        or history[-1].get("state") != "draft_created"
        or str(card.get("message_id")) != PARENT_MESSAGE_ID
        or card.get("state") != "published"
    ):
        fail("card-154 is no longer the sealed current edit-recovery card", current)
    if current["delivery_count"] != 0:
        fail("delivery is nonzero before Edit recovery", current)
    if current["provider_count"] != 2:
        fail("provider receipt count drifted before Edit recovery", current)
    if current["gateway_state"] != "running" or current["telegram_state"] != "connected":
        fail("gateway is not running and Telegram-connected", current)
    active = subprocess.run(
        ["systemctl", "--user", "is-active", "--quiet", "hermes-gateway-dualcoachtest.service"],
        check=False,
    )
    if active.returncode != 0:
        fail("gateway systemd service is not active", current)
    baseline = {
        "draft_tokens": set(drafts),
        "card_tokens": set(cards),
        "parent_generation_tip": history[-1].get("record_digest"),
        "parent_card": card,
        "provider_count": current["provider_count"],
        "delivery_count": current["delivery_count"],
        "events": current["events"],
    }
    bound = {
        "repair_record_sha256": hashlib.sha256(REPAIR_RECORD.read_bytes()).hexdigest(),
        "parent_token": PARENT_TOKEN,
        "parent_message_id": PARENT_MESSAGE_ID,
        "parent_generation_tip": baseline["parent_generation_tip"],
        "draft_tokens": sorted(baseline["draft_tokens"]),
        "card_tokens": sorted(baseline["card_tokens"]),
        "provider_count": baseline["provider_count"],
        "delivery_count": baseline["delivery_count"],
    }
    seal = hashlib.sha256(json.dumps(bound, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return baseline, seal


def assess_success(baseline: dict[str, Any], current: dict[str, Any]) -> tuple[str, str] | None:
    if current["delivery_count"] != baseline["delivery_count"]:
        fail("Edit recovery created a delivery", current)
    if current["provider_count"] != baseline["provider_count"]:
        fail("Edit recovery invoked or changed a provider generation receipt", current)
    new_drafts = sorted(set(current["drafts"]) - baseline["draft_tokens"])
    new_cards = sorted(set(current["cards"]) - baseline["card_tokens"])
    if len(new_drafts) > 1 or len(new_cards) > 1:
        fail("Edit recovery created duplicate immutable drafts or cards", current)
    if not new_drafts:
        return None
    child = new_drafts[0]
    draft = current["drafts"].get(child)
    history = current["generations"].get(child)
    card = current["cards"].get(child)
    parent = current["drafts"].get(PARENT_TOKEN)
    parent_card = current["cards"].get(PARENT_TOKEN)
    if not isinstance(draft, dict) or not isinstance(history, list) or not history or not isinstance(card, dict):
        return None
    if not isinstance(parent, dict) or parent_card != baseline["parent_card"]:
        fail("card 154 was not preserved as historical evidence", current)
    if draft.get("parent_draft_id") != PARENT_TOKEN:
        fail("edited child does not bind card-154 token", current)
    if hashlib.sha256(str(draft.get("text", "")).encode()).hexdigest() != hashlib.sha256(EDIT_TEXT.encode()).hexdigest():
        fail("Owner Edit text did not match the sealed exact text", current)
    first = history[0]
    if (
        first.get("lineage_parent_token") != PARENT_TOKEN
        or first.get("lineage_predecessor_digest") != baseline["parent_generation_tip"]
    ):
        fail("edited child generation does not bind card-154 parent tip", current)
    if (
        parent.get("status") != "superseded"
        or parent.get("superseded_by_draft_id") != child
        or history[-1].get("state") != "draft_created"
        or card.get("state") != "published"
        or not card.get("message_id")
        or len(new_cards) != 1
    ):
        return None
    if (
        current["events"].get("draft_created", 0) != baseline["events"].get("draft_created", 0) + 1
        or current["events"].get("draft_edited", 0) != baseline["events"].get("draft_edited", 0) + 1
    ):
        fail("Edit recovery durable ingress is not exactly one authenticated immutable revision", current)
    return str(card["message_id"]), child


def run() -> int:
    baseline, expected_seal = sealed_baseline()
    if os.environ.get("TASK24_EDIT_RECOVERY_APPROVAL") != APPROVAL:
        raise SystemExit("BLOCK: missing exact Owner Edit recovery approval")
    if os.environ.get("TASK24_EDIT_RECOVERY_SEAL") != expected_seal:
        raise SystemExit("BLOCK: supplied Owner Edit recovery 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,  # CLOSE_WRITE | MOVED_TO | ONLYDIR
        )
        if watch < 0:
            raise OSError(ctypes.get_errno(), "inotify_add_watch")
        current = state()
        if current["delivery_count"] != 0 or current["provider_count"] != baseline["provider_count"]:
            fail("state drifted while arming Edit recovery observer", current)
        checkpoint("WAITING_OWNER_EDIT_154", PARENT_MESSAGE_ID, PARENT_TOKEN, current)
        print("WAITING_OWNER_EDIT_154", flush=True)
        deadline = time.monotonic() + 1800
        while True:
            ready, _, _ = select.select([fd], [], [], deadline - time.monotonic())
            if not ready:
                fail("timed out waiting for the sealed Owner Edit recovery", state())
            os.read(fd, 65536)
            current = state()
            result = assess_success(baseline, current)
            if result is None:
                continue
            message_id, child = result
            checkpoint(f"WAITING_OWNER_APPROVE_{message_id}", message_id, child, current)
            atomic_json(
                OBSERVATION,
                {
                    "schema": "task24-edit-recovery-observation-v1",
                    "status": "PASS",
                    "parent_message_id": PARENT_MESSAGE_ID,
                    "parent_token": PARENT_TOKEN,
                    "child_message_id": message_id,
                    "child_token": child,
                    "provider_count_delta": 0,
                    "delivery_count": current["delivery_count"],
                    "new_draft_count": 1,
                    "new_card_count": 1,
                    "timestamp": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
                },
            )
            print(f"WAITING_OWNER_APPROVE {message_id} {child}", 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 = sealed_baseline()
    if args.seal:
        print(seal)
        return 0
    return run()


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