"""Immutable fork runtime for the Task22 canonical launcher."""
from __future__ import annotations

import ctypes
import fcntl
import hashlib
import json
import os
import stat
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Protocol, cast
from task22_child_protocol import (
    await_child, ensure_pidfd_support, run_child, validate_completion,
)
from task22_dependency_closure import RuntimeSeal, attest_runtime
from task22_lifecycle import supervise
from task22_resource_ownership import (
    ChildOwnership, LockHandoff, assert_parent_profile_free, await_gate,
    open_exact_lock_residue, revalidate_lock_residue, validate_lock_residue,
)
Entries = dict[str, list[object]]
class MemfdCreate(Protocol):
    def __call__(self, name: bytes, flags: int, /) -> int: ...
class Syscall(Protocol):
    def __call__(self, number: int, argument: int, flags: int, /) -> int: ...
def sha256(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()

def sealed_memfd(name: str, content: bytes) -> int:
    create = cast(MemfdCreate, ctypes.CDLL(None, use_errno=True).memfd_create)
    descriptor = create(name.encode(), 0x0001 | 0x0002)
    if descriptor < 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error))
    try:
        view = memoryview(content)
        while view:
            written = os.write(descriptor, view)
            if written <= 0:
                raise RuntimeError(f"failed to populate sealed snapshot {name}")
            view = view[written:]
        _ = os.lseek(descriptor, 0, os.SEEK_SET)
        seals = 0x0001 | 0x0002 | 0x0004 | 0x0008
        _ = fcntl.fcntl(descriptor, 1033, seals)
        if fcntl.fcntl(descriptor, 1034) != seals:
            raise RuntimeError(f"sealed snapshot {name} is not immutable")
        if sha256(os.read(descriptor, len(content) + 1)) != sha256(content):
            raise RuntimeError(f"sealed snapshot {name} failed final hash revalidation")
        _ = os.lseek(descriptor, 0, os.SEEK_SET)
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def trusted_user_bus() -> tuple[str, str]:
    uid, gid = os.geteuid(), os.getegid()
    boundary, runtime = Path("/run/user"), Path("/run/user") / str(uid)
    bus = runtime / "bus"
    boundary_info, runtime_info, bus_info = boundary.lstat(), runtime.lstat(), bus.lstat()
    safe = stat.S_ISDIR(boundary_info.st_mode) and boundary_info.st_uid == 0
    safe = safe and boundary_info.st_gid == 0 and stat.S_IMODE(boundary_info.st_mode) == 0o755
    safe = safe and stat.S_ISDIR(runtime_info.st_mode) and runtime_info.st_uid == uid
    safe = safe and runtime_info.st_gid == gid and stat.S_IMODE(runtime_info.st_mode) == 0o700
    safe = safe and stat.S_ISSOCK(bus_info.st_mode) and bus_info.st_uid == uid
    safe = safe and bus_info.st_gid == gid and stat.S_IMODE(bus_info.st_mode) == 0o666
    if not safe:
        raise RuntimeError("systemd user bus boundary is unsafe")
    return str(runtime), f"unix:path={bus}"


def strict_environment() -> dict[str, str]:
    runtime, bus = trusted_user_bus()
    return {
        "HOME": "/home/cube", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8",
        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
        "PYTHONDONTWRITEBYTECODE": "1", "XDG_RUNTIME_DIR": runtime,
        "DBUS_SESSION_BUS_ADDRESS": bus,
    }


def pidfd_open(pid: int) -> int:
    descriptor = cast(Syscall, ctypes.CDLL(None, use_errno=True).syscall)(434, pid, 0)
    if descriptor < 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error))
    return descriptor


def _read_sealed(descriptor: int, digest: str, label: str) -> tuple[str, bytes]:
    path = f"/proc/self/fd/{descriptor}"
    with open(path, "rb", buffering=0) as stream:
        source = stream.read()
    if sha256(source) != digest:
        raise RuntimeError(f"sealed {label} digest mismatch immediately before execution")
    return path, source


def _child_environment(cwd: Path, argv: list[str]) -> None:
    os.environ.clear()
    os.environ.update(strict_environment())
    os.chdir(cwd)
    sys.argv = argv


def run_sealed_source(
    descriptor: int, digest: str, argv: list[str], cwd: Path
) -> tuple[int, str, str]:
    ensure_pidfd_support()
    with ChildOwnership() as owned:
        stdout_read, stdout_write = owned.pipe()
        stderr_read, stderr_write = owned.pipe()
        gate_read, gate_write = owned.pipe()
        receipt_read, receipt_write = owned.pipe()
        pid = os.fork()
        if pid == 0:
            os.close(stdout_read)
            os.close(stderr_read)
            os.close(gate_write)
            os.close(receipt_read)
            _ = os.dup2(stdout_write, 1)
            _ = os.dup2(stderr_write, 2)
            os.close(stdout_write)
            os.close(stderr_write)

            def body() -> None:
                await_gate(gate_read, "sealed child")
                _child_environment(cwd, argv)
                path, source = _read_sealed(descriptor, digest, "verifier")
                namespace = {"__name__": "__main__", "__file__": path, "__builtins__": __builtins__}
                exec(compile(source, path, "exec", dont_inherit=True), namespace)

            run_child(body, receipt_write, sys.stdout.flush, sys.stderr.flush)
        owned.forked(pid, gate_write)
        owned.close(stdout_write)
        owned.close(stderr_write)
        owned.close(gate_read)
        owned.close(receipt_write)
        pidfd = owned.own(pidfd_open(pid))
        _ = os.write(gate_write, b"1")
        owned.close(gate_write)
        code = await_child(pid, pidfd, receipt_read)
        owned.reaped()
        stdout = os.read(stdout_read, 65536).decode(errors="replace")
        stderr = os.read(stderr_read, 65536).decode(errors="replace")
        result = code, stdout, stderr
    return result


def _clear_application_modules() -> None:
    for name in tuple(sys.modules):
        if name == "gateway" or name.startswith("gateway."):
            del sys.modules[name]
        elif name == "checkin_cli" or name.startswith("checkin_cli."):
            del sys.modules[name]


def run_snapshot_child(
    entries: Entries,
    cli_descriptor: int,
    cli_digest: str,
    bootstrap: bytes,
    attest: Callable[[], RuntimeSeal],
    after_fork: Callable[[], None] | None = None,
    lock_handoff: LockHandoff | None = None,
) -> int:
    ensure_pidfd_support()
    try:
        seal = attest()
    except BaseException:
        if lock_handoff is not None:
            os.close(lock_handoff[2])
        raise
    bootstrap_fd = sealed_memfd("task22-removal-bootstrap", bootstrap)
    if lock_handoff is not None:
        profile_root, lock_path, lock_fd, lock_expected = lock_handoff
        try:
            _ = lock_path.relative_to(profile_root)
        except ValueError as exc:
            raise RuntimeError("transaction lock is outside the profile root") from exc
        validate_lock_residue(lock_path, lock_fd, lock_expected)

    def worker(gate_read: int, receipt: int) -> None:
        def body() -> None:
            await_gate(gate_read, "snapshot worker")
            handoff: dict[str, object] | None = None
            if lock_handoff is not None:
                _, child_lock, child_fd, child_expected = lock_handoff
                validate_lock_residue(child_lock, child_fd, child_expected)
                handoff = {"path": str(child_lock), "fd": child_fd,
                           "expected": child_expected}
            _child_environment(Path.cwd(), [])
            _clear_application_modules()
            bootstrap_path, source = _read_sealed(
                bootstrap_fd, sha256(bootstrap), "bootstrap"
            )
            cli_path = f"/proc/self/fd/{cli_descriptor}"
            sys.argv = [bootstrap_path, json.dumps(entries, sort_keys=True, separators=(",", ":")),
                        cli_path, cli_digest, json.dumps(handoff, separators=(",", ":"))]
            namespace = {"__name__": "__main__", "__builtins__": __builtins__}
            exec(compile(source, bootstrap_path, "exec", dont_inherit=True), namespace)
        run_child(body, receipt, sys.stdout.flush, sys.stderr.flush)

    def close_supervisor_references() -> None:
        seal.close()
        os.close(bootstrap_fd)
        if lock_handoff is not None:
            os.close(lock_handoff[2])
            assert_parent_profile_free(lock_handoff[0])

    parent_lock_closed = False

    def parent_release() -> None:
        nonlocal parent_lock_closed
        error: BaseException | None = None
        try:
            if after_fork is not None:
                after_fork()
            if lock_handoff is not None:
                validate_lock_residue(lock_handoff[1], lock_handoff[2], lock_handoff[3])
        except BaseException as exc:
            error = exc
        finally:
            if lock_handoff is not None:
                os.close(lock_handoff[2])
                parent_lock_closed = True
                assert_parent_profile_free(lock_handoff[0])
        if error is not None:
            raise error

    try:
        return supervise(
            pidfd_open, worker, close_supervisor_references, parent_release
        )
    finally:
        seal.close()
        os.close(bootstrap_fd)
        if lock_handoff is not None and not parent_lock_closed:
            os.close(lock_handoff[2])
            assert_parent_profile_free(lock_handoff[0])


__all__ = [
    "RuntimeSeal", "attest_runtime", "open_exact_lock_residue", "revalidate_lock_residue",
    "run_sealed_source", "run_snapshot_child", "sealed_memfd", "strict_environment",
    "validate_completion",
]