#!/usr/bin/env python3
"""Permission-sealed, archive-first profile reset. It has no service or network actions."""
from __future__ import annotations

import argparse
import fnmatch
import hashlib
import json
import os
import stat
import sys
import tempfile
from pathlib import Path
from typing import Any

APPROVAL_SCHEMA = "task26-profile-reset-permission-v1"
MANIFEST_SCHEMA = "task26-profile-reset-archive-manifest-v1"
RECEIPT_SCHEMA = "task26-profile-reset-execution-receipt-v1"
O_FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0)


class ResetError(RuntimeError):
    pass


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


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


def safe_external_file(path: Path) -> tuple[bytes, os.stat_result]:
    fd = os.open(path, os.O_RDONLY | O_FLAGS)
    try:
        before = os.fstat(fd)
        if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_uid != os.getuid():
            raise ResetError(f"unsafe sealed file: {path}")
        if stat.S_IMODE(before.st_mode) not in {0o400, 0o500, 0o600, 0o700}:
            raise ResetError(f"sealed file is not private: {path}")
        chunks: list[bytes] = []
        while chunk := os.read(fd, 1024 * 1024):
            chunks.append(chunk)
        after = os.fstat(fd)
        if stable(before) != stable(after):
            raise ResetError(f"sealed file changed during read: {path}")
        return b"".join(chunks), before
    finally:
        os.close(fd)


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


def open_root(path: Path) -> int:
    absolute = path.absolute()
    fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY | O_FLAGS)
    try:
        for part in absolute.parts[1:]:
            nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS, dir_fd=fd)
            if not stat.S_ISDIR(os.fstat(nxt).st_mode):
                os.close(nxt)
                raise ResetError(f"unsafe profile directory component: {part}")
            os.close(fd)
            fd = nxt
        info = os.fstat(fd)
        if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
            raise ResetError("profile root must be owner-held mode 0700")
        return fd
    except Exception:
        os.close(fd)
        raise


def open_parent(root_fd: int, relative: str) -> tuple[int, str]:
    parts = relative.split("/")
    fd = os.dup(root_fd)
    try:
        for part in parts[:-1]:
            nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS, dir_fd=fd)
            info = os.fstat(nxt)
            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
                os.close(nxt)
                raise ResetError(f"unsafe authority directory: {relative}")
            os.close(fd)
            fd = nxt
        return fd, parts[-1]
    except Exception:
        os.close(fd)
        raise


def lstat_at(root_fd: int, relative: str) -> os.stat_result | None:
    parent, name = open_parent(root_fd, relative)
    try:
        try:
            return os.stat(name, dir_fd=parent, follow_symlinks=False)
        except FileNotFoundError:
            return None
    finally:
        os.close(parent)


def read_regular(root_fd: int, relative: str) -> tuple[bytes, os.stat_result]:
    parent, name = open_parent(root_fd, relative)
    try:
        fd = os.open(name, os.O_RDONLY | O_FLAGS, dir_fd=parent)
    finally:
        os.close(parent)
    try:
        before = os.fstat(fd)
        if (not stat.S_ISREG(before.st_mode) or before.st_uid != os.getuid() or
                before.st_nlink != 1 or stat.S_IMODE(before.st_mode) != 0o600):
            raise ResetError(f"authority file must be regular, private, owner-held, and single-link: {relative}")
        chunks: list[bytes] = []
        while chunk := os.read(fd, 1024 * 1024):
            chunks.append(chunk)
        after = os.fstat(fd)
        if stable(before) != stable(after):
            raise ResetError(f"authority changed during stable read: {relative}")
        return b"".join(chunks), before
    finally:
        os.close(fd)


def walk_authority(root_fd: int, relative: str) -> list[dict[str, Any]]:
    info = lstat_at(root_fd, relative)
    if info is None:
        return []
    if stat.S_ISREG(info.st_mode):
        raw, observed = read_regular(root_fd, relative)
        return [{"path": relative, "sha256": digest_bytes(raw), "size": len(raw),
                 "mode": format(stat.S_IMODE(observed.st_mode), "04o")}]
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
        raise ResetError(f"authority root is not a private directory: {relative}")
    parent, name = open_parent(root_fd, relative)
    try:
        directory = os.open(name, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS, dir_fd=parent)
    finally:
        os.close(parent)
    try:
        rows: list[dict[str, Any]] = []
        for child in sorted(os.listdir(directory)):
            rows.extend(walk_authority(root_fd, f"{relative}/{child}"))
        if stable(info) != stable(os.fstat(directory)):
            raise ResetError(f"authority directory changed during traversal: {relative}")
        return rows
    finally:
        os.close(directory)


def tree_digest(path: Path) -> str:
    rows: list[bytes] = []
    for root, dirs, files in os.walk(path, topdown=True, followlinks=False):
        dirs.sort(); files.sort()
        base = Path(root)
        for name in dirs + files:
            item = base / name
            rel = item.relative_to(path).as_posix()
            info = item.lstat()
            if stat.S_ISLNK(info.st_mode):
                rows.append(b"L\0" + rel.encode() + b"\0" + os.readlink(item).encode())
            elif stat.S_ISDIR(info.st_mode):
                rows.append(b"D\0" + rel.encode() + b"\0" + format(stat.S_IMODE(info.st_mode), "o").encode())
            elif stat.S_ISREG(info.st_mode):
                fd = os.open(item, os.O_RDONLY | O_FLAGS)
                try:
                    before = os.fstat(fd); h = hashlib.sha256()
                    while chunk := os.read(fd, 1024 * 1024): h.update(chunk)
                    after = os.fstat(fd)
                    if stable(before) != stable(after): raise ResetError(f"tree changed during read: {item}")
                    rows.append(b"F\0" + rel.encode() + b"\0" + h.hexdigest().encode())
                finally: os.close(fd)
            else:
                raise ResetError(f"unsupported tree entry: {item}")
    return digest_bytes(b"\n".join(rows))


def atomic_write(path: Path, raw: bytes) -> None:
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(prefix=".reset-", dir=path.parent)
    try:
        os.fchmod(fd, 0o600)
        offset = 0
        while offset < len(raw): offset += os.write(fd, raw[offset:])
        os.fsync(fd); os.close(fd); fd = -1
        os.replace(temporary, path)
        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS)
        try: os.fsync(directory)
        finally: os.close(directory)
    finally:
        if fd >= 0: os.close(fd)
        try: os.unlink(temporary)
        except FileNotFoundError: pass


def copy_stable(root_fd: int, relative: str, destination: Path) -> dict[str, Any]:
    raw, info = read_regular(root_fd, relative)
    atomic_write(destination, raw)
    if digest_bytes(destination.read_bytes()) != digest_bytes(raw):
        raise ResetError(f"archive copy mismatch: {relative}")
    return {"path": relative, "sha256": digest_bytes(raw), "size": len(raw),
            "mode": format(stat.S_IMODE(info.st_mode), "04o")}


def validate_json_schemas(root_fd: int, rows: list[dict[str, Any]], contract: dict[str, Any]) -> int:
    schemas = contract["json_schemas"]
    bootstrap_count = 0
    for row in rows:
        relative = row["path"]
        if not relative.endswith(".json"):
            continue
        if relative not in schemas:
            raise ResetError(f"unknown JSON schema: {relative}")
        raw, _ = read_regular(root_fd, relative)
        try: value = json.loads(raw)
        except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ResetError(f"invalid JSON schema: {relative}") from exc
        rule = schemas[relative]
        if rule["type"] == "object" and not isinstance(value, dict): raise ResetError(f"wrong JSON root: {relative}")
        if not set(rule.get("required_keys", ())).issubset(value): raise ResetError(f"unknown/incomplete schema: {relative}")
        if "schema_value" in rule and value.get("schema") != rule["schema_value"]: raise ResetError(f"unknown schema value: {relative}")
        if relative.endswith("telegram-customer-bootstrap-v1/ledger.json"):
            sessions = value.get("sessions")
            if not isinstance(sessions, list): raise ResetError("customer-bootstrap sessions is not a list")
            terminal = set(contract["bootstrap_terminal_states"])
            for session in sessions:
                if (not isinstance(session, dict) or session.get("state") not in terminal or
                        session.get("role_claims") or session.get("recovery_attempts")):
                    raise ResetError("customer-bootstrap ledger contains nonterminal authority")
            bootstrap_count = len(sessions)
    return bootstrap_count


def effective_scopes(root_fd: int, contract: dict[str, Any]) -> list[str]:
    scopes = list(contract["approved_clear_scopes"])
    data_fd = os.open("data", os.O_RDONLY | os.O_DIRECTORY | O_FLAGS, dir_fd=root_fd)
    try: children = sorted(os.listdir(data_fd))
    finally: os.close(data_fd)
    protected = set(contract["protected_data_roots"])
    approved_top = {p.split("/", 1)[1].split("/", 1)[0] for p in scopes if p.startswith("data/")}
    patterns = contract["dynamic_clear_patterns"]
    for child in children:
        relative = f"data/{child}"
        if child in protected or child in approved_top: continue
        if any(fnmatch.fnmatchcase(relative, pattern) for pattern in patterns): scopes.append(relative); continue
        raise ResetError(f"unknown data authority root: {relative}")
    present = []
    for scope in scopes:
        if lstat_at(root_fd, scope) is not None: present.append(scope)
    # Keep only outermost scope; nested paths are represented by the containing directory.
    result: list[str] = []
    for scope in sorted(set(present), key=lambda p: (p.count("/"), p)):
        if not any(scope == parent or scope.startswith(parent + "/") for parent in result): result.append(scope)
    return sorted(result)


def validate_inputs(args: argparse.Namespace) -> tuple[dict[str, Any], dict[str, Any]]:
    contract_raw, _ = safe_external_file(args.contract); permission_raw, _ = safe_external_file(args.permission)
    try: contract = json.loads(contract_raw); permission = json.loads(permission_raw)
    except json.JSONDecodeError as exc: raise ResetError("contract or permission is not valid JSON") from exc
    if contract.get("schema") != "task26-profile-reset-contract-v1": raise ResetError("unknown reset contract")
    expected = {"schema": APPROVAL_SCHEMA, "approval": args.approval, "candidate_digest": args.candidate,
                "wheel_sha256": args.wheel_sha256, "plan_sha256": args.plan_sha256,
                "controller_sha256": digest_bytes(Path(__file__).read_bytes()),
                "contract_sha256": digest_bytes(contract_raw), "execute_allowed": True}
    if permission != expected: raise ResetError("permission seal or exact pins mismatch")
    if args.candidate != "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925":
        raise ResetError("candidate is not the exact approved successor")
    if len(args.run_id) < 3 or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for c in args.run_id):
        raise ResetError("invalid deterministic run id")
    return contract, permission


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


def prior_archives(args: argparse.Namespace) -> list[dict[str, str]]:
    roots = [args.profile / "data/rehearsal-reset-archives", args.archive_root]
    archives: list[dict[str, str]] = []
    for root in roots:
        if not root.exists():
            continue
        if root.is_symlink() or not root.is_dir():
            raise ResetError(f"unsafe archive collection root: {root}")
        for child in sorted(root.iterdir(), key=lambda p: p.name):
            if child.name.startswith(".pending-") or (root == args.archive_root and child.name == args.run_id):
                continue
            if child.is_symlink() or not child.is_dir():
                raise ResetError(f"unsafe prior archive root: {child}")
            archives.append({"path": child.relative_to(args.profile).as_posix(), "sha256": tree_digest(child)})
    return archives


def preflight(args: argparse.Namespace, contract: dict[str, Any]) -> tuple[int, list[str], list[dict[str, Any]], int, list[dict[str, str]], dict[str, str]]:
    root_fd = open_root(args.profile)
    try:
        scopes = effective_scopes(root_fd, contract)
        rows: list[dict[str, Any]] = []
        for scope in scopes: rows.extend(walk_authority(root_fd, scope))
        rows.sort(key=lambda row: row["path"])
        existing = {row["path"] for row in rows}
        missing = sorted(set(contract["required_current_scopes"]) - existing)
        if missing and args.mode != "verify": raise ResetError(f"required current authority missing: {missing}")
        bootstrap_count = validate_json_schemas(root_fd, rows, contract)
    except Exception:
        os.close(root_fd); raise
    archives = prior_archives(args)
    other = {str(path.absolute()): tree_digest(path) for path in args.other_profile}
    return root_fd, scopes, rows, bootstrap_count, archives, other


def manifest_value(args: argparse.Namespace, rows: list[dict[str, Any]], count: int,
                   archives: list[dict[str, str]], other: dict[str, str]) -> dict[str, Any]:
    value = {"schema": MANIFEST_SCHEMA, "run_id": args.run_id, "candidate_digest": args.candidate,
             "wheel_sha256": args.wheel_sha256, "plan_sha256": args.plan_sha256,
             "approval": args.approval, "bootstrap_ledger_covered": any(
                 row["path"] == "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" for row in rows),
             "bootstrap_session_count": count, "entries": rows, "prior_archives": archives,
             "other_profiles_pre_sha256": other, "restore_or_prepopulate": False}
    value["evidence_digest"] = digest_bytes(canonical(value))
    return value


def move_scope(root_fd: int, scope: str, quarantine: Path) -> None:
    source_parent, name = open_parent(root_fd, scope)
    destination = quarantine / scope
    destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    dest_parent = os.open(destination.parent, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS)
    try: os.rename(name, destination.name, src_dir_fd=source_parent, dst_dir_fd=dest_parent)
    finally: os.close(source_parent); os.close(dest_parent)


def rollback(root_fd: int, scopes: list[str], quarantine: Path) -> None:
    for scope in reversed(scopes):
        source = quarantine / scope
        if not source.exists(): continue
        destination_parent, name = open_parent(root_fd, scope)
        src_parent = os.open(source.parent, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS)
        try: os.rename(source.name, name, src_dir_fd=src_parent, dst_dir_fd=destination_parent)
        finally: os.close(src_parent); os.close(destination_parent)


def execute(args: argparse.Namespace, contract: dict[str, Any]) -> dict[str, Any]:
    if processes_for(args.profile): raise ResetError("profile still has running processes")
    root_fd, scopes, rows, count, prior, other = preflight(args, contract)
    pending = args.archive_root / f".pending-{args.run_id}"
    final = args.archive_root / args.run_id
    if pending.exists() or final.exists(): os.close(root_fd); raise ResetError("archive run id already exists")
    pending.mkdir(mode=0o700, parents=True)
    payload = pending / "payload"; payload.mkdir(mode=0o700)
    moved: list[str] = []
    committed = False
    try:
        copied = [copy_stable(root_fd, row["path"], payload / row["path"]) for row in rows]
        if copied != rows: raise ResetError("staged archive manifest mismatch")
        manifest = manifest_value(args, rows, count, prior, other)
        atomic_write(pending / "manifest.json", canonical(manifest))
        for scope in scopes:
            move_scope(root_fd, scope, pending / "rollback-originals"); moved.append(scope)
        for scope in scopes:
            if lstat_at(root_fd, scope) is not None: raise ResetError(f"clear mismatch: {scope}")
        if args.test_fail_before_commit: raise ResetError("injected precommit mismatch")
        other_after = {str(path.absolute()): tree_digest(path) for path in args.other_profile}
        if other_after != other: raise ResetError("other profile changed before commit")
        if prior_archives(args) != prior:
            raise ResetError("prior archive changed before commit")
        receipt = {"schema": RECEIPT_SCHEMA, "status": "PASS", "mode": "execute", "run_id": args.run_id,
                   "manifest_sha256": digest_bytes(canonical(manifest)), "evidence_digest": manifest["evidence_digest"],
                   "bootstrap_ledger_covered": manifest["bootstrap_ledger_covered"], "post_reset_empty_baseline": True,
                   "other_profiles_unchanged": True, "rollback_boundary": "archive directory rename (commit)"}
        atomic_write(pending / "receipt.json", canonical(receipt))
        os.rename(pending, final); committed = True
        return receipt
    except Exception:
        if not committed: rollback(root_fd, moved, pending / "rollback-originals")
        raise
    finally:
        os.close(root_fd)
        if not committed and pending.exists():
            import shutil
            shutil.rmtree(pending)


def verify_archive(args: argparse.Namespace, contract: dict[str, Any]) -> dict[str, Any]:
    archive = args.archive or (args.archive_root / args.run_id)
    manifest_raw, _ = safe_external_file(archive / "manifest.json")
    receipt_raw, _ = safe_external_file(archive / "receipt.json")
    manifest = json.loads(manifest_raw); receipt = json.loads(receipt_raw)
    if manifest.get("schema") != MANIFEST_SCHEMA or receipt.get("schema") != RECEIPT_SCHEMA: raise ResetError("unknown archive schema")
    evidence = dict(manifest); claimed = evidence.pop("evidence_digest", None)
    if claimed != digest_bytes(canonical(evidence)): raise ResetError("manifest evidence digest mismatch")
    if receipt.get("manifest_sha256") != digest_bytes(manifest_raw): raise ResetError("receipt manifest pin mismatch")
    for row in manifest["entries"]:
        raw, _ = safe_external_file(archive / "payload" / row["path"])
        if len(raw) != row["size"] or digest_bytes(raw) != row["sha256"]: raise ResetError(f"archive payload mismatch: {row['path']}")
    root_fd = open_root(args.profile)
    try:
        scopes = effective_scopes(root_fd, contract)
        if scopes: raise ResetError(f"post-reset authority is not empty: {scopes}")
    finally: os.close(root_fd)
    current_prior = []
    for row in manifest["prior_archives"]:
        path = args.profile / row["path"]
        current_prior.append({"path": row["path"], "sha256": tree_digest(path)})
    if current_prior != manifest["prior_archives"]: raise ResetError("prior archive hash mismatch")
    other = {str(path.absolute()): tree_digest(path) for path in args.other_profile}
    if other != manifest["other_profiles_pre_sha256"]: raise ResetError("other profile mismatch")
    return {"schema": "task26-profile-reset-independent-verification-v1", "status": "PASS", "mode": "verify",
            "archive": str(archive), "manifest_sha256": digest_bytes(manifest_raw),
            "bootstrap_ledger_covered": manifest["bootstrap_ledger_covered"], "post_reset_empty_baseline": True}


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


def main() -> int:
    args = parser().parse_args()
    try:
        contract, _ = validate_inputs(args)
        if args.mode == "execute": value = execute(args, contract)
        elif args.mode == "verify": value = verify_archive(args, contract)
        else:
            root_fd, scopes, rows, count, prior, other = preflight(args, contract)
            os.close(root_fd)
            value = {"schema": "task26-profile-reset-dry-run-v1", "status": "PASS", "mode": "dry-run",
                     "profile": str(args.profile.absolute()), "scopes": scopes,
                     "manifest": manifest_value(args, rows, count, prior, other), "mutations": 0}
        sys.stdout.buffer.write(canonical(value)); return 0
    except (ResetError, OSError, ValueError, KeyError, TypeError) as exc:
        sys.stderr.write(f"FAIL: {exc}\n"); return 2


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