#!/usr/bin/env python3
"""Candidate-bound archive-first cleanup for exactly one completed ACTIVE lifecycle."""

from __future__ import annotations
import argparse
import fnmatch
import hashlib
import json
import os
import shutil
import stat
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any

FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0)


class CleanupError(RuntimeError):
    pass


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


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


def stable(s: os.stat_result) -> tuple[int, ...]:
    return (
        s.st_dev,
        s.st_ino,
        s.st_uid,
        s.st_gid,
        s.st_mode,
        s.st_nlink,
        s.st_size,
        s.st_mtime_ns,
        s.st_ctime_ns,
    )


def safe_external(p: Path) -> bytes:
    fd = os.open(p, os.O_RDONLY | FLAGS)
    try:
        a = os.fstat(fd)
        if (
            not stat.S_ISREG(a.st_mode)
            or a.st_uid != os.getuid()
            or a.st_nlink != 1
            or stat.S_IMODE(a.st_mode) not in {0o400, 0o500, 0o600, 0o700}
        ):
            raise CleanupError(f"unsafe sealed input: {p}")
        out = b""
        while c := os.read(fd, 1048576):
            out += c
        if stable(a) != stable(os.fstat(fd)):
            raise CleanupError(f"concurrent sealed input: {p}")
        return out
    finally:
        os.close(fd)


def open_root(p: Path) -> int:
    fd = os.open(p, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
    s = os.fstat(fd)
    if s.st_uid != os.getuid() or stat.S_IMODE(s.st_mode) != 0o700:
        os.close(fd)
        raise CleanupError("unsafe profile root")
    return fd


def parent_fd(root: int, rel: str) -> tuple[int, str]:
    parts = rel.split("/")
    fd = os.dup(root)
    try:
        for part in parts[:-1]:
            n = os.open(part, os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=fd)
            s = os.fstat(n)
            if (
                not stat.S_ISDIR(s.st_mode)
                or s.st_uid != os.getuid()
                or stat.S_IMODE(s.st_mode) != 0o700
            ):
                os.close(n)
                raise CleanupError(f"unsafe directory: {rel}")
            os.close(fd)
            fd = n
        return fd, parts[-1]
    except Exception:
        os.close(fd)
        raise


def lst(root: int, rel: str) -> os.stat_result | None:
    p, n = parent_fd(root, rel)
    try:
        try:
            return os.stat(n, dir_fd=p, follow_symlinks=False)
        except FileNotFoundError:
            return None
    finally:
        os.close(p)


def read_file(root: int, rel: str) -> bytes:
    p, n = parent_fd(root, rel)
    try:
        fd = os.open(n, os.O_RDONLY | FLAGS, dir_fd=p)
    finally:
        os.close(p)
    try:
        a = os.fstat(fd)
        if (
            not stat.S_ISREG(a.st_mode)
            or a.st_uid != os.getuid()
            or a.st_nlink != 1
            or stat.S_IMODE(a.st_mode) != 0o600
        ):
            raise CleanupError(f"unsafe authority file: {rel}")
        out = b""
        while c := os.read(fd, 1048576):
            out += c
        if stable(a) != stable(os.fstat(fd)):
            raise CleanupError(f"concurrent authority file: {rel}")
        return out
    finally:
        os.close(fd)


def walk(root: int, rel: str) -> list[dict[str, Any]]:
    s = lst(root, rel)
    if s is None:
        return []
    if stat.S_ISREG(s.st_mode):
        raw = read_file(root, rel)
        return [{"path": rel, "size": len(raw), "sha256": sha(raw), "mode": "0600"}]
    if (
        not stat.S_ISDIR(s.st_mode)
        or s.st_uid != os.getuid()
        or stat.S_IMODE(s.st_mode) != 0o700
    ):
        raise CleanupError(f"unsafe authority directory: {rel}")
    p, n = parent_fd(root, rel)
    try:
        fd = os.open(n, os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=p)
    finally:
        os.close(p)
    try:
        rows = []
        for child in sorted(os.listdir(fd)):
            rows += walk(root, f"{rel}/{child}")
        if stable(s) != stable(os.fstat(fd)):
            raise CleanupError(f"concurrent authority directory: {rel}")
        return rows
    finally:
        os.close(fd)


def jload(root: int, rel: str) -> dict[str, Any]:
    try:
        v = json.loads(read_file(root, rel))
    except json.JSONDecodeError as e:
        raise CleanupError(f"invalid JSON: {rel}") from e
    if not isinstance(v, dict):
        raise CleanupError(f"unknown schema: {rel}")
    return v


def atomic(p: Path, raw: bytes) -> None:
    p.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd, tmp = tempfile.mkstemp(prefix=".cleanup-", dir=p.parent)
    try:
        os.fchmod(fd, 0o600)
        os.write(fd, raw)
        os.fsync(fd)
        os.close(fd)
        fd = -1
        os.replace(tmp, p)
        d = os.open(p.parent, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
        os.fsync(d)
        os.close(d)
    finally:
        if fd >= 0:
            os.close(fd)
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass


def tree_digest(p: Path) -> str:
    rows = []
    for root, dirs, files in os.walk(p, followlinks=False):
        dirs.sort()
        files.sort()
        for n in dirs + files:
            x = Path(root) / n
            r = x.relative_to(p).as_posix()
            s = x.lstat()
            if stat.S_ISLNK(s.st_mode):
                raise CleanupError(f"symlink in protected tree: {x}")
            if stat.S_ISDIR(s.st_mode):
                rows.append(f"D\0{r}\0{stat.S_IMODE(s.st_mode):o}")
            elif stat.S_ISREG(s.st_mode):
                rows.append(f"F\0{r}\0{sha(safe_external(x))}")
            else:
                raise CleanupError(f"unsupported protected entry: {x}")
    return sha("\n".join(rows).encode())


def scopes(root: int, c: dict[str, Any]) -> list[str]:
    values = list(c["approved_clear_scopes"])
    d = os.open("data", os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=root)
    try:
        children = sorted(os.listdir(d))
    finally:
        os.close(d)
    protected = set(c["protected_data_roots"])
    tops = {
        x.split("/", 1)[1].split("/", 1)[0] for x in values if x.startswith("data/")
    }
    for child in children:
        rel = f"data/{child}"
        if child in protected or child in tops:
            continue
        if any(fnmatch.fnmatchcase(rel, p) for p in c["dynamic_clear_patterns"]):
            values.append(rel)
        else:
            raise CleanupError(f"unknown data root: {rel}")
    present = [x for x in values if lst(root, x) is not None]
    outer = []
    for x in sorted(set(present), key=lambda y: (y.count("/"), y)):
        if not any(x == p or x.startswith(p + "/") for p in outer):
            outer.append(x)
    return sorted(outer)


def prior_archives(a: argparse.Namespace) -> list[dict[str, str]]:
    result = []
    for root in [
        a.profile / "data/rehearsal-reset-archives",
        a.profile / "data/profile-reset-archives",
        a.archive_root,
    ]:
        if not root.exists():
            continue
        if root.is_symlink() or not root.is_dir():
            raise CleanupError(f"unsafe archive root: {root}")
        for child in sorted(root.iterdir()):
            if child.name.startswith(".pending-") or (
                root == a.archive_root and child.name == a.run_id
            ):
                continue
            if child.is_symlink() or not child.is_dir():
                raise CleanupError(f"invalid historical archive: {child}")
            manifests = list(child.glob("manifest.json"))
            if not manifests:
                raise CleanupError(f"historical archive lacks manifest: {child}")
            try:
                v = json.loads(manifests[0].read_text())
            except Exception as e:
                raise CleanupError(f"invalid historical manifest: {child}") from e
            if not isinstance(v, dict):
                raise CleanupError(f"unknown historical archive schema: {child}")
            result.append(
                {
                    "path": child.relative_to(a.profile).as_posix(),
                    "sha256": tree_digest(child),
                }
            )
    return result


def processes(profile: Path) -> list[int]:
    needle = b"HERMES_HOME=" + os.fsencode(profile.absolute())
    out = []
    for p in Path("/proc").iterdir():
        if not p.name.isdecimal():
            continue
        try:
            v = (p / "environ").read_bytes().split(b"\0")
        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
            continue
        if needle in v:
            out.append(int(p.name))
    return sorted(out)


def lifecycle_gate(
    a: argparse.Namespace, root: int, c: dict[str, Any]
) -> dict[str, str]:
    ledger = jload(root, "data/onboarding/telegram-customer-bootstrap-v1/ledger.json")
    if ledger.get("schema") != "telegram-customer-bootstrap-v1" or not isinstance(
        ledger.get("sessions"), list
    ):
        raise CleanupError("unknown customer-bootstrap schema")
    active = [
        x
        for x in ledger["sessions"]
        if isinstance(x, dict) and x.get("state") == "ACTIVE"
    ]
    if len(active) != 1:
        raise CleanupError("requires exactly one authorized ACTIVE lifecycle")
    for x in ledger["sessions"]:
        if x is active[0]:
            continue
        if (
            not isinstance(x, dict)
            or x.get("state") not in c["terminal_bootstrap_states"]
            or x.get("role_claims")
            or x.get("recovery_attempts")
        ):
            raise CleanupError("all other bootstrap sessions must be terminal")
    x = active[0]
    claims = x.get("role_claims")
    draft = x.get("customer_draft")
    if (
        not isinstance(claims, list)
        or len(claims) != 1
        or not isinstance(claims[0], dict)
        or claims[0].get("role") != "customer"
        or not isinstance(draft, dict)
    ):
        raise CleanupError("ACTIVE claim schema mismatch")
    ids = {
        "actor_id": str(claims[0].get("user_id")),
        "session_id": str(x.get("session_id")),
        "customer_key": str(draft.get("customer_key")),
    }
    if (
        ids["actor_id"] != a.actor_id
        or str(claims[0].get("chat_id")) != a.actor_id
        or str(draft.get("customer_user_id")) != a.actor_id
    ):
        raise CleanupError("actor/session/customer identity mismatch")
    reg = jload(root, "customers/registry.json")
    customers = reg.get("customers")
    if (
        not isinstance(customers, list)
        or len(customers) != 1
        or customers[0].get("customer_key") != ids["customer_key"]
        or str(customers[0].get("user_id")) != a.actor_id
        or customers[0].get("enabled") is not False
        or customers[0].get("status") != "disabled"
    ):
        raise CleanupError("registry/customer is not uniquely disabled")
    jobs = jload(root, "cron/jobs.json").get("jobs")
    if jobs != []:
        raise CleanupError("future cron jobs remain")
    config = read_file(root, "config.yaml")
    if b"delivery_enabled: false" not in config or b"delivery_enabled: true" in config:
        raise CleanupError("delivery gate is not disabled")
    deliveries = jload(root, "data/owner-actions/draft-deliveries.json")
    if len(deliveries) != 1:
        raise CleanupError("requires exactly one delivery receipt")
    delivery = next(iter(deliveries.values()))
    if (
        not isinstance(delivery, dict)
        or delivery.get("status") != "sent_audited"
        or not delivery.get("provider_receipt")
        or not delivery.get("message_id")
    ):
        raise CleanupError("delivery/provider outcome is not terminal sent_audited")
    ids["draft_id"] = str(delivery.get("draft_id"))
    for key in ("customer_key", "session_id", "actor_id"):
        if str(delivery.get(key)) != ids[key]:
            raise CleanupError("delivery identity mismatch")
    states = jload(root, "data/owner-actions/customer-service-state.json").get("states")
    if not isinstance(states, dict) or list(states) != [ids["customer_key"]]:
        raise CleanupError("owner state is not unique")
    owner = states[ids["customer_key"]]
    if (
        not isinstance(owner, dict)
        or owner.get("state") not in c["owner_terminal_states"]
        or owner.get("owner_action_state") not in c["owner_terminal_states"]
        or str(owner.get("session_id")) != ids["session_id"]
        or str(owner.get("actor_id")) != ids["actor_id"]
    ):
        raise CleanupError("owner action is nonterminal or mismatched")
    gens = jload(root, "data/owner-actions/draft-generations.json")
    if (
        list(gens) != [ids["draft_id"]]
        or not isinstance(gens[ids["draft_id"]], list)
        or not gens[ids["draft_id"]]
    ):
        raise CleanupError("generation lineage mismatch")
    latest = gens[ids["draft_id"]][-1]
    if (
        not isinstance(latest, dict)
        or latest.get("state") != "completed"
        or latest.get("claim_id") is not None
        or not latest.get("generation_provider_receipt")
    ):
        raise CleanupError("pending/unknown generation provider outcome")
    drafts = jload(root, "data/owner-actions/drafts.json")
    if (
        list(drafts) != [ids["draft_id"]]
        or drafts[ids["draft_id"]].get("status") != "approved"
        or drafts[ids["draft_id"]].get("customer_key") != ids["customer_key"]
        or drafts[ids["draft_id"]].get("session_id") != ids["session_id"]
    ):
        raise CleanupError("draft is not reconciled")
    cards = jload(root, "data/owner-actions/draft-generation-cards.json")
    if len(cards) != 1:
        raise CleanupError("card reconciliation is not unique")
    card = next(iter(cards.values()))
    if (
        card.get("state") != "published"
        or card.get("claim_id") is not None
        or card.get("token") != ids["draft_id"]
        or card.get("session_id") != ids["session_id"]
    ):
        raise CleanupError("card is pending or mismatched")
    outbox = jload(root, "data/onboarding/telegram-publication-outbox-v1/ledger.json")
    if (
        outbox.get("schema") != "telegram-publication-outbox-v1"
        or not isinstance(outbox.get("records"), list)
        or any(
            r.get("state") != "published" or r.get("session_id") != ids["session_id"]
            for r in outbox["records"]
            if isinstance(r, dict)
        )
    ):
        raise CleanupError("outbox/publication is not reconciled")
    emergency = jload(
        root, "data/onboarding/telegram-publication-outbox-v1/emergency.json"
    )
    if emergency.get("records") != []:
        raise CleanupError("emergency publication remains")
    return ids


def runtime_gate(a: argparse.Namespace, root: int) -> None:
    if lst(root, "gateway.lock") is not None:
        raise CleanupError("gateway lock remains")
    state = jload(root, "gateway_state.json")
    if state.get("gateway_state") != "stopped" or state.get("pid") != 0:
        raise CleanupError("gateway state is not stopped pid0")
    if processes(a.profile):
        raise CleanupError("profile process remains")
    tool = safe_external(a.systemctl)
    if not (tool.startswith(b"\x7fELF") or tool.startswith(b"#!")):
        raise CleanupError("invalid systemctl executable")
    r = subprocess.run(
        [
            str(a.systemctl),
            "--user",
            "show",
            a.service,
            "--property=ActiveState,SubState,MainPID",
            "--value",
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    if r.returncode != 0 or r.stdout.splitlines() != ["inactive", "dead", "0"]:
        raise CleanupError("service is not inactive/dead MainPID0")


def preflight(
    a: argparse.Namespace, c: dict[str, Any]
) -> tuple[
    int,
    list[str],
    list[dict[str, Any]],
    dict[str, str],
    list[dict[str, str]],
    dict[str, str],
]:
    root = open_root(a.profile)
    try:
        ss = scopes(root, c)
        rows = []
        for s in ss:
            rows += walk(root, s)
        ids = lifecycle_gate(a, root, c)
        files = {r["path"] for r in rows}
        missing = sorted(set(c["required_evidence_files"]) - files)
        if missing:
            raise CleanupError(f"missing required lifecycle evidence: {missing}")
        runtime_gate(a, root)
        for r in rows:
            if fnmatch.fnmatchcase(
                r["path"], "data/.scheduled-delivery-attempt-*.lock"
            ) or r["path"].startswith("data/customer-schedule-claims/"):
                raise CleanupError("claim/lease/attempt authority remains")
        prior = prior_archives(a)
        other = {str(p.absolute()): tree_digest(p) for p in a.other_profile}
        return root, ss, sorted(rows, key=lambda x: x["path"]), ids, prior, other
    except Exception:
        os.close(root)
        raise


def validate_inputs(a: argparse.Namespace) -> dict[str, Any]:
    cr = safe_external(a.contract)
    pr = safe_external(a.permission)
    c = json.loads(cr)
    p = json.loads(pr)
    expected = {
        "schema": "task26-post-lifecycle-cleanup-permission-v2",
        "approval": a.approval,
        "candidate_digest": a.candidate,
        "wheel_sha256": a.wheel_sha256,
        "plan_sha256": a.plan_sha256,
        "actor_id": a.actor_id,
        "controller_sha256": sha(Path(__file__).read_bytes()),
        "contract_sha256": sha(cr),
        "execute_allowed": True,
        "lifecycle_binding": "derive_unique_active_and_seal_in_archive",
    }
    if p != expected:
        raise CleanupError("permission seal/pins mismatch")
    if (
        c.get("schema") != "task26-post-lifecycle-cleanup-contract-v2"
        or c.get("candidate_digest") != a.candidate
        or c.get("wheel_sha256") != a.wheel_sha256
    ):
        raise CleanupError("contract/candidate mismatch")
    return c


def move(root: int, rel: str, dest: Path) -> None:
    p, n = parent_fd(root, rel)
    q = dest / rel
    q.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    d = os.open(q.parent, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
    try:
        os.rename(n, q.name, src_dir_fd=p, dst_dir_fd=d)
    finally:
        os.close(p)
        os.close(d)


def rollback(root: int, moved: list[str], dest: Path) -> None:
    for rel in reversed(moved):
        src = dest / rel
        if not src.exists():
            continue
        p, n = parent_fd(root, rel)
        d = os.open(src.parent, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
        try:
            os.rename(src.name, n, src_dir_fd=d, dst_dir_fd=p)
        finally:
            os.close(p)
            os.close(d)


def execute(a: argparse.Namespace, c: dict[str, Any]) -> dict[str, Any]:
    root, ss, rows, ids, prior, other = preflight(a, c)
    pending = a.archive_root / f".pending-{a.run_id}"
    final = a.archive_root / a.run_id
    if pending.exists() or final.exists():
        os.close(root)
        raise CleanupError("archive id exists")
    pending.mkdir(parents=True, mode=0o700)
    payload = pending / "payload"
    payload.mkdir(mode=0o700)
    committed = False
    moved = []
    try:
        for row in rows:
            raw = read_file(root, row["path"])
            atomic(payload / row["path"], raw)
            if sha(raw) != row["sha256"]:
                raise CleanupError("archive copy mismatch")
        manifest = {
            "schema": "task26-post-lifecycle-cleanup-archive-v2",
            "run_id": a.run_id,
            "candidate_digest": a.candidate,
            "wheel_sha256": a.wheel_sha256,
            "plan_sha256": a.plan_sha256,
            "approval": a.approval,
            "active_lifecycle": ids,
            "active_ledger_path": "payload/data/onboarding/telegram-customer-bootstrap-v1/ledger.json",
            "active_ledger_byte_for_byte": True,
            "entries": rows,
            "prior_archives": prior,
            "other_profiles_pre_sha256": other,
            "archive_committed_before_clear": True,
            "restore_or_prepopulate": False,
        }
        manifest["evidence_digest"] = sha(canon(manifest))
        atomic(pending / "manifest.json", canon(manifest))
        if a.test_fail_before_archive_commit:
            raise CleanupError("injected archive commit failure")
        os.rename(pending, final)
        committed = True
        quarantine = final / "cleared-originals"
        quarantine.mkdir(mode=0o700)
        try:
            for s in ss:
                move(root, s, quarantine)
                moved.append(s)
            for s in ss:
                if lst(root, s) is not None:
                    raise CleanupError(f"clear mismatch: {s}")
            if {str(p.absolute()): tree_digest(p) for p in a.other_profile} != other:
                raise CleanupError("other profile changed")
        except Exception:
            rollback(root, moved, quarantine)
            raise
        receipt = {
            "schema": "task26-post-lifecycle-cleanup-receipt-v2",
            "status": "PASS",
            "mode": "execute",
            "run_id": a.run_id,
            "manifest_sha256": sha(canon(manifest)),
            "archive_committed_before_clear": True,
            "post_cleanup_empty": True,
            "active_lifecycle": ids,
            "rollback_boundary": "after archive commit, all live renames roll back before terminal receipt",
        }
        atomic(final / "receipt.json", canon(receipt))
        return receipt
    finally:
        os.close(root)
        if not committed and pending.exists():
            shutil.rmtree(pending)


def verify(a: argparse.Namespace, c: dict[str, Any]) -> dict[str, Any]:
    archive = a.archive or a.archive_root / a.run_id
    mraw = safe_external(archive / "manifest.json")
    m = json.loads(mraw)
    receipt = json.loads(safe_external(archive / "receipt.json"))
    if (
        m.get("schema") != "task26-post-lifecycle-cleanup-archive-v2"
        or receipt.get("status") != "PASS"
    ):
        raise CleanupError("unknown archive/receipt schema")
    claimed = m["evidence_digest"]
    x = dict(m)
    x.pop("evidence_digest")
    if claimed != sha(canon(x)) or receipt.get("manifest_sha256") != sha(mraw):
        raise CleanupError("archive seal mismatch")
    for row in m["entries"]:
        raw = safe_external(archive / "payload" / row["path"])
        if len(raw) != row["size"] or sha(raw) != row["sha256"]:
            raise CleanupError("payload mismatch")
    root = open_root(a.profile)
    try:
        if scopes(root, c):
            raise CleanupError("live cleanup scope is not empty")
    finally:
        os.close(root)
    return {
        "schema": "task26-post-lifecycle-cleanup-verification-v2",
        "status": "PASS",
        "mode": "verify",
        "archive": str(archive),
        "active_lifecycle": m["active_lifecycle"],
    }


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser()
    p.add_argument("mode", choices=["dry-run", "execute", "verify"])
    p.add_argument("--profile", type=Path, required=True)
    p.add_argument("--other-profile", type=Path, action="append", default=[])
    p.add_argument("--archive-root", type=Path, required=True)
    p.add_argument("--archive", type=Path)
    p.add_argument("--contract", type=Path, required=True)
    p.add_argument("--permission", type=Path, required=True)
    p.add_argument("--candidate", required=True)
    p.add_argument("--wheel-sha256", required=True)
    p.add_argument("--plan-sha256", required=True)
    p.add_argument("--approval", required=True)
    p.add_argument("--actor-id", required=True)
    p.add_argument("--run-id", required=True)
    p.add_argument("--systemctl", type=Path, default=Path("/usr/bin/systemctl"))
    p.add_argument("--service", default="hermes-gateway-dualcoachtest.service")
    p.add_argument(
        "--test-fail-before-archive-commit", action="store_true", help=argparse.SUPPRESS
    )
    return p


def main() -> int:
    a = parser().parse_args()
    try:
        c = validate_inputs(a)
        if a.mode == "execute":
            v = execute(a, c)
        elif a.mode == "verify":
            v = verify(a, c)
        else:
            root, ss, rows, ids, prior, other = preflight(a, c)
            os.close(root)
            v = {
                "schema": "task26-post-lifecycle-cleanup-dry-run-v2",
                "status": "PASS",
                "mode": "dry-run",
                "mutations": 0,
                "active_lifecycle": ids,
                "scopes": ss,
                "entry_count": len(rows),
                "prior_archives": prior,
                "other_profiles_sha256": other,
            }
        sys.stdout.buffer.write(canon(v))
        return 0
    except (CleanupError, OSError, ValueError, KeyError, TypeError) as e:
        sys.stderr.write(f"FAIL: {e}\n")
        return 2


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