#!/usr/bin/python3
"""Disposable root/kernel fixture; requires its own private run/home/data/net.

No native compilation. The protocol client implements the frozen C proxy wire.
The access lease fixture is synthetic; it is NOT production access qualification.
"""
import array
import ctypes
import json
import os
from pathlib import Path
import select
import shutil
import signal
import socket
import struct
import subprocess
import sys
import time

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import c01_runtime as runtime
from c01_runtime_view import RuntimeLease
from c01_runtime_groups import current_group

LIBC = ctypes.CDLL(None, use_errno=True)
UID, MANAGER, AUTHORITY, OPERATOR = 61000, 61001, 61002, 61003
HOME = Path('/home/runtime-fixture')
BACKING = Path('/var/lib/runtime-home')
RUN = Path(f'/run/user/{UID}')
DATA = Path('/var/lib/runtime-data')
CODE = Path('/run/runtime-code')
UNITS = Path('/run/runtime-units')


def require(value, message='fixture assertion failed'):
    if not value:
        raise AssertionError(message)


def wait_for(check, timeout=20):
    end = time.monotonic() + timeout
    while time.monotonic() < end:
        result = check()
        if result:
            return result
        time.sleep(.05)
    raise AssertionError('fixture deadline expired')


def drop(uid):
    os.setgroups([]); os.setresgid(uid, uid, uid); os.setresuid(uid, uid, uid)


def mount(source, target, kind=None, flags=4096, data=None):
    require(LIBC.mount(os.fsencode(source) if source else None, os.fsencode(target),
                      os.fsencode(kind) if kind else None, flags, os.fsencode(data) if data else None) == 0,
            f'mount failed: {source} -> {target}: errno {ctypes.get_errno()}')


def owned(path, uid=UID):
    path.mkdir(parents=True, exist_ok=True)
    path.chmod(0o700); os.chown(path, uid, uid)


def packet(argv, cwd=str(HOME), env=()):
    def string(text):
        raw = text.encode(); return struct.pack('!I', len(raw)) + raw
    raw = struct.pack('!IIII', runtime.runtime_owner.MAGIC, 1, len(argv), len(env))
    raw += b''.join(string(text) for text in [cwd, *argv, *env])
    return struct.pack('!I', len(raw)) + raw


def client(argv, uid=MANAGER, gid=None, disconnect=False):
    # Fork gives real SO_PEERCRED. Regular fixture files avoid pipe-EOF ambiguity
    # from an intentionally persistent container/tmux process retaining stdout.
    out = os.memfd_create('runtime-output', os.MFD_CLOEXEC)
    err = os.memfd_create('runtime-error', os.MFD_CLOEXEC)
    report_read, report_write = os.pipe2(os.O_CLOEXEC)
    cancel_read, cancel_write = os.pipe2(os.O_CLOEXEC)
    pid = os.fork()
    if pid == 0:
        try:
            os.close(report_read)
            os.close(cancel_write)
            os.setgroups([MANAGER] if gid is not None else [])
            os.setresgid(gid or uid, gid or uid, gid or uid); os.setresuid(uid, uid, uid)
            channel = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            channel.settimeout(45)
            channel.connect('/run/archipelago-managed/runtime.sock')
            # Exactly what runtime-proxy.c requires; root creator must remain >1.
            peer_pid, peer_uid, peer_gid = runtime.peer(channel)
            require(peer_pid > 1 and (peer_uid, peer_gid) == (0, 0), 'root listener credential mismatch')
            with open('/dev/null', 'rb') as source:
                raw = packet(argv)
                count = channel.sendmsg([raw], [(socket.SOL_SOCKET, socket.SCM_RIGHTS,
                                                array.array('i', [source.fileno(), out, err]))])
                channel.sendall(raw[count:])
            if disconnect:
                require(os.read(cancel_read, 1) == b'x', 'fixture cancellation barrier')
                channel.close(); os.write(report_write, b'cancelled')
            else:
                raw = b''
                while len(raw) < 12:
                    chunk = channel.recv(12 - len(raw))
                    if not chunk:
                        break
                    raw += chunk
                os.write(report_write, raw)
            os._exit(0)
        except (PermissionError, ConnectionResetError, BrokenPipeError):
            if (uid, gid or uid) != (MANAGER, MANAGER):
                os.write(report_write, b'REFUSED'); os._exit(0)
            import traceback; traceback.print_exc(); os._exit(111)
        except BaseException:
            import traceback; traceback.print_exc(); os._exit(111)
    os.close(report_write)
    os.close(cancel_read)
    if disconnect:
        wait_for(lambda: (DATA / 'descendant.json').exists(), 5)
        os.write(cancel_write, b'x')
    os.close(cancel_write)
    poll = select.poll(); handle = os.pidfd_open(pid); poll.register(handle, select.POLLIN)
    require(poll.poll(50000), 'fixture client timeout')
    _, status = os.waitpid(pid, 0); os.close(handle)
    result = os.read(report_read, 128); os.close(report_read)
    os.lseek(out, 0, 0); os.lseek(err, 0, 0)
    output, error = os.read(out, 1024 * 1024), os.read(err, 1024 * 1024)
    os.close(out); os.close(err)
    require(os.waitstatus_to_exitcode(status) == 0, f'client failed: {error!r}')
    if disconnect:
        return result
    return struct.unpack('!III', result)[1:] if len(result) == 12 else None, output, error


def command(argv):
    out = os.memfd_create('static-proxy-output', os.MFD_CLOEXEC)
    err = os.memfd_create('static-proxy-error', os.MFD_CLOEXEC)
    try:
        result = subprocess.run(['/opt/archipelago-runtime-proxy', '--', *argv], cwd='/', preexec_fn=lambda: drop(MANAGER),
                                stdin=subprocess.DEVNULL, stdout=out, stderr=err,
                                env={'PATH': '/usr/bin:/bin'}, timeout=45)
        os.lseek(out, 0, 0); os.lseek(err, 0, 0)
        output, error = os.read(out, 1024 * 1024), os.read(err, 1024 * 1024)
        require(result.returncode == 0, f'{argv}: {result.returncode}: {error.decode(errors="replace")}')
        return output.decode().strip()
    finally:
        os.close(out); os.close(err)


def asynchronous_command(source):
    return subprocess.Popen(['/opt/archipelago-runtime-proxy', '--', '/usr/bin/python3', '-c', source],
                            cwd='/', preexec_fn=lambda: drop(MANAGER),
                            stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                            env={'PATH': '/usr/bin:/bin'})


def rendered_home_view(repo):
    """Exercise the rendered template's home/runtime binds over covered roots.

    This checks kernel visibility/inodes with the actual bus/services below.
    It is NOT a systemd SYSTEM-manager/drop-in dependency-order qualification;
    that still requires a disposable boot of the rendered user@ composition.
    """
    text = (repo / 'image-recipe/configs/80-archipelago-runtime-user.conf.in').read_text()
    for key, value in {'ORIGINAL_RUNTIME_UID': str(UID), 'ORIGINAL_HOME': str(HOME),
                       'SANITIZED_RUNTIME_HOME': str(BACKING),
                       'MANAGER_BACKING_PARENT': '/var/lib/private-manager',
                       'SNAPSHOT_BACKING_PARENT': '/var/lib/private-snapshot',
                       'AUTHORITY_BACKING_PARENT': '/var/lib/private-authority'}.items():
        text = text.replace('@' + key + '@', value)
    require('ProtectHome=tmpfs' in text)
    binds = []
    for line in text.splitlines():
        if line.startswith('BindPaths='):
            source, _, target = line.split('=', 1)[1].partition(':')
            binds.append((os.open(source, os.O_PATH | os.O_DIRECTORY | os.O_CLOEXEC), Path(target or source)))
    try:
        for target in ('/home', '/root', '/run/user'):
            mount('tmpfs', target, 'tmpfs', 0, 'mode=755,size=4M')
        for fd, target in binds:
            target.mkdir(parents=True, exist_ok=True)
            mount('/proc/self/fd/' + str(fd), target)
        for target in ('/home', '/root', '/run/user'):
            mount(None, target, flags=32 | 1)  # readonly cover, writable sub-binds
        require(RUN.stat().st_ino == next(os.fstat(fd).st_ino for fd, path in binds if path == RUN))
        require(not Path('/run/user/61003').exists(), 'foreign user runtime exposed')
    finally:
        for fd, _ in binds:
            os.close(fd)


def native_terminal_contract(identity):
    binary = Path('/opt/runtime-terminal-tests')
    if not binary.exists():
        print('NOT RUN native handler/runtime workspace: supply --terminal-tests rebuilt test binary', flush=True)
        return
    require(not (DATA / 'native-terminal-pass').exists(), 'stale native terminal success marker')
    work_identity = (BACKING / 'Work').stat()
    original = [(path.stat().st_dev, path.stat().st_ino, path.read_bytes())
                for path in (BACKING / 'Work/preserved', BACKING / 'Work/explicit ; literal/preserved')]
    view = RuntimeLease(MANAGER, MANAGER, UID, UID, identity['configuration_sha256'])
    child = os.fork()
    if child == 0:
        try:
            require(LIBC.unshare(0x00020000) == 0, 'private terminal manager namespace unavailable')
            mount(None, '/', flags=16384 | (1 << 18))
            mount('tmpfs', '/home', 'tmpfs', 0, 'mode=755,size=4M')
            mount('tmpfs', '/run', 'tmpfs', 0, 'mode=755,size=16M')
            view.install(); view.close()
            home = Path('/run/archipelago-c01/manager')
            owned(home, MANAGER)
            aliases = Path('/run/archipelago-c01/bin'); aliases.mkdir()
            for name in ('sh', 'tmux'):
                (aliases / name).symlink_to('/opt/archipelago-runtime-proxy')
            drop(MANAGER)
            os.chdir(home)  # Handler commands must explicitly use runtime-visible /.
            os.execve(binary, [str(binary), 'api::handler::terminal::runtime_workspace_tests::actual_runtime_workspace_contract',
                              '--exact', '--ignored', '--nocapture', '--test-threads=1'],
                      {'PATH': str(aliases), 'HOME': str(home), 'ARCHY_SESSION_STATE_DIR': str(home / 'sessions'),
                       'ARCHIPELAGO_RUNTIME_WORKSPACE': str(HOME / 'Work'), 'ARCHY_RUNTIME_TERMINAL_FIXTURE': '1'})
        except BaseException:
            import traceback; traceback.print_exc(); os._exit(111)
    try:
        poll = select.poll(); handle = os.pidfd_open(child); poll.register(handle, select.POLLIN)
        reaped = False
        try:
            require(poll.poll(60000), 'native terminal contract deadline')
            _, status = os.waitpid(child, 0)
            reaped = True
            require(os.waitstatus_to_exitcode(status) == 0, 'native terminal contract failed')
        finally:
            try:
                if not reaped:
                    os.kill(child, signal.SIGKILL)  # Still pinned by our unreaped child ownership.
                    require(poll.poll(5000), 'native terminal fixture retirement deadline')
                    os.waitpid(child, 0)
            finally:
                os.close(handle)
    finally:
        view.close()
    after = [(path.stat().st_dev, path.stat().st_ino, path.read_bytes())
             for path in (BACKING / 'Work/preserved', BACKING / 'Work/explicit ; literal/preserved')]
    require(after == original, 'preserved workspace inode/content changed')
    require((work_identity.st_dev, work_identity.st_ino) ==
            ((BACKING / 'Work').stat().st_dev, (BACKING / 'Work').stat().st_ino), 'preserved Work inode replaced')
    require((DATA / 'native-terminal-pass').read_text() == 'passed', 'native test did not run to completion')
    print('PASS native handler -> proxy -> runtime workspace -> persistent tmux; default/explicit pwd, preserved files, reconnect metadata', flush=True)


def main():
    require(os.geteuid() == 0 and os.environ.get('ARCHY_RUNTIME_FIXTURE') == '1', 'isolated root fixture required')
    rows = Path('/proc/self/mountinfo').read_text().splitlines()
    for target in ('/run', '/home', '/var/lib'):
        require(any(row.split()[4] == target and ' - tmpfs ' in row for row in rows), 'private fixture mount missing')
    Path('/tmp').chmod(0o1777)
    Path('/var/tmp').chmod(0o1777)
    require('INTERP' not in subprocess.check_output(['/usr/bin/readelf', '-l', '/opt/archipelago-runtime-proxy'], text=True),
            'existing proxy is not static')
    root = current_group()
    require('c01-runtime-fixture-' in str(root), 'owned transient unit required')
    control = root / 'control'; control.mkdir(); (control / 'cgroup.procs').write_text(str(os.getpid()))
    (root / 'cgroup.subtree_control').write_text('+cpu +memory +pids')
    user_group = root / 'runtime-user.service'; user_group.mkdir()
    for name in ('', 'cgroup.procs', 'cgroup.threads', 'cgroup.subtree_control'):
        os.chown(user_group / name, UID, UID)
    CODE.mkdir(); CODE.chmod(0o755)
    repo = Path(__file__).resolve().parents[2]
    subprocess.run(['/usr/bin/python3', '-B', '-m', 'unittest', 'discover', '-s', str(repo / 'scripts/tests'),
                    '-p', 'test*runtime*.py', '-v'], check=True, timeout=10)
    for name in ('c01_runtime.py', 'c01_runtime_groups.py', 'runtime_owner.py'):
        shutil.copyfile(repo / 'scripts' / name, CODE / name)
        (CODE / name).chmod(0o644)
    for path in (BACKING, RUN, DATA):
        owned(path)
    owned(Path('/run/user/61003'), OPERATOR)
    for path in (BACKING / 'Work', BACKING / 'Work/explicit ; literal',
                 BACKING / 'Work/literal##dir', BACKING / 'Work/literal#{pane_id}###'):
        owned(path)
        (path / 'preserved').write_text('preserved-work-inode\n')
        os.chown(path / 'preserved', UID, UID)
    (BACKING / 'root-alias').symlink_to('/', target_is_directory=True)
    HOME.mkdir(parents=True)
    for path in (Path('/run/archipelago-runtime'), Path('/run/archipelago-managed'),
                 Path('/run/archipelago-c01'), Path('/run/systemd/system')):
        path.mkdir(parents=True, exist_ok=True); path.chmod(0o755)
    # Root-only backing parents retain synthetic old-UID private file ownership.
    for name in ('manager', 'authority'):
        parent = Path('/var/lib/private-' + name); parent.mkdir(mode=0o700)
        (parent / 'secret').write_text('synthetic-private-state')
        os.chown(parent / 'secret', UID, UID)
    config_dir = Path('/run/runtime-etc'); config_dir.mkdir()
    for name, content in {
        'passwd': f'root:x:0:0:root:/root:/bin/sh\nruntime-fixture:x:{UID}:{UID}::/home/runtime-fixture:/usr/sbin/nologin\n',
        'group': f'root:x:0:\nruntime-fixture:x:{UID}:\n',
        'subuid': 'runtime-fixture:2000000:65536\n',
        'subgid': 'runtime-fixture:2000000:65536\n',
    }.items():
        source = config_dir / name; source.write_text(content); source.chmod(0o644)
        mount(source, '/etc/' + name)
    # Minimal private /etc, synthesized rather than copying host configuration.
    etc = Path('/run/etc-view'); etc.mkdir()
    for name in ('passwd', 'group', 'subuid', 'subgid'):
        shutil.copyfile(config_dir / name, etc / name)
    (etc / 'machine-id').write_text('684f9c0509564804bc9ad0816083f71d\n')
    (etc / 'nsswitch.conf').write_text('passwd: files\ngroup: files\nhosts: files\n')
    (etc / 'hosts').write_text('127.0.0.1 localhost\n')
    (etc / 'resolv.conf').touch()
    (etc / 'os-release').symlink_to('/usr/lib/os-release')
    (etc / 'dbus-1').mkdir()
    # --session loads /usr/share/dbus-1/session.conf itself; that distro file
    # optionally includes /etc/dbus-1/session.conf. Do not create a recursive alias.
    (etc / 'dbus-1/session.d').mkdir()
    (etc / 'containers').mkdir()
    (etc / 'containers/policy.json').write_text('{"default":[{"type":"insecureAcceptAnything"}]}')
    (etc / 'archipelago-c01').mkdir()
    mount(etc, '/etc')
    value = {'version': 1, 'runtime_user': 'runtime-fixture', 'runtime_uid': UID, 'runtime_gid': UID,
             'manager_uid': MANAGER, 'manager_gid': MANAGER, 'authority_uid': AUTHORITY, 'authority_gid': AUTHORITY,
             'operator_uid': OPERATOR, 'operator_gid': OPERATOR, 'home': str(HOME), 'home_backing': str(BACKING),
             'runtime_dir': str(RUN), 'runtime_data_dir': str(DATA), 'graphroot': str(DATA / 'storage'),
             'subuids': [[2000000, 65536]], 'subgids': [[2000000, 65536]],
             'user_cgroup': str(user_group).removeprefix('/sys/fs/cgroup'),
             'private_parents': ['/var/lib/private-manager', '/var/lib/private-authority']}
    runtime.CONFIG = Path('/etc/archipelago-c01/runtime.json')
    runtime.CONFIG.write_text(json.dumps(value)); runtime.CONFIG.chmod(0o600)
    shutil.copyfile(repo / 'image-recipe/configs/archipelago-runtime-tmux.conf',
                    '/etc/archipelago-c01/runtime-tmux.conf')
    containers = BACKING / '.config/containers'; containers.mkdir(parents=True)
    (containers / 'storage.conf').write_text('[storage]\ndriver="vfs"\ngraphroot="' + str(DATA / 'storage') +
                                           '"\nrunroot="' + str(RUN / 'containers') + '"\n')
    for path in (BACKING / '.config', containers, containers / 'storage.conf'):
        os.chown(path, UID, UID)
    UNITS.mkdir(); UNITS.chmod(0o755)
    for name in ('archipelago-runtime-worker.service', 'archipelago-runtime-tmux.service', 'archipelago-runtime-podman.service'):
        text = (repo / 'image-recipe/configs' / name).read_text().replace('/usr/lib/archipelago/', str(CODE) + '/')
        text = text.replace('[Service]', '[Service]\nStandardOutput=append:/var/lib/runtime-data/' + name + '.log\nStandardError=inherit')
        (UNITS / name).write_text(text)
    for name in ('basic', 'sockets', 'paths', 'timers', 'shutdown', 'exit'):
        (UNITS / (name + '.target')).write_text('[Unit]\nDescription=Private fixture ' + name + '\nDefaultDependencies=no\n')
    (UNITS / 'default.target').write_text('[Unit]\nRequires=dbus.socket archipelago-runtime-tmux.service\n')
    # User bus: distro dbus-daemon with the distro session policy, private socket.
    (UNITS / 'dbus.socket').write_text('[Socket]\nListenStream=%t/bus\nSocketMode=0600\n')
    (UNITS / 'dbus.service').write_text('[Unit]\nRequires=dbus.socket\n[Service]\nType=notify\n'
        'StandardOutput=append:/var/lib/runtime-data/dbus.log\nStandardError=inherit\n'
        'ExecStart=/usr/bin/dbus-daemon --session --address=systemd: --nofork --nopidfile --systemd-activation\n')
    log = open('/run/fixture-service.log', 'wb', buffering=0)
    # Actual dependency refusal, not a saved qualification flag or mocked answer.
    refused = subprocess.run(['/usr/bin/python3', '-I', '-S', str(CODE / 'c01_runtime.py'), 'broker'],
                             stdout=log, stderr=log, env={'PATH': '/usr/bin:/bin'}, timeout=5)
    require(refused.returncode != 0 and not (runtime.PUBLIC / 'runtime.sock').exists(),
            'missing access owner published a command listener')
    def rejects():
        try:
            runtime.load_config()
        except (ValueError, KeyError):
            return
        raise AssertionError('invalid runtime prerequisite accepted')
    original = runtime.CONFIG.read_text()
    for key, wrong in (('runtime_uid', MANAGER), ('subuids', [[2000001, 65536]]), ('graphroot', '/var/lib/private-manager/storage')):
        runtime.CONFIG.write_text(json.dumps({**value, key: wrong})); rejects()
    runtime.CONFIG.write_text(original)
    group_file = Path('/etc/group'); group_original = group_file.read_text()
    group_file.write_text(group_original + 'sudo:x:27:runtime-fixture\n'); rejects(); group_file.write_text(group_original)
    (BACKING / '.ssh').mkdir(); rejects(); (BACKING / '.ssh').rmdir()
    runtime.load_config()
    print('PASS fail-closed missing live access owner, overlapping UID/private root, changed subuids, supplementary privilege, credential home', flush=True)
    access = runtime.listen(runtime.ACCESS, 0, socket.SOCK_SEQPACKET)
    broker = subprocess.Popen(['/usr/bin/python3', '-I', '-S', str(CODE / 'c01_runtime.py'), 'broker'],
                              stdout=log, stderr=log, env={'PATH': '/usr/bin:/bin'})
    require(select.select([access], [], [], 5)[0], 'broker never requested live lease')
    lease, _ = access.accept(); lease.settimeout(2)
    request, _ = runtime.receive(lease); runtime.send(lease, request)
    wait_for(lambda: (runtime.BASE / 'grant.sock').exists())
    user = os.fork()
    if user == 0:
        try:
            (user_group / 'cgroup.procs').write_text(str(os.getpid()))
            require(LIBC.unshare(0x00020000) == 0, 'user private view unavailable')
            mount(None, '/', flags=16384 | (1 << 18))
            rendered_home_view(repo)
            for fd in (1, 2):
                os.dup2(log.fileno(), fd)
            drop(UID)
            os.chdir(HOME)
            os.execve('/usr/lib/systemd/systemd', ['systemd', '--user'],
                {'HOME': str(HOME), 'USER': 'runtime-fixture', 'LOGNAME': 'runtime-fixture',
                 'PATH': '/usr/bin:/bin', 'XDG_RUNTIME_DIR': str(RUN), 'DBUS_SESSION_BUS_ADDRESS': 'unix:path=' + str(RUN / 'bus'),
                 'SYSTEMD_UNIT_PATH': str(UNITS), 'SYSTEMD_LOG_TARGET': 'console'})
        except BaseException:
            import traceback; traceback.print_exc(); os._exit(111)
    watch = None
    try:
        user_environment = {'HOME': str(HOME), 'XDG_RUNTIME_DIR': str(RUN),
                            'DBUS_SESSION_BUS_ADDRESS': 'unix:path=' + str(RUN / 'bus'), 'PATH': '/usr/bin:/bin'}
        def user_control(*args):
            return subprocess.run(['/usr/bin/systemctl', '--user', *args], env=user_environment,
                                  preexec_fn=lambda: drop(UID), capture_output=True, timeout=10)
        wait_for(lambda: (RUN / 'systemd/private').exists())
        bus = user_control('start', 'dbus.service')
        require(bus.returncode == 0, 'bus preflight: ' + bus.stderr.decode())
        bus = subprocess.run(['/usr/bin/busctl', '--user', '--no-pager', 'list'], env=user_environment,
                             preexec_fn=lambda: drop(UID), capture_output=True, timeout=10)
        require(bus.returncode == 0 and b'org.freedesktop.systemd1' in bus.stdout,
                'user bus systemd name unavailable: ' + bus.stderr.decode())
        check = user_control('is-active', 'archipelago-runtime-tmux.service')
        require(check.returncode == 0, 'tmux preflight: ' + check.stdout.decode() + check.stderr.decode())
        wait_for(lambda: (Path('/tmp') / f'tmux-{UID}' / 'default').is_socket(), 5)
        print('PASS staged private user-manager/tmux startup preflight', flush=True)
        podman = user_control('--no-block', 'start', 'archipelago-runtime-podman.service')
        require(podman.returncode == 0, podman.stderr.decode())
        wait_for(lambda: user_control('is-active', 'archipelago-runtime-podman.service').returncode == 0, 35)
        print('PASS independent USER-unit Podman bootstrap preflight', flush=True)
        started = user_control('--no-block', 'start', 'archipelago-runtime-worker.service')
        require(started.returncode == 0, started.stderr.decode())
        def ready():
            nonlocal watch
            require(broker.poll() is None, 'broker exited: ' + Path('/run/fixture-service.log').read_text())
            channel = runtime.root_connect(runtime.PUBLIC / 'readiness.sock')
            try:
                result, rights = runtime.receive(channel, 3)
                require(result['version'] == 2, 'runtime detached-mount ABI mismatch')
                for fd in rights:
                    os.close(fd)
                watch = channel
                return result
            except (ValueError, ConnectionResetError):
                channel.close(); return None
        identity = wait_for(ready, 40)
        print('PASS live root broker/user-worker readiness:', json.dumps(identity), flush=True)
        info = json.loads(command(['/usr/bin/python3', '-c',
            'import os,json;print(json.dumps([os.getuid(),os.getgid(),os.getgroups(),os.environ["HOME"],os.environ["XDG_RUNTIME_DIR"]]))']))
        require(info == [UID, UID, [], str(HOME), str(RUN)], str(info))
        runtime_meta = json.loads(command(['/usr/bin/python3', '-c',
            'import os,json;from pathlib import Path;p=Path(os.environ["XDG_RUNTIME_DIR"]);s=p.stat();'
            'assert not Path("/run/user/61003").exists();print(json.dumps([s.st_dev,s.st_ino,s.st_uid,s.st_mode & 511]))']))
        meta = RUN.stat()
        require(runtime_meta == [meta.st_dev, meta.st_ino, UID, 0o700], 'runtime directory identity changed under ProtectHome cover')
        print('PASS rendered home/runtime bindings over /home /root /run/user covers preserve exact runtime inode, owner, mode and real bus', flush=True)
        require(client(['/usr/bin/true'], uid=MANAGER, gid=AUTHORITY)[0] is None, 'foreign GID accepted')
        require(client(['/usr/bin/true'], uid=AUTHORITY, gid=MANAGER)[0] is None, 'foreign UID accepted')
        command(['/usr/bin/python3', '-c', 'from pathlib import Path\n'
                 'for p in ["/var/lib/private-manager/secret","/var/lib/private-authority/secret"]:\n'
                 ' try:Path(p).read_text()\n except PermissionError:continue\n raise SystemExit(1)'])
        print('PASS actual permanent identity, exact peer UID/GID, private-state denial', flush=True)
        fds = len(list(Path('/proc/self/fd').iterdir()))
        try:
            RuntimeLease(MANAGER, MANAGER, UID, UID, '0' * 64)
        except ValueError:
            pass
        else:
            raise AssertionError('stale configuration accepted')
        require(len(list(Path('/proc/self/fd').iterdir())) == fds, 'refused readiness leaked capabilities')
        view = RuntimeLease(MANAGER, MANAGER, UID, UID, identity['configuration_sha256'])
        child = os.fork()
        if child == 0:
            try:
                require(LIBC.unshare(0x00020000) == 0, 'private manager view unavailable')
                mount(None, '/', flags=16384 | (1 << 18))
                mount('tmpfs', '/run', 'tmpfs', 0, 'mode=755,size=16M')
                view.install()
                require(command(['/usr/bin/id', '-u']) == str(UID), 'mounted static proxy delegation failed')
                os._exit(0)
            except BaseException:
                import traceback; traceback.print_exc(); os._exit(111)
        _, status = os.waitpid(child, 0)
        require(os.waitstatus_to_exitcode(status) == 0, 'detached socket view failed')
        view.close()
        print('PASS exact socket capability survives covered /run via detached mount; actual static proxy executes', flush=True)
        escaped = ('import os,time,json\nfrom pathlib import Path\n'
                   'p=os.fork()\n'
                   'if p==0:\n os.setsid();Path("/var/lib/runtime-data/descendant.json").write_text(json.dumps([os.getpid()]));time.sleep(120)\n'
                   'else:time.sleep(120)')
        require(client(['/usr/bin/python3', '-c', escaped], disconnect=True) == b'cancelled')
        descendant = json.loads((DATA / 'descendant.json').read_text())[0]
        wait_for(lambda: not Path(f'/proc/{descendant}').exists(), 5)
        wait_for(lambda: not list(Path(identity['cgroup']).parent.glob('job-*')), 5)
        print('PASS disconnected job retires escaped setsid descendant and cgroup', flush=True)
        (DATA / 'descendant.json').unlink()
        command(['/usr/bin/python3', '-c',
                 'import os,time,json\nfrom pathlib import Path\nr,w=os.pipe();p=os.fork()\n'
                 'if p==0:\n os.close(r);os.setsid();Path("/var/lib/runtime-data/descendant.json").write_text(json.dumps([os.getpid()]));os.write(w,b"x");time.sleep(120)\n'
                 'else:\n os.close(w);os.read(r,1);os._exit(0)'])
        descendant = json.loads((DATA / 'descendant.json').read_text())[0]
        wait_for(lambda: not Path(f'/proc/{descendant}').exists(), 5)
        wait_for(lambda: not list(Path(identity['cgroup']).parent.glob('job-*')), 5)
        print('PASS normal leader exit also retires escaped descendant before completion', flush=True)
        # Keep one leader alive while it repeatedly double-forks exited helpers.
        # Each wave must disappear BEFORE releasing the leader; unrelated work
        # must still be admitted. A killed leader cannot make this case pass.
        producer = asynchronous_command(
            'import os,time,json\nfrom pathlib import Path\n'
            'root=Path("/var/lib/runtime-data")\n'
            'root.joinpath("producer-pid").write_text(str(os.getpid()))\n'
            'for wave in range(4):\n'
            ' pids=[]\n'
            ' for i in range(32):\n'
            '  r,w=os.pipe(); helper=os.fork()\n'
            '  if helper==0:\n'
            '   os.close(r); orphan=os.fork()\n'
            '   if orphan==0: os.setsid();os._exit(0)\n'
            '   os.write(w,str(orphan).encode());os._exit(0)\n'
            '  os.close(w);pids.append(int(os.read(r,32)));os.close(r);os.waitpid(helper,0)\n'
            ' tmp=root/"wave-pending";tmp.write_text(json.dumps(pids));tmp.rename(root/("wave-"+str(wave)))\n'
            ' while not root.joinpath("release-"+str(wave)).exists(): time.sleep(.02)\n'
            'while not root.joinpath("producer-done").exists(): time.sleep(.02)')
        try:
            for wave in range(4):
                path = DATA / ('wave-' + str(wave))
                wait_for(path.exists, 5)
                pids = json.loads(path.read_text())
                wait_for(lambda: all(not Path(f'/proc/{pid}').exists() for pid in pids), 5)
                require(producer.poll() is None, 'producer exited before live-leader reaping assertion')
                require(Path('/proc/' + (DATA / 'producer-pid').read_text()).exists())
                command(['/usr/bin/true'])
                (DATA / ('release-' + str(wave))).touch()
            # Nested empty cgroups must retire without cancelling that leader.
            command(['/usr/bin/python3', '-c',
                'from pathlib import Path\n'
                'root=Path("/sys/fs/cgroup")/Path("/proc/self/cgroup").read_text().strip()[4:]\n'
                'for i in range(80): (root/str(i)/"nested").mkdir(parents=True)'])
            require(producer.poll() is None, 'nested retirement cancelled unrelated producer')
            # A genuinely unreadable descendant retains only its own slot until
            # the fixture repairs permissions; successful response is withheld.
            blocked = asynchronous_command(
                'from pathlib import Path\n'
                'root=Path("/sys/fs/cgroup")/Path("/proc/self/cgroup").read_text().strip()[4:]\n'
                'child=root/"unreadable";child.mkdir();child.chmod(0)\n'
                'tmp=Path("/var/lib/runtime-data/blocked-pending");tmp.write_text(str(child));tmp.rename(tmp.with_name("blocked-group"))')
            locked = None
            try:
                wait_for(lambda: (DATA / 'blocked-group').exists(), 5)
                locked = Path((DATA / 'blocked-group').read_text())
                time.sleep(.2)
                require(blocked.poll() is None and locked.exists(), 'failed cleanup released ownership')
                command(['/usr/bin/true'])
                require(producer.poll() is None, 'cleanup failure cancelled unrelated producer')
                locked.chmod(0o755)
                require(blocked.wait(timeout=10) == 0, 'retained cleanup did not recover')
                require(not locked.parent.exists(), 'nested invocation root remained')
            finally:
                if locked is not None and locked.exists():
                    locked.chmod(0o755)
                if blocked.poll() is None:
                    blocked.terminate(); blocked.wait(timeout=5)
            (DATA / 'producer-done').touch()
            require(producer.wait(timeout=10) == 0)
            wait_for(lambda: not list(Path(identity['cgroup']).parent.glob('job-*')), 5)
        finally:
            if producer.poll() is None:
                producer.terminate(); producer.wait(timeout=5)
        print('PASS live-leader orphan waves, nested empty cgroups, retained unreadable cleanup and concurrent admission', flush=True)
        native_terminal_contract(identity)
        command(['/usr/bin/tmux', 'new-session', '-d', '-s', 'fixture', '/bin/sh'])
        command(['/usr/bin/tmux', 'send-keys', '-t', 'fixture', 'printf "runtime-terminal-ok\\n"', 'Enter'])
        command(['/usr/bin/tmux', 'resize-window', '-t', 'fixture', '-x', '90', '-y', '30'])
        wait_for(lambda: 'runtime-terminal-ok' in command(['/usr/bin/tmux', 'capture-pane', '-p', '-t', 'fixture']))
        print('PASS actual persistent tmux new-session/send-keys/resize/capture', flush=True)
        command(['/usr/bin/python3', '-c', 'import tempfile\n'
                 'for p in ["/tmp","/var/tmp"]:\n'
                 ' with tempfile.TemporaryFile(dir=p) as f:f.write(b"private-temp-probe")'])
        command(['/usr/bin/podman', 'load', '-i', '/run/public.oci.tar'])
        image = command(['/usr/bin/podman', 'images', '--format', '{{.ID}}']).splitlines()[0]
        cid = command(['/usr/bin/podman', 'run', '-d', '--name', 'runtime-fixture', '--network=none',
                       '--entrypoint=/bin/sh', image, '-c', 'while :; do sleep 1; done'])
        require(command(['/usr/bin/podman', 'exec', cid, '/bin/sh', '-c', 'printf actual-podman-exec']) == 'actual-podman-exec')
        before = command(['/usr/bin/podman', 'inspect', '--format', '{{.Id}} {{.State.Pid}} {{.State.StartedAt}}', cid])
        # Real manager surrogate service, separate from all runtime user groups.
        manager_group = root / 'manager-surrogate'; manager_group.mkdir()
        for _ in range(2):
            manager = os.fork()
            if manager == 0:
                (manager_group / 'cgroup.procs').write_text(str(os.getpid()))
                drop(MANAGER); time.sleep(60); os._exit(0)
            wait_for(lambda: (manager_group / 'cgroup.procs').read_text().strip())
            (manager_group / 'cgroup.kill').write_text('1'); os.waitpid(manager, 0)
        manager_group.rmdir()
        require(command(['/usr/bin/podman', 'inspect', '--format', '{{.Id}} {{.State.Pid}} {{.State.StartedAt}}', cid]) == before)
        require(command(['/usr/bin/tmux', 'has-session', '-t', 'fixture']) == '')
        require(command(['/usr/bin/podman', 'exec', cid, '/bin/true']) == '')
        print('PASS actual Podman USER-unit load/run/exec; container and tmux survive manager cgroup stop', flush=True)
        live = RuntimeLease(MANAGER, MANAGER, UID, UID, identity['configuration_sha256'])
        stop_case = os.environ.get('ARCHY_RUNTIME_STOP', 'revoke')
        if stop_case == 'stall':
            os.kill(broker.pid, signal.SIGSTOP)
            time.sleep(3.3)
            try:
                live.check()
            except ValueError:
                pass
            else:
                raise AssertionError('stalled root observer retained readiness')
            finally:
                os.kill(broker.pid, signal.SIGCONT)
        else:
            lease.close()
        wait_for(lambda: broker.poll() is not None, 5)
        def revoked():
            for _ in range(128):
                if not select.select([watch], [], [], 0)[0]:
                    return False
                if watch.recv(16384) == b'':
                    return True
            return False
        wait_for(revoked, 5)
        try:
            live.check()
        except ValueError:
            pass
        else:
            raise AssertionError('revoked root lease remained valid')
        live.close()
        wait_for(lambda: not Path(f'/proc/{identity["pid"]}').exists(), 5)
        def persistent_command(argv):
            # Actual transient USER unit; never system-unit --uid Podman exec.
            result = subprocess.run(['/usr/bin/systemd-run', '--user', '--quiet', '--wait', '--pipe', '--collect', *argv],
                                    env=user_environment, preexec_fn=lambda: drop(UID), capture_output=True, timeout=15)
            require(result.returncode == 0, result.stderr.decode())
            return result.stdout.decode().strip()
        require(persistent_command(['/usr/bin/podman', 'inspect', '--format', '{{.Id}} {{.State.Pid}} {{.State.StartedAt}}', cid]) == before)
        persistent_command(['/usr/bin/podman', 'exec', cid, '/bin/true'])
        persistent_command(['/usr/bin/tmux', 'has-session', '-t', 'fixture'])
        persistent_command(['/usr/bin/podman', 'rm', '-f', cid])
        persistent_command(['/usr/bin/tmux', 'kill-session', '-t', 'fixture'])
        print('PASS ' + stop_case + ' revokes broker/readiness/worker; persistent container and tmux survive worker exit too', flush=True)
    finally:
        for path in DATA.glob('*.log'):
            print(path.name + ':', path.read_text(), flush=True)
        for path in user_group.rglob('cgroup.procs'):
            for pid in path.read_text().split():
                try:
                    print('OWNED TASK', pid, path, Path('/proc/' + pid + '/cmdline').read_bytes(),
                          Path('/proc/' + pid + '/wchan').read_text(), flush=True)
                except FileNotFoundError:
                    pass
        if watch is not None:
            watch.close()
        if broker.poll() is None:
            broker.terminate(); broker.wait(timeout=5)
        # Fixture-owned hierarchy only, AFTER workload lifecycle assertions.
        (user_group / 'cgroup.kill').write_text('1')
        os.waitpid(user, 0)
        print('SERVICE LOG:', Path('/run/fixture-service.log').read_text(), flush=True)


if __name__ == '__main__':
    main()