#!/usr/bin/env python3
"""Independent read-only verifier for Task26 cleanup v5."""

from __future__ import annotations

import ast
import hashlib
import json
import os
import stat
import subprocess
from pathlib import Path
from typing import Any

ROOT = Path(__file__).parent
V4 = ROOT.parent / "task26-post-lifecycle-cleanup-v4-4a6c7ee5-st_01a00e68"
PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest")
ARCHIVE = PROFILE / "data/post-lifecycle-cleanup-archives/task26-current-4a6c7ee5-20260817"
SERVICE = "hermes-gateway-dualcoachtest.service"
V4_SEAL = "188e9e630b53e9329ae185fc8c007aecb7988c16d174d325d68e8e004edb38e9"
ARCHIVE_ROOT = "9f5d72840b4cc1fcb57720a61ebc53529fa00202d9201615e4ebdbb3e047a635"
OBSERVER_NAMES = ("gateway_start_observer.py", "lifecycle_observer.py", "observer_v2.py", "observer_v3.py", "observer_v4.py", "observer_v5.py", "observer_v6.py", "observer_v61.py", "observer_v62.py", "observer_v63.py", "observer_v64.py")


def canon(value: Any) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()


def sha(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def check(condition: bool, message: str) -> None:
    if not condition:
        raise RuntimeError(message)


def archive_inventory() -> tuple[str, int]:
    rows: list[dict[str, str]] = []
    for base, dirs, files in os.walk(ARCHIVE, followlinks=False):
        dirs.sort()
        files.sort()
        for name in dirs:
            path = Path(base) / name
            st = path.lstat()
            check(stat.S_ISDIR(st.st_mode), f"unsafe archive directory: {path}")
            rows.append({"path": path.relative_to(ARCHIVE).as_posix(), "type": "D", "mode": f"{stat.S_IMODE(st.st_mode):04o}"})
        for name in files:
            path = Path(base) / name
            st = path.lstat()
            check(stat.S_ISREG(st.st_mode), f"unsafe archive entry: {path}")
            rows.append({"path": path.relative_to(ARCHIVE).as_posix(), "type": "F", "mode": f"{stat.S_IMODE(st.st_mode):04o}", "sha256": sha(path.read_bytes())})
    return sha(canon(rows)), len(rows)


def parse_stat(raw: bytes, pid: int) -> tuple[int, int, int]:
    try:
        prefix, suffix = raw.decode("ascii").strip().rsplit(") ", 1)
        claimed, _comm = prefix.split(" (", 1)
        fields = suffix.split()
        check(int(claimed) == pid and len(fields) >= 20, "stat shape")
        return int(fields[1]), int(fields[3]), int(fields[19])
    except (UnicodeDecodeError, ValueError) as exc:
        raise RuntimeError(f"malformed /proc/{pid}/stat") from exc


def reasons(argv: list[str]) -> list[str]:
    if not argv:
        return []
    joined = "\0".join(argv)
    lower = joined.lower()
    exe = Path(argv[0]).name.lower()
    context = SERVICE in joined or str(PROFILE) in joined or "dualcoachtest" in lower or "task26" in lower
    found: list[str] = []
    if SERVICE in joined:
        found.append("service")
    if str(PROFILE) in joined:
        found.append("profile")
    if any(name in joined for name in OBSERVER_NAMES) or ("live-v" in lower and "-events.jsonl" in lower):
        found.append("observer")
    follow = any(arg == "--follow" or (arg.startswith("-") and not arg.startswith("--") and "f" in arg[1:]) for arg in argv[1:])
    if exe == "journalctl" and follow and context:
        found.append("journal_follower")
    if exe == "inotifywait" and context:
        found.append("inotify_watcher")
    if "task26" in lower and any(word in lower for word in ("monitor", "watch", "follow", "observer")):
        found.append("task_monitor")
    return found


def process_snapshot() -> dict[str, Any]:
    rows: list[dict[str, Any]] = []
    races: list[dict[str, Any]] = []
    for name in sorted((item for item in os.listdir("/proc") if item.isdecimal()), key=int):
        pid = int(name)
        root = Path("/proc") / name
        try:
            if root.stat().st_uid != os.getuid():
                continue
            before = parse_stat((root / "stat").read_bytes(), pid)
            raw = (root / "cmdline").read_bytes()
            after = parse_stat((root / "stat").read_bytes(), pid)
        except FileNotFoundError:
            races.append({"pid": pid, "race": "disappeared"})
            continue
        except PermissionError:
            continue
        check(before[2] == after[2], f"PID reuse: {pid}")
        check(not raw or raw.endswith(b"\0"), f"malformed cmdline: {pid}")
        argv = [] if not raw else [part.decode("utf-8", "surrogateescape") for part in raw[:-1].split(b"\0")]
        rows.append({"pid": pid, "ppid": after[0], "session_id": after[1], "start_time_ticks": after[2], "cmdline": argv})
    by_pid = {row["pid"]: row for row in rows}
    ancestors: set[int] = set()
    cursor = os.getpid()
    while cursor in by_pid and cursor not in ancestors:
        ancestors.add(cursor)
        cursor = by_pid[cursor]["ppid"]
    violations: list[dict[str, Any]] = []
    excluded: list[dict[str, int]] = []
    for row in rows:
        is_verifier = row["pid"] == os.getpid() or (row["pid"] in ancestors and any(Path(__file__).name in arg for arg in row["cmdline"]))
        if is_verifier:
            excluded.append({"pid": row["pid"], "start_time_ticks": row["start_time_ticks"]})
            continue
        found = reasons(row["cmdline"])
        if found:
            violations.append({**row, "reasons": found})
    return {"same_uid_count": len(rows), "races": races, "violations": violations, "excluded_verifier_identities": excluded}


def verify_write_surface() -> None:
    tree = ast.parse((ROOT / "cleanup_controller.py").read_text())
    owners: list[str] = []
    functions = [node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))]
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) and node.func.value.id == "os" and node.func.attr == "write":
            owners.append(next((function.name for function in functions if node in ast.walk(function)), "module"))
    check(owners == ["write_all"], f"unexpected writes: {owners}")


def main() -> int:
    check(sha((V4 / "SEAL.json").read_bytes()) == V4_SEAL, "v4 seal drift")
    contract_raw = (ROOT / "schema-contract-v5.json").read_bytes()
    contract = json.loads(contract_raw)
    permission = json.loads((ROOT / "permission-seal-v5.json").read_text())
    check(contract["schema"] == "task26-post-lifecycle-cleanup-contract-v5", "contract schema")
    check(contract["predecessor"]["v4_seal_sha256"] == V4_SEAL, "predecessor binding")
    check(permission["execute_allowed_once"] is False, "execute permission")
    check(permission["contract_sha256"] == sha(contract_raw), "permission contract")
    check(permission["controller_sha256"] == sha((ROOT / "cleanup_controller.py").read_bytes()), "permission controller")
    archive_root, archive_entries = archive_inventory()
    check(archive_root == ARCHIVE_ROOT, "archive root")
    for rel in ("customers", "data/customers", "data/onboarding", "data/owner-actions", "gateway.lock", "gateway.pid", "gateway_state.json", "state.db", "state.db-shm", "state.db-wal"):
        check(not (PROFILE / rel).exists(), f"live residue: {rel}")
    show = subprocess.run(["/usr/bin/systemctl", "--user", "show", SERVICE, "--property=ActiveState,SubState,MainPID,ControlPID"], capture_output=True, text=True, check=True)
    service = dict(line.split("=", 1) for line in show.stdout.splitlines() if "=" in line)
    check(service == {"MainPID": "0", "ControlPID": "0", "ActiveState": "inactive", "SubState": "dead"}, "service state")
    first = process_snapshot()
    second = process_snapshot()
    check(not first["violations"] and not second["violations"], f"session processes: {first['violations'] + second['violations']}")
    receipt = json.loads((ROOT / "controller-verification-receipt.json").read_text())
    proof = receipt["poststate"]["session_process_proof"]
    check(receipt["status"] == "PASS" and proof["passes"] == 2 and not proof["target_processes"], "controller receipt")
    verify_write_surface()
    result = {"schema": "task26-post-lifecycle-independent-verification-v5", "status": "PASS", "v4_seal_sha256": V4_SEAL, "archive_root_sha256": archive_root, "archive_entries": archive_entries, "service": service, "process_passes": 2, "process_violations": 0, "same_uid_processes_second_pass": second["same_uid_count"], "process_races": first["races"] + second["races"], "write_all_only": True, "cleanup_replayed": False, "live_mutations": 0}
    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
    return 0


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