"""Canonical, private Issue runtime allocation (not cache/worktree admission)."""
from __future__ import annotations

import argparse
import grp
import json
import os
import pwd
import re
import stat
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

if __package__ in (None, ''):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from harness.lib.managed_lock import ManagedLockSet, validate_held_flock

_IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,254}")
_RUN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC


def _identity(root, owner, repository, issue, run_id=None):
    raw = os.fspath(root)
    if not isinstance(raw, str) or not raw.startswith('/') or '\x00' in raw:
        raise ValueError('invalid harness root')
    if any(part in {'', '.', '..'} for part in raw.split('/')[1:]):
        raise ValueError('invalid harness root')
    for value in (owner, repository):
        if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
            raise ValueError('invalid repository identity')
    if isinstance(issue, bool) or not isinstance(issue, int) or issue <= 0:
        raise ValueError('invalid Issue')
    if run_id is not None and (not isinstance(run_id, str) or not _RUN.fullmatch(run_id)):
        raise ValueError('invalid workflow run id')
    return Path(raw)


def issue_runtime_path(root, owner, repository, issue, run_id):
    root = _identity(root, owner, repository, issue, run_id)
    return root / 'repositories' / owner / repository / 'worktrees' / f'issue-{issue}' / 'runtime' / run_id


def _open_absolute(path):
    fd = os.open('/', _DIR_FLAGS)
    try:
        for part in path.parts[1:]:
            child = os.open(part, _DIR_FLAGS, dir_fd=fd)
            os.close(fd)
            fd = child
        return fd
    except BaseException:
        os.close(fd)
        raise


def _private(st):
    if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o700:
        raise ValueError('unsafe private directory')


def _repository_container(st):
    mode = stat.S_IMODE(st.st_mode)
    if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid() or mode & 0o002:
        raise ValueError('unsafe repositories directory')
    if mode & 0o020:
        try:
            group = grp.getgrgid(st.st_gid)
            members = {pwd.getpwnam(name).pw_uid for name in group.gr_mem}
            members.update(user.pw_uid for user in pwd.getpwall() if user.pw_gid == st.st_gid)
        except KeyError as exc:
            raise ValueError('unresolved repositories group') from exc
        if members != {os.getuid()}:
            raise ValueError('unsafe repositories group')


def _existing(parent, name, check=_private):
    child = os.open(name, _DIR_FLAGS, dir_fd=parent)
    try:
        check(os.fstat(child))
        named = os.stat(name, dir_fd=parent, follow_symlinks=False)
        opened = os.fstat(child)
        if (named.st_dev, named.st_ino) != (opened.st_dev, opened.st_ino):
            raise ValueError('directory identity changed')
        return child
    except BaseException:
        os.close(child)
        raise


def _cache_fd(root, owner, repository):
    fd = _open_absolute(root)
    try:
        if os.fstat(fd).st_mode & 0o002:
            raise ValueError('world-writable harness root')
        for name, check in [('repositories', _repository_container), (owner, _private), (repository, _private)]:
            child = _existing(fd, name, check)
            os.close(fd)
            fd = child
        return fd
    except BaseException:
        os.close(fd)
        raise


def _new_child(parent, name, created):
    os.mkdir(name, 0o700, dir_fd=parent)
    initial = os.stat(name, dir_fd=parent, follow_symlinks=False)
    created.append((os.dup(parent), name, (initial.st_dev, initial.st_ino)))
    # Correct only a directory just created by this call, including restrictive umask.
    if not stat.S_ISDIR(initial.st_mode) or initial.st_uid != os.getuid():
        raise ValueError('new directory identity changed')
    os.chmod(name, 0o700, dir_fd=parent, follow_symlinks=False)
    fd = _existing(parent, name)
    opened = os.fstat(fd)
    if (opened.st_dev, opened.st_ino) != (initial.st_dev, initial.st_ino):
        os.close(fd)
        raise ValueError('new directory identity changed')
    return fd


@dataclass(frozen=True)
class RuntimeDirectory:
    path: Path
    fd: int  # Caller owns this descriptor and must close it.


def prepare_issue_runtime(root, owner, repository, issue, run_id, *, lock_set):
    path = issue_runtime_path(root, owner, repository, issue, run_id)
    expected = ManagedLockSet.paths_for_issue(root, owner, repository, issue)
    if (not isinstance(lock_set, ManagedLockSet) or not lock_set.issue.held or not lock_set.repository.held
            or (lock_set.issue.path, lock_set.repository.path) != expected):
        raise ValueError('matching Issue and repository leases required')
    for lock in (lock_set.issue, lock_set.repository):
        lock_fd = lock.fd
        if lock_fd is None or not validate_held_flock(lock_fd, lock.path):
            raise ValueError('kernel lease unavailable')
        parent = _open_absolute(lock.path.parent)
        try:
            _private(os.fstat(parent))
            named = os.stat(lock.path.name, dir_fd=parent, follow_symlinks=False)
            opened = os.fstat(lock_fd)
            if (named.st_dev, named.st_ino) != (opened.st_dev, opened.st_ino):
                raise ValueError('kernel lease path identity mismatch')
        finally:
            os.close(parent)
    fd = _cache_fd(Path(root), owner, repository)
    created = []
    success = False
    try:
        for name in ('worktrees', f'issue-{issue}', 'runtime'):
            try:
                child = _existing(fd, name)
            except FileNotFoundError:
                child = _new_child(fd, name, created)
            os.close(fd)
            fd = child
        # Runs are immutable identities; even a safe, empty stale run is a collision.
        try:
            fd_run = _new_child(fd, run_id, created)
        except FileExistsError as exc:
            raise ValueError('runtime run already exists') from exc
        os.close(fd)
        fd = fd_run
        success = True
        return RuntimeDirectory(path, fd)
    finally:
        if not success:
            os.close(fd)
            for parent, name, identity in reversed(created):
                try:
                    current = os.stat(name, dir_fd=parent, follow_symlinks=False)
                    if (current.st_dev, current.st_ino) == identity and stat.S_ISDIR(current.st_mode):
                        os.rmdir(name, dir_fd=parent)  # Empty, owned inode only; never recursive.
                except OSError:
                    pass
        for parent, _, _ in created:
            os.close(parent)


def inspect_issue_worktree_admission(root, owner, repository, issue):
    """Classify container shape; caller still authenticates state/base/registration.

    This read-only check is used under Issue + repository leases. It does not
    authorize Git writes or declare an existing checkout valid.
    """
    root = _identity(root, owner, repository, issue)
    fd = _cache_fd(root, owner, repository)
    issue_fd = None
    try:
        bare = root / 'repositories' / owner / repository / 'repository.git'
        bare_fd = os.open('repository.git', _DIR_FLAGS, dir_fd=fd)
        try:
            bare_st = os.fstat(bare_fd)
            if bare_st.st_uid != os.getuid() or bare_st.st_mode & 0o022:
                raise ValueError('unsafe bare repository')
            result = subprocess.run(['git', '--git-dir', f'/proc/self/fd/{bare_fd}', 'worktree', 'list', '--porcelain', '-z'],
                                    pass_fds=(bare_fd,), capture_output=True, text=True, timeout=30, check=False)
        finally:
            os.close(bare_fd)
        if result.returncode:
            raise ValueError('Git registration unavailable')
        checkout = bare.parent / 'worktrees' / f'issue-{issue}' / 'checkout'
        registrations = [field[9:] for field in result.stdout.split('\x00') if field.startswith('worktree ')]
        registered = registrations.count(str(checkout))
        try:
            worktrees = _existing(fd, 'worktrees')
        except FileNotFoundError:
            return 'invalid' if registered else 'absent'
        try:
            try:
                issue_fd = _existing(worktrees, f'issue-{issue}')
            except FileNotFoundError:
                return 'invalid' if registered else 'absent'
        finally:
            os.close(worktrees)
        entries = set(os.listdir(issue_fd))
        for name in entries & {'runtime', 'verification'}:
            child = _existing(issue_fd, name)
            os.close(child)
        if entries <= {'runtime', 'verification'}:
            return 'invalid' if registered else 'runtime-only'
        if entries != (entries & {'runtime', 'verification'}) | {'checkout', 'state.json'} or registered != 1:
            return 'invalid'
        source = os.stat('checkout', dir_fd=issue_fd, follow_symlinks=False)
        state = os.stat('state.json', dir_fd=issue_fd, follow_symlinks=False)
        if (not stat.S_ISDIR(source.st_mode) or source.st_uid != os.getuid() or source.st_mode & 0o022
                or not stat.S_ISREG(state.st_mode) or state.st_uid != os.getuid()
                or stat.S_IMODE(state.st_mode) != 0o600 or state.st_nlink != 1):
            return 'invalid'
        return 'checkout-candidate'
    finally:
        if issue_fd is not None:
            os.close(issue_fd)
        os.close(fd)


def main():
    # Direct-script bootstrap is below; imports are delayed to avoid broker cycles.
    parser = argparse.ArgumentParser()
    parser.add_argument('command', choices=['inspect-worktree'])
    for name in ('root', 'owner', 'repository', 'handle'):
        parser.add_argument('--' + name, required=True)
    parser.add_argument('--issue', required=True, type=int)
    args = parser.parse_args()
    try:
        from harness.lib import managed_lock_broker as broker, verification_root
        root = _identity(args.root, args.owner, args.repository, args.issue)
        verification_root._lease(Path(args.handle), root, args.owner, args.repository, args.issue)
        if broker.request_handle(args.handle, 'status').get('status') != 'repository-held':
            raise ValueError('repository lease required')
        shape = inspect_issue_worktree_admission(root, args.owner, args.repository, args.issue)
        print(json.dumps({'container': shape}))
        return 1 if shape == 'invalid' else 0
    except (ValueError, OSError, subprocess.SubprocessError):
        print(json.dumps({'error': 'worktree-container-unavailable'}))
        return 2


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