"""Per-seat implementer system identity, ACL grants, and path validation."""

from __future__ import annotations

import hashlib
import grp
import os
import pwd
import shutil
import shlex
import stat
import subprocess
import errno
import re

from seat_common import (
    CONTROL_BASE_MODE,
    CONTROL_DIR_MODE,
    CONTROL_OWNER_MARKER_MODE,
    CONTROL_NETNS_INODE_MODE,
    CONTROL_RE,
    IMPLEMENTER_USER_PREFIX,
    LAUNCH_AUTHORITY_FILES,
    LINUX_USERNAME_RE,
    MAX_LINUX_USERNAME,
    NOLOGIN_SHELL,
    OWNER_MARKER_PREFIX,
    SEAT_GIT_EMAIL_DOMAIN,
    SEAT_ID_RE,
    SEAT_LAUNCHER_BIN,
    SEAT_TRUSTED_CONFIG_DIR,
    die,
    implementer_runtime_base,
    seat_control_base,
    seat_test_mode,
)

_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)

MSG_PREFIX = "overdeck-seat-identity:"


def _die(msg: str, code: int = 2) -> None:
    die(f"{MSG_PREFIX} {msg}", code)


def _lstat_strict(path: str) -> os.stat_result:
    try:
        return os.lstat(path)
    except OSError as exc:
        _die(f"path-missing:{path}:{exc.strerror}", 1)


def implementer_username_for_seat(seat_id: str) -> str:
    if not SEAT_ID_RE.match(seat_id):
        _die("invalid seat id")
    digest = hashlib.sha256(seat_id.encode("utf-8")).hexdigest()[:16]
    if (
        seat_id != seat_id.lower()
        or "." in seat_id
        or len(f"{IMPLEMENTER_USER_PREFIX}{seat_id.lower()}") > MAX_LINUX_USERNAME
    ):
        return f"{IMPLEMENTER_USER_PREFIX}{digest}"
    candidate = f"{IMPLEMENTER_USER_PREFIX}{seat_id.lower()}"
    if LINUX_USERNAME_RE.match(candidate):
        return candidate
    return f"{IMPLEMENTER_USER_PREFIX}{digest}"


def implementer_home_for_seat(seat_id: str) -> str:
    return f"{implementer_runtime_base()}/{seat_id}/home"


def implementer_runtime_dir(seat_id: str) -> str:
    return f"{implementer_runtime_base()}/{seat_id}"


def control_dir_for_seat(seat_id: str) -> str:
    return f"{seat_control_base()}/{seat_id}"


def owner_marker_path(seat_id: str) -> str:
    return f"{control_dir_for_seat(seat_id)}/owner"


def tmux_conf_path(seat_id: str) -> str:
    return f"{control_dir_for_seat(seat_id)}/tmux-seat.conf"


def netns_inode_marker_path(seat_id: str) -> str:
    return f"{control_dir_for_seat(seat_id)}/netns-inode"


def read_netns_inode(path: str) -> str:
    st = os.stat(path)
    return f"{st.st_dev}:{st.st_ino}"


def record_netns_inode(seat_id: str) -> None:
    inode = read_netns_inode("/proc/self/ns/net")
    atomic_root_control_write(seat_id, "netns-inode", f"{inode}\n", mode=CONTROL_NETNS_INODE_MODE)


def read_recorded_netns_inode(seat_id: str) -> str:
    path = netns_inode_marker_path(seat_id)
    if not os.path.isfile(path):
        die("netns-inode-missing", 1)
    st = os.lstat(path)
    if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
        die("netns-inode-unreadable", 1)
    if not seat_test_mode() and (st.st_uid != 0 or st.st_gid != 0):
        die("netns-inode-owner", 1)
    if not seat_test_mode() and stat.S_IMODE(st.st_mode) != CONTROL_NETNS_INODE_MODE:
        die("netns-inode-mode", 1)
    with open(path, encoding="utf-8") as fh:
        raw = fh.read().strip()
    if not raw or not re.fullmatch(r"\d+:\d+", raw):
        die("netns-inode-nonnumeric", 1)
    return raw


def expected_socket_for_seat(operator_home: str, seat_id: str) -> str:
    return f"{operator_home}/.local/state/overdeck/seats/{seat_id}/tmux.sock"


def assert_safe_path(path: str, *, label: str) -> None:
    if not path.startswith("/") or "//" in path or CONTROL_RE.search(path):
        _die(f"invalid-{label}")
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"symlink-forbidden:{path}", 1)


def assert_owned_regular(path: str, *, uid: int, gid: int, mode: int) -> None:
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
        _die(f"not-regular-file:{path}", 1)
    if st.st_uid != uid or st.st_gid != gid:
        _die(f"ownership-drift:{path}", 1)
    if stat.S_IMODE(st.st_mode) != mode:
        _die(f"mode-drift:{path}", 1)


def ancestor_paths(path: str) -> list[str]:
    parts: list[str] = []
    current = os.path.normpath(path)
    while True:
        parts.append(current)
        parent = os.path.dirname(current)
        if parent == current:
            break
        current = parent
    return list(reversed(parts))


def _relpath_from_home(path: str, operator_home: str) -> str:
    rel = os.path.relpath(path, operator_home)
    if rel.startswith("..") or rel == ".":
        _die(f"path-outside-home:{path}", 1)
    return rel


def _allowed_symlink(parent: str, name: str, operator_home: str) -> bool:
    rel_parent = _relpath_from_home(parent, operator_home)
    if rel_parent == ".local/share/overdeck/seat-guard" and name == "current":
        return True
    if rel_parent == ".local/bin" and name == "claude":
        return True
    return False


def _read_symlink_target(link_path: str, parent: str) -> str:
    target = os.readlink(link_path)
    if target.startswith("/"):
        resolved = os.path.normpath(target)
    else:
        resolved = os.path.normpath(os.path.join(parent, target))
    if not resolved.startswith("/") or "//" in resolved or CONTROL_RE.search(resolved):
        _die(f"symlink-target-invalid:{link_path}", 1)
    return resolved


def resolve_managed_path(
    operator_home: str,
    logical_path: str,
    *,
    operator_uid: int,
    operator_gid: int,
) -> str:
    """Resolve one-hop managed symlinks; validate ownership on every step."""
    if not logical_path.startswith("/"):
        logical_path = os.path.join(operator_home, logical_path)
    logical_path = os.path.normpath(logical_path)
    if not logical_path.startswith(operator_home + "/") and logical_path != operator_home:
        _die(f"managed-path-outside-home:{logical_path}", 1)

    current = operator_home
    parts = logical_path[len(operator_home) :].strip("/").split("/")
    idx = 0
    while idx < len(parts):
        part = parts[idx]
        candidate = os.path.join(current, part)
        st = _lstat_strict(candidate)
        if stat.S_ISLNK(st.st_mode):
            if not _allowed_symlink(os.path.dirname(candidate), part, operator_home):
                _die(f"symlink-unmanaged:{candidate}", 1)
            parent = os.path.dirname(candidate)
            resolved = _read_symlink_target(candidate, parent)
            if not resolved.startswith(operator_home + "/"):
                _die(f"symlink-escape:{candidate}", 1)
            target_st = _lstat_strict(resolved)
            if stat.S_ISLNK(target_st.st_mode):
                _die(f"symlink-chained:{candidate}", 1)
            if target_st.st_uid != operator_uid or target_st.st_gid != operator_gid:
                _die(f"symlink-target-owner:{resolved}", 1)
            current = resolved
            idx += 1
            continue
        if st.st_uid != operator_uid or st.st_gid != operator_gid:
            _die(f"path-owner-mismatch:{candidate}", 1)
        current = candidate
        idx += 1
    return current


def managed_runtime_targets(
    operator_home: str,
    seat_id: str,
    *,
    repo: str,
    profile: str,
    state_dir: str,
    guard_bin: str,
    claude_bin: str,
    launcher: str,
) -> tuple[list[str], list[str], list[str]]:
    seat_root = f"{operator_home}/seats/{seat_id}"
    traverse_only = [
        operator_home,
        f"{operator_home}/seats",
        seat_root,
        f"{operator_home}/.claudex-accounts",
        os.path.dirname(profile),
        f"{operator_home}/.local",
        f"{operator_home}/.local/state",
        f"{operator_home}/.local/state/overdeck",
        f"{operator_home}/.local/state/overdeck/seats",
        f"{operator_home}/.local/share",
        f"{operator_home}/.local/share/overdeck",
        f"{operator_home}/.local/share/overdeck/seat-guard",
        f"{operator_home}/.local/bin",
    ]
    rw_roots = [
        os.path.normpath(state_dir),
        os.path.normpath(repo),
        os.path.normpath(profile),
    ]
    executables = [guard_bin, claude_bin]
    return traverse_only, rw_roots, executables


def require_setfacl() -> None:
    if shutil.which("setfacl") is None:
        _die("setfacl-missing", 1)


def _run_setfacl(argv: list[str], *, label: str) -> None:
    proc = subprocess.run(argv, capture_output=True, text=True)
    if proc.returncode != 0:
        detail = (proc.stderr or proc.stdout or "").strip()
        _die(f"{label}:{detail}", 1)


def grant_traverse_acl(path: str, username: str) -> None:
    if not os.path.lexists(path):
        return
    assert_safe_path(path, label="acl-target")
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"symlink-forbidden:{path}", 1)
    _run_setfacl(["setfacl", "-m", f"u:{username}:x", path], label=f"setfacl-failed:{path}")


def trusted_config_dir(state_dir: str) -> str:
    return os.path.join(os.path.normpath(state_dir), SEAT_TRUSTED_CONFIG_DIR)


def launch_authority_paths(state_dir: str) -> list[str]:
    base = trusted_config_dir(state_dir)
    return [os.path.join(base, name) for name in LAUNCH_AUTHORITY_FILES]


def seat_git_author_name(seat_id: str) -> str:
    return f"Overdeck Seat {seat_id}"


def seat_git_author_email(seat_id: str) -> str:
    return f"{seat_id}@{SEAT_GIT_EMAIL_DOMAIN}"


def configure_seat_git_author(checkout: str, seat_id: str) -> None:
    if not os.path.isdir(checkout):
        _die("checkout-missing", 1)
    name = seat_git_author_name(seat_id)
    email = seat_git_author_email(seat_id)
    for key, value in (("user.name", name), ("user.email", email)):
        proc = subprocess.run(
            ["git", "-C", checkout, "config", "--local", key, value],
            capture_output=True,
            text=True,
        )
        if proc.returncode != 0:
            detail = (proc.stderr or proc.stdout or "").strip()
            _die(f"seat-git-config-failed:{key}:{detail}", 1)


def verify_seat_git_author(checkout: str, seat_id: str) -> None:
    if not os.path.isdir(checkout):
        _die("checkout-missing", 1)
    want_name = seat_git_author_name(seat_id)
    want_email = seat_git_author_email(seat_id)
    for key, want in (("user.name", want_name), ("user.email", want_email)):
        proc = subprocess.run(
            ["git", "-C", checkout, "config", "--local", "--get", key],
            capture_output=True,
            text=True,
        )
        if proc.returncode != 0 or proc.stdout.strip() != want:
            _die(f"seat-git-config-drift:{key}", 1)


def validate_implementer_group_membership(username: str, primary_gid: int) -> None:
    try:
        pw = pwd.getpwnam(username)
    except KeyError:
        _die("implementer-user-missing", 1)
    if pw.pw_gid != primary_gid:
        _die("implementer-primary-gid-mismatch", 1)
    try:
        primary = grp.getgrgid(primary_gid)
    except KeyError:
        _die("implementer-primary-group-missing", 1)
    if primary.gr_name != username:
        _die("implementer-primary-group-name-mismatch", 1)
    for group in grp.getgrall():
        if username in group.gr_mem:
            _die(f"implementer-supplementary-group:{group.gr_name}", 1)


def deny_operator_credentials(implementer_user: str, operator_home: str) -> None:
    for path in operator_credential_paths(operator_home):
        if not os.path.lexists(path):
            continue
        assert_safe_path(path, label="credential-deny-target")
        st = _lstat_strict(path)
        if stat.S_ISLNK(st.st_mode):
            _die(f"symlink-forbidden:{path}", 1)
        _run_setfacl(
            ["setfacl", "-m", f"u:{implementer_user}:---", path],
            label=f"credential-deny-failed:{path}",
        )


def grant_rw_root_acl(path: str, implementer_user: str, operator_user: str) -> None:
    if not os.path.exists(path):
        return
    assert_safe_path(path, label="acl-rw-target")
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"symlink-forbidden:{path}", 1)
    for username in (implementer_user, operator_user):
        perm = "rwX" if stat.S_ISDIR(st.st_mode) else "rw-"
        _run_setfacl(["setfacl", "-m", f"u:{username}:{perm}", path], label=f"setfacl-failed:{path}")
        if stat.S_ISDIR(st.st_mode):
            _run_setfacl(
                ["setfacl", "-R", "-m", f"u:{username}:rwX", path],
                label=f"setfacl-recursive-failed:{path}",
            )
            _run_setfacl(
                ["setfacl", "-R", "-d", "-m", f"u:{username}:rwX", path],
                label=f"setfacl-default-recursive-failed:{path}",
            )


def grant_trusted_config_read_boundary(
    state_dir: str,
    implementer_user: str,
    operator_user: str,
) -> None:
    trusted = trusted_config_dir(state_dir)
    if not os.path.lexists(trusted):
        os.makedirs(trusted, mode=0o700, exist_ok=True)
    assert_safe_path(trusted, label="trusted-config")
    st = _lstat_strict(trusted)
    if stat.S_ISLNK(st.st_mode):
        _die(f"symlink-forbidden:{trusted}", 1)
    _run_setfacl(
        ["setfacl", "-m", f"u:{operator_user}:rwX", trusted],
        label=f"trusted-config-operator:{trusted}",
    )
    _run_setfacl(
        ["setfacl", "-d", "-m", f"u:{operator_user}:rwX", trusted],
        label=f"trusted-config-operator-default:{trusted}",
    )
    _run_setfacl(
        ["setfacl", "-m", f"u:{implementer_user}:r-x", trusted],
        label=f"trusted-config-implementer:{trusted}",
    )
    _run_setfacl(
        ["setfacl", "-d", "-m", f"u:{implementer_user}:r--", trusted],
        label=f"trusted-config-implementer-default:{trusted}",
    )
    if os.path.isdir(trusted):
        for root, dirnames, filenames in os.walk(trusted):
            for name in dirnames:
                dir_path = os.path.join(root, name)
                _run_setfacl(
                    ["setfacl", "-m", f"u:{implementer_user}:r-x", dir_path],
                    label=f"trusted-config-implementer-dir:{dir_path}",
                )
            for name in filenames:
                file_path = os.path.join(root, name)
                _run_setfacl(
                    ["setfacl", "-m", f"u:{implementer_user}:r--", file_path],
                    label=f"trusted-config-implementer-file:{file_path}",
                )


def _verify_implementer_path_boundary(
    implementer_user: str,
    paths: list[str],
    *,
    sudo: str,
    require_readable: bool,
    label: str,
) -> None:
    for path in paths:
        if not os.path.lexists(path):
            continue
        if require_readable:
            proc = subprocess.run(
                [sudo, "-n", "-u", implementer_user, "test", "-r", path],
                capture_output=True,
                text=True,
            )
            if proc.returncode != 0:
                _die(f"{label}-unreadable:{path}", 1)
        for mode in ("-r", "-w"):
            if require_readable and mode == "-r":
                continue
            proc = subprocess.run(
                [sudo, "-n", "-u", implementer_user, "test", mode, path],
                capture_output=True,
                text=True,
            )
            if proc.returncode == 0:
                _die(f"{label}-readable:{path}" if mode == "-r" else f"{label}-writable:{path}", 1)
        proc = subprocess.run(
            [sudo, "-n", "-u", implementer_user, "rm", "-f", path],
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            _die(f"{label}-deletable:{path}", 1)
        proc = subprocess.run(
            [sudo, "-n", "-u", implementer_user, "sh", "-c", f"printf x >{shlex.quote(path)}"],
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            _die(f"{label}-replaceable:{path}", 1)


def verify_implementer_launch_authority_boundary(
    implementer_user: str,
    state_dir: str,
    *,
    sudo: str = "sudo",
) -> None:
    trusted = trusted_config_dir(state_dir)
    _verify_implementer_path_boundary(
        implementer_user,
        launch_authority_paths(state_dir),
        sudo=sudo,
        require_readable=True,
        label="launch-authority",
    )
    for mode in ("-w",):
        proc = subprocess.run(
            [sudo, "-n", "-u", implementer_user, "test", mode, trusted],
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            _die(f"launch-authority-dir-writable:{trusted}", 1)
    for path in launch_authority_paths(state_dir):
        if not os.path.lexists(path):
            continue
        for mode in ("-w",):
            proc = subprocess.run(
                [sudo, "-n", "-u", implementer_user, "test", mode, path],
                capture_output=True,
                text=True,
            )
            if proc.returncode == 0:
                _die(f"launch-authority-writable:{path}", 1)


def grant_executable_acl(path: str, username: str) -> None:
    if not os.path.lexists(path):
        _die(f"executable-missing:{path}", 1)
    assert_safe_path(path, label="acl-exec-target")
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"symlink-forbidden:{path}", 1)
    perm = "rx" if stat.S_ISREG(st.st_mode) else "x"
    _run_setfacl(["setfacl", "-m", f"u:{username}:{perm}", path], label=f"setfacl-failed:{path}")


def grant_link_ancestors_acl(path: str, username: str, operator_home: str) -> None:
    for ancestor in ancestor_paths(path):
        if ancestor == path:
            continue
        if ancestor != operator_home and not ancestor.startswith(operator_home + "/"):
            continue
        grant_traverse_acl(ancestor, username)


def verify_root_launcher(path: str) -> None:
    if path != SEAT_LAUNCHER_BIN:
        _die("launcher-mismatch", 1)
    if not os.path.lexists(path):
        _die(f"launcher-missing:{path}", 1)
    st = _lstat_strict(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"launcher-symlink:{path}", 1)
    if not stat.S_ISREG(st.st_mode):
        _die(f"launcher-not-file:{path}", 1)
    if stat.S_IMODE(st.st_mode) != 0o755:
        _die(f"launcher-mode:{path}", 1)
    if not seat_test_mode() and (st.st_uid != 0 or st.st_gid != 0):
        _die(f"launcher-owner:{path}", 1)
    if not os.access(path, os.X_OK):
        _die(f"launcher-not-executable:{path}", 1)


def grant_managed_runtime_access(
    implementer_user: str,
    operator_user: str,
    operator_home: str,
    seat_id: str,
    *,
    operator_uid: int,
    operator_gid: int,
    repo: str,
    profile: str,
    state_dir: str,
    guard_bin: str,
    claude_bin: str,
    launcher: str,
) -> None:
    require_setfacl()
    deny_operator_credentials(implementer_user, operator_home)
    resolved_guard = resolve_managed_path(
        operator_home, guard_bin, operator_uid=operator_uid, operator_gid=operator_gid
    )
    resolved_claude = resolve_managed_path(
        operator_home, claude_bin, operator_uid=operator_uid, operator_gid=operator_gid
    )
    verify_root_launcher(launcher)
    traverse_only, rw_roots, _executables = managed_runtime_targets(
        operator_home,
        seat_id,
        repo=repo,
        profile=profile,
        state_dir=state_dir,
        guard_bin=resolved_guard,
        claude_bin=resolved_claude,
        launcher=launcher,
    )
    seen: set[str] = set()
    for path in traverse_only:
        norm = os.path.normpath(path)
        if norm in seen:
            continue
        seen.add(norm)
        grant_traverse_acl(norm, implementer_user)
    for path in rw_roots:
        grant_rw_root_acl(path, implementer_user, operator_user)
    grant_trusted_config_read_boundary(state_dir, implementer_user, operator_user)
    grant_link_ancestors_acl(resolved_guard, implementer_user, operator_home)
    grant_executable_acl(resolved_guard, implementer_user)
    grant_link_ancestors_acl(resolved_claude, implementer_user, operator_home)
    grant_executable_acl(resolved_claude, implementer_user)


def _test_mode() -> bool:
    from seat_common import seat_test_mode

    return seat_test_mode()


def _ensure_root_control_dir(
    path: str,
    mode: int,
    *,
    symlink_msg: str,
    invariant_msg: str,
) -> str:
    if os.path.lexists(path):
        st = _lstat_strict(path)
        if stat.S_ISLNK(st.st_mode):
            _die(symlink_msg, 1)
        if not stat.S_ISDIR(st.st_mode):
            _die(f"{invariant_msg}-not-dir", 1)
        if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0):
            _die(invariant_msg, 1)
        if stat.S_IMODE(st.st_mode) != mode:
            os.chmod(path, mode, follow_symlinks=False)
            st = os.lstat(path)
            if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
                _die(symlink_msg, 1)
        if not _test_mode():
            if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode:
                _die(invariant_msg, 1)
        return path

    os.makedirs(path, mode=mode)
    st = os.lstat(path)
    if stat.S_ISLNK(st.st_mode):
        _die(symlink_msg, 1)
    if not stat.S_ISDIR(st.st_mode):
        _die(f"{invariant_msg}-not-dir", 1)
    if stat.S_IMODE(st.st_mode) != mode:
        os.chmod(path, mode, follow_symlinks=False)
        st = os.lstat(path)
    if not _test_mode():
        if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode:
            _die(invariant_msg, 1)
    return path


def ensure_control_base() -> str:
    return _ensure_root_control_dir(
        seat_control_base(),
        CONTROL_BASE_MODE,
        symlink_msg="control-base-symlink",
        invariant_msg="control-base-invariant",
    )


def ensure_control_dir(seat_id: str) -> str:
    ensure_control_base()
    return _ensure_root_control_dir(
        control_dir_for_seat(seat_id),
        CONTROL_DIR_MODE,
        symlink_msg="control-dir-symlink",
        invariant_msg="control-dir-invariant",
    )


def atomic_root_control_write(seat_id: str, name: str, content: str, *, mode: int) -> str:
    control = ensure_control_dir(seat_id)
    final = os.path.join(control, name)
    if os.path.lexists(final):
        st = os.lstat(final)
        if stat.S_ISLNK(st.st_mode):
            _die(f"control-symlink:{final}", 1)
        if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0):
            _die(f"control-owner:{final}", 1)
    dir_fd = os.open(control, os.O_RDONLY | os.O_DIRECTORY | _O_NOFOLLOW)
    try:
        tmp = f".{name}.{os.getpid()}.tmp"
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW
        try:
            fd = os.open(tmp, flags, mode, dir_fd=dir_fd)
        except OSError as exc:
            if exc.errno in (errno.EEXIST, errno.ELOOP):
                _die(f"control-temp-create:{final}:{exc.strerror}", 1)
            raise
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                fh.write(content)
                fh.flush()
                os.fchmod(fh.fileno(), mode)
                os.fsync(fh.fileno())
        except OSError:
            try:
                os.unlink(tmp, dir_fd=dir_fd)
            except OSError:
                pass
            raise
        os.replace(tmp, name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
        os.fsync(dir_fd)
    finally:
        os.close(dir_fd)
    st = os.lstat(final)
    if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
        _die(f"control-not-regular:{final}", 1)
    if _test_mode():
        return final
    if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode:
        _die(f"control-finalize:{final}", 1)
    return final


def read_owner_marker(seat_id: str) -> str:
    path = owner_marker_path(seat_id)
    if not os.path.isfile(path):
        _die("implementer-owner-marker-missing", 1)
    st = os.lstat(path)
    if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
        _die("implementer-owner-marker-invalid", 1)
    if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0):
        _die("implementer-owner-marker-invalid", 1)
    with open(path, encoding="utf-8") as fh:
        return fh.read().strip()


def write_owner_marker(seat_id: str, operator_user: str) -> None:
    atomic_root_control_write(seat_id, "owner", operator_user, mode=CONTROL_OWNER_MARKER_MODE)


def pw_comment_matches(username: str, operator_user: str) -> bool:
    try:
        pw = pwd.getpwnam(username)
    except KeyError:
        return False
    return pw.pw_gecos == f"{OWNER_MARKER_PREFIX}{operator_user}"


def validate_implementer_passwd(pw: pwd.struct_passwd, operator_user: str, seat_id: str) -> None:
    want_user = implementer_username_for_seat(seat_id)
    if pw.pw_name != want_user:
        _die(f"implementer-username-mismatch:{pw.pw_name}", 1)
    if pw.pw_uid == 0:
        _die("implementer-uid-root", 1)
    if pw.pw_shell != NOLOGIN_SHELL:
        _die("implementer-shell-invalid", 1)
    want_home = implementer_home_for_seat(seat_id)
    if pw.pw_dir != want_home:
        _die("implementer-home-mismatch", 1)
    if pw.pw_gecos != f"{OWNER_MARKER_PREFIX}{operator_user}":
        _die("implementer-gecos-mismatch", 1)


def ensure_implementer_user(seat_id: str, operator_user: str) -> tuple[str, int, int]:
    username = implementer_username_for_seat(seat_id)
    home = implementer_home_for_seat(seat_id)
    marker = f"{OWNER_MARKER_PREFIX}{operator_user}"
    ensure_control_dir(seat_id)
    try:
        pw = pwd.getpwnam(username)
        validate_implementer_passwd(pw, operator_user, seat_id)
        validate_implementer_group_membership(username, pw.pw_gid)
        marker_owner = read_owner_marker(seat_id)
        if marker_owner != operator_user:
            _die("implementer-owner-marker-mismatch", 1)
        return username, pw.pw_uid, pw.pw_gid
    except KeyError:
        pass
    runtime = implementer_runtime_dir(seat_id)
    os.makedirs(runtime, mode=0o700, exist_ok=True)
    os.chown(runtime, 0, 0)
    proc = subprocess.run(
        [
            "useradd",
            "--system",
            "--user-group",
            "--home-dir",
            home,
            "--shell",
            NOLOGIN_SHELL,
            "--comment",
            marker,
            username,
        ],
        capture_output=True,
        text=True,
    )
    if proc.returncode != 0:
        detail = (proc.stderr or proc.stdout or "").strip()
        _die(f"implementer-useradd-failed:{detail}", 1)
    pw = pwd.getpwnam(username)
    validate_implementer_passwd(pw, operator_user, seat_id)
    validate_implementer_group_membership(username, pw.pw_gid)
    write_owner_marker(seat_id, operator_user)
    os.makedirs(home, mode=0o700, exist_ok=True)
    os.chown(home, pw.pw_uid, pw.pw_gid)
    return username, pw.pw_uid, pw.pw_gid


def resolve_implementer_for_seat(seat_id: str, operator_user: str) -> tuple[str, int, int]:
    username = implementer_username_for_seat(seat_id)
    try:
        pw = pwd.getpwnam(username)
    except KeyError:
        _die("implementer-user-missing", 1)
    validate_implementer_passwd(pw, operator_user, seat_id)
    validate_implementer_group_membership(username, pw.pw_gid)
    marker_owner = read_owner_marker(seat_id)
    if marker_owner != operator_user:
        _die("implementer-owner-marker-mismatch", 1)
    return username, pw.pw_uid, pw.pw_gid


def operator_credential_paths(operator_home: str) -> list[str]:
    return [
        f"{operator_home}/.claude/.credentials.json",
        f"{operator_home}/.ssh",
        f"{operator_home}/.config/gh",
        f"{operator_home}/.git-credentials",
        f"{operator_home}/.gitconfig",
    ]


def verify_implementer_seat_metadata_boundary(
    implementer_user: str,
    operator_home: str,
    seat_id: str,
    *,
    sudo: str = "sudo",
) -> None:
    seat_root = os.path.normpath(f"{operator_home}/seats/{seat_id}")
    seat_json = os.path.join(seat_root, "seat.json")
    _verify_implementer_path_boundary(
        implementer_user,
        [seat_json],
        sudo=sudo,
        require_readable=False,
        label="seat-metadata",
    )
    if os.path.isdir(seat_root):
        proc = subprocess.run(
            [sudo, "-n", "-u", implementer_user, "test", "-w", seat_root],
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            _die(f"seat-root-writable:{seat_root}", 1)


def verify_implementer_credential_boundary(
    implementer_user: str,
    operator_home: str,
    *,
    sudo: str = "sudo",
) -> None:
    for path in operator_credential_paths(operator_home):
        if not os.path.exists(path):
            continue
        proc = subprocess.run(
            [sudo, "-n", "-u", implementer_user, "test", "-r", path],
            capture_output=True,
            text=True,
        )
        if proc.returncode == 0:
            _die(f"operator-credential-readable:{path}", 1)


def verify_implementer_systemd_absent(implementer_user: str, *, sudo: str = "sudo") -> None:
    proc = subprocess.run(
        [sudo, "-n", "-u", implementer_user, "systemctl", "--user", "is-system-running"],
        capture_output=True,
        text=True,
    )
    if proc.returncode == 0:
        _die("user-systemd-available", 1)


def _sudo_readable_as_user(path: str, username: str, *, sudo: str = "sudo") -> bool:
    proc = subprocess.run(
        [sudo, "-n", "-u", username, "test", "-r", path],
        capture_output=True,
        text=True,
    )
    return proc.returncode == 0


def _assert_secure_implementer_runtime_dir(
    path: str,
    *,
    uid: int,
    gid: int,
) -> None:
    if not os.path.lexists(path):
        return
    st = os.lstat(path)
    if stat.S_ISLNK(st.st_mode):
        _die(f"implementer-runtime-symlink:{path}", 1)
    if not stat.S_ISDIR(st.st_mode):
        _die(f"implementer-runtime-not-dir:{path}", 1)
    if st.st_uid != uid or st.st_gid != gid:
        _die(f"implementer-runtime-ownership-drift:{path}", 1)
    if stat.S_IMODE(st.st_mode) & 0o077:
        _die(f"implementer-runtime-insecure-mode:{path}", 1)


def _assert_path_unreadable_as_implementer(
    path: str,
    implementer_user: str,
    *,
    label: str,
    sudo: str = "sudo",
) -> None:
    if not os.path.lexists(path):
        return
    if _sudo_readable_as_user(path, implementer_user, sudo=sudo):
        _die(f"{label}:{path}", 1)


def _assert_implementer_bus_absent(bus_path: str) -> None:
    if os.path.lexists(bus_path):
        _die(f"implementer-bus-present:{bus_path}", 1)


def verify_implementer_runtime_bus_absent(
    implementer_user: str,
    implementer_uid: int,
    implementer_gid: int,
    operator_uid: int | None,
    *,
    sudo: str = "sudo",
) -> None:
    runtime = f"/run/user/{implementer_uid}"
    _assert_secure_implementer_runtime_dir(
        runtime,
        uid=implementer_uid,
        gid=implementer_gid,
    )
    _assert_implementer_bus_absent(f"{runtime}/bus")
    if operator_uid is not None:
        operator_runtime = f"/run/user/{operator_uid}"
        for path in (operator_runtime, f"{operator_runtime}/bus"):
            _assert_path_unreadable_as_implementer(
                path,
                implementer_user,
                label="operator-runtime-readable",
                sudo=sudo,
            )
