Capture redacted guest serial while waiting for shutdown

bin/hydra

@@ -0,0 +1,9 @@
+#!/usr/bin/env python3
+"""Run HydraOS from this checkout without installing a host package."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from hydra.cli import main
+
+raise SystemExit(main())

bin/hydra-install

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Source checkout entry point; installed copies use the installed Python package."""
+from pathlib import Path
+import sys
+
+checkout = Path(__file__).resolve().parent.parent
+if (checkout / 'hydra/installer.py').is_file():
+    sys.path.insert(0, str(checkout))
+from hydra.installer import main
+
+raise SystemExit(main())

dev/iso-smoke.py

@@ -0,0 +1,911 @@
+#!/usr/bin/env python3
+"""Boot/install/reboot acceptance on a fresh, isolated UEFI/KVM VM only.
+
+No existing disk option, host sudo, SSH, host TTY, or GUI is used. Artifacts and
+logs remain in a new private scratch directory even after failure.
+"""
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import inspect
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import select
+import shlex
+import shutil
+import socket
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from hydra.qmp import QMPClient, QMPError
+
+
+class SmokeError(RuntimeError):
+    pass
+
+
+def lifecycle_socket(path, *, connect_socket=True):
+    path = Path(path)
+    info = path.lstat()
+    if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.getuid():
+        raise SmokeError('Lifecycle socket is not owned by the installed user')
+    identity = [info.st_dev, info.st_ino, info.st_uid, info.st_mode]
+    if connect_socket:
+        with socket.socket(socket.AF_UNIX) as connection:
+            connection.settimeout(2)
+            connection.connect(str(path))
+        after = path.lstat()
+        if identity != [after.st_dev, after.st_ino, after.st_uid, after.st_mode]:
+            raise SmokeError('Lifecycle socket changed while connecting')
+    return identity
+
+
+def lifecycle_snapshot(context):
+    """Prove live identity and owned runtime endpoints, never trust PID alone."""
+    try:
+        if context['uid'] != os.getuid():
+            raise SmokeError('Lifecycle probe is not the installed normal user')
+        runtime = Path(context['runtime'])
+        run = runtime / context['name']
+        directories = {}
+        for directory in (runtime, run):
+            info = directory.lstat()
+            if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid()
+                    or stat.S_IMODE(info.st_mode) != 0o700):
+                raise SmokeError('Lifecycle runtime must remain private and owned')
+            directories[str(directory)] = [info.st_dev, info.st_ino, info.st_mode]
+        metadata = json.loads((run / 'identity.json').read_text())
+        if metadata != {'name': context['name'], 'uuid': context['uuid']}:
+            raise SmokeError('Lifecycle metadata is stale or foreign')
+        pid = int((run / 'qemu.pid').read_text())
+        process = Path('/proc') / str(pid)
+        if process.stat().st_uid != os.getuid():
+            raise SmokeError('Lifecycle process belongs to another user')
+        fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
+        if fields[0] == 'Z':
+            raise SmokeError('Lifecycle process has exited')
+        argv = (process / 'cmdline').read_bytes().split(b'\0')
+        for option, expected in ((b'-name', context['name']), (b'-uuid', context['uuid'])):
+            if argv[argv.index(option) + 1] != expected.encode():
+                raise SmokeError('Lifecycle PID is not the owned VM identity')
+        endpoints = {'qmp': lifecycle_socket(run / 'qmp.sock', connect_socket=False)}
+        with QMPClient(run / 'qmp.sock', timeout=2) as client:
+            if (client.command('query-name') != {'name': context['name']}
+                    or client.command('query-uuid') != {'UUID': context['uuid']}):
+                raise SmokeError('Lifecycle QMP identity is foreign')
+            if client.command('query-status').get('status') != 'running':
+                raise SmokeError('Lifecycle VM is not running')
+            if not client.command('query-kvm').get('enabled'):
+                raise SmokeError('Lifecycle VM did not initialise nested KVM')
+        endpoints['spice'] = lifecycle_socket(run / 'spice.sock')
+        audio = Path(context['user_runtime'])
+        endpoints['pipewire'] = lifecycle_socket(audio / 'pipewire-0')
+        endpoints['pulse'] = lifecycle_socket(audio / 'pulse/native')
+        services = {}
+        for service in context.get('audio_services', []):
+            subprocess.run(['systemctl', '--user', 'is-active', service], check=True,
+                           capture_output=True, timeout=5)
+            services[service] = subprocess.run(['systemctl', '--user', 'show', service,
+                '-p', 'InvocationID', '-p', 'ExecMainPID'], check=True,
+                capture_output=True, text=True, timeout=5).stdout
+        return {'name': context['name'], 'uuid': context['uuid'], 'pid': pid,
+                'process_start': fields[19], 'directories': directories,
+                'sockets': endpoints, 'audio_services': services}
+    except (OSError, ValueError, KeyError, IndexError, QMPError, subprocess.SubprocessError) as exc:
+        raise SmokeError('Lifecycle runtime or process was lost: ' + str(exc)) from exc
+
+
+def verify_lifecycle_survival(before, after):
+    if before != after:
+        raise SmokeError('VM identity, private runtime, or audio endpoints changed across logout')
+
+
+def login_session_records(uid):
+    """Read logind's public properties, avoiding display-table row semantics."""
+    def query(*args):
+        return subprocess.run(['loginctl', *args], check=True, capture_output=True,
+                              text=True, timeout=5).stdout
+    try:
+        ids = query('show-user', str(uid), '-p', 'Sessions', '--value').split()
+        records = []
+        for ident in ids:
+            output = query('show-session', ident, '--all', '--no-pager',
+                           '-p', 'Id', '-p', 'User', '-p', 'State', '-p', 'Class',
+                           '-p', 'Leader', '-p', 'TTY')
+            record = dict(line.split('=', 1) for line in output.splitlines())
+            if not {'Id', 'User', 'State', 'Class', 'Leader', 'TTY'} <= record.keys():
+                raise SmokeError('Incomplete logind session identity')
+            record['User'], record['Leader'] = int(record['User']), int(record['Leader'])
+            if record['Id'] != ident or record['User'] != uid:
+                raise SmokeError('Logind returned a foreign session identity')
+            records.append(record)
+        return records
+    except (OSError, ValueError, subprocess.SubprocessError) as exc:
+        raise SmokeError('Cannot establish guest login session state: ' + str(exc)) from exc
+
+
+def is_open_login(record):
+    # systemd v258: closing session scopes can remain while user processes
+    # survive. Manager-class records describe bookkeeping, not a PAM login.
+    users = {'user', 'user-early', 'user-incomplete', 'user-light', 'user-early-light'}
+    bookkeeping = {'manager', 'manager-early', 'background', 'background-light',
+                   'greeter', 'lock-screen', 'none'}
+    if record['State'] not in {'opening', 'online', 'active', 'closing'}:
+        raise SmokeError('Unknown logind session state')
+    if record['Class'] not in users | bookkeeping:
+        raise SmokeError('Unknown logind session class')
+    return record['Class'] in users and record['State'] != 'closing'
+
+
+def require_initial_login(uid, ident):
+    opened = [record for record in login_session_records(uid) if is_open_login(record)]
+    if (len(opened) != 1 or opened[0]['Id'] != ident or opened[0]['TTY'] != 'ttyS0'
+            or opened[0]['Leader'] <= 0 or opened[0]['State'] not in {'online', 'active'}):
+        raise SmokeError('Lifecycle test requires exactly its actual serial PAM login')
+    return opened[0]
+
+
+def no_open_logins(context):
+    original = context['login_session']
+    records = login_session_records(context['uid'])
+    for record in records:
+        if record['Id'] == original['Id']:
+            if (record['User'] != original['User'] or record['TTY'] != original['TTY']
+                    or record['Class'] != original['Class']
+                    or record['Leader'] not in {0, original['Leader']}):
+                raise SmokeError('Original PAM session identity was replaced')
+            if record['State'] != 'closing':
+                return False
+        if is_open_login(record):
+            return False
+    return True
+
+
+def capture_logout_snapshot(context):
+    if not no_open_logins(context):
+        raise SmokeError('An open login overlaps the logout sample')
+    during = lifecycle_snapshot(context)
+    if not no_open_logins(context):
+        raise SmokeError('A login opened while the logout sample was running')
+    verify_lifecycle_survival(context['before'], during)
+    return during
+
+
+LIFECYCLE_BODY = r'''
+def run(*args):
+    return subprocess.run(args, check=True, capture_output=True, text=True, timeout=40).stdout
+
+def cli(context, *args):
+    return json.loads(run('hydra', '--config', context['config'], '--state', context['state'],
+                          '--runtime', context['runtime'], *args))
+
+def write_report(path, data):
+    temporary = path.with_suffix('.tmp')
+    temporary.write_text(json.dumps(data, indent=2) + '\n')
+    temporary.chmod(0o600)
+    temporary.replace(path)
+
+mode, base_arg = sys.argv[1:]
+base = Path(base_arg)
+context_path = base / 'context.json'
+observation_path = base / 'logout-observation.json'
+if mode == 'start':
+    if not Path('/dev/kvm').exists():
+        print('SKIP: installed guest nested KVM unavailable')
+        raise SystemExit(0)
+    if os.getuid() == 0 or not os.access('/dev/kvm', os.R_OK | os.W_OK):
+        raise SmokeError('Lifecycle test needs installed-user nested KVM access')
+    user_runtime = Path(os.environ['XDG_RUNTIME_DIR'])
+    if user_runtime != Path('/run/user') / str(os.getuid()):
+        raise SmokeError('Lifecycle test needs the real login runtime directory')
+    if run('loginctl', 'show-user', str(os.getuid()), '-p', 'Linger', '--value').strip() != 'no':
+        raise SmokeError('Lingering would mask the installed logout policy')
+    login_session = require_initial_login(os.getuid(), os.environ['XDG_SESSION_ID'])
+    name = 'logout-' + base.name.rsplit('-', 1)[1]
+    state = base / 'state'
+    runtime = user_runtime / base.name
+    context = {'uid': os.getuid(), 'name': name, 'config': str(base / 'config'),
+               'state': str(state), 'runtime': str(runtime), 'user_runtime': str(user_runtime),
+               'uuid': str(uuid.uuid5(uuid.NAMESPACE_URL, str(state / name / 'disk.qcow2'))),
+               'login_session': login_session,
+               'audio_services': ['pipewire.service', 'pipewire-pulse.service', 'wireplumber.service']}
+    run('systemctl', '--user', 'start', *context['audio_services'])
+    cli(context, 'add', name, '--memory', '256', '--cpus', '1', '--disk-size', '64M',
+        '--graphics', 'software', '--network', 'bridge', '--bridge', 'hydrabr0')
+    if cli(context, 'start', name).get('status') != 'running':
+        raise SmokeError('Real installed hydra CLI failed to start lifecycle VM')
+    try:
+        disk = Path(context['state']) / name / 'disk.qcow2'
+        info = disk.lstat()
+        if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid():
+            raise SmokeError('Lifecycle VM disk is not a new owned regular qcow2')
+        context['before'] = lifecycle_snapshot(context)
+    except Exception:
+        # CLI control is scoped to the newly created private identity.
+        cli(context, 'stop', name, '--force')
+        raise
+    write_report(context_path, context)
+    with open(base / 'monitor.log', 'ab', buffering=0) as log:
+        subprocess.Popen([sys.executable, __file__, 'monitor', str(base)],
+                         stdin=subprocess.DEVNULL, stdout=log, stderr=log,
+                         start_new_session=True)
+    print('READY: normal-user Hydra VM and audio sockets recorded before real logout')
+elif mode == 'monitor':
+    context = json.loads(context_path.read_text())
+    try:
+        deadline = time.monotonic() + 35
+        while not no_open_logins(context):
+            if time.monotonic() >= deadline:
+                raise SmokeError('Serial session did not actually log out')
+            time.sleep(.2)
+        # Exceed systemd's usual 10-second user-manager stop delay. No helper
+        # login or lingering is allowed to keep the tested user session alive.
+        time.sleep(15)
+        during = capture_logout_snapshot(context)
+        write_report(observation_path, {'status': 'passed', 'no_open_logins': True,
+                     'survival_seconds': 15, 'during_logout': during})
+    except Exception as exc:
+        write_report(observation_path, {'status': 'failed', 'error': str(exc)})
+        raise
+elif mode == 'finish':
+    context = json.loads(context_path.read_text())
+    try:
+        if os.environ.get('XDG_RUNTIME_DIR') != context['user_runtime']:
+            raise SmokeError('Relogin replaced the user runtime path')
+        deadline = time.monotonic() + 10
+        while not observation_path.is_file():
+            if time.monotonic() >= deadline:
+                raise SmokeError('Logout monitor died or did not record survival')
+            time.sleep(.2)
+        observation = json.loads(observation_path.read_text())
+        if observation.get('status') != 'passed' or observation.get('no_open_logins') is not True:
+            raise SmokeError('No-open-login observation failed: ' + json.dumps(observation))
+        after = lifecycle_snapshot(context)
+        verify_lifecycle_survival(context['before'], after)
+        if cli(context, 'status', context['name']).get('status') != 'running':
+            raise SmokeError('Real installed hydra CLI could not reconnect after login')
+        print(json.dumps({'status': 'passed', 'real_logout': True, 'no_open_logins': True,
+              'survival_seconds': observation['survival_seconds'], 'same_vm': after,
+              'audio_scope': 'Socket and service continuity only; signal/playback untested'}))
+    finally:
+        # Real product CLI verifies metadata and live QMP name/UUID before quit.
+        # Never signal a saved PID or control a VM outside these private paths.
+        cleanup_status = cli(context, 'status', context['name']).get('status')
+        if cleanup_status == 'running':
+            run_path = Path(context['runtime']) / context['name']
+            if json.loads((run_path / 'identity.json').read_text()) != {'name': context['name'], 'uuid': context['uuid']}:
+                raise SmokeError('Refusing lifecycle cleanup for a foreign identity')
+            with QMPClient(run_path / 'qmp.sock', timeout=2) as client:
+                if (client.command('query-name') != {'name': context['name']}
+                        or client.command('query-uuid') != {'UUID': context['uuid']}):
+                    raise SmokeError('Refusing lifecycle cleanup for foreign live QMP')
+            if cli(context, 'stop', context['name'], '--force').get('status') != 'stopped':
+                raise SmokeError('Owned lifecycle VM did not stop')
+        elif cleanup_status != 'stopped':
+            raise SmokeError('Cannot safely verify the lifecycle VM for cleanup')
+else:
+    raise SmokeError('Unknown lifecycle mode')
+'''
+
+
+def lifecycle_guest_script():
+    helpers = (lifecycle_socket, lifecycle_snapshot, verify_lifecycle_survival,
+               login_session_records, is_open_login, require_initial_login,
+               no_open_logins, capture_logout_snapshot)
+    return ('import json, os, socket, stat, subprocess, sys, time, uuid\n'
+            'from pathlib import Path\nsys.path.insert(0, "/usr/lib/hydra")\n'
+            'from hydra.qmp import QMPClient, QMPError\n'
+            'class SmokeError(RuntimeError): pass\n' +
+            '\n'.join(inspect.getsource(helper) for helper in helpers) + LIFECYCLE_BODY)
+
+
+def installed_admin(console, password):
+    console.buffer = b''
+    prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
+    console.send(f'sudo -k -p "$(printf %s {prompt_encoded} | base64 -d)" -i\n')
+    console.expect(rb'HYDRA_SUDO_PASSWORD: ', 30)
+    console.send(password + '\n')
+    console.expect(rb'root[^\r\n]*[#$] ', 30)
+
+
+def installed_logout_probe(console, password):
+    # Exit only the guest sudo shell, then work in its actual getty PAM login.
+    console.send('exit\n')
+    console.expect(rb'hydratest[^\r\n]*[#$] ', 30)
+    if console.command('id -un', 10).strip() != b'hydratest':
+        raise SmokeError('Logout probe did not return to installed user login')
+    base = '/home/hydratest/hydra-logout-' + secrets.token_hex(4)
+    script = base + '/probe.py'
+    preparation = (f'umask 077; mkdir {shlex.quote(base)} && cat > {shlex.quote(script)} '
+                   "<<'HYDRA_LIFECYCLE_SOURCE'\n" + lifecycle_guest_script() + '\nHYDRA_LIFECYCLE_SOURCE\n' +
+                   f'python3 {shlex.quote(script)} start {shlex.quote(base)}')
+    ready = console.command(preparation, 120)
+    if ready.startswith(b'SKIP:'):
+        installed_admin(console, password)
+        return ready
+    if b'READY:' not in ready:
+        raise SmokeError('Lifecycle VM did not become ready before logout')
+    console.buffer = b''
+    console.send('exit\n')
+    console.expect(rb'login: ', 30)
+    # This is intentionally a period at the actual login prompt with no
+    # replacement login. The guest monitor independently confirms no open logins.
+    time.sleep(20)
+    console.send('hydratest\n')
+    console.expect(rb'Password: ', 30)
+    console.send(password + '\n')
+    console.expect(rb'hydratest[^\r\n]*[#$] ', 60)
+    result = console.command(f'python3 {shlex.quote(script)} finish {shlex.quote(base)}', 90)
+    installed_admin(console, password)
+    return result
+
+
+def reserve_disk(path):
+    try:
+        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        os.close(fd)
+    except OSError as exc:
+        raise SmokeError('Test disk must be a newly created regular file') from exc
+
+
+def owned_file(path, scratch):
+    path, scratch = Path(path), Path(scratch)
+    try:
+        info = path.lstat()
+    except OSError as exc:
+        raise SmokeError('Missing disposable drive') from exc
+    if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1
+            or path.parent.resolve() != scratch.resolve()
+            or info.st_uid != os.getuid()):
+        raise SmokeError('Drive must be an owned, unlinked regular file directly in scratch')
+
+
+def qemu_path(path):
+    # QEMU keyval arguments use commas as separators. Refuse ambiguous paths.
+    if any(c in str(path) for c in ',\n\r\x00'):
+        raise SmokeError('QEMU file paths cannot contain comma or control characters')
+    return str(path)
+
+
+def qemu_argv(qemu, scratch, disk, code, variables, iso, memory):
+    scratch = Path(scratch)
+    owned_file(disk, scratch)
+    owned_file(variables, scratch)
+    argv = [qemu, '-machine', 'q35,accel=kvm', '-cpu', 'host', '-smp', '2',
+            '-m', str(memory), '-display', 'none', '-monitor', 'none',
+            '-no-reboot', '-nic', 'user,model=virtio-net-pci',
+            '-drive', f'if=pflash,format=raw,readonly=on,file={qemu_path(code)}',
+            '-drive', f'if=pflash,format=raw,file={qemu_path(variables)}',
+            '-drive', f'if=virtio,format=qcow2,file={qemu_path(disk)}',
+            '-chardev', f'socket,id=console,path={qemu_path(scratch / "serial.sock")},server=on,wait=off',
+            '-serial', 'chardev:console', '-qmp', f'unix:{qemu_path(scratch / "qmp.sock")},server=on,wait=off']
+    if iso is not None:
+        if not Path(iso).is_file():
+            raise SmokeError('Trusted ISO must exist as a regular file')
+        argv += ['-drive', f'file={qemu_path(iso)},media=cdrom,format=raw,readonly=on',
+                 '-boot', 'order=d,strict=on']
+    else:
+        argv += ['-boot', 'order=c,strict=on']
+    return argv
+
+
+def frame_command(command, token):
+    if not re.fullmatch(r'[a-zA-Z0-9]+', token):
+        raise SmokeError('Invalid framing token')
+    encoded = base64.b64encode(command.encode()).decode()
+    # Linux canonical serial TTY input is bounded per physical line. Continue
+    # the shell expression across short lines for substantial guest probes.
+    chunks = ' \\\n'.join(shlex.quote(encoded[i:i + 768]) for i in range(0, len(encoded), 768)) or "''"
+    # Literal control bytes appear only in output, never in the echoed command.
+    return (f"printf '\\036{token}:BEGIN\\037'; "
+            f"sh -c \"$(printf %s {chunks} | base64 -d)\"; "
+            f"rc=$?; printf '\\036{token}:END:%s\\037' \"$rc\"")
+
+
+def parse_frame(data, token):
+    begin = b'\x1e' + token.encode() + b':BEGIN\x1f'
+    start = data.find(begin)
+    if start < 0:
+        return None
+    end = re.search(rb'\x1e' + token.encode() + rb':END:(\d+)\x1f', data[start + len(begin):])
+    if end is None:
+        return None
+    output = data[start + len(begin):start + len(begin) + end.start()]
+    return int(end.group(1)), output
+
+
+def installer_command(token):
+    # A literal control-byte return frame cannot be forged by terminal echo.
+    # Keep the installer on this controlling TTY for getpass, then mark every
+    # shell return, including an unexpected status zero without completion.
+    return ('sudo -n hydra-install --disk /dev/vda --user hydratest --execute '
+            '--confirm-erase /dev/vda; rc=$?; '
+            f"printf '\\036{token}:INSTALL_EXIT:%s\\037' \"$rc\"")
+
+
+class Console:
+    def __init__(self, connection, logfile, passwords):
+        self.sock = connection
+        self.log = open(logfile, 'ab', buffering=0)
+        self.passwords = [p.encode() for p in passwords if p]
+        self.pending_log = b''
+        self.buffer = b''
+        self.closed = False
+
+    def record(self, data, final=False):
+        self.pending_log += data
+        for password in self.passwords:
+            self.pending_log = self.pending_log.replace(password, b'[REDACTED]')
+        keep = 0
+        if not final:
+            for password in self.passwords:
+                for length in range(1, len(password)):
+                    if self.pending_log.endswith(password[:length]):
+                        keep = max(keep, length)
+        if keep:
+            self.log.write(self.pending_log[:-keep])
+            self.pending_log = self.pending_log[-keep:]
+        else:
+            self.log.write(self.pending_log)
+            self.pending_log = b''
+
+    def close(self):
+        if not self.closed:
+            self.record(b'', final=True)
+            self.log.close()
+            self.sock.close()
+            self.closed = True
+
+    def send(self, data):
+        self.sock.sendall(data.encode() if isinstance(data, str) else data)
+
+    def read(self, timeout):
+        ready, _, _ = select.select([self.sock], [], [], max(0, timeout))
+        if not ready:
+            return b''
+        data = self.sock.recv(65536)
+        if not data:
+            raise SmokeError('Guest console closed')
+        self.record(data)
+        self.buffer += data
+        if len(self.buffer) > 16 * 1024 * 1024:
+            raise SmokeError('Console exceeded bounded capture size')
+        return data
+
+    def expect(self, pattern, timeout, failure_pattern=None):
+        deadline = time.monotonic() + timeout
+        while True:
+            match = re.search(pattern, self.buffer)
+            failure = re.search(failure_pattern, self.buffer) if failure_pattern else None
+            if failure and (not match or failure.start() < match.start()):
+                evidence = failure.group()
+                for password in self.passwords:
+                    evidence = evidence.replace(password, b'[REDACTED]')
+                raise SmokeError('Installer exited before expected console state: ' +
+                                 evidence.decode(errors='replace'))
+            if match:
+                self.buffer = self.buffer[match.end():]
+                return match
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Timed out waiting for guest console state')
+            self.read(min(remaining, 1))
+
+    def expect_installer(self, pattern, timeout, token):
+        failure = (rb'hydra-install: [^\r\n]*[\r\n]|\x1e' +
+                   re.escape(token.encode()) + rb':INSTALL_EXIT:[0-9]+\x1f')
+        return self.expect(pattern, timeout, failure_pattern=failure)
+
+    def finish_installer(self, token, timeout=30):
+        # Completion text can arrive before the process exits and readline
+        # resumes. Input sent during that gap can be discarded by the TTY.
+        deadline = time.monotonic() + timeout
+        returned = self.expect(rb'\x1e' + re.escape(token.encode()) +
+                               rb':INSTALL_EXIT:([0-9]+)\x1f', timeout)
+        status = int(returned.group(1))
+        if status:
+            raise SmokeError(f'Installer returned status {status} after completion')
+        self.expect(rb'hydra-live[^\r\n]*[#$] ', max(0, deadline - time.monotonic()))
+
+    def command(self, command, timeout=60):
+        token = secrets.token_hex(12)
+        self.buffer = b''
+        self.send(frame_command(command, token) + '\n')
+        deadline = time.monotonic() + timeout
+        while True:
+            parsed = parse_frame(self.buffer, token)
+            if parsed is not None:
+                status, output = parsed
+                self.buffer = b''
+                if status:
+                    raise SmokeError(f'Guest probe failed with status {status}: {output.decode(errors="replace")[-3000:]}')
+                return output.replace(b'\r\n', b'\n')
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Timed out waiting for framed guest command')
+            self.read(min(remaining, 1))
+
+
+def connect(path, process, timeout=30):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise SmokeError(f'Test QEMU exited with status {process.returncode}')
+        connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        try:
+            connection.connect(str(path))
+            return connection
+        except OSError:
+            connection.close()
+            time.sleep(0.1)
+    raise SmokeError('Timed out connecting to private test socket')
+
+
+class VM:
+    def __init__(self, args, scratch, disk, iso, password, phase):
+        self.scratch = scratch
+        for name in ('serial.sock', 'qmp.sock'):
+            (scratch / name).unlink(missing_ok=True)
+        self.stderr = open(scratch / f'{phase}-qemu.log', 'ab', buffering=0)
+        self.process = subprocess.Popen(qemu_argv(args.qemu, scratch, disk,
+            args.ovmf_code, scratch / 'OVMF_VARS.fd', iso, args.memory),
+            stdin=subprocess.DEVNULL, stdout=self.stderr, stderr=self.stderr)
+        self.console = None
+        self.qmp = None
+        try:
+            self.console = Console(connect(scratch / 'serial.sock', self.process),
+                                   scratch / f'{phase}-serial.log', [password])
+            ready = connect(scratch / 'qmp.sock', self.process)
+            ready.close()
+            self.qmp = QMPClient(scratch / 'qmp.sock', timeout=5)
+            self.kvm = self.qmp.command('query-kvm')
+            if not self.kvm.get('enabled'):
+                raise SmokeError('Disposable QEMU is not using KVM')
+        except BaseException:
+            self.close()
+            raise
+
+    def wait_poweroff(self, timeout):
+        deadline = time.monotonic() + timeout
+        serial_open = True
+        while self.process.poll() is None:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Guest failed to power off in time')
+            if serial_open:
+                try:
+                    self.console.read(min(remaining, 0.25))
+                except SmokeError as exc:
+                    if str(exc) != 'Guest console closed':
+                        raise
+                    serial_open = False
+            else:
+                try:
+                    self.process.wait(timeout=remaining)
+                except subprocess.TimeoutExpired as exc:
+                    raise SmokeError('Guest failed to power off in time') from exc
+        # Capture bytes queued immediately before exit, even if poll observed
+        # the child exiting before the first serial read.
+        while serial_open:
+            try:
+                if not self.console.read(0):
+                    break
+            except SmokeError as exc:
+                if str(exc) != 'Guest console closed':
+                    raise
+                serial_open = False
+        rc = self.process.returncode
+        if rc:
+            raise SmokeError(f'Test QEMU failed during poweroff ({rc})')
+
+    def close(self):
+        # Only the Popen child owned by this harness is signalled.
+        if self.qmp:
+            if self.process.poll() is None:
+                try:
+                    self.qmp.command('quit')
+                    self.process.wait(timeout=5)
+                except (QMPError, subprocess.TimeoutExpired):
+                    pass
+            self.qmp.close()
+        if self.process.poll() is None:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait(timeout=10)
+        if self.console:
+            self.console.close()
+        self.stderr.close()
+
+
+LIVE_CHECK = r'''set -eu
+[ "$(id -un)" = hydra-live ]
+test -e /sys/firmware/efi
+test -f /etc/hydra/live-mode/active
+findmnt -n -o FSTYPE / | grep -qx overlay
+test -f /run/initramfs/live/LiveOS/squashfs.img
+sudo -n true
+for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
+for service in NetworkManager hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
+'''
+
+INSTALLED_CHECK = r'''set -eu
+assert_no_match() {
+    if grep "$@"; then
+        printf 'Forbidden live configuration remains\n' >&2
+        exit 1
+    else
+        status=$?
+        [ "$status" -eq 1 ] || exit "$status"
+    fi
+}
+[ "$(findmnt -n -o FSTYPE /)" = ext4 ]
+[ "$(readlink -f "$(findmnt -n -o SOURCE /)")" = /dev/vda2 ]
+assert_no_match -Eq 'root=live:|rd\.live' /proc/cmdline
+test ! -e /etc/hydra/live-mode
+test ! -L /etc/hydra/live-mode
+if getent passwd hydra-live; then
+    printf 'Live account remains\n' >&2
+    exit 1
+else
+    status=$?
+    [ "$status" -eq 2 ] || exit "$status"
+fi
+remaining=$(find /etc/systemd/system /etc/sudoers.d \( -name 90-hydra-live.conf -o -name 90-hydra-live \) -print)
+[ -z "$remaining" ]
+test ! -e /etc/greetd/live.toml
+test ! -L /etc/greetd/live.toml
+for directory in /etc/systemd/system/getty@*.service.d /etc/systemd/system/serial-getty@*.service.d /etc/sudoers.d; do
+    if [ -d "$directory" ]; then
+        assert_no_match -R -E -- '--autologin|NOPASSWD' "$directory"
+    elif [ -e "$directory" ] || [ -L "$directory" ]; then
+        printf 'Unexpected non-directory configuration path\n' >&2
+        exit 1
+    fi
+done
+getent shadow root | cut -d: -f2 | grep -Eq '^[!*]'
+getent shadow hydratest | cut -d: -f2 | grep -Eq '^\$'
+for service in NetworkManager greetd hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
+[ "$(systemctl show hydra-bridge-permissions -p Result --value)" = success ]
+[ "$(systemctl show hydra-bridge-permissions -p ExecMainStatus --value)" = 0 ]
+ip -4 -o addr show hydrabr0 | grep -F '10.203.0.1/24'
+ip -4 route show default | grep '^default '
+nft list table ip hydra_nat
+nmcli -t -f STATE general | grep -E '^connected'
+getent ahostsv4 gentoo.org
+for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
+hydra --help
+qemu-system-x86_64 --version
+qemu-img --version
+test -r /etc/greetd/config.toml
+find /usr/share -name OVMF_CODE.fd -print -quit | grep .
+'''
+
+NETWORK_CHECK = r'''set -eu
+[ "$(cat /proc/sys/net/ipv4/ip_forward)" = 1 ]
+python3 - <<'NETWORK'
+import pathlib, secrets, subprocess, time
+tag = secrets.token_hex(3)
+lan, guest = 'hl' + tag, 'hv' + tag
+uplink, lanpeer = 'hu' + tag, 'hp' + tag
+bridgepeer, guestpeer = 'hb' + tag, 'hg' + tag
+created = []
+links = []
+server = None
+def run(*args):
+    return subprocess.run(args, check=True, capture_output=True, text=True, timeout=20).stdout
+try:
+    for name in (lan, guest):
+        run('ip', 'netns', 'add', name)
+        created.append(name)
+    run('ip', 'link', 'add', uplink, 'type', 'veth', 'peer', 'name', lanpeer)
+    links.append(uplink)
+    run('ip', 'link', 'set', lanpeer, 'netns', lan)
+    run('ip', 'addr', 'add', '198.18.0.1/30', 'dev', uplink)
+    run('ip', 'link', 'set', uplink, 'up')
+    run('ip', '-n', lan, 'addr', 'add', '198.18.0.2/30', 'dev', lanpeer)
+    run('ip', '-n', lan, 'link', 'set', lanpeer, 'up')
+    run('ip', '-n', lan, 'link', 'set', 'lo', 'up')
+    run('ip', '-n', lan, 'route', 'add', '10.203.0.0/24', 'via', '198.18.0.1')
+    run('ip', 'link', 'add', bridgepeer, 'type', 'veth', 'peer', 'name', guestpeer)
+    links.append(bridgepeer)
+    run('ip', 'link', 'set', guestpeer, 'netns', guest)
+    run('ip', 'link', 'set', bridgepeer, 'master', 'hydrabr0')
+    run('ip', 'link', 'set', bridgepeer, 'up')
+    run('ip', '-n', guest, 'addr', 'add', '10.203.0.250/24', 'dev', guestpeer)
+    run('ip', '-n', guest, 'link', 'set', guestpeer, 'up')
+    run('ip', '-n', guest, 'link', 'set', 'lo', 'up')
+    run('ip', '-n', guest, 'route', 'add', 'default', 'via', '10.203.0.1')
+    port = 43117
+    server_code = "import socket; s=socket.socket(); s.settimeout(15); s.bind(('10.203.0.250',43117)); s.listen(1); c,a=s.accept(); c.sendall(b'HYDRA_LAN_TO_VM_OK'); c.close(); s.close()"
+    server = subprocess.Popen(['ip', 'netns', 'exec', guest, 'python3', '-c', server_code], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    client_code = "import socket,time; deadline=time.monotonic()+10\nwhile True:\n try:\n  s=socket.create_connection(('10.203.0.250',43117),1); break\n except OSError:\n  if time.monotonic()>deadline: raise\n  time.sleep(.1)\nexpected=b'HYDRA_LAN_TO_VM_OK'; received=b''\nwhile len(received)<len(expected):\n remaining=deadline-time.monotonic()\n if remaining<=0: raise TimeoutError('TCP marker deadline exceeded')\n s.settimeout(remaining)\n part=s.recv(len(expected)-len(received))\n if not part: raise RuntimeError('Truncated TCP marker')\n received+=part\nassert received==expected; s.close(); print('PASS: routed new LAN-to-VM TCP connection on arbitrary port 43117')"
+    print(run('ip', 'netns', 'exec', lan, 'python3', '-c', client_code), end='')
+    assert server.wait(timeout=10) == 0
+finally:
+    if server is not None and server.poll() is None:
+        server.terminate()
+        try: server.wait(timeout=5)
+        except subprocess.TimeoutExpired: server.kill(); server.wait(timeout=5)
+    for link in reversed(links):
+        subprocess.run(['ip', 'link', 'del', link], check=False, capture_output=True, timeout=10)
+    for name in reversed(created):
+        subprocess.run(['ip', 'netns', 'del', name], check=False, capture_output=True, timeout=10)
+NETWORK
+'''
+
+# A nested VM hardware smoke proves KVM initialisation/QMP, not guest OS boot.
+NESTED_CHECK = r'''set -eu
+if ! test -c /dev/kvm; then printf 'SKIP: guest /dev/kvm unavailable\n'; exit 0; fi
+python3 - <<'NESTED'
+import json, pathlib, socket, subprocess, tempfile, time
+with tempfile.TemporaryDirectory(prefix='hydra-nested-') as tmp:
+    root = pathlib.Path(tmp)
+    disk = root / 'fresh.qcow2'
+    subprocess.run(['qemu-img', 'create', '-f', 'qcow2', str(disk), '64M'], check=True, timeout=30)
+    log = open(root / 'qemu.log', 'wb')
+    proc = subprocess.Popen(['qemu-system-x86_64', '-machine', 'q35,accel=kvm', '-cpu', 'host', '-m', '128', '-S', '-display', 'none', '-nodefaults', '-nic', 'bridge,br=hydrabr0,model=virtio-net-pci', '-drive', 'if=virtio,format=qcow2,file=' + str(disk), '-qmp', 'unix:' + str(root / 'qmp') + ',server=on,wait=off'], stdin=subprocess.DEVNULL, stdout=log, stderr=log)
+    sock = socket.socket(socket.AF_UNIX)
+    sock.settimeout(10)
+    try:
+        deadline = time.monotonic() + 15
+        while True:
+            try: sock.connect(str(root / 'qmp')); break
+            except OSError:
+                if proc.poll() is not None or time.monotonic() >= deadline: raise RuntimeError('Nested KVM start failed')
+                time.sleep(.1)
+        stream = sock.makefile('rwb', buffering=0)
+        assert 'QMP' in json.loads(stream.readline())
+        for request in ({'execute':'qmp_capabilities'}, {'execute':'query-kvm'}, {'execute':'quit'}):
+            stream.write(json.dumps(request).encode() + b'\n')
+            while True:
+                response = json.loads(stream.readline())
+                if 'return' in response or 'error' in response: break
+            assert 'error' not in response, response
+            if request['execute'] == 'query-kvm':
+                assert response['return']['enabled'], response
+                print('PASS: normal-user nested KVM VM and bridge helper; fresh qcow2; private QMP')
+        assert proc.wait(timeout=10) == 0
+    finally:
+        sock.close()
+        if proc.poll() is None:
+            proc.terminate()
+            try: proc.wait(timeout=5)
+            except subprocess.TimeoutExpired: proc.kill(); proc.wait(timeout=5)
+        log.close()
+NESTED
+'''
+
+
+def digest(path):
+    result = hashlib.sha256()
+    with open(path, 'rb') as source:
+        for block in iter(lambda: source.read(1024 * 1024), b''):
+            result.update(block)
+    return result.hexdigest()
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--iso', required=True, type=Path)
+    parser.add_argument('--ovmf-code', required=True, type=Path)
+    parser.add_argument('--ovmf-vars', required=True, type=Path)
+    parser.add_argument('--qemu', default='qemu-system-x86_64')
+    parser.add_argument('--qemu-img', default='qemu-img')
+    parser.add_argument('--scratch-parent', type=Path, default=Path(tempfile.gettempdir()))
+    parser.add_argument('--memory', type=int, default=4096)
+    parser.add_argument('--boot-timeout', type=int, default=360)
+    parser.add_argument('--install-timeout', type=int, default=1800)
+    args = parser.parse_args(argv)
+    for path in (args.iso, args.ovmf_code, args.ovmf_vars):
+        if not path.is_file():
+            parser.error(f'Missing input file: {path}')
+    if args.memory < 2048 or args.boot_timeout <= 0 or args.install_timeout <= 0:
+        parser.error('Memory must be >=2048 MiB and timeouts positive')
+    args.iso = args.iso.resolve()
+    args.ovmf_code = args.ovmf_code.resolve()
+    args.ovmf_vars = args.ovmf_vars.resolve()
+    scratch = Path(tempfile.mkdtemp(prefix='hydra-iso-smoke-', dir=args.scratch_parent))
+    if len(str(scratch / 'serial.sock').encode()) >= 104:
+        parser.error('Scratch path too long for Unix sockets; use /tmp')
+    print(f'Acceptance artifacts: {scratch}', flush=True)
+    report = {'schema': 1, 'status': 'running', 'scratch': str(scratch),
+              'iso': str(args.iso), 'iso_sha256': digest(args.iso),
+              'checks': {}, 'physical_hardware': 'pending',
+              'nested_scope': 'Normal-user KVM VM launch, bridge helper and QMP; no nested guest OS boot',
+              'ovmf_code_sha256': digest(args.ovmf_code),
+              'ovmf_vars_input_sha256': digest(args.ovmf_vars)}
+    password = secrets.token_urlsafe(24)
+    sentinel = secrets.token_hex(32)
+    vm = None
+    def checkpoint(name, output):
+        report['checks'][name] = {'status': 'skipped' if output.startswith(b'SKIP:') else 'passed',
+                                  'output': output.decode(errors='replace')}
+        (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
+    try:
+        disk = scratch / 'fresh-install.qcow2'
+        reserve_disk(disk)
+        subprocess.run([args.qemu_img, 'create', '-f', 'qcow2', str(disk), '32G'], check=True, timeout=30,
+                       stdout=subprocess.DEVNULL)
+        shutil.copyfile(args.ovmf_vars, scratch / 'OVMF_VARS.fd')
+        os.chmod(scratch / 'OVMF_VARS.fd', 0o600)
+        vm = VM(args, scratch, disk, args.iso, password, 'live')
+        checkpoint('live_qmp_kvm', json.dumps(vm.kvm).encode())
+        console = vm.console
+        console.expect(rb'hydra-live[^\r\n]*[#$] ', args.boot_timeout)
+        checkpoint('live_boot', console.command(LIVE_CHECK, 90))
+        console.buffer = b''
+        installer_token = secrets.token_hex(12)
+        console.send(installer_command(installer_token) + '\n')
+        console.expect_installer(rb'Installed account password: ', 120, installer_token)
+        console.send(password + '\n')
+        console.expect_installer(rb'Repeat password: ', 30, installer_token)
+        console.send(password + '\n')
+        console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                 args.install_timeout, installer_token)
+        console.finish_installer(installer_token)
+        checkpoint('real_installer', b'Installer completed via guest controlling TTY')
+        console.command("sudo -n mkdir -p /mnt/hydra-validation && sudo -n mount /dev/vda2 /mnt/hydra-validation && printf %s " +
+                        shlex.quote(sentinel) + " | sudo -n tee /mnt/hydra-validation/var/lib/hydra-acceptance-sentinel >/dev/null && sudo -n umount /mnt/hydra-validation", 60)
+        console.send('sudo -n poweroff\n')
+        vm.wait_poweroff(90)
+        vm.close()
+        vm = None
+        checkpoint('live_poweroff', b'QEMU exited before ISO-free reboot')
+        vm = VM(args, scratch, disk, None, password, 'installed')
+        checkpoint('installed_qmp_kvm', json.dumps(vm.kvm).encode())
+        console = vm.console
+        console.expect(rb'login: ', args.boot_timeout)
+        console.send('hydratest\n')
+        console.expect(rb'Password: ', 30)
+        console.send(password + '\n')
+        console.expect(rb'hydratest[^\r\n]*[#$] ', 60)
+        output = console.command('id -un; if sudo -n true 2>/dev/null; then exit 1; fi', 30)
+        if output.strip() != b'hydratest':
+            raise SmokeError('Installed account identity mismatch')
+        checkpoint('installed_password_login', output)
+        console.buffer = b''
+        prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
+        console.send(f'sudo -p "$(printf %s {prompt_encoded} | base64 -d)" -i\n')
+        console.expect(rb'HYDRA_SUDO_PASSWORD: ', 30)
+        console.send(password + '\n')
+        console.expect(rb'root[^\r\n]*[#$] ', 30)
+        observed = console.command('cat /var/lib/hydra-acceptance-sentinel', 30)
+        if observed != sentinel.encode():
+            raise SmokeError('Persistence sentinel did not survive installed disk boot')
+        checkpoint('disk_persistence', b'Sentinel content verified from command output frame')
+        checkpoint('installed_contract', console.command(INSTALLED_CHECK, 120))
+        checkpoint('lan_to_vm_tcp', console.command(NETWORK_CHECK, 120))
+        checkpoint('nested_kvm', console.command('runuser -u hydratest -- sh -c ' + shlex.quote(NESTED_CHECK), 90))
+        checkpoint('installed_user_logout', installed_logout_probe(console, password))
+        console.send('poweroff\n')
+        vm.wait_poweroff(90)
+        checkpoint('installed_poweroff', b'Normal installed guest shutdown')
+        report['status'] = 'passed'
+        return 0
+    except (SmokeError, QMPError, OSError, subprocess.SubprocessError) as exc:
+        report['status'] = 'failed'
+        report['error'] = str(exc).replace(password, '[REDACTED]')
+        print(f'Acceptance failed: {report["error"]}', file=sys.stderr)
+        return 1
+    finally:
+        if vm:
+            vm.close()
+        report['completed_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
+        (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
+        print(f'Report: {scratch / "report.json"}', flush=True)
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

dev/tools.nix

@@ -0,0 +1,23 @@
+# Temporary tools only: nix-shell dev/tools.nix. No host/profile installation.
+let
+  pkgs = import (builtins.fetchTarball {
+    url = "https://github.com/NixOS/nixpkgs/archive/7c8764b7c7b09b34f632464276218ef9090eaa11.tar.gz";
+    sha256 = "sha256-MWhktfFtFE8Jl7bf9Pkp7TR5MSdYCjCswSo0Gkb4ywc=";
+  }) {};
+  graphics = import (builtins.fetchTarball {
+    url = "https://github.com/NixOS/nixpkgs/archive/7fd4004d0511991fff3a0524856a9be09e28c5f0.tar.gz";
+    sha256 = "sha256-i5qk1kLiit98rRbh1AXmWBNT9QXC/y4l5Wk/u5EdpDQ=";
+  }) {};
+  viewer = pkgs.virt-viewer.overrideAttrs (old: {
+    src = pkgs.fetchurl {
+      url = "https://releases.pagure.org/virt-viewer/virt-viewer-11.0.tar.xz";
+      hash = "sha256-pD+iMlxMHHelyMmAZaww7wURohrJjlkPIjQIabrZq9A=";
+    };
+    patches = (old.patches or []) ++ [ ../patches/virt-viewer-hydra-session.patch ];
+    doCheck = true;
+  });
+in pkgs.mkShell {
+  packages = [ pkgs.python3 pkgs.qemu viewer graphics.labwc pkgs.foot
+               pkgs.brightnessctl pkgs.wayland-utils pkgs.wlr-randr ];
+  passthru = { inherit viewer; labwc = graphics.labwc; };
+}

docs/install.md

@@ -0,0 +1,85 @@
+# Install HydraOS
+
+HydraOS targets x86_64 UEFI systems. Disable Secure Boot for this first image.
+Boot the ISO and open the recovery terminal. The live session uses a locked
+`hydra-live` account with local sudo access; changes in the live filesystem are
+volatile. The image provides no default SSH login or universal installed password.
+
+`hydra-install` lists disks without selecting one. Inspect the devices, then ask
+for a plan using the whole disk you intend to erase:
+
+```sh
+hydra-install
+hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER
+```
+
+Replace the placeholders with an actual whole disk and your desired account
+name. The installer rejects mounted devices, the running root, the live medium,
+and their backing disks. These checks help prevent mistakes; verify the chosen
+disk and preserve backups before executing the plan. Physical disk installation
+requires the operator's explicit choice.
+
+To execute, enter the identical resolved whole-disk path in the confirmation:
+
+```sh
+sudo hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER \
+  --execute --confirm-erase /dev/CHOSEN_DISK
+```
+
+Execution erases the disk, creates a 1 GiB FAT32 EFI partition and an ext4 root,
+extracts the prepared payload, and prompts twice on the local terminal for your
+installed account password. Passwords have no command-line or configuration
+option. The installed root account stays locked. Defaults are Europe/Moscow and
+English UTF-8, with Russian UTF-8 also available; `--timezone` accepts an IANA
+zone present in the payload.
+
+The installer removes the live account, local password-free sudo rule, and
+live autologin overrides, then builds a normal disk initramfs and fallback UEFI
+loader. Wait for `Installation completed`, power off, remove the ISO, and boot
+the installed disk. Log in with the account and password you supplied. On
+failure, read the printed live diagnostic path; when a target root was mounted,
+a nonsecret snapshot is also saved at `/var/log/hydra-install.log` on that disk.
+
+## Disposable VM acceptance
+
+Run the repository harness with QEMU/KVM, `qemu-img`, Python 3, and an OVMF
+CODE/VARS pair. Use trusted image and firmware files. For example:
+
+```sh
+python3 dev/iso-smoke.py --iso /absolute/path/to/hydra-os.iso \
+  --ovmf-code /absolute/path/to/OVMF_CODE.fd \
+  --ovmf-vars /absolute/path/to/OVMF_VARS.fd
+```
+
+The harness creates a private directory under `/tmp`, a new 32 GiB sparse
+qcow2 disk, and a private copy of OVMF variables. There is no existing disk
+argument. QEMU attaches the ISO read-only; serial and QMP use private Unix
+sockets. Tool overrides `--qemu` and `--qemu-img` support temporary development
+environments. `--scratch-parent`, `--memory`, `--boot-timeout`, and
+`--install-timeout` control retained artifacts and bounded resource use.
+
+The test boots the live image, runs the real installer through the guest's
+controlling terminal using a generated password, writes a persistence sentinel,
+shuts down, and launches a second QEMU with the ISO absent. It verifies password
+login, disk root, removed live overrides, services, bridge/NAT configuration,
+DNS, recovery tools, and a routed new TCP connection from a synthetic LAN to a
+synthetic VM endpoint. Network namespaces and veth interfaces exist only inside
+the disposable HydraOS guest. When guest KVM is available, it launches a further
+fresh-disk QEMU as the installed account to exercise KVM, the bridge helper,
+and private QMP; this checks VM launch, not a nested operating system boot.
+
+`report.json` records each observed check, ISO/firmware SHA256, failures, and
+skipped nested virtualization. Serial/QEMU logs and the qcow2 remain in the
+printed scratch directory for review; passwords are redacted even across serial
+read boundaries. Cleanup controls only the QEMU process created by this run.
+A passing unit suite establishes harness behavior, not ISO acceptance. An
+actual successful report from the built ISO is required for boot/install/reboot
+acceptance.
+
+## Checks requiring the laptop
+
+Disposable QEMU results cannot establish ASUS FX517ZC hardware support. Before
+using HydraOS daily, boot the live ISO and observe Wi-Fi, touchpad, audio,
+brightness, suspend/resume, USB device selection, input capture, and actual
+display refresh. Physical installation and these checks remain pending until
+explicitly performed and recorded.

hydra/__init__.py

@@ -0,0 +1 @@
+"""HydraOS independent VM management."""

hydra/build.py

@@ -0,0 +1,392 @@
+"""Verified, resumable Gentoo builds in a dedicated rootless Podman store."""
+import argparse
+import contextlib
+import fcntl
+import hashlib
+import json
+import lzma
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.parse
+import urllib.request
+
+
+class BuildError(RuntimeError):
+    pass
+
+
+def package_job_script(command, state='/var/lib/hydra-build', package_db='/var/db/pkg', inputs=''):
+    """The contained job owns its log/result even if the outer caller disconnects."""
+    return f'''import fcntl, json, os, pathlib, subprocess, sys, time
+state = pathlib.Path({str(state)!r})
+database = pathlib.Path({str(package_db)!r})
+state.mkdir(parents=True, exist_ok=True)
+lock = (state / 'packages.lock').open('a')
+try:
+    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+except BlockingIOError:
+    sys.exit(75)
+def write(name, data):
+    temporary = state / (name + '.tmp')
+    temporary.write_text(json.dumps(data, indent=2) + '\\n')
+    temporary.replace(state / name)
+status = {{'running': True, 'pid': os.getpid(), 'inputs': {inputs!r}, 'started': int(time.time())}}
+write('packages.status.json', status)
+with (state / 'packages.log').open('a') as log:
+    try:
+        result = subprocess.run({command!r}, stdout=log, stderr=subprocess.STDOUT)
+        code = result.returncode
+    except OSError as error:
+        log.write(str(error) + '\\n')
+        code = 127
+if code == 0:
+    inventory = []
+    for use in sorted(database.glob('*/*/USE')):
+        package = use.parent
+        inventory.append({{'package': str(package.relative_to(database)), 'use': use.read_text().split(),
+                          'build_time': (package / 'BUILD_TIME').read_text().strip() if (package / 'BUILD_TIME').exists() else None}})
+    write('packages.json', inventory)
+status.update(running=False, exit_code=code, finished=int(time.time()))
+write('packages.status.json', status)
+sys.exit(code)
+'''
+
+
+def verify_hash(path, expected):
+    digest = hashlib.sha256()
+    with Path(path).open('rb') as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b''):
+            digest.update(chunk)
+    if digest.hexdigest() != expected:
+        raise BuildError(f'SHA256 mismatch: {path}')
+
+
+def require_signature(status, fingerprint):
+    valid = [line.split() for line in status.splitlines() if line.startswith('[GNUPG:] VALIDSIG ')]
+    if not valid or not any(fields[-1].upper() == fingerprint.upper() for fields in valid):
+        raise BuildError('Required primary signature fingerprint is missing')
+    if any(token in status for token in ('BADSIG', 'ERRSIG', 'REVKEYSIG', 'EXPKEYSIG')):
+        raise BuildError('Invalid release signature')
+
+
+class Builder:
+    def __init__(self, project, work, cache, podman=None, gpg=None, source_dir=None):
+        if os.geteuid() == 0:
+            raise BuildError('Run as the normal user, never host root')
+        self.project = Path(project).resolve()
+        self.source_dir = Path(source_dir or project).resolve()
+        self.work, self.cache = Path(work).resolve(), Path(cache).resolve()
+        for path in (self.work, self.cache):
+            if path == self.project or not path.is_relative_to(self.project):
+                raise BuildError('Work and cache paths must be inside the project, below its root')
+        if self.work == self.cache or self.work.is_relative_to(self.cache) or self.cache.is_relative_to(self.work):
+            raise BuildError('Work and cache must be separate directories')
+        self.podman = podman or shutil.which('podman') or 'podman'
+        self.gpg = gpg or shutil.which('gpg') or 'gpg'
+        runtime = Path(os.environ.get('XDG_RUNTIME_DIR', f'/run/user/{os.getuid()}'))
+        identifier = hashlib.sha256(str(self.work).encode()).hexdigest()[:12]
+        self.runroot = runtime / ('hydra-build-podman' if self.work == self.project / 'build' else f'hydra-build-{identifier}')
+        self.name = 'hydra-builder-' + identifier
+        self.image = 'localhost/hydra-stage3:20261011'
+        self._stage3_layers = {}
+
+    def podman_argv(self, *args):
+        return [self.podman, '--root', str(self.work / 'podman'), '--runroot', str(self.runroot), '--storage-driver', 'overlay', *args]
+
+    def container_run_args(self):
+        return ['run', '--detach', '--name', self.name, '--hostname', 'hydra-builder',
+                '--network', 'host', '--cap-add', 'SYS_ADMIN', '--security-opt', 'seccomp=unconfined',
+                '--volume', f'{self.source_dir}:/hydra-source:ro', '--volume', f'{self.cache}:/hydra-cache:ro',
+                self.image, '/bin/sleep', 'infinity']
+
+    def run(self, argv, capture=False, check=True):
+        print('+ ' + ' '.join(str(arg) for arg in argv), file=sys.stderr, flush=True)
+        result = subprocess.run(argv, text=True, stdout=subprocess.PIPE if capture else None,
+                                stderr=subprocess.PIPE if capture else None)
+        if check and result.returncode:
+            raise BuildError(f'Command failed ({result.returncode}): {argv[0]}\n{result.stderr or ""}')
+        return result
+
+    def pod(self, *args, **kwargs):
+        return self.run(self.podman_argv(*args), **kwargs)
+
+    def exec(self, command, capture=False):
+        return self.pod('exec', self.name, '/bin/bash', '-euc', command, capture=capture)
+
+    def inspect(self, kind, name):
+        result = self.pod(kind, 'inspect', name, capture=True, check=False)
+        if result.returncode:
+            raise BuildError(f'{kind} provenance is unavailable; use a fresh work directory')
+        entries = json.loads(result.stdout)
+        if not isinstance(entries, list) or len(entries) != 1 or not isinstance(entries[0], dict):
+            raise BuildError(f'Unexpected {kind} provenance')
+        return entries[0]
+
+    def check_mounts(self, container):
+        for destination, path, label in (('/hydra-source', self.source_dir, 'source'),
+                                         ('/hydra-cache', self.cache, 'cache')):
+            mounts = [item for item in container.get('Mounts', []) if item.get('Destination') == destination]
+            if len(mounts) != 1 or mounts[0].get('Source') != str(path) or mounts[0].get('RW') is not False:
+                raise BuildError(f'Existing builder {label} mount differs; use its original checkout/cache or a fresh work directory')
+
+    def stage3_layer(self, inputs):
+        stage = inputs['stage3']
+        archive = self.cache / Path(urllib.parse.urlparse(stage['url']).path).name
+        verify_hash(archive, stage['sha256'])
+        if stage['sha256'] not in self._stage3_layers:
+            digest = hashlib.sha256()
+            with lzma.open(archive, 'rb') as stream:
+                for chunk in iter(lambda: stream.read(1024 * 1024), b''):
+                    digest.update(chunk)
+            self._stage3_layers[stage['sha256']] = 'sha256:' + digest.hexdigest()
+        return self._stage3_layers[stage['sha256']]
+
+    def check_base(self, image, inputs):
+        expected = self.stage3_layer(inputs)
+        if image.get('RootFS', {}).get('Layers') != [expected]:
+            raise BuildError('Existing builder stage3 differs from pinned archive; use a fresh work directory')
+        return expected
+
+    def read_container_file(self, path):
+        # Podman cp also reads a stopped container, so provenance precedes start.
+        with tempfile.TemporaryDirectory(prefix='hydra-read-') as directory:
+            target = Path(directory) / 'data'
+            result = self.pod('cp', f'{self.name}:{path}', str(target), capture=True, check=False)
+            return target.read_text() if result.returncode == 0 else None
+
+    def check_snapshot(self, inputs, allow_pending=False):
+        marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot')
+        if marker is None and allow_pending:
+            marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot-input')
+        if marker is None or marker.strip() != inputs['portage']['commit']:
+            raise BuildError('Existing builder snapshot differs or is unverified; use a fresh work directory')
+
+    def active_package_job(self):
+        if not (self.work / 'podman').is_dir():
+            return False
+        result = self.pod('top', self.name, 'hpid', 'args', capture=True, check=False)
+        if result.returncode == 0 and any(re.search(r'(?:^|\s)(?:\S*/)?emerge(?:\s|$)', line)
+                                         for line in result.stdout.splitlines()[1:]):
+            return True
+        probe = self.pod('exec', self.name, 'python3', '-c', '''import fcntl, pathlib, sys
+path = pathlib.Path('/var/lib/hydra-build/packages.lock')
+if path.exists():
+    with path.open('r') as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        except BlockingIOError:
+            sys.exit(75)
+''', capture=True, check=False)
+        return probe.returncode == 75
+
+    def manifest(self):
+        path = self.work / 'manifest.json'
+        return json.loads(path.read_text()) if path.exists() else {}
+
+    def input_id(self):
+        digest = hashlib.sha256()
+        for path in sorted((self.source_dir / 'image').rglob('*')):
+            if path.is_file():
+                digest.update(str(path.relative_to(self.source_dir)).encode())
+                digest.update(path.read_bytes())
+        return digest.hexdigest()
+
+    def record(self, step, **details):
+        data = self.manifest()
+        data[step] = {'complete': True, 'inputs': self.input_id(), **details}
+        data['builder'] = self.name
+        path = self.work / 'manifest.json'
+        temp = path.with_suffix('.tmp')
+        temp.write_text(json.dumps(data, indent=2) + '\n')
+        temp.replace(path)
+
+    def complete(self, step):
+        entry = self.manifest().get(step, {})
+        if not entry.get('complete') or entry.get('inputs') != self.input_id():
+            return False
+        if step == 'packages':
+            if not (self.work / 'artifacts/packages.json').is_file():
+                return False
+            return self.complete('prepare')
+        if step == 'prepare':
+            try:
+                container = self.inspect('container', self.name)
+                self.check_mounts(container)
+                inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
+                self.check_base(self.inspect('image', container['Image']), inputs)
+                self.check_snapshot(inputs)
+                return True
+            except (BuildError, OSError, ValueError, KeyError, lzma.LZMAError):
+                return False
+        if step == 'fetch':
+            inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
+            try:
+                for name in ('stage3', 'portage'):
+                    item = inputs[name]
+                    path = self.cache / item.get('filename', Path(urllib.parse.urlparse(item['url']).path).name)
+                    verify_hash(path, item['sha256'])
+            except (OSError, BuildError):
+                return False
+            return True
+        return False
+
+    def download(self, url, filename=None):
+        parsed = urllib.parse.urlparse(url)
+        if parsed.scheme != 'https':
+            raise BuildError('Build input URLs must use HTTPS')
+        target = self.cache / (filename or Path(parsed.path).name)
+        if not target.exists():
+            temp = target.with_suffix(target.suffix + '.partial')
+            print(f'Downloading {url}', flush=True)
+            with urllib.request.urlopen(url, timeout=120) as response, temp.open('wb') as stream:
+                shutil.copyfileobj(response, stream)
+            temp.replace(target)
+        return target
+
+    def fetch(self):
+        inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
+        stage = inputs['stage3']
+        archive = self.download(stage['url'])
+        verify_hash(archive, stage['sha256'])
+        snapshot = self.download(inputs['portage']['url'], inputs['portage']['filename'])
+        verify_hash(snapshot, inputs['portage']['sha256'])
+        signature = self.download(stage['signature_url'])
+        keyring = self.download(inputs['keyring_url'])
+        verify_hash(keyring, inputs['keyring_sha256'])
+        with tempfile.TemporaryDirectory(prefix='hydra-gpg-', dir=self.work) as home:
+            result = self.run([self.gpg, '--batch', '--homedir', home, '--no-default-keyring', '--keyring', str(keyring),
+                               '--status-fd', '1', '--verify', str(signature), str(archive)], capture=True)
+            require_signature(result.stdout, stage['fingerprint'])
+            (self.work / 'artifacts/stage3-signature.txt').write_text(result.stdout + result.stderr)
+        self.record('fetch', sources=inputs)
+        return inputs
+
+    def prepare(self):
+        existing = self.pod('container', 'exists', self.name, capture=True, check=False).returncode == 0
+        container = self.inspect('container', self.name) if existing else None
+        if container is not None:
+            self.check_mounts(container)  # Reject another worktree before changing any checkpoint.
+            if self.active_package_job():
+                raise BuildError('A contained package job is running; wait for it to finish before preparation')
+        inputs = self.fetch()  # Reverify cached bytes even when resuming.
+        if container is not None:
+            layer = self.check_base(self.inspect('image', container['Image']), inputs)
+            self.check_snapshot(inputs, allow_pending=True)
+        elif self.pod('image', 'exists', self.image, capture=True, check=False).returncode:
+            self.pod('import', str(self.cache / Path(urllib.parse.urlparse(inputs['stage3']['url']).path).name), self.image)
+            layer = self.check_base(self.inspect('image', self.image), inputs)
+        else:
+            layer = self.check_base(self.inspect('image', self.image), inputs)
+        if container is None:
+            self.pod(*self.container_run_args())
+        else:
+            self.pod('start', self.name)
+        snapshot = inputs['portage']['filename']
+        # Snapshot extraction happens only inside the container; host paths are read-only.
+        self.exec(f'''mkdir -p /var/db/repos/gentoo
+if [ ! -f /var/db/repos/gentoo/.hydra-snapshot ]; then
+  printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot-input
+  tar --extract --gzip --file /hydra-cache/{snapshot} --strip-components=1 --directory /var/db/repos/gentoo
+  printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot
+fi
+cp -a /hydra-source/image/portage/. /etc/portage/
+if [ -d /etc/portage/gnupg ] && [ ! -s /etc/portage/gnupg/mykeyid ]; then
+  mv /etc/portage/gnupg /etc/portage/gnupg.hydra-untrusted
+fi
+getuto
+gpg --batch --homedir /etc/portage/gnupg --import /hydra-cache/service-keys.gpg
+chmod a+r /etc/portage/gnupg/pubring.kbx /etc/portage/gnupg/trustdb.gpg
+''')
+        self.record('prepare', image=container['Image'] if container else self.image,
+                    stage3_sha256=inputs['stage3']['sha256'], stage3_layer=layer,
+                    source_dir=str(self.source_dir), cache_dir=str(self.cache), snapshot=inputs['portage']['commit'])
+
+    def packages(self, pretend=False):
+        self.prepare()
+        atoms = [line.strip() for line in (self.source_dir / 'image/packages').read_text().splitlines()
+                 if line.strip() and not line.startswith('#')]
+        import shlex
+        command = ['emerge', '--verbose', '--update', '--newuse', '--deep', '--backtrack=100']
+        if pretend:
+            self.exec(shlex.join([*command, '--pretend', *atoms]))
+            return
+        inputs = self.input_id()
+        self.record('packages', complete=False, job_log='/var/lib/hydra-build/packages.log')
+        script = package_job_script([*command, *atoms], inputs=inputs)
+        self.exec(shlex.join(['python3', '-c', script]))
+        if self.input_id() != inputs:
+            raise BuildError('Image inputs changed during package installation; completion was not recorded')
+        inventory = self.exec('cat /var/lib/hydra-build/packages.json', capture=True).stdout
+        (self.work / 'artifacts/packages.json').write_text(inventory)
+        self.record('packages', inventory='artifacts/packages.json')
+
+    def status(self):
+        data = self.manifest()
+        for step in ('fetch', 'prepare', 'packages', 'rootfs', 'iso'):
+            if step in data:
+                data[step]['recorded_complete'] = data[step].get('complete', False)
+                data[step]['complete'] = self.complete(step)
+        data['mutation_lock_held'] = self.build_running()
+        data['package_job_running'] = self.active_package_job()
+        data['running'] = data['mutation_lock_held'] or data['package_job_running']
+        print(json.dumps(data, indent=2))
+
+    def build_running(self):
+        try:
+            with (self.work / '.hydra-build.lock').open('r') as lock:
+                try:
+                    fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
+                except BlockingIOError:
+                    return True
+        except FileNotFoundError:
+            pass
+        return False
+
+    @contextlib.contextmanager
+    def locked(self):
+        self.work.mkdir(parents=True, exist_ok=True)
+        self.cache.mkdir(parents=True, exist_ok=True)
+        (self.work / 'artifacts').mkdir(exist_ok=True)
+        with (self.work / '.hydra-build.lock').open('a') as lock:
+            try:
+                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError as exc:
+                raise BuildError('Another Hydra build is running in this work directory') from exc
+            yield
+
+
+def main(argv=None):
+    source = Path(__file__).resolve().parents[1]
+    # Linked worktrees share project-owned caches/build state in the primary checkout.
+    common = subprocess.run(['git', '-C', str(source), 'rev-parse', '--path-format=absolute', '--git-common-dir'],
+                            text=True, capture_output=True)
+    project = Path(common.stdout.strip()).parent if common.returncode == 0 else source
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('step', choices=('fetch', 'prepare', 'packages', 'rootfs', 'iso', 'status'))
+    parser.add_argument('--work', type=Path, default=project / 'build')
+    parser.add_argument('--cache', type=Path, default=project / 'cache')
+    parser.add_argument('--podman')
+    parser.add_argument('--gpg')
+    parser.add_argument('--pretend', action='store_true', help='resolve packages without installing')
+    options = parser.parse_args(argv)
+    try:
+        builder = Builder(project, options.work, options.cache, options.podman, options.gpg, source)
+        if options.step == 'status':
+            builder.status()
+            return 0
+        with builder.locked():
+            if options.step in ('rootfs', 'iso'):
+                raise BuildError(f'{options.step} assembly is not configured yet')
+            if options.step == 'packages':
+                builder.packages(options.pretend)
+            else:
+                getattr(builder, options.step)()
+        return 0
+    except (BuildError, OSError, ValueError, lzma.LZMAError) as exc:
+        print(f'hydra-build: {exc}', file=sys.stderr)
+        return 1

hydra/cli.py

@@ -0,0 +1,76 @@
+"""Portable source-checkout command line interface."""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from .profiles import Profile
+from .runtime import Runtime, RuntimeError
+from .session import launch_session, session_helper, view
+
+
+def parser() -> argparse.ArgumentParser:
+    cli = argparse.ArgumentParser(prog='hydra', description='Independent HydraOS VM manager')
+    cli.add_argument('--config', type=Path, default=Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'hydra')
+    cli.add_argument('--state', type=Path, default=Path(os.environ.get('XDG_STATE_HOME', Path.home() / '.local/state')) / 'hydra')
+    cli.add_argument('--runtime', type=Path, default=Path(os.environ.get('XDG_RUNTIME_DIR', f'/tmp/hydra-{os.getuid()}')) / 'hydra')
+    commands = cli.add_subparsers(dest='command', required=True)
+    add = commands.add_parser('add', help='Create a VM profile without starting it')
+    add.add_argument('name')
+    add.add_argument('--iso', type=Path, required=True)
+    add.add_argument('--memory', type=int, default=4096)
+    add.add_argument('--cpus', type=int, default=2)
+    add.add_argument('--disk-size', default='40G')
+    add.add_argument('--graphics', choices=('auto', 'software'), default='auto')
+    add.add_argument('--network', choices=('user', 'bridge'), default='user')
+    add.add_argument('--bridge', default='hydrabr0')
+    commands.add_parser('list', help='List configured VMs')
+    status = commands.add_parser('status', help='Query VM state through private QMP')
+    status.add_argument('name', nargs='?')
+    for command in ('start', 'shutdown', 'stop'):
+        cmd = commands.add_parser(command)
+        cmd.add_argument('name')
+        if command == 'stop':
+            cmd.add_argument('--force', action='store_true', required=True)
+    viewer = commands.add_parser('view', help='Connect a viewer without changing VM power')
+    viewer.add_argument('name')
+    viewer.add_argument('--windowed', action='store_true')
+    session = commands.add_parser('session', help='Start the minimal HydraOS labwc session')
+    session.add_argument('--nested', action='store_true')
+    helper = commands.add_parser('_session-helper', help=argparse.SUPPRESS)
+    helper.add_argument('operation')
+    helper.add_argument('manifest', type=Path)
+    return cli
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parser().parse_args(argv)
+    runtime = Runtime(args.config, args.state, args.runtime)
+    try:
+        if args.command == 'view':
+            return view(runtime, args.name, windowed=args.windowed)
+        elif args.command == 'session':
+            return launch_session(runtime, nested=args.nested)
+        elif args.command == '_session-helper':
+            return session_helper(args.operation, args.manifest)
+        elif args.command == 'add':
+            profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size,
+                              args.graphics, args.network, args.bridge)
+            runtime.add(profile)
+            result = {'name': profile.name, 'status': 'added'}
+        elif args.command == 'list':
+            result = [{'name': profile.name, 'iso': str(profile.iso)} for profile in runtime.profiles()]
+        elif args.command == 'status':
+            result = runtime.status(args.name) if args.name else [runtime.status(p.name) for p in runtime.profiles()]
+        elif args.command == 'stop':
+            result = runtime.stop(args.name, force=args.force)
+        else:
+            result = getattr(runtime, args.command)(args.name)
+        print(json.dumps(result, indent=2))
+        return 0
+    except (RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc:
+        print(f'hydra: {exc}', file=sys.stderr)
+        return 1

hydra/installer.py

@@ -0,0 +1,721 @@
+"""Explicit whole-disk installer. Importing this module never mutates storage.
+
+Live payload contract: /etc/hydra/live-mode/active and remove-on-install.json
+must be regular files. The manifest is a JSON array of absolute paths matching
+LIVE_REMOVABLE. Store all other live-only configuration in live-mode itself;
+ordinary installed configuration must already be present in the payload.
+"""
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import getpass
+import json
+import os
+from pathlib import Path, PurePosixPath
+import re
+import shlex
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import warnings
+from typing import TextIO
+
+DEFAULT_PAYLOAD = '/run/initramfs/live/LiveOS/squashfs.img'
+LIVE_REMOVABLE = (
+    '/etc/systemd/system/getty@*.service.d/90-hydra-live.conf',
+    '/etc/systemd/system/serial-getty@*.service.d/90-hydra-live.conf',
+    '/etc/systemd/system/greetd.service.d/90-hydra-live.conf',
+    '/etc/sudoers.d/90-hydra-live',
+    '/etc/greetd/live.toml',
+    '/etc/NetworkManager/system-connections/90-hydra-live.nmconnection',
+)
+DEVICE = re.compile(r'/dev/[A-Za-z0-9_.+:/-]+\Z')
+KERNEL = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.+-]*\Z')
+KNOWN_TYPES = {'disk', 'part', 'loop', 'rom', 'crypt', 'lvm', 'mpath', 'dm',
+               'raid0', 'raid1', 'raid4', 'raid5', 'raid6', 'raid10', 'md', 'zram'}
+
+
+class InstallError(RuntimeError):
+    """A failed safety check or operation; never suppress this in execution."""
+
+
+@dataclass
+class Request:
+    disk: str
+    user: str
+    payload: str = DEFAULT_PAYLOAD
+    execute: bool = False
+    confirm_erase: str | None = None
+    timezone: str = 'Europe/Moscow'
+    kernel: str | None = None
+
+
+class Runner:
+    """Checked argv-only commands. Secret-bearing commands redact all output."""
+    def run(self, argv, *, input=None, sensitive=False):
+        try:
+            result = subprocess.run(list(argv), input=input, text=True,
+                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                    shell=False, check=False)
+        except OSError as exc:
+            raise InstallError(f'Cannot run {argv[0]}: {exc.strerror}') from exc
+        if result.returncode:
+            detail = '' if sensitive else ': ' + result.stderr.strip()[:2000]
+            raise InstallError(f'Command failed ({result.returncode}): {shlex.join(argv)}{detail}')
+        return '' if sensitive else result.stdout
+
+
+class Topology:
+    def __init__(self):
+        self.nodes = {}
+        self.edges = {}
+        self.protected = set()
+
+    @classmethod
+    def from_json(cls, raw):
+        try:
+            data = json.loads(raw)
+        except (ValueError, TypeError) as exc:
+            raise InstallError('Malformed lsblk JSON') from exc
+        if not isinstance(data, dict) or not isinstance(data.get('blockdevices'), list) or not data['blockdevices']:
+            raise InstallError('lsblk must return a nonempty blockdevices array')
+        result = cls()
+        parents = []
+
+        def visit(node, ancestor=None, trail=()):
+            if not isinstance(node, dict):
+                raise InstallError('Malformed lsblk node')
+            path, kind = node.get('path'), node.get('type')
+            if not isinstance(path, str) or not DEVICE.fullmatch(path) or '..' in PurePosixPath(path).parts:
+                raise InstallError('Invalid lsblk device path')
+            if path in trail or kind not in KNOWN_TYPES:
+                raise InstallError('Cyclic or unknown lsblk topology')
+            mounts = node.get('mountpoints')
+            size, identity = node.get('size'), node.get('maj:min')
+            if not isinstance(mounts, list) or any(m is not None and not isinstance(m, str) for m in mounts):
+                raise InstallError('Malformed lsblk mountpoints')
+            if isinstance(size, bool) or not isinstance(size, int) or size < 0:
+                raise InstallError('Invalid lsblk size')
+            empty_optical = (kind == 'rom' and not any(mounts)
+                             and not node.get('children') and node.get('pkname') is None
+                             and ancestor is None)
+            if size == 0 and not empty_optical:
+                raise InstallError('Zero-size device is not an unrelated empty optical drive')
+            if not isinstance(identity, str) or not re.fullmatch(r'\d+:\d+', identity):
+                raise InstallError('Invalid lsblk device identity')
+            if path in result.nodes and any(result.nodes[path][key] != node[key] for key in ('type', 'size', 'maj:min', 'mountpoints')):
+                raise InstallError('Inconsistent duplicate lsblk node')
+            result.nodes[path] = node
+            result.edges.setdefault(path, set())
+            if any(m for m in mounts):
+                result.protected.add(path)
+            pkname = node.get('pkname')
+            if pkname is not None:
+                if not isinstance(pkname, str) or not pkname:
+                    raise InstallError('Invalid lsblk parent')
+                parent = pkname if pkname.startswith('/dev/') else '/dev/' + pkname
+                if not DEVICE.fullmatch(parent):
+                    raise InstallError('Invalid lsblk parent path')
+                parents.append((path, parent))
+            if ancestor:
+                parents.append((path, ancestor))
+            children = node.get('children', [])
+            if not isinstance(children, list):
+                raise InstallError('Malformed lsblk children')
+            for child in children:
+                visit(child, path, trail + (path,))
+
+        for node in data['blockdevices']:
+            visit(node)
+        identities = {}
+        for path, node in result.nodes.items():
+            identity = node['maj:min']
+            if identity in identities and identities[identity] != path:
+                raise InstallError('Conflicting block-device identities')
+            identities[identity] = path
+        directed = {path: set() for path in result.nodes}
+        for child, parent in parents:
+            result.link(child, parent)
+            directed[child].add(parent)
+        visiting, visited = set(), set()
+        def check_cycle(path):
+            if path in visiting:
+                raise InstallError('Cyclic block-device parent references')
+            if path in visited:
+                return
+            visiting.add(path)
+            for parent in directed[path]:
+                check_cycle(parent)
+            visiting.remove(path)
+            visited.add(path)
+        for path in directed:
+            check_cycle(path)
+        for path, node in result.nodes.items():
+            if node['size'] == 0 and result.edges[path]:
+                raise InstallError('Empty optical drive has block-device dependencies')
+            # An undirected child edge does not establish backing ancestry.
+            # Each layer must identify its own parent, even when another
+            # apparently unrelated disk is the requested installation target.
+            if node['type'] not in {'disk', 'rom', 'loop', 'zram'} and not directed[path]:
+                raise InstallError(f"Device has no known backing parent: {path} ({node['type']})")
+        return result
+
+    def link(self, child, parent):
+        if child == parent or child not in self.nodes or parent not in self.nodes:
+            raise InstallError('Unresolved or cyclic block-device parent')
+        self.edges[child].add(parent)
+        self.edges[parent].add(child)
+
+    def protect(self, path):
+        if path not in self.nodes:
+            raise InstallError(f'Cannot resolve mounted/live source: {path}')
+        if self.nodes[path]['size'] == 0:
+            raise InstallError(f'Empty optical drive unexpectedly backs mounted/live storage: {path}')
+        self.protected.add(path)
+
+    def component(self, path):
+        seen, pending = set(), [path]
+        while pending:
+            current = pending.pop()
+            if current not in seen:
+                seen.add(current)
+                pending.extend(self.edges[current] - seen)
+        return seen
+
+
+class SystemProbe:
+    """Read-only Linux topology probes, including loop backing files."""
+    def __init__(self, runner=None):
+        self.runner = runner or Runner()
+
+    def resolve(self, path):
+        return os.path.realpath(path)
+
+    def physical_disk(self, path):
+        try:
+            info = os.stat(path)
+            if not stat.S_ISBLK(info.st_mode):
+                return False
+            device = Path(f'/sys/dev/block/{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}')
+            if not device.exists() or (device / 'partition').exists():
+                return False
+            # Reject loop, device mapper, md, ram and other virtual targets.
+            return '/virtual/' not in str(device.resolve()) and (device / 'device').exists()
+        except OSError:
+            return False
+
+    def snapshot(self, payload):
+        raw = self.runner.run(['lsblk', '--json', '--paths', '--bytes', '--output',
+                               'PATH,TYPE,PKNAME,MOUNTPOINTS,SIZE,MAJ:MIN'])
+        # lsblk PATH can use /dev/mapper aliases while findmnt uses /dev/dm-N.
+        try:
+            decoded = json.loads(raw)
+            def canonicalize(node):
+                if not isinstance(node, dict):
+                    raise ValueError()
+                if not isinstance(node.get('path'), str) or not DEVICE.fullmatch(node['path']) or '..' in PurePosixPath(node['path']).parts:
+                    raise ValueError()
+                node['path'] = self.resolve(node['path'])
+                if isinstance(node.get('pkname'), str):
+                    parent = node['pkname']
+                    node['pkname'] = self.resolve(parent if parent.startswith('/dev/') else '/dev/' + parent)
+                children = node.get('children', [])
+                if not isinstance(children, list):
+                    raise ValueError()
+                for child in children:
+                    canonicalize(child)
+            for node in decoded['blockdevices']:
+                canonicalize(node)
+            topo = Topology.from_json(json.dumps(decoded))
+        except (ValueError, TypeError, KeyError) as exc:
+            raise InstallError('Malformed lsblk JSON') from exc
+        mounts = self._mounts(self.runner.run(['findmnt', '--json', '--output',
+                                             'SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS']))
+        for mount in mounts:
+            source = self._block_source(mount, topo)
+            if source:
+                topo.protect(source)
+        # Mounted loops may hide the physical filesystem that stores their file.
+        for path, node in topo.nodes.items():
+            if node['type'] == 'loop' and path in topo.protected:
+                backing_file = Path('/sys/dev/block') / node['maj:min'] / 'loop/backing_file'
+                try:
+                    backing = backing_file.read_text().strip()
+                except OSError as exc:
+                    raise InstallError(f'Cannot inspect mounted loop backing file: {path}') from exc
+                backing = re.sub(r'\\([0-7]{3})', lambda m: chr(int(m[1], 8)), backing)
+                if not backing.startswith('/'):
+                    backing = '/' + backing
+                source = self._source_for_path(backing, topo)
+                if not source:
+                    raise InstallError(f'Unknown physical backing for mounted loop: {path}')
+                topo.link(path, source)
+        if not Path(payload).is_file():
+            raise InstallError(f'Prepared squashfs payload is not a regular file: {payload}')
+        source = self._source_for_path(self.resolve(payload), topo)
+        if not source:
+            # tmpfs/overlay payload location cannot establish live-medium ancestry.
+            raise InstallError('Cannot identify the payload backing block device; use --payload on the mounted live medium')
+        topo.protect(source)
+        return topo
+
+    @staticmethod
+    def _mounts(raw):
+        try:
+            data = json.loads(raw)
+            roots = data['filesystems']
+            if not isinstance(roots, list) or not roots:
+                raise ValueError()
+            result = []
+
+            def visit(item):
+                if not isinstance(item, dict) or any(not isinstance(item.get(k), str) for k in ('source', 'target', 'fstype', 'maj:min')):
+                    raise ValueError()
+                result.append(item)
+                children = item.get('children', [])
+                if not isinstance(children, list):
+                    raise ValueError()
+                for child in children:
+                    visit(child)
+            for root in roots:
+                visit(root)
+            return result
+        except (ValueError, TypeError, KeyError) as exc:
+            raise InstallError('Malformed findmnt topology') from exc
+
+    def _block_source(self, mount, topo):
+        if mount['fstype'] in {'zfs', 'btrfs'}:
+            raise InstallError('Cannot safely resolve all backing disks of mounted ' + mount['fstype'])
+        source = mount['source'].split('[', 1)[0]
+        identity = mount['maj:min']
+        matches = [p for p, n in topo.nodes.items() if n['maj:min'] == identity]
+        if source.startswith('/dev/'):
+            path = self.resolve(source)
+            if path in topo.nodes:
+                if topo.nodes[path]['maj:min'] != identity:
+                    raise InstallError('Inconsistent mounted block identity')
+                topo.protect(path)
+                return path
+            if not matches:
+                raise InstallError(f'Cannot resolve mounted/live source: {source}')
+        if matches:
+            topo.protect(matches[0])
+            return matches[0]
+        if identity.split(':', 1)[0] != '0' and mount['fstype'] not in {'zfs', 'nfs', 'nfs4', 'cifs', '9p', 'virtiofs'}:
+            raise InstallError(f'Unknown mounted block source: {source}')
+        return None
+
+    def _source_for_path(self, path, topo):
+        raw = self.runner.run(['findmnt', '--json', '--target', path, '--output',
+                               'SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS'])
+        mounts = self._mounts(raw)
+        if len(mounts) != 1:
+            raise InstallError('Ambiguous backing filesystem')
+        return self._block_source(mounts[0], topo)
+
+
+def validate_request(request, probe):
+    if not isinstance(request.disk, str) or not DEVICE.fullmatch(request.disk) or '..' in PurePosixPath(request.disk).parts:
+        raise InstallError('Choose an explicit /dev whole-disk device')
+    if not re.fullmatch(r'[a-z_][a-z0-9_-]{0,31}', request.user) or request.user in {'root', 'hydra-live', 'greetd'}:
+        raise InstallError('Choose a valid nonreserved lowercase Unix username')
+    if not re.fullmatch(r'[A-Za-z_+-]+(?:/[A-Za-z0-9_+-]+)+', request.timezone):
+        raise InstallError('Invalid timezone; use an IANA name such as Europe/Moscow')
+    if not Path(request.payload).is_absolute() or any(ord(c) < 32 for c in request.payload):
+        raise InstallError('Payload must be an absolute path without control characters')
+    if request.kernel is not None and not KERNEL.fullmatch(request.kernel):
+        raise InstallError('Invalid kernel version')
+    target = probe.resolve(request.disk)
+    if not DEVICE.fullmatch(target) or not probe.physical_disk(target):
+        raise InstallError(f'Target is not a physical whole block disk: {target}')
+    if request.execute and request.confirm_erase != target:
+        raise InstallError(f'Execution requires --confirm-erase {target} exactly')
+    topo = probe.snapshot(request.payload)
+    if target not in topo.nodes or topo.nodes[target]['type'] != 'disk':
+        raise InstallError('Target is not a known whole disk')
+    if topo.nodes[target]['size'] < 4 * 1024 ** 3:
+        raise InstallError('Target disk is smaller than the minimum 4 GiB')
+    forbidden = topo.component(target) & topo.protected
+    if forbidden:
+        raise InstallError('Target backs mounted/root/live storage: ' + ', '.join(sorted(forbidden)))
+    return target
+
+
+def partitions(target):
+    suffix = 'p' if target[-1].isdigit() else ''
+    return target + suffix + '1', target + suffix + '2'
+
+
+def partition_commands(target):
+    esp, root = partitions(target)
+    return [
+        ['parted', '--script', target, 'mklabel', 'gpt'],
+        ['parted', '--script', target, 'unit', 'MiB', 'mkpart', 'ESP', 'fat32', '1MiB', '1025MiB'],
+        ['parted', '--script', target, 'set', '1', 'esp', 'on'],
+        ['parted', '--script', target, 'unit', 'MiB', 'mkpart', 'root', 'ext4', '1025MiB', '100%'],
+        ['partprobe', target], ['udevadm', 'settle'],
+        ['mkfs.vfat', '-F', '32', '-n', 'HYDRA_EFI', esp],
+        ['mkfs.ext4', '-F', '-L', 'HYDRA_ROOT', root],
+    ]
+
+
+def grub_command(root):
+    return ['chroot', str(root), 'grub-install', '--target=x86_64-efi',
+            '--efi-directory=/boot/efi', '--bootloader-id=HydraOS', '--removable', '--no-nvram']
+
+
+def render_plan(request, target, root):
+    esp, part = partitions(target)
+    commands = partition_commands(target) + [
+        ['mount', part, str(root)], ['unsquashfs', '-f', '-d', str(root), request.payload],
+        ['mount', esp, str(Path(root) / 'boot/efi')],
+        ['blkid', '-s', 'UUID', '-o', 'value', part],
+        ['blkid', '-s', 'UUID', '-o', 'value', esp],
+        ['mount', '--rbind', '/dev', str(Path(root) / 'dev')],
+        ['mount', '--make-rslave', str(Path(root) / 'dev')],
+        ['mount', '-t', 'proc', 'proc', str(Path(root) / 'proc')],
+        ['mount', '--rbind', '/sys', str(Path(root) / 'sys')],
+        ['mount', '--make-rslave', str(Path(root) / 'sys')],
+        ['chroot', str(root), 'locale-gen'],
+        ['chroot', str(root), 'useradd', '--create-home', '--groups', 'wheel,kvm,video,render,audio', '--shell', '/bin/bash', request.user],
+        ['chroot', str(root), 'chpasswd'],
+        ['chroot', str(root), 'passwd', '--lock', 'root'],
+        ['chroot', str(root), 'systemd-machine-id-setup'],
+        ['chroot', str(root), 'dracut', '--force', '--no-hostonly', '--omit', 'dmsquash-live',
+         '/boot/initramfs-<kernel>.img', '<kernel>'],
+        grub_command(root), ['chroot', str(root), 'grub-mkconfig', '-o', '/boot/grub/grub.cfg'],
+    ]
+    lines = [f'HydraOS {"EXECUTION" if request.execute else "dry-run"}: erase all data on {target}.',
+             'GPT: exactly 1 GiB FAT32 ESP, remaining space ext4 root.',
+             f'Prepared payload: {request.payload}; account: {request.user}; timezone: {request.timezone}.',
+             'Commands (UUIDs/kernel discovered after unpacking; no commands run in dry-run):']
+    lines.extend('  ' + shlex.join(c) for c in commands)
+    lines.extend(['Write UUID fstab, en_US.UTF-8 and ru_RU.UTF-8, timezone, installed dracut/GRUB defaults.',
+                  'Remove allowlisted live-mode files, hydra-live account and /home/hydra-live; reset machine-id.',
+                  'Ensure wheel,kvm,video,render,audio groups exist before creating the installed account.',
+                  'chpasswd receives the TTY-entered password on private stdin; root remains locked.',
+                  'Unmount tracked sys/proc/dev/ESP/root in reverse order on success or failure.',
+                  f'To execute, add --execute --confirm-erase {target} (as root; no automatic sudo).'])
+    return '\n'.join(lines)
+
+
+def target_path(root, relative, *, symlinks=False):
+    """Resolve a path inside a target, interpreting absolute symlinks as chroot paths."""
+    root = Path(root).resolve()
+    parts = list(PurePosixPath(relative).parts)
+    if parts and parts[0] == '/':
+        parts.pop(0)
+    resolved, hops = [], 0
+    while parts:
+        part = parts.pop(0)
+        if part in ('', '.'):
+            continue
+        if part == '..':
+            if not resolved:
+                raise InstallError('Target path escapes root')
+            resolved.pop()
+            continue
+        candidate = root.joinpath(*resolved, part)
+        if candidate.is_symlink():
+            if not symlinks:
+                raise InstallError(f'Refusing symlink traversal in target: {relative}')
+            hops += 1
+            if hops > 40:
+                raise InstallError('Target symlink cycle')
+            link = PurePosixPath(os.readlink(candidate))
+            extra = list(link.parts)
+            if link.is_absolute():
+                resolved = []
+                extra.pop(0)
+            parts = extra + parts
+        else:
+            resolved.append(part)
+    return root.joinpath(*resolved)
+
+
+def remove_live_files(root):
+    directory = target_path(root, 'etc/hydra/live-mode')
+    if not directory.is_dir():
+        raise InstallError('Payload missing /etc/hydra/live-mode contract')
+    manifest = target_path(root, 'etc/hydra/live-mode/remove-on-install.json')
+    marker = target_path(root, 'etc/hydra/live-mode/active')
+    if not manifest.is_file() or not marker.is_file():
+        raise InstallError('Payload live-mode marker/manifest missing')
+    try:
+        paths = json.loads(manifest.read_text())
+    except (OSError, ValueError) as exc:
+        raise InstallError('Invalid live-mode removal manifest') from exc
+    if not isinstance(paths, list):
+        raise InstallError('Live removal manifest must be a JSON array')
+    removals = []
+    for path in paths:
+        if not isinstance(path, str) or not path.startswith('/') or '..' in PurePosixPath(path).parts or str(PurePosixPath(path)) != path:
+            raise InstallError('Invalid path in live removal manifest')
+        if not any(re.fullmatch(re.escape(pattern).replace(r'\*', r'[A-Za-z0-9_.-]+'), path) for pattern in LIVE_REMOVABLE):
+            raise InstallError(f'Live manifest path is not allowlisted: {path}')
+        leaf = PurePosixPath(path)
+        parent = target_path(root, str(leaf.parent))
+        candidate = parent / leaf.name
+        if candidate.exists() and not candidate.is_file() and not candidate.is_symlink():
+            raise InstallError('Live manifest may only remove files/symlinks')
+        removals.append(candidate)
+    for path in removals:
+        path.unlink(missing_ok=True)
+    # shutil.rmtree never follows directory symlinks inside this directory.
+    shutil.rmtree(directory)
+
+
+def remove_live_account(root, run):
+    """Remove the payload's fixed local live identity, before binding live /proc."""
+    passwd = target_path(root, 'etc/passwd')
+    try:
+        rows = [row.split(':') for row in passwd.read_text().splitlines() if row.split(':', 1)[0] == 'hydra-live']
+    except OSError as exc:
+        raise InstallError('Payload missing readable /etc/passwd') from exc
+    if len(rows) > 1:
+        raise InstallError('Duplicate live account in payload')
+    if rows:
+        row = rows[0]
+        if len(row) != 7 or not row[2].isdigit() or int(row[2]) < 1000 or row[5] != '/home/hydra-live':
+            raise InstallError('Live account must have a nonroot UID and exactly /home/hydra-live')
+    home = target_path(root, 'home/hydra-live')
+    if home.exists() and not home.is_dir():
+        raise InstallError('Live home must be an ordinary directory')
+    if rows:
+        # /proc is not mounted yet, so the live session's identical UID cannot
+        # make userdel mistake a running live process for a target process.
+        run(['chroot', str(root), 'userdel', 'hydra-live'])
+    if home.exists():
+        shutil.rmtree(home)
+    for spool in ('var/mail/hydra-live', 'var/spool/mail/hydra-live'):
+        parent = target_path(root, str(PurePosixPath(spool).parent), symlinks=True)
+        leaf = parent / 'hydra-live'
+        if leaf.is_file() or leaf.is_symlink():
+            leaf.unlink()
+
+
+def select_kernel(root, requested):
+    modules = target_path(root, '/lib/modules', symlinks=True)
+    if not modules.is_dir():
+        raise InstallError('Payload has no installed kernel modules')
+    versions = sorted(p.name for p in modules.iterdir() if not p.is_symlink() and p.is_dir() and KERNEL.fullmatch(p.name))
+    if requested:
+        if requested not in versions:
+            raise InstallError(f'Requested kernel not installed: {requested}')
+        version = requested
+    elif len(versions) == 1:
+        version = versions[0]
+    else:
+        raise InstallError('Select --kernel explicitly when payload has zero or multiple kernels: ' + ', '.join(versions))
+    if not target_path(root, '/boot/vmlinuz-' + version, symlinks=True).is_file():
+        raise InstallError(f'Payload missing /boot/vmlinuz-{version}')
+    return version
+
+
+def write_config(root, relative, value, mode=0o644):
+    path = target_path(root, relative)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(value)
+    path.chmod(mode)
+
+
+def install(request, runner, probe, root, password, log: TextIO):
+    """Execute into an empty mount directory; inject a runner for disk-free tests."""
+    if not request.execute:
+        raise InstallError('Installation requires --execute')
+    if not isinstance(password, str) or not password or any(c in password for c in '\r\n\x00'):
+        raise InstallError('Password must be nonempty and contain no newline or NUL')
+    target = validate_request(request, probe)
+    root = Path(root)
+    if not root.is_absolute() or root.is_symlink() or not root.is_dir() or any(root.iterdir()):
+        raise InstallError('Target mount directory must be an empty absolute directory')
+    mounted = []
+    error = None
+    journal = []
+
+    def emit(message):
+        journal.append(message)
+        print(message, file=log, flush=True)
+
+    def run(argv, **kw):
+        emit('+ ' + shlex.join(argv))
+        return runner.run(argv, **kw)
+
+    def mount(argv, path, recursive=False):
+        run(argv)
+        mounted.append((str(path), recursive))
+
+    try:
+        run(['unsquashfs', '-s', request.payload])
+        # Recheck after preflight, directly before the first destructive command.
+        if validate_request(request, probe) != target:
+            raise InstallError('Target device changed during validation')
+        for argv in partition_commands(target):
+            run(argv)
+        esp, part = partitions(target)
+        mount(['mount', part, str(root)], root)
+        run(['unsquashfs', '-f', '-d', str(root), request.payload])
+        kernel = select_kernel(root, request.kernel)
+        remove_live_files(root)
+        remove_live_account(root, run)
+        if not target_path(root, '/usr/share/zoneinfo/' + request.timezone, symlinks=True).is_file():
+            raise InstallError('Timezone does not exist in payload: ' + request.timezone)
+        for folder in ('boot/efi', 'dev', 'proc', 'sys'):
+            target_path(root, folder).mkdir(parents=True, exist_ok=True)
+        mount(['mount', esp, str(root / 'boot/efi')], root / 'boot/efi')
+        root_uuid = run(['blkid', '-s', 'UUID', '-o', 'value', part]).strip()
+        esp_uuid = run(['blkid', '-s', 'UUID', '-o', 'value', esp]).strip()
+        if any(not re.fullmatch(r'[A-Fa-f0-9-]+', value) for value in (root_uuid, esp_uuid)):
+            raise InstallError('Missing or invalid partition UUID')
+        write_config(root, 'etc/fstab', f'UUID={root_uuid} / ext4 defaults 0 1\nUUID={esp_uuid} /boot/efi vfat umask=0077 0 2\n')
+        write_config(root, 'etc/locale.gen', 'en_US.UTF-8 UTF-8\nru_RU.UTF-8 UTF-8\n')
+        write_config(root, 'etc/locale.conf', 'LANG=en_US.UTF-8\n')
+        write_config(root, 'etc/timezone', request.timezone + '\n')
+        localtime = target_path(root, 'etc/localtime', symlinks=True)
+        # Replace even an absolute /etc/localtime symlink without touching the host.
+        localtime_link = target_path(root, 'etc') / 'localtime'
+        if localtime_link.is_symlink():
+            localtime_link.unlink()
+        elif localtime.exists():
+            localtime.unlink()
+        (root / 'etc/localtime').symlink_to('/usr/share/zoneinfo/' + request.timezone)
+        write_config(root, 'etc/machine-id', '')
+        dbus_machine_id = target_path(root, 'var/lib/dbus') / 'machine-id'
+        dbus_machine_id.unlink(missing_ok=True)
+        write_config(root, 'etc/dracut.conf.d/90-hydra-installed.conf',
+                     'hostonly="no"\nhostonly_cmdline="no"\nomit_dracutmodules+=" dmsquash-live "\n# No root=live: or rd.live arguments on the installed disk.\n')
+        write_config(root, 'etc/default/grub', 'GRUB_DISTRIBUTOR="HydraOS"\nGRUB_CMDLINE_LINUX=""\nGRUB_TIMEOUT=3\n')
+        write_config(root, 'etc/sudoers.d/90-hydra-admin', '%wheel ALL=(ALL:ALL) ALL\n', 0o440)
+        mount(['mount', '--rbind', '/dev', str(root / 'dev')], root / 'dev', True)
+        run(['mount', '--make-rslave', str(root / 'dev')])
+        mount(['mount', '-t', 'proc', 'proc', str(root / 'proc')], root / 'proc')
+        mount(['mount', '--rbind', '/sys', str(root / 'sys')], root / 'sys', True)
+        run(['mount', '--make-rslave', str(root / 'sys')])
+        def chroot(*args, **kw):
+            return run(['chroot', str(root), *args], **kw)
+        chroot('locale-gen')
+        for group in ('wheel', 'kvm', 'video', 'render', 'audio'):
+            chroot('groupadd', '--force', '--system', group)
+        chroot('useradd', '--create-home', '--groups', 'wheel,kvm,video,render,audio', '--shell', '/bin/bash', request.user)
+        chroot('chpasswd', input=f'{request.user}:{password}\n', sensitive=True)
+        password = None
+        chroot('passwd', '--lock', 'root')
+        chroot('systemd-machine-id-setup')
+        chroot('dracut', '--force', '--no-hostonly', '--omit', 'dmsquash-live',
+               '/boot/initramfs-' + kernel + '.img', kernel)
+        run(grub_command(root))
+        chroot('grub-mkconfig', '-o', '/boot/grub/grub.cfg')
+        run(['sync'])
+    except (InstallError, OSError, KeyboardInterrupt) as exc:
+        error = exc
+        emit(f'Installation stopped: {exc}. Target data retained for diagnosis.')
+    finally:
+        password = None
+        if any(path == str(root) for path, _ in mounted):
+            try:
+                write_config(root, 'var/log/hydra-install.log', '\n'.join(journal) + '\n', 0o600)
+            except (InstallError, OSError) as exc:
+                emit(f'Unable to retain diagnostic in target: {exc}')
+        cleanup_errors = []
+        for path, recursive in reversed(mounted):
+            try:
+                run(['umount', *(['--recursive'] if recursive else []), path])
+            except (InstallError, OSError) as exc:
+                cleanup_errors.append(str(exc))
+                emit(f'cleanup failed: {exc}')
+        if cleanup_errors and error is None:
+            error = InstallError('Mount cleanup failed: ' + '; '.join(cleanup_errors))
+    if error:
+        raise InstallError(str(error)) from error
+    print('Installation finished; remove the live ISO before booting the installed disk.', file=log, flush=True)
+
+
+def preflight(request, runner):
+    if any(Path(marker).exists() for marker in ('/.dockerenv', '/run/.containerenv', '/run/systemd/container')):
+        raise InstallError('Refusing disk installation in a container; boot the live ISO in a disposable VM')
+    if os.geteuid() != 0:
+        raise InstallError('Execution requires root inside the live environment; no automatic sudo')
+    if not sys.stdin.isatty():
+        raise InstallError('Execution requires a TTY for password entry')
+    if not Path('/sys/firmware/efi').is_dir():
+        raise InstallError('Boot the live medium in UEFI mode before installation')
+    for command in ('lsblk', 'findmnt', 'parted', 'partprobe', 'udevadm', 'mkfs.vfat',
+                    'mkfs.ext4', 'mount', 'umount', 'unsquashfs', 'blkid', 'chroot', 'sync'):
+        if shutil.which(command) is None:
+            raise InstallError('Required live installer tool missing: ' + command)
+    runner.run(['unsquashfs', '-s', request.payload])
+
+
+def prompt_password():
+    try:
+        # Only the prompt output stream is needed here; getpass opens its own
+        # raw controlling-TTY input. Buffered r+ requires seekability.
+        with open('/dev/tty', 'w') as tty:
+            if not tty.isatty():
+                raise InstallError('Password entry requires a controlling TTY')
+            with warnings.catch_warnings():
+                warnings.simplefilter('error', getpass.GetPassWarning)
+                password = getpass.getpass('Installed account password: ', stream=tty)
+                if password != getpass.getpass('Repeat password: ', stream=tty):
+                    raise InstallError('Passwords differ')
+            return password
+    except (OSError, getpass.GetPassWarning) as exc:
+        raise InstallError('Cannot securely read password from the controlling TTY') from exc
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description='HydraOS explicit whole-disk installer; dry-run by default')
+    parser.add_argument('--disk', help='explicit physical whole-disk target; never automatically selected')
+    parser.add_argument('--user', help='installed administrator username')
+    parser.add_argument('--payload', default=DEFAULT_PAYLOAD, help='prepared live squashfs root payload')
+    parser.add_argument('--execute', action='store_true')
+    parser.add_argument('--confirm-erase', help='must exactly match the resolved /dev target with --execute')
+    parser.add_argument('--timezone', default='Europe/Moscow')
+    parser.add_argument('--kernel', help='required if payload contains multiple kernel module versions')
+    args = parser.parse_args(argv)
+    runner = Runner()
+    try:
+        if not args.disk:
+            if args.execute or args.confirm_erase:
+                raise InstallError('--execute requires an explicit --disk and --user')
+            topo = Topology.from_json(runner.run(['lsblk', '--json', '--paths', '--bytes', '--output',
+                                                'PATH,TYPE,PKNAME,MOUNTPOINTS,SIZE,MAJ:MIN']))
+            print('Disks (no target selected; mounted disks are ineligible):')
+            for path, node in sorted(topo.nodes.items()):
+                if node['type'] == 'disk':
+                    state = 'in use' if topo.component(path) & topo.protected else 'unmounted; requires full guard check'
+                    print(f'  {path}  {node["size"] / 1024**3:.1f} GiB  {state}')
+            print('Choose --disk DEVICE --user USER to inspect the dry-run plan.')
+            return 0
+        if not args.user:
+            raise InstallError('--user is required with --disk')
+        request = Request(**vars(args))
+        probe = SystemProbe(runner)
+        target = validate_request(request, probe)
+        print(render_plan(request, target, '/mnt/hydra-target'))
+        if not request.execute:
+            return 0
+        preflight(request, runner)
+        password = prompt_password()
+        state = Path(tempfile.mkdtemp(prefix='hydra-install-', dir='/run'))
+        state.chmod(0o700)
+        root = state / 'target'
+        root.mkdir()
+        logfile = state / 'installer.log'
+        print(f'Diagnostic log: {logfile}')
+        with logfile.open('w') as log:
+            log.write(render_plan(request, target, str(root)) + '\n')
+            install(request, runner, probe, root, password, log)
+        password = None
+        print(f'Installation completed. Diagnostic log: {logfile}')
+        return 0
+    except (InstallError, OSError, KeyboardInterrupt) as exc:
+        print(f'hydra-install: {exc}', file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

hydra/profiles.py

@@ -0,0 +1,72 @@
+"""Validated, independently editable VM profiles."""
+
+import json
+import re
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+
+
+def validate_name(name: str) -> str:
+    if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', name):
+        raise ValueError('VM name must match [a-z][a-z0-9-]{0,31}')
+    return name
+
+
+@dataclass(frozen=True)
+class Profile:
+    name: str
+    iso: Path
+    memory: int = 4096
+    cpus: int = 2
+    disk_size: str = '40G'
+    graphics: str = 'auto'
+    network: str = 'user'
+    bridge: str = 'hydrabr0'
+
+    def __post_init__(self):
+        validate_name(self.name)
+        iso = Path(self.iso).expanduser().resolve()
+        if not iso.is_file():
+            raise ValueError(f'ISO does not exist or is not a file: {iso}')
+        object.__setattr__(self, 'iso', iso)
+        for key, upper in (('memory', 262144), ('cpus', 256)):
+            value = getattr(self, key)
+            if type(value) is not int or not 1 <= value <= upper:
+                raise ValueError(f'{key} must be an integer between 1 and {upper}')
+        if not isinstance(self.disk_size, str) or not re.fullmatch(r'[1-9][0-9]{0,5}[MGT]', self.disk_size):
+            raise ValueError('disk_size must be a positive size with M, G or T suffix')
+        if self.graphics not in ('auto', 'software'):
+            raise ValueError('graphics must be auto or software')
+        if self.network not in ('user', 'bridge'):
+            raise ValueError('network must be user or bridge')
+        if not isinstance(self.bridge, str) or not re.fullmatch(r'[a-zA-Z0-9_-]{1,15}', self.bridge):
+            raise ValueError('bridge must be a valid Linux interface name')
+
+
+def load_profile(path: Path) -> Profile:
+    path = Path(path)
+    with path.open('rb') as stream:
+        data = tomllib.load(stream)
+    unknown = data.keys() - Profile.__dataclass_fields__.keys()
+    if unknown:
+        raise ValueError(f'Unknown profile fields: {", ".join(sorted(unknown))}')
+    name = data.pop('name', path.stem)
+    if name != path.stem:
+        raise ValueError('Profile name must match its filename')
+    try:
+        return Profile(name=name, **data)
+    except TypeError as exc:
+        raise ValueError(f'Invalid profile: {exc}') from exc
+
+
+def save_profile(path: Path, profile: Profile) -> None:
+    path = Path(path)
+    if path.stem != profile.name:
+        raise ValueError('Profile name must match its filename')
+    lines = []
+    for field in Profile.__dataclass_fields__:
+        value = getattr(profile, field)
+        lines.append(f'{field} = {value if type(value) is int else json.dumps(str(value), ensure_ascii=False)}\n')
+    with path.open('x', encoding='utf-8') as stream:
+        stream.write(''.join(lines))

hydra/qmp.py

@@ -0,0 +1,91 @@
+"""Bounded JSON QMP over private Unix sockets, with reply ID matching."""
+
+import json
+import socket
+import time
+from pathlib import Path
+
+
+class QMPError(RuntimeError):
+    pass
+
+
+class QMPClient:
+    def __init__(self, path: Path, timeout: float = 2.0):
+        if timeout <= 0:
+            raise ValueError('QMP timeout must be positive')
+        self.timeout = timeout
+        self.socket = socket.socket(socket.AF_UNIX)
+        self.buffer = bytearray()
+        self.counter = 0
+        deadline = time.monotonic() + timeout
+        try:
+            self.socket.settimeout(timeout)
+            self.socket.connect(str(path))
+            greeting = self._receive(deadline)
+            if not isinstance(greeting.get('QMP'), dict):
+                raise QMPError('Invalid QMP greeting')
+            self._command('qmp_capabilities', None, deadline)
+        except (OSError, QMPError) as exc:
+            self.close()
+            if isinstance(exc, QMPError):
+                raise
+            raise QMPError(f'QMP connection failed: {exc}') from exc
+
+    def _receive(self, deadline: float) -> dict:
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise QMPError('QMP deadline exceeded')
+            if b'\n' in self.buffer:
+                line, _, rest = self.buffer.partition(b'\n')
+                self.buffer = bytearray(rest)
+                try:
+                    message = json.loads(line)
+                except (ValueError, UnicodeDecodeError) as exc:
+                    raise QMPError('Malformed QMP JSON') from exc
+                if not isinstance(message, dict):
+                    raise QMPError('QMP response must be an object')
+                return message
+            self.socket.settimeout(remaining)
+            try:
+                chunk = self.socket.recv(65536)
+            except OSError as exc:
+                raise QMPError(f'QMP receive failed: {exc}') from exc
+            if not chunk:
+                raise QMPError('QMP connection closed')
+            self.buffer.extend(chunk)
+            if len(self.buffer) > 1048576:
+                raise QMPError('QMP message exceeds 1 MiB')
+
+    def _command(self, name: str, arguments: dict | None, deadline: float) -> dict:
+        self.counter += 1
+        request = {'execute': name, 'id': self.counter}
+        if arguments is not None:
+            request['arguments'] = arguments
+        try:
+            self.socket.settimeout(max(0.001, deadline - time.monotonic()))
+            self.socket.sendall(json.dumps(request).encode() + b'\r\n')
+        except OSError as exc:
+            raise QMPError(f'QMP send failed: {exc}') from exc
+        while True:
+            response = self._receive(deadline)
+            if response.get('id') != self.counter:
+                continue
+            if 'error' in response:
+                raise QMPError(f'QMP {name}: {response["error"]}')
+            if 'return' not in response:
+                raise QMPError('QMP reply has no return value')
+            return response['return']
+
+    def command(self, name: str, arguments: dict | None = None) -> dict:
+        return self._command(name, arguments, time.monotonic() + self.timeout)
+
+    def close(self):
+        self.socket.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()

hydra/runtime.py

@@ -0,0 +1,297 @@
+"""Independent QEMU lifecycle; all destructive control requires QMP identity."""
+
+import contextlib
+import errno
+import fcntl
+import json
+import os
+import socket
+import stat
+import subprocess
+import tempfile
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+
+from .profiles import Profile, load_profile, save_profile, validate_name
+from .qmp import QMPClient, QMPError
+
+
+class RuntimeError(Exception):
+    """A VM lifecycle operation could not be safely completed."""
+
+
+@dataclass(frozen=True)
+class VMPaths:
+    name: str
+    disk: Path
+    log: Path
+    lock: Path
+    qmp: Path
+    spice: Path
+    pid: Path
+    metadata: Path
+
+    @property
+    def uuid(self) -> str:
+        return str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.disk)))
+
+
+def _qemu_value(value: Path | str) -> str:
+    # QEMU's key/value parser uses doubled commas for literal commas.
+    return str(value).replace(',', ',,')
+
+
+def _intel_render_node() -> Path | None:
+    for node in sorted(Path('/dev/dri').glob('renderD*')):
+        vendor = Path('/sys/class/drm') / node.name / 'device/vendor'
+        try:
+            if vendor.read_text().strip() == '0x8086' and os.access(node, os.R_OK | os.W_OK):
+                return node
+        except OSError:
+            continue
+    return None
+
+
+def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
+    accel = 'kvm' if os.access('/dev/kvm', os.R_OK | os.W_OK) else 'tcg'
+    argv = [os.environ.get('HYDRA_QEMU', 'qemu-system-x86_64'), '-name', profile.name,
+            '-uuid', paths.uuid, '-machine', f'q35,accel={accel}',
+            '-m', str(profile.memory), '-smp', str(profile.cpus),
+            '-daemonize', '-pidfile', str(paths.pid), '-nodefaults',
+            '-chardev', f'socket,id=qmp,path={_qemu_value(paths.qmp)},server=on,wait=off',
+            '-mon', 'chardev=qmp,mode=control']
+    if accel == 'kvm':
+        argv += ['-cpu', 'host']
+    for options in (
+        {'driver': 'file', 'filename': str(paths.disk), 'node-name': 'disk-file'},
+        {'driver': 'qcow2', 'file': 'disk-file', 'node-name': 'disk'},
+        {'driver': 'file', 'filename': str(profile.iso), 'node-name': 'iso-file', 'read-only': True},
+        {'driver': 'raw', 'file': 'iso-file', 'node-name': 'iso', 'read-only': True},
+    ):
+        argv += ['-blockdev', json.dumps(options)]
+    argv += ['-device', 'virtio-blk-pci,drive=disk', '-device', 'ide-cd,drive=iso',
+             '-boot', 'order=dc', '-device', 'virtio-net-pci,netdev=net',
+             '-netdev', 'user,id=net' if profile.network == 'user'
+             else f'bridge,id=net,br={profile.bridge}',
+             '-device', 'qemu-xhci,id=usb', '-device', 'usb-tablet', '-device', 'usb-kbd']
+    argv += ['-audiodev', 'spice,id=audio', '-device', 'ich9-intel-hda',
+             '-device', 'hda-duplex,audiodev=audio']
+    for channel in range(3):
+        argv += ['-chardev', f'spicevmc,id=usbredir{channel},name=usbredir',
+                 '-device', f'usb-redir,chardev=usbredir{channel}']
+    spice = (f'unix=on,addr={_qemu_value(paths.spice)},disable-ticketing=on,'
+             'disable-copy-paste=on,disable-agent-file-xfer=on')
+    render = _intel_render_node() if profile.graphics == 'auto' else None
+    if render:
+        spice += f',gl=on,rendernode={_qemu_value(render)}'
+        argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl']
+    else:
+        argv += ['-display', 'none', '-device', 'virtio-vga']
+    return argv + ['-spice', spice]
+
+
+def _private_dir(path: Path) -> None:
+    if path.is_symlink():
+        raise RuntimeError(f'Private directory must not be a symlink: {path}')
+    path.mkdir(parents=True, mode=0o700, exist_ok=True)
+    info = path.stat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
+        raise RuntimeError(f'Private directory is not owned by this user: {path}')
+    path.chmod(0o700)
+
+
+class Runtime:
+    def __init__(self, config: Path, state: Path, runtime: Path, boot_timeout: float = 10):
+        # Keep final components lexical so symlink checks still see them.
+        self.config = Path(config).expanduser().absolute()
+        self.state = Path(state).expanduser().absolute()
+        self.runtime = Path(runtime).expanduser().absolute()
+        self.boot_timeout = boot_timeout
+
+    def paths(self, name: str) -> VMPaths:
+        validate_name(name)
+        state = self.state / name
+        run = self.runtime / name
+        paths = VMPaths(name, state / 'disk.qcow2', state / 'qemu.log', state / 'lock',
+                        run / 'qmp.sock', run / 'spice.sock', run / 'qemu.pid', run / 'identity.json')
+        if any(len(os.fsencode(path)) >= 108 for path in (paths.qmp, paths.spice)):
+            raise RuntimeError('Unix socket path is too long; choose a shorter --runtime path')
+        return paths
+
+    def prepare(self, paths: VMPaths) -> None:
+        for directory in (self.config, self.state, self.runtime, paths.disk.parent, paths.qmp.parent):
+            _private_dir(directory)
+
+    def add(self, profile: Profile) -> None:
+        _private_dir(self.config)
+        save_profile(self.config / f'{profile.name}.toml', profile)
+
+    def profiles(self) -> list[Profile]:
+        if not self.config.exists():
+            return []
+        return [load_profile(path) for path in sorted(self.config.glob('*.toml'))]
+
+    def profile(self, name: str) -> Profile:
+        validate_name(name)
+        return load_profile(self.config / f'{name}.toml')
+
+    @contextlib.contextmanager
+    def _locked(self, paths: VMPaths):
+        self.prepare(paths)
+        fd = os.open(paths.lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX)
+            yield
+        finally:
+            os.close(fd)
+
+    def _owned_process_alive(self, paths: VMPaths) -> bool:
+        # PID is only a diagnostic hint. Never send a signal to it.
+        try:
+            pid = int(paths.pid.read_text())
+            args = Path(f'/proc/{pid}/cmdline').read_bytes().split(b'\0')
+            wanted_uuid = paths.uuid.encode()
+            wanted_name = paths.name.encode()
+            return (b'-uuid' in args and args[args.index(b'-uuid') + 1] == wanted_uuid
+                    and b'-name' in args and args[args.index(b'-name') + 1] == wanted_name)
+        except (OSError, ValueError, IndexError):
+            return False
+
+    def _connect(self, paths: VMPaths) -> QMPClient:
+        try:
+            record = json.loads(paths.metadata.read_text())
+            if not isinstance(record, dict) or record.get('name') != paths.name or record.get('uuid') != paths.uuid:
+                raise RuntimeError('VM identity record does not match selected VM')
+            client = QMPClient(paths.qmp, timeout=min(2, self.boot_timeout))
+            try:
+                name_result = client.command('query-name')
+                uuid_result = client.command('query-uuid')
+                if not isinstance(name_result, dict) or not isinstance(uuid_result, dict):
+                    raise RuntimeError('Malformed QMP identity reply')
+                name = name_result.get('name')
+                actual_uuid = uuid_result.get('UUID')
+                if name != paths.name or actual_uuid != record['uuid']:
+                    raise RuntimeError('QMP VM identity mismatch; refusing control')
+            except Exception:
+                client.close()
+                raise
+            return client
+        except (OSError, ValueError, QMPError) as exc:
+            raise RuntimeError(f'Cannot verify VM identity: {exc}') from exc
+
+    def _listening(self, path: Path) -> bool:
+        try:
+            mode = path.lstat().st_mode
+        except FileNotFoundError:
+            return False
+        if not stat.S_ISSOCK(mode):
+            raise RuntimeError(f'Refusing non-socket runtime path: {path}')
+        with socket.socket(socket.AF_UNIX) as probe:
+            probe.settimeout(min(1, self.boot_timeout))
+            try:
+                probe.connect(str(path))
+                return True
+            except OSError as exc:
+                if exc.errno in (errno.ENOENT, errno.ECONNREFUSED):
+                    return False
+                raise RuntimeError(f'Cannot establish socket ownership: {exc}') from exc
+
+    def status(self, name: str) -> dict:
+        paths = self.paths(name)
+        try:
+            if not self._listening(paths.qmp):
+                if self._owned_process_alive(paths):
+                    return {'name': name, 'status': 'unknown', 'error': 'VM process exists without working QMP'}
+                return {'name': name, 'status': 'stopped'}
+            with self._connect(paths) as client:
+                status = client.command('query-status')
+            if not isinstance(status, dict) or not isinstance(status.get('status'), str):
+                raise RuntimeError('Malformed QMP status reply')
+            return {'name': name, **status}
+        except (RuntimeError, QMPError) as exc:
+            return {'name': name, 'status': 'unknown', 'error': str(exc)}
+
+    def _create_disk(self, profile: Profile, paths: VMPaths) -> None:
+        if paths.disk.exists() or paths.disk.is_symlink():
+            if paths.disk.is_symlink() or not paths.disk.is_file():
+                raise RuntimeError(f'Refusing non-regular disk: {paths.disk}')
+            return
+        fd, temporary = tempfile.mkstemp(prefix='.disk-', suffix='.qcow2', dir=paths.disk.parent)
+        os.close(fd)
+        temp = Path(temporary)
+        temp.unlink()  # qemu-img creates a new file; final publication never replaces a disk.
+        try:
+            subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'create', '-f', 'qcow2',
+                            str(temp), profile.disk_size], check=True, capture_output=True,
+                           text=True, timeout=30)
+            temp.chmod(0o600)
+            os.link(temp, paths.disk)
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise RuntimeError(f'Disk creation failed: {exc}') from exc
+        finally:
+            temp.unlink(missing_ok=True)
+
+    def start(self, name: str) -> dict:
+        profile = self.profile(name)
+        paths = self.paths(name)
+        with self._locked(paths):
+            current = self.status(name)
+            if current['status'] != 'stopped':
+                raise RuntimeError(f'Cannot start {name}: {current["status"]}')
+            for path in (paths.qmp, paths.spice):
+                if self._listening(path):
+                    raise RuntimeError(f'Runtime socket is already in use: {path}')
+                path.unlink(missing_ok=True)
+            self._create_disk(profile, paths)
+            fd, temporary = tempfile.mkstemp(prefix='.identity-', dir=paths.metadata.parent)
+            try:
+                with os.fdopen(fd, 'w') as record:
+                    record.write(json.dumps({'name': name, 'uuid': paths.uuid}) + '\n')
+                os.replace(temporary, paths.metadata)
+            finally:
+                Path(temporary).unlink(missing_ok=True)
+            paths.pid.unlink(missing_ok=True)
+            try:
+                log_fd = os.open(paths.log, os.O_CREAT | os.O_APPEND | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
+                with os.fdopen(log_fd, 'ab') as log:
+                    subprocess.run(build_qemu_argv(profile, paths), stdin=subprocess.DEVNULL,
+                                   stdout=log, stderr=log, start_new_session=True,
+                                   check=True, timeout=self.boot_timeout,
+                                   umask=0o077)
+                deadline = time.monotonic() + self.boot_timeout
+                while time.monotonic() < deadline:
+                    current = self.status(name)
+                    if current['status'] not in ('stopped', 'unknown'):
+                        return current
+                    time.sleep(0.05)
+                raise RuntimeError(f'QEMU did not become ready; inspect {paths.log}')
+            except (OSError, subprocess.SubprocessError) as exc:
+                raise RuntimeError(f'QEMU launch failed; inspect {paths.log}: {exc}') from exc
+
+    def shutdown(self, name: str) -> dict:
+        paths = self.paths(name)
+        try:
+            with self._locked(paths), self._connect(paths) as client:
+                client.command('system_powerdown')
+        except QMPError as exc:
+            raise RuntimeError(f'Guest shutdown request failed: {exc}') from exc
+        return {'name': name, 'status': 'shutdown-requested'}
+
+    def stop(self, name: str, force: bool = False) -> dict:
+        if not force:
+            raise RuntimeError('Abrupt stop requires --force; use shutdown for guest ACPI shutdown')
+        paths = self.paths(name)
+        with self._locked(paths):
+            try:
+                with self._connect(paths) as client:
+                    client.command('quit')
+            except QMPError as exc:
+                raise RuntimeError(f'QEMU quit request failed: {exc}') from exc
+            deadline = time.monotonic() + self.boot_timeout
+            while time.monotonic() < deadline:
+                if self.status(name)['status'] == 'stopped':
+                    return {'name': name, 'status': 'stopped'}
+                time.sleep(0.05)
+            raise RuntimeError('QEMU accepted quit but has not stopped within the deadline')

hydra/session.py

@@ -0,0 +1,263 @@
+"""Minimal labwc session; viewer and compositor lifetimes never control QEMU."""
+
+import fcntl
+import json
+import os
+import re
+import shlex
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+from urllib.parse import quote
+from xml.etree import ElementTree as ET
+
+from .profiles import validate_name
+from .runtime import Runtime, VMPaths, _private_dir
+
+
+class SessionError(ValueError):
+    """An actionable viewer/session error safe to present at the CLI."""
+
+
+def viewer_title(name: str) -> str:
+    return 'HydraOS: ' + validate_name(name)
+
+
+def build_viewer_argv(paths: VMPaths, viewer: str, *, windowed=False) -> list[str]:
+    argv = [viewer, '--title', viewer_title(paths.name), '--hotkeys',
+            'release-cursor=ctrl+alt+shift+r,usb-device-select=ctrl+alt+shift+u',
+            '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0',
+            '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0']
+    if not windowed:
+        argv += ['--full-screen', '--hide-toolbar']
+    # QEMU denies clipboard/file transfer; no shared-dir/CD flag is added here.
+    return argv + ['spice+unix://' + quote(str(paths.spice), safe='/')]
+
+
+def build_labwc_config(names: list[str], recovery_argv: list[str],
+                       brightness='brightnessctl', *, initial: str | None = None) -> str:
+    names = [validate_name(name) for name in names]
+    if len(set(names)) != len(names):
+        raise SessionError('Duplicate VM workspace name')
+    if initial is not None and initial not in names:
+        raise SessionError('Initial VM workspace must be configured')
+    root = ET.Element('labwc_config')
+    core = ET.SubElement(root, 'core')
+    ET.SubElement(core, 'primarySelection').text = 'no'
+    desktops = ET.SubElement(root, 'desktops', number=str(max(1, len(names))))
+    labels = ET.SubElement(desktops, 'names')
+    for name in names:
+        ET.SubElement(labels, 'name').text = name
+    if initial is not None:
+        ET.SubElement(desktops, 'initial').text = str(names.index(initial) + 1)
+    ET.SubElement(desktops, 'popupTime').text = '0'
+    theme = ET.SubElement(root, 'theme')
+    ET.SubElement(theme, 'keepBorder').text = 'no'
+    keyboard = ET.SubElement(root, 'keyboard')
+    for key, direction in (('C-A-S-Left', 'left'), ('C-A-S-Right', 'right')):
+        binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes')
+        ET.SubElement(binding, 'action', name='GoToDesktop', to=direction, wrap='yes')
+    commands = [('C-A-S-Escape', recovery_argv),
+                ('XF86MonBrightnessUp', [brightness, '-c', 'backlight', 'set', '+5%']),
+                ('XF86MonBrightnessDown', [brightness, '-c', 'backlight', 'set', '5%-'])]
+    for key, argv in commands:
+        binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes')
+        ET.SubElement(binding, 'action', name='Execute', command=shlex.join(argv))
+    # An empty mouse section restores labwc defaults. A binding with no action
+    # suppresses that fallback without consuming guest window pointer events.
+    context = ET.SubElement(ET.SubElement(root, 'mouse'), 'context', name='Root')
+    ET.SubElement(context, 'mousebind', button='Left', action='Press')
+    device = ET.SubElement(ET.SubElement(root, 'libinput'), 'device', category='touchpad')
+    ET.SubElement(device, 'tap').text = 'yes'
+    rules = ET.SubElement(root, 'windowRules')
+    for index, name in enumerate(names, 1):
+        rule = ET.SubElement(rules, 'windowRule', title=viewer_title(name),
+                             serverDecoration='no')
+        ET.SubElement(rule, 'action', name='SendToDesktop', to=str(index), follow='no')
+        ET.SubElement(rule, 'action', name='SetDecorations', decorations='none')
+    ET.indent(root)
+    return ET.tostring(root, encoding='unicode', xml_declaration=True) + '\n'
+
+
+def _tool(variable: str, default: str) -> str:
+    value = os.environ.get(variable, default)
+    executable = shutil.which(value)
+    if not executable:
+        raise SessionError(f'Missing {default} tool ({variable}); enter the temporary dev/tools.nix environment')
+    return executable
+
+
+def _display() -> None:
+    if not os.environ.get('WAYLAND_DISPLAY') and not os.environ.get('DISPLAY'):
+        raise SessionError('No graphical display; start a session or use a graphical host terminal')
+
+
+def _viewer() -> str:
+    viewer = _tool('HYDRA_VIEWER', 'remote-viewer')
+    result = subprocess.run([viewer, '--help'], capture_output=True, text=True, timeout=10)
+    if result.returncode or '--hide-toolbar' not in result.stdout:
+        raise SessionError('Viewer lacks HydraOS toolbar/USB adaptation; use dev/tools.nix')
+    return viewer
+
+
+def _socket(paths: VMPaths) -> None:
+    try:
+        mode = paths.spice.lstat().st_mode
+    except FileNotFoundError as exc:
+        raise SessionError(f'{paths.name}: no SPICE socket; run hydra start {paths.name}, then retry') from exc
+    if not stat.S_ISSOCK(mode):
+        raise SessionError(f'{paths.name}: SPICE path is not a Unix socket')
+
+
+def _viewer_environment(directory: Path) -> dict[str, str]:
+    config = directory / 'config'
+    settings = config / 'virt-viewer'
+    settings.mkdir(parents=True, mode=0o700)
+    (settings / 'settings').write_text('[virt-viewer]\nshare-clipboard=false\n', encoding='utf-8')
+    env = dict(os.environ, XDG_CONFIG_HOME=str(config), GSETTINGS_BACKEND='memory')
+    # A dedicated backend prevents accidental XWayland input behavior in HydraOS.
+    if env.get('WAYLAND_DISPLAY'):
+        env['GDK_BACKEND'] = 'wayland'
+    return env
+
+
+def view(runtime: Runtime, name: str, *, windowed=False) -> int:
+    _display()
+    viewer = _viewer()
+    runtime.profile(name)
+    paths = runtime.paths(name)
+    _socket(paths)
+    _private_dir(runtime.runtime)
+    with tempfile.TemporaryDirectory(prefix='viewer-', dir=runtime.runtime) as tmp:
+        return subprocess.run(build_viewer_argv(paths, viewer, windowed=windowed),
+                              env=_viewer_environment(Path(tmp))).returncode
+
+
+def compositor_environment(environ: dict[str, str], *, nested=False) -> dict[str, str]:
+    env = dict(environ)
+    if nested:
+        if env.get('WAYLAND_DISPLAY'):
+            env['WLR_BACKENDS'] = 'wayland'
+        elif env.get('DISPLAY'):
+            env['WLR_BACKENDS'] = 'x11'
+        else:
+            raise SessionError('Nested session requires an existing graphical display')
+    else:
+        env.pop('WAYLAND_DISPLAY', None)
+        env.pop('DISPLAY', None)
+        env['WLR_BACKENDS'] = 'drm,libinput'
+    env['XDG_CURRENT_DESKTOP'] = 'HydraOS'
+    return env
+
+
+def _module_argv(command: str, manifest: Path) -> list[str]:
+    # Absolute source launcher also works when labwc changes the working directory.
+    return [sys.executable, str(Path(__file__).resolve().parents[1] / 'bin/hydra'),
+            '_session-helper', command, str(manifest)]
+
+
+def _recovery_argv(manifest: Path, terminal: str) -> list[str]:
+    return [terminal, '--title=HydraOS-recovery'] + _module_argv('recover', manifest)
+
+
+def launch_session(runtime: Runtime, *, nested=False) -> int:
+    env = compositor_environment(dict(os.environ), nested=nested)
+    labwc = _tool('HYDRA_LABWC', 'labwc')
+    terminal = _tool('HYDRA_TERMINAL', 'foot')
+    brightness = _tool('HYDRA_BRIGHTNESS', 'brightnessctl')
+    version = subprocess.run([labwc, '--version'], capture_output=True, text=True, timeout=10)
+    match = re.search(r'(\d+)\.(\d+)\.(\d+)', version.stdout)
+    if version.returncode or not match or tuple(map(int, match.groups())) < (0, 20, 0):
+        raise SessionError('HydraOS requires labwc >= 0.20.0 for reserved host input bindings')
+    profiles = runtime.profiles()
+    initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None)
+    _private_dir(runtime.runtime)
+    with tempfile.TemporaryDirectory(prefix='session-', dir=runtime.runtime) as tmp:
+        directory = Path(tmp)
+        config = directory / 'labwc'
+        config.mkdir(mode=0o700)
+        manifest = directory / 'session.json'
+        manifest.write_text(json.dumps({'config': str(runtime.config), 'state': str(runtime.state),
+            'runtime': str(runtime.runtime), 'names': [p.name for p in profiles],
+            'terminal': terminal, 'directory': str(directory)}), encoding='utf-8')
+        (config / 'rc.xml').write_text(build_labwc_config([p.name for p in profiles],
+                                    _recovery_argv(manifest, terminal), brightness,
+                                    initial=initial), encoding='utf-8')
+        (config / 'autostart').write_text('#!/bin/sh\n' + shlex.join(_module_argv('clients', manifest)) + ' &\n', encoding='utf-8')
+        # Empty environment files prevent loading unrelated /etc/xdg defaults.
+        (config / 'environment').touch()
+        env['PYTHONUNBUFFERED'] = '1'
+        return subprocess.run([labwc, '-C', str(config)], env=env).returncode
+
+
+def session_helper(command: str, manifest: Path) -> int:
+    data = json.loads(manifest.read_text(encoding='utf-8'))
+    if (not isinstance(data, dict) or
+        any(not isinstance(data.get(key), str) or not Path(data[key]).is_absolute()
+            for key in ('config', 'state', 'runtime', 'terminal')) or
+        not isinstance(data.get('names'), list)):
+        raise SessionError('Invalid session manifest')
+    for name in data['names']:
+        validate_name(name)
+    if command not in ('clients', 'recover') and (
+        not command.startswith('reconnect:') or command.split(':', 1)[1] not in data['names']):
+        raise SessionError('Invalid session helper operation')
+    runtime = Runtime(Path(data['config']), Path(data['state']), Path(data['runtime']))
+    if command == 'recover':
+        print('HydraOS local recovery. VMs keep running independently.\n'
+              'Ctrl+Alt+Shift+Left/Right switches workspaces; Ctrl+Alt+Shift+R releases input; Ctrl+Alt+Shift+U opens USB.\n'
+              'Enter a VM name to reconnect, :shell for a host shell, :quit to close this terminal.')
+        while True:
+            try:
+                print(json.dumps([runtime.status(name) for name in data['names']], indent=2))
+                choice = input('hydra recovery> ').strip()
+            except (EOFError, KeyboardInterrupt):
+                return 0
+            if choice == ':quit':
+                return 0
+            if choice == ':shell':
+                subprocess.run([os.environ.get('SHELL', '/bin/sh')])
+            elif choice in data['names']:
+                subprocess.Popen(_module_argv('reconnect:' + choice, manifest), start_new_session=True)
+            else:
+                print('Enter a configured VM name, :shell, or :quit.')
+    if command.startswith('reconnect:'):
+        name = command.split(':', 1)[1]
+        # One viewer per VM inside this session; independent host view commands
+        # retain their separate lifecycle and never share VM manager locks.
+        with (manifest.parent / (name + '.viewer.lock')).open('a') as lock:
+            try:
+                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                print(f'{name}: viewer is already connected; close recovery or switch workspace')
+                return 0
+            return view(runtime, name)
+    processes = []
+    recovery = None
+    def open_recovery():
+        nonlocal recovery
+        if recovery is None or recovery.poll() is not None:
+            recovery = subprocess.Popen(_recovery_argv(manifest, data['terminal']))
+    for name in data['names']:
+        try:
+            if runtime.status(name)['status'] == 'running':
+                processes.append(subprocess.Popen(_module_argv('reconnect:' + name, manifest)))
+        except (OSError, ValueError) as exc:
+            print(f'hydra: {name}: {exc}', file=sys.stderr)
+    if not processes:
+        open_recovery()
+    # No signals to QEMU or viewers: compositor disconnect tears down clients,
+    # while the VM manager's detached QEMU processes retain their lifetime.
+    while processes or (recovery is not None and recovery.poll() is None):
+        for process in processes[:]:
+            code = process.poll()
+            if code is not None:
+                processes.remove(process)
+                print(f'hydra: viewer exited ({code}); reconnect from local recovery', file=sys.stderr)
+                open_recovery()
+        time.sleep(0.2)
+    return 0

image/README.md

@@ -0,0 +1,38 @@
+HydraOS builds start from the pinned amd64 systemd stage3 and pinned Gentoo
+repository in `sources.json`. Run the builder as the normal user with Python 3,
+Podman and GnuPG available; temporary tool environments are sufficient. It
+refuses host root and work/cache paths outside the primary project checkout.
+
+```
+scripts/hydra-build fetch
+scripts/hydra-build prepare
+scripts/hydra-build packages --pretend
+scripts/hydra-build packages
+scripts/hydra-build status
+```
+
+`--work PATH`, `--cache PATH`, `--podman PATH` and `--gpg PATH` override defaults.
+Linked worktrees use the primary checkout's `build/` and `cache/`. The build
+lock prevents simultaneous writes in the same work directory. Interrupted
+package installation resumes through Portage; successful steps write
+`build/manifest.json`, signature evidence and the installed package/version/USE
+inventory in `build/artifacts/`. Downloaded inputs are reverified on every
+prepare; missing or changed files cannot be trusted from a checkpoint alone.
+
+The dedicated rootless Podman store is `build/podman`, separate from default
+Podman and Distrobox storage. The build container mounts inputs and repository
+read-only and retains its writable layer across failures. Host networking makes
+the existing local VPN/proxy available. `SYS_ADMIN` operates within the rootless
+user namespace; package sandbox features incompatible with it are disabled.
+No host disks or `/dev/kvm` are passed into the builder. Hydra never prunes
+containers/images or changes host services.
+
+Official Gentoo binpkgs require signatures. USE mismatches and selective testing
+keywords can cause source builds; concurrency is four compiler processes and
+one emerge job. Keep licenses and Portage's `/var/db/pkg`, world and repository
+for installed-system updates. Rolling binhost contents can change; resulting
+versions/USE are recorded without claiming bit-identical output. The pinned
+service-keyring hash may need deliberate review when Gentoo rotates keys.
+
+`rootfs` and `iso` are reserved assembly commands and currently return an
+explicit error until the live-image assembly implementation is integrated.

image/packages

@@ -0,0 +1,30 @@
+# Host, boot medium, and recovery tools. Keep Portage's installed database/world.
+app-emulation/qemu
+~sys-firmware/edk2-bin-202408
+>=gui-wm/labwc-0.20.0
+app-emulation/virt-viewer
+gui-apps/foot
+media-fonts/dejavu
+gui-libs/greetd
+net-misc/networkmanager
+net-dns/dnsmasq
+net-firewall/nftables
+media-video/pipewire
+media-video/wireplumber
+app-admin/sudo
+sys-kernel/gentoo-kernel-bin
+sys-kernel/linux-firmware
+sys-firmware/sof-firmware
+sys-firmware/intel-microcode
+sys-kernel/dracut
+sys-boot/grub
+sys-fs/squashfs-tools
+sys-fs/dosfstools
+sys-fs/e2fsprogs
+sys-block/parted
+sys-apps/util-linux
+sys-fs/mtools
+dev-libs/libisoburn
+dev-lang/python
+sec-keys/openpgp-keys-gentoo-release
+sec-keys/openpgp-keys-gentoo-auth

image/portage/binrepos.conf/gentoo.conf

@@ -0,0 +1,4 @@
+[gentoo]
+sync-uri = https://distfiles.gentoo.org/releases/amd64/binpackages/23.0/x86-64/
+priority = 10
+verify-signature = yes

image/portage/make.conf

@@ -0,0 +1,16 @@
+COMMON_FLAGS="-O2 -pipe -march=x86-64 -mtune=generic"
+CFLAGS="${COMMON_FLAGS}"
+CXXFLAGS="${COMMON_FLAGS}"
+FCFLAGS="${COMMON_FLAGS}"
+FFLAGS="${COMMON_FLAGS}"
+MAKEOPTS="-j4"
+EMERGE_DEFAULT_OPTS="--jobs=1 --getbinpkg --binpkg-respect-use=y --autounmask=n"
+FEATURES="getbinpkg binpkg-request-signature -usersandbox -sandbox -network-sandbox -ipc-sandbox -pid-sandbox"
+ACCEPT_LICENSE="@FREE @BINARY-REDISTRIBUTABLE linux-fw-redistributable no-source-code Intel-SDP"
+VIDEO_CARDS="intel"
+USE="X dbus wayland opengl alsa"
+QEMU_SOFTMMU_TARGETS="x86_64"
+QEMU_USER_TARGETS=""
+GRUB_PLATFORMS="efi-64"
+BINPKG_GPG_VERIFY_GPG_HOME="/etc/portage/gnupg"
+PORTAGE_NICENESS="10"

image/portage/package.accept_keywords/hydra

@@ -0,0 +1,2 @@
+>=gui-wm/labwc-0.20.0 ~amd64
+gui-libs/wlroots ~amd64

image/portage/package.use/hydra

@@ -0,0 +1,8 @@
+app-emulation/qemu spice usb usbredir virgl opengl wayland pipewire -gtk -sdl
+app-emulation/virt-viewer spice -libvirt
+net-misc/spice-gtk usbredir
+media-video/pipewire sound-server bluetooth dbus pipewire-alsa
+net-misc/networkmanager wifi
+sys-boot/grub mount
+media-libs/freetype harfbuzz
+sys-kernel/installkernel dracut

image/portage/repos.conf/gentoo.conf

@@ -0,0 +1,5 @@
+[gentoo]
+location = /var/db/repos/gentoo
+sync-type = git
+sync-uri = https://github.com/gentoo/gentoo.git
+auto-sync = no

image/sources.json

@@ -0,0 +1,17 @@
+{
+  "stage3": {
+    "url": "https://distfiles.gentoo.org/releases/amd64/autobuilds/20261011T170049Z/stage3-amd64-systemd-20261011T170049Z.tar.xz",
+    "sha256": "50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b",
+    "signature_url": "https://distfiles.gentoo.org/releases/amd64/autobuilds/20261011T170049Z/stage3-amd64-systemd-20261011T170049Z.tar.xz.asc",
+    "fingerprint": "13EBBDBEDE7A12775DFDB1BABB572E0E2D182910"
+  },
+  "portage": {
+    "url": "https://github.com/gentoo/gentoo/archive/4696089e782c20b5f0c04952f6698d6a8cc1acb4.tar.gz",
+    "filename": "gentoo-4696089e782c20b5f0c04952f6698d6a8cc1acb4.tar.gz",
+    "sha256": "d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca",
+    "commit": "4696089e782c20b5f0c04952f6698d6a8cc1acb4"
+  },
+  "keyring_url": "https://qa-reports.gentoo.org/output/service-keys.gpg",
+  "keyring_sha256": "9fe40114c956c4c4d593f6e961ac9a994b0772863f4e685608fbdd8d77e39748",
+  "binhost": "https://distfiles.gentoo.org/releases/amd64/binpackages/23.0/x86-64/"
+}

patches/virt-viewer-hydra-session.patch

@@ -0,0 +1,112 @@
+--- a/src/virt-viewer-app.c
++++ b/src/virt-viewer-app.c
+@@ -134,6 +134,7 @@
+     gboolean authretry;
+     gboolean started;
+     gboolean fullscreen;
++    gboolean hide_toolbar;
+     gboolean attach;
+     gboolean shared;
+     gboolean quitting;
+@@ -951,6 +952,13 @@
+ {
+     gchar *subtitle = NULL;
+     const gchar *title = virt_viewer_app_get_title(app);
++
++    /* Hidden fullscreen windows use the requested title as stable identity. */
++    if (title != NULL && virt_viewer_app_get_hide_toolbar(app) &&
++        virt_viewer_app_get_fullscreen(app)) {
++        g_object_set(window, "subtitle", title, NULL);
++        return;
++    }
+ 
+     if (title != NULL) {
+         VirtViewerDisplay *display = virt_viewer_window_get_display(window);
+@@ -2134,6 +2142,7 @@
+ static gboolean opt_verbose = FALSE;
+ static gboolean opt_debug = FALSE;
+ static gboolean opt_fullscreen = FALSE;
++static gboolean opt_hide_toolbar = FALSE;
+ static gboolean opt_kiosk = FALSE;
+ static gboolean opt_kiosk_quit = FALSE;
+ static gchar *opt_cursor = NULL;
+@@ -2437,6 +2446,7 @@
+     { "smartcard-remove", "app.smartcard-remove", {"<Shift>F9", NULL, NULL} },
+     { "secure-attention", "win.secure-attention", {"<Ctrl><Alt>End", NULL, NULL} },
+     { "usb-device-reset", "win.usb-device-reset", {"<Ctrl><Shift>r", NULL, NULL} },
++    { "usb-device-select", "win.usb-device-select", {NULL, NULL, NULL} },
+ };
+ 
+ static gchar **hotkey_names;
+@@ -2484,6 +2494,7 @@
+     priv->resource = virt_viewer_get_resource();
+ 
+     virt_viewer_app_set_debug(opt_debug);
++    priv->hide_toolbar = opt_hide_toolbar;
+     virt_viewer_app_set_fullscreen(self, opt_fullscreen);
+ 
+     virt_viewer_app_set_keymap(self, opt_keymap);
+@@ -3098,6 +3109,14 @@
+ }
+ 
+ gboolean
++virt_viewer_app_get_hide_toolbar(VirtViewerApp *self)
++{
++    VirtViewerAppPrivate *priv = virt_viewer_app_get_instance_private(self);
++
++    return priv->hide_toolbar;
++}
++
++gboolean
+ virt_viewer_app_get_fullscreen(VirtViewerApp *self)
+ {
+     g_return_val_if_fail(VIRT_VIEWER_IS_APP(self), FALSE);
+@@ -3557,6 +3576,8 @@
+           N_("Zoom level of window, in percentage"), "ZOOM" },
+         { "full-screen", 'f', 0, G_OPTION_ARG_NONE, &opt_fullscreen,
+           N_("Open in full screen mode (adjusts guest resolution to fit the client)"), NULL },
++        { "hide-toolbar", '\0', 0, G_OPTION_ARG_NONE, &opt_hide_toolbar,
++          N_("Hide the toolbar in full screen mode"), NULL },
+         { "hotkeys", 'H', 0, G_OPTION_ARG_STRING, &opt_hotkeys,
+           N_("Customise hotkeys"), NULL },
+         { "auto-resize", '\0', 0, G_OPTION_ARG_STRING, &opt_resize,
+--- a/src/virt-viewer-app.h
++++ b/src/virt-viewer-app.h
+@@ -100,6 +100,7 @@
+ GList* virt_viewer_app_get_windows(VirtViewerApp *self);
+ VirtViewerSession* virt_viewer_app_get_session(VirtViewerApp *self);
+ gboolean virt_viewer_app_get_fullscreen(VirtViewerApp *app);
++gboolean virt_viewer_app_get_hide_toolbar(VirtViewerApp *app);
+ void virt_viewer_app_clear_hotkeys(VirtViewerApp *app);
+ GList* virt_viewer_app_get_initial_displays(VirtViewerApp* self);
+ gint virt_viewer_app_get_initial_monitor_for_display(VirtViewerApp* self, gint display);
+--- a/src/virt-viewer-window.c
++++ b/src/virt-viewer-window.c
+@@ -740,9 +740,10 @@
+         return;
+     }
+ 
+-    if (!self->kiosk) {
+-        virt_viewer_timed_revealer_force_reveal(self->revealer, TRUE);
+-    }
++    gboolean hide_toolbar = virt_viewer_app_get_hide_toolbar(self->app);
++    gtk_widget_set_visible(GTK_WIDGET(self->revealer), !hide_toolbar);
++    virt_viewer_timed_revealer_force_reveal(self->revealer,
++                                          !self->kiosk && !hide_toolbar);
+ 
+     if (self->display) {
+         virt_viewer_display_set_monitor(self->display, monitor);
+@@ -1354,6 +1355,13 @@
+         title = g_strdup(g_get_application_name());
+     }
+ 
++    /* Keep exact workspace identity across pointer/keyboard grab changes. */
++    if (self->app && virt_viewer_app_get_hide_toolbar(self->app) &&
++        virt_viewer_app_get_fullscreen(self->app) && self->subtitle) {
++        g_free(title);
++        title = g_strdup(self->subtitle);
++    }
++
+     gtk_window_set_title(GTK_WINDOW(self->window), title);
+     if (self->subtitle) {
+         gtk_header_bar_set_title(GTK_HEADER_BAR(header), self->subtitle);

scripts/hydra-build

@@ -0,0 +1,9 @@
+#!/usr/bin/env python3
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from hydra.build import main
+
+if __name__ == '__main__':
+    raise SystemExit(main())

session/hydra-session

@@ -0,0 +1,4 @@
+#!/bin/sh
+# Portable source launcher. Desktop/session packaging may call this directly.
+HYDRA_CHECKOUT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) || exit 1
+exec "$HYDRA_CHECKOUT/bin/hydra" session "$@"

tests/test_build.py

@@ -0,0 +1,320 @@
+import hashlib
+import contextlib
+import io
+import importlib.util
+import json
+import lzma
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+class BuilderTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.project = Path(self.tmp.name)
+        spec = importlib.util.spec_from_file_location('hydra_build', Path(__file__).parents[1] / 'hydra/build.py')
+        self.assertIsNotNone(spec, 'builder module is required')
+        self.assertTrue(Path(spec.origin).is_file(), 'builder implementation is missing')
+        self.module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.module)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def builder(self):
+        return self.module.Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
+
+    def test_bad_cached_hash_is_rejected(self):
+        artifact = self.project / 'archive'
+        artifact.write_bytes(b'tampered')
+        with self.assertRaisesRegex(self.module.BuildError, 'SHA256'):
+            self.module.verify_hash(artifact, hashlib.sha256(b'authentic').hexdigest())
+
+    def test_wrong_primary_signature_is_rejected(self):
+        status = '[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 DEADBEEF\n'
+        with self.assertRaisesRegex(self.module.BuildError, 'fingerprint'):
+            self.module.require_signature(status, '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910')
+
+    def test_signature_accepts_only_successful_matching_primary(self):
+        fingerprint = '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910'
+        self.module.require_signature('[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 ' + fingerprint, fingerprint)
+        with self.assertRaises(self.module.BuildError):
+            self.module.require_signature('[GNUPG:] BADSIG ABC test', fingerprint)
+
+    def test_refuses_root_and_outside_project(self):
+        with patch.object(self.module.os, 'geteuid', return_value=0):
+            with self.assertRaisesRegex(self.module.BuildError, 'root'):
+                self.builder()
+        with patch.object(self.module.os, 'geteuid', return_value=1000):
+            with self.assertRaisesRegex(self.module.BuildError, 'project'):
+                self.module.Builder(self.project, self.project.parent / 'unsafe', self.project / 'cache')
+
+    def test_podman_store_is_scoped_to_work(self):
+        argv = self.builder().podman_argv('container', 'exists', 'builder')
+        self.assertEqual(argv[0], 'podman')
+        self.assertEqual(argv[argv.index('--root') + 1], str(self.project / 'build/podman'))
+        self.assertIn('--runroot', argv)
+        self.assertEqual(argv[-3:], ['container', 'exists', 'builder'])
+
+    def test_resume_requires_actual_inventory(self):
+        builder = self.builder()
+        builder.work.mkdir()
+        (builder.work / 'manifest.json').write_text(json.dumps({'packages': {'complete': True}}))
+        self.assertFalse(builder.complete('packages'))
+
+    def test_symlink_escape_is_rejected(self):
+        (self.project / 'escape').symlink_to(self.project.parent, target_is_directory=True)
+        with self.assertRaisesRegex(self.module.BuildError, 'project'):
+            self.module.Builder(self.project, self.project / 'escape/unsafe', self.project / 'cache')
+
+    def test_changed_configuration_invalidates_fetch_checkpoint(self):
+        image = self.project / 'image'
+        image.mkdir()
+        config = image / 'sources.json'
+        config.write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
+        builder = self.builder()
+        with builder.locked():
+            (builder.cache / 'stage.tar').write_bytes(b'good')
+            (builder.cache / 'repo.tar').write_bytes(b'repo')
+            builder.record('fetch')
+            self.assertTrue(builder.complete('fetch'))
+            config.write_text('{"changed":true}')
+            self.assertFalse(builder.complete('fetch'))
+
+    def test_missing_fetch_archive_invalidates_checkpoint(self):
+        image = self.project / 'image'
+        image.mkdir()
+        (image / 'sources.json').write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
+        builder = self.builder()
+        with builder.locked():
+            builder.record('fetch')
+            self.assertFalse(builder.complete('fetch'))
+
+    def test_second_builder_cannot_share_active_work(self):
+        first, second = self.builder(), self.builder()
+        with first.locked():
+            with self.assertRaisesRegex(self.module.BuildError, 'running'):
+                with second.locked():
+                    self.fail('overlapping build was allowed')
+
+    def test_container_uses_host_proxy_network_and_readonly_inputs(self):
+        builder = self.builder()
+        args = builder.container_run_args()
+        self.assertEqual(args[args.index('--network') + 1], 'host')
+        self.assertIn(str(self.project) + ':/hydra-source:ro', args)
+        self.assertIn(str(self.project / 'cache') + ':/hydra-cache:ro', args)
+        self.assertNotIn('--privileged', args)
+
+    def test_distinct_workspaces_have_distinct_podman_runtime_storage(self):
+        other = self.module.Builder(self.project, self.project / 'other-build', self.project / 'cache')
+        self.assertNotEqual(self.builder().runroot, other.runroot)
+
+    def existing_builder(self, source=None, cache=None, layer=None, snapshot='old-snapshot'):
+        """A real executable models only the external Podman boundary."""
+        source = source or self.project
+        cache = cache or self.project / 'cache'
+        image = self.project / 'image'
+        image.mkdir(exist_ok=True)
+        inputs = {'stage3': {'url': 'https://example.test/stage.tar.xz',
+                            'sha256': hashlib.sha256(lzma.compress(b'original layer')).hexdigest()},
+                  'portage': {'filename': 'repo.tar.gz', 'commit': 'old-snapshot'}}
+        (image / 'sources.json').write_text(json.dumps(inputs))
+        builder = self.builder()
+        (builder.work / 'podman').mkdir(parents=True)
+        fixture = self.project / 'podman-fixture.json'
+        fixture.write_text(json.dumps({'Image': 'original-image-id',
+            'Mounts': [{'Source': str(source), 'Destination': '/hydra-source', 'RW': False},
+                       {'Source': str(cache), 'Destination': '/hydra-cache', 'RW': False}],
+            'RootFS': {'Layers': [layer or 'sha256:' + hashlib.sha256(b'original layer').hexdigest()]},
+            'snapshot': snapshot}))
+        executable = self.project / 'fixture-podman'
+        executable.write_text('''#!/usr/bin/env python3
+import json, pathlib, sys
+root = pathlib.Path(__file__).parent
+data = json.loads((root / 'podman-fixture.json').read_text())
+args = sys.argv[7:]
+if 'exists' in args:
+    sys.exit(0)
+if 'inspect' in args:
+    print(json.dumps([data]))
+elif args[0] == 'top':
+    print('HPID ARGS')
+    if data.get('active_job'):
+        print('123 /usr/bin/python3 /usr/lib/python-exec/python3/emerge --update package')
+elif args[0] == 'cp':
+    snapshot = data.get('snapshot_pending') if args[1].endswith('.hydra-snapshot-input') else data['snapshot']
+    if snapshot is None:
+        sys.exit(1)
+    pathlib.Path(args[-1]).write_text(snapshot + '\\n')
+elif args[0] == 'exec' and data.get('stopped'):
+    sys.exit(125)
+elif args[0] == 'exec' and 'packages.lock' in args[-1]:
+    sys.exit(75 if data.get('job_lock') else 0)
+elif args[0] == 'exec' and args[-1].startswith('cat '):
+    snapshot = data['snapshot'] or (data.get('snapshot_pending') if '.hydra-snapshot-input' in args[-1] else None)
+    if snapshot is None:
+        sys.exit(1)
+    print(snapshot)
+else:
+    if args[0] == 'start':
+        data['stopped'] = False
+        (root / 'podman-fixture.json').write_text(json.dumps(data))
+    with (root / 'mutations.jsonl').open('a') as stream:
+        stream.write(json.dumps(args) + '\\n')
+''')
+        executable.chmod(0o755)
+        builder.podman = str(executable)
+        with builder.locked():
+            (builder.cache / 'stage.tar.xz').write_bytes(lzma.compress(b'original layer'))
+        return builder, inputs
+
+    def assert_prepare_refused(self, builder, inputs, reason):
+        with patch.object(builder, 'fetch', return_value=inputs):
+            with self.assertRaisesRegex(self.module.BuildError, reason):
+                builder.prepare()
+        self.assertFalse((self.project / 'mutations.jsonl').exists(), 'refusal must precede container/config mutations')
+        self.assertNotIn('prepare', builder.manifest(), 'refusal must not record false provenance')
+
+    def test_other_worktree_mount_is_refused_before_preparation(self):
+        builder, inputs = self.existing_builder(source=self.project / 'other-worktree')
+        self.assert_prepare_refused(builder, inputs, 'source')
+
+    def test_other_cache_mount_is_refused_before_preparation(self):
+        builder, inputs = self.existing_builder(cache=self.project / 'other-cache')
+        self.assert_prepare_refused(builder, inputs, 'cache')
+
+    def test_changed_stage3_pin_refuses_old_container(self):
+        builder, inputs = self.existing_builder()
+        archive = lzma.compress(b'new stage layer')
+        (builder.cache / 'stage.tar.xz').write_bytes(archive)
+        inputs['stage3']['sha256'] = hashlib.sha256(archive).hexdigest()
+        self.assert_prepare_refused(builder, inputs, 'stage3')
+
+    def test_changed_snapshot_pin_refuses_existing_tree(self):
+        builder, inputs = self.existing_builder()
+        inputs['portage']['commit'] = 'new-snapshot'
+        self.assert_prepare_refused(builder, inputs, 'snapshot')
+
+    def test_matching_existing_builder_retains_compatible_provenance(self):
+        builder, inputs = self.existing_builder()
+        with patch.object(builder, 'fetch', return_value=inputs):
+            builder.prepare()
+        self.assertEqual(builder.manifest()['prepare'].get('stage3_sha256'), inputs['stage3']['sha256'])
+        self.assertEqual(builder.manifest()['prepare']['snapshot'], 'old-snapshot')
+
+    def test_status_invalidates_missing_package_inventory(self):
+        builder = self.builder()
+        with builder.locked():
+            builder.record('packages')
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            builder.status()
+        self.assertFalse(json.loads(output.getvalue())['packages']['complete'])
+
+    def test_status_cli_runs_during_active_build_without_mutations(self):
+        builder = self.builder()
+        self.module.subprocess.run(['git', 'init', '--quiet', str(self.project)], check=True)
+        output = io.StringIO()
+        with builder.locked():
+            builder.record('packages')
+            manifest_before = (builder.work / 'manifest.json').read_bytes()
+            with patch.object(self.module, '__file__', str(self.project / 'hydra/build.py')):
+                with contextlib.redirect_stdout(output):
+                    result = self.module.main(['status'])
+            self.assertEqual(result, 0, 'status must not acquire the exclusive build lock')
+            data = json.loads(output.getvalue())
+            self.assertTrue(data['running'])
+            self.assertFalse(data['packages']['complete'])
+            self.assertEqual((builder.work / 'manifest.json').read_bytes(), manifest_before)
+
+    def test_detached_emerge_refuses_preparation_before_fetch_or_writes(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['active_job'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs) as fetch:
+            with self.assertRaisesRegex(self.module.BuildError, 'package job'):
+                builder.prepare()
+            fetch.assert_not_called()
+        self.assertFalse((self.project / 'mutations.jsonl').exists())
+
+    def test_contained_job_lock_refuses_preparation_before_fetch(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['job_lock'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs) as fetch:
+            with self.assertRaisesRegex(self.module.BuildError, 'package job'):
+                builder.prepare()
+            fetch.assert_not_called()
+
+    def test_status_detects_detached_emerge_without_outer_lock(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['active_job'] = True
+        fixture.write_text(json.dumps(data))
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            builder.status()
+        status = json.loads(output.getvalue())
+        self.assertTrue(status['running'])
+        self.assertTrue(status['package_job_running'])
+        self.assertFalse(status['mutation_lock_held'])
+
+    def test_package_job_records_failure_and_log_without_host_collection(self):
+        self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
+        script = self.module.package_job_script(['/bin/sh', '-c', 'echo diagnostic; exit 7'],
+                                               self.project / 'job', self.project / 'pkgdb', 'inputs-one')
+        result = self.module.subprocess.run(['python3', '-c', script])
+        self.assertEqual(result.returncode, 7)
+        status = json.loads((self.project / 'job/packages.status.json').read_text())
+        self.assertFalse(status['running'])
+        self.assertEqual(status['exit_code'], 7)
+        self.assertIn('diagnostic', (self.project / 'job/packages.log').read_text())
+        self.assertFalse((self.project / 'job/packages.json').exists())
+
+    def test_package_job_persists_inventory_after_success_without_host_collection(self):
+        self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
+        package = self.project / 'pkgdb/category/example-1.0'
+        package.mkdir(parents=True)
+        (package / 'USE').write_text('wayland opengl')
+        (package / 'BUILD_TIME').write_text('123')
+        script = self.module.package_job_script(['/bin/sh', '-c', 'exit 0'], self.project / 'job',
+                                               self.project / 'pkgdb', 'inputs-one')
+        result = self.module.subprocess.run(['python3', '-c', script])
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(json.loads((self.project / 'job/packages.json').read_text()),
+                         [{'package': 'category/example-1.0', 'use': ['wayland', 'opengl'], 'build_time': '123'}])
+        self.assertEqual(json.loads((self.project / 'job/packages.status.json').read_text())['exit_code'], 0)
+
+    def test_matching_stopped_builder_checks_provenance_before_start(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['stopped'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs):
+            try:
+                builder.prepare()
+            except self.module.BuildError as error:
+                self.fail(f'Compatible stopped builder must resume: {error}')
+        self.assertTrue(builder.manifest()['prepare']['complete'])
+
+    def test_unfinished_snapshot_is_not_reported_as_prepared(self):
+        builder, inputs = self.existing_builder(snapshot=None)
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['snapshot_pending'] = 'old-snapshot'
+        fixture.write_text(json.dumps(data))
+        builder.record('prepare')
+        self.assertFalse(builder.complete('prepare'))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_installer.py

@@ -0,0 +1,559 @@
+"""Installer tests never invoke a real partition, mount or bootloader command."""
+import contextlib
+import io
+import json
+import os
+import pty
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+try:
+    from hydra import installer as i
+except ImportError:
+    i = None
+
+
+def disk(path='/dev/vda', children=None, mounts=None, kind='disk', pkname=None):
+    return {'path': path, 'type': kind, 'pkname': pkname, 'mountpoints': [] if mounts is None else mounts,
+            'size': 32 * 1024 ** 3, 'maj:min': '252:' + str(sum((n + 1) * ord(c) for n, c in enumerate(path))), 'children': children or []}
+
+
+def topology(*nodes):
+    return i.Topology.from_json(json.dumps({'blockdevices': list(nodes)}))
+
+
+class FakeProbe:
+    def __init__(self, topo=None):
+        self.topo = topo or topology(disk())
+        self.calls = 0
+
+    def snapshot(self, payload):
+        self.calls += 1
+        return self.topo
+
+    def resolve(self, path):
+        return {'/dev/disk/by-id/test': '/dev/vda'}.get(path, path)
+
+    def physical_disk(self, path):
+        return path == '/dev/vda'
+
+
+class FakeRunner:
+    def __init__(self, root, fail=None):
+        self.root, self.fail, self.calls, self.inputs = root, fail, [], []
+
+    def run(self, argv, *, input=None, sensitive=False):
+        self.calls.append(list(argv))
+        self.inputs.append((input, sensitive))
+        if self.fail == argv[0] or (argv[0] == 'chroot' and self.fail == argv[2]) or (self.fail == 'unsquashfs-extract' and argv[0] == 'unsquashfs' and '-d' in argv):
+            raise i.InstallError('command failed: ' + argv[0])
+        if argv[0] == 'unsquashfs' and '-d' in argv:
+            for directory in ('etc/hydra/live-mode', 'etc/systemd/system/getty@tty1.service.d',
+                              'etc/sudoers.d', 'usr/share/zoneinfo/Europe', 'etc/locale.gen', 'lib/modules/6.12.1', 'boot'):
+                if directory != 'etc/locale.gen':
+                    (self.root / directory).mkdir(parents=True, exist_ok=True)
+            (self.root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
+            (self.root / 'usr/share/zoneinfo/Europe/Moscow').touch()
+            (self.root / 'etc/locale.gen').write_text('# en_US.UTF-8 UTF-8\n')
+            (self.root / 'etc/hydra/live-mode/active').touch()
+            (self.root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps([
+                '/etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf',
+                '/etc/sudoers.d/90-hydra-live']))
+            (self.root / 'etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf').touch()
+            (self.root / 'etc/sudoers.d/90-hydra-live').touch()
+            (self.root / 'boot/vmlinuz-6.12.1').touch()
+        if argv[0] == 'blkid':
+            return '11111111-2222-3333-4444-555555555555\n'
+        return ''
+
+
+class GuardTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(i, 'installer module must implement the guarded installer')
+
+    def request(self, **kw):
+        return i.Request(**({'disk': '/dev/vda', 'user': 'emil'} | kw))
+
+    def test_confirmation_missing_and_mismatched(self):
+        for confirmation in (None, '/dev/vdb', '/dev/disk/by-id/test'):
+            with self.subTest(confirmation=confirmation), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(execute=True, confirm_erase=confirmation), FakeProbe())
+        self.assertEqual(i.validate_request(self.request(execute=True, confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
+
+    def test_alias_resolved_but_confirmation_must_name_resolved_target(self):
+        self.assertEqual(i.validate_request(self.request(disk='/dev/disk/by-id/test', execute=True,
+                                                        confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
+
+    def test_disk_mounted_child_root_lvm_and_live_source_rejected(self):
+        examples = [
+            topology(disk(mounts=['/mnt'])),
+            topology(disk(children=[disk('/dev/vda1', kind='part', mounts=['/'])])),
+            topology(disk(children=[disk('/dev/vda2', kind='part', children=[disk('/dev/mapper/root', kind='lvm', mounts=['/'])])])),
+        ]
+        live = topology(disk(children=[disk('/dev/vda1', kind='part')]))
+        live.protect('/dev/vda1')
+        examples.append(live)
+        for topo in examples:
+            with self.subTest(topo=topo), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(), FakeProbe(topo))
+
+    def test_loop_backing_mount_protects_parent_disk(self):
+        topo = topology(disk(children=[disk('/dev/vda1', kind='part')]), disk('/dev/loop0', kind='loop', mounts=['/run/live']))
+        topo.link('/dev/loop0', '/dev/vda1')
+        with self.assertRaises(i.InstallError):
+            i.validate_request(self.request(), FakeProbe(topo))
+
+    def test_malformed_topology_fails_closed(self):
+        bad = ['{}', '{', '{"blockdevices":null}', '{"blockdevices":[]}',
+               json.dumps({'blockdevices': [disk(kind='mystery')]}),
+               json.dumps({'blockdevices': [disk(mounts='')] }),
+               json.dumps({'blockdevices': [disk(pkname='/dev/missing')]}),
+               json.dumps({'blockdevices': [disk(children=[disk('/dev/vda', kind='part')])]}),
+               json.dumps({'blockdevices': [disk('/dev/vda;reboot')]}),
+               json.dumps({'blockdevices': [disk(children='bad')]} )]
+        for data in bad:
+            with self.subTest(data=data), self.assertRaises(i.InstallError):
+                i.Topology.from_json(data)
+
+    def test_nonphysical_unknown_or_partition_target_rejected(self):
+        for path in ('/dev/vdb', '/tmp/fake', '/dev/vda1', '/dev/loop0', '/dev/vda;touch /tmp/x'):
+            with self.subTest(path=path), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(disk=path), FakeProbe())
+
+    def test_username_timezone_and_payload_injection_rejected(self):
+        for username in ('root', '-u', 'a;reboot', 'a\nb', '../x', 'A', ''):
+            with self.subTest(username=username), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(user=username), FakeProbe())
+        for timezone in ('../../etc/passwd', '/etc/passwd', 'a;reboot'):
+            with self.subTest(timezone=timezone), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(timezone=timezone), FakeProbe())
+        for payload in ('-evil', '/tmp/a\nb'):
+            with self.subTest(payload=payload), self.assertRaises(i.InstallError):
+                i.validate_request(self.request(payload=payload), FakeProbe())
+
+    def test_plan_is_nonsecret_and_has_exact_partition_boot_commands(self):
+        plan = i.render_plan(self.request(), '/dev/vda', '/mnt/hydra-target')
+        self.assertIn('1MiB', plan)
+        self.assertIn('1025MiB', plan)
+        self.assertIn('--removable --no-nvram', plan)
+        self.assertIn('/run/initramfs/live/LiveOS/squashfs.img', plan)
+        self.assertIn('dry-run', plan)
+        self.assertNotIn('chpasswd:', plan)
+
+
+class PipelineTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(i, 'installer module must implement the guarded installer')
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name) / 'target'
+        self.root.mkdir()
+        self.probe = FakeProbe()
+        self.log = io.StringIO()
+        self.request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
+
+    def install(self, runner, password='distinctive-private-password'):
+        i.install(self.request, runner, self.probe, self.root, password, self.log)
+
+    def test_full_fake_pipeline_sets_installed_defaults_and_cleans_live_files(self):
+        runner = FakeRunner(self.root)
+        self.install(runner)
+        self.assertIn('UUID=', (self.root / 'etc/fstab').read_text())
+        self.assertIn('ru_RU.UTF-8 UTF-8', (self.root / 'etc/locale.gen').read_text())
+        self.assertFalse((self.root / 'etc/hydra/live-mode').exists())
+        self.assertFalse((self.root / 'etc/sudoers.d/90-hydra-live').exists())
+        self.assertIn(['chroot', str(self.root), 'grub-install', '--target=x86_64-efi', '--efi-directory=/boot/efi', '--bootloader-id=HydraOS', '--removable', '--no-nvram'], runner.calls)
+        password_commands = [(args, data) for args, data in zip(runner.calls, runner.inputs) if 'chpasswd' in args]
+        self.assertEqual(len(password_commands), 1)
+        self.assertEqual(password_commands[0][1], ('emil:distinctive-private-password\n', True))
+        self.assertNotIn('distinctive-private-password', self.log.getvalue())
+        self.assertGreaterEqual(self.probe.calls, 2)
+        self.assertIn('hostonly_cmdline="no"', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
+        self.assertIn('omit_dracutmodules+=" dmsquash-live "', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
+        self.assertNotIn('root=live:', (self.root / 'etc/default/grub').read_text())
+        dracut = next(c for c in runner.calls if 'dracut' in c)
+        self.assertIn('6.12.1', dracut)
+        self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
+
+    def test_partition_format_and_extract_failures_abort_and_cleanup_only_mounted(self):
+        for failing in ('parted', 'mkfs.vfat', 'mkfs.ext4', 'unsquashfs-extract'):
+            with self.subTest(failing=failing):
+                runner = FakeRunner(self.root, failing)
+                with self.assertRaises(i.InstallError):
+                    self.install(runner)
+                self.assertFalse(any('grub-install' in c for c in runner.calls))
+                index = next(n for n, c in enumerate(runner.calls) if c[0] == failing or (failing == 'unsquashfs-extract' and c[0] == 'unsquashfs' and '-d' in c))
+                self.assertTrue(all(c[0] == 'umount' for c in runner.calls[index + 1:]))
+                if failing == 'unsquashfs-extract':
+                    self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
+                self.assertIn(failing.split('-extract')[0], self.log.getvalue())
+
+    def test_unmount_failure_is_visible(self):
+        runner = FakeRunner(self.root, 'umount')
+        with self.assertRaises(i.InstallError):
+            self.install(runner)
+        self.assertIn('cleanup', self.log.getvalue())
+
+    def test_password_newlines_rejected_before_commands(self):
+        for password in ('x\ny', 'x\ry', '', 'x\x00y'):
+            runner = FakeRunner(self.root)
+            with self.subTest(password=password), self.assertRaises(i.InstallError):
+                self.install(runner, password)
+            self.assertEqual(runner.calls, [])
+
+    def test_dry_run_cannot_enter_install_pipeline(self):
+        runner = FakeRunner(self.root)
+        self.request.execute = False
+        with self.assertRaises(i.InstallError):
+            self.install(runner)
+        self.assertEqual(runner.calls, [])
+
+    def test_live_manifest_arbitrary_paths_and_symlink_parents_refused(self):
+        for path in ('/etc/passwd', '/home/emil/data', '/../etc/sudoers.d/90-hydra-live'):
+            directory = self.root / 'etc/hydra/live-mode'
+            directory.mkdir(parents=True, exist_ok=True)
+            (directory / 'active').touch()
+            (directory / 'remove-on-install.json').write_text(json.dumps([path]))
+            with self.subTest(path=path), self.assertRaises(i.InstallError):
+                i.remove_live_files(self.root)
+        outside = Path(self.tmp.name) / 'outside'
+        outside.mkdir()
+        (outside / '90-hydra-live').write_text('must remain')
+        (self.root / 'etc/sudoers.d').symlink_to(outside)
+        (directory / 'remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
+        with self.assertRaises(i.InstallError):
+            i.remove_live_files(self.root)
+        self.assertTrue((outside / '90-hydra-live').exists())
+
+    def test_missing_or_multiple_kernels_are_explicit_errors(self):
+        with self.assertRaises(i.InstallError):
+            i.select_kernel(self.root, None)
+        for version in ('6.12.1', '6.12.2'):
+            (self.root / 'lib/modules' / version).mkdir(parents=True)
+            (self.root / 'boot').mkdir(exist_ok=True)
+            (self.root / 'boot' / ('vmlinuz-' + version)).touch()
+        with self.assertRaises(i.InstallError):
+            i.select_kernel(self.root, None)
+        self.assertEqual(i.select_kernel(self.root, '6.12.2'), '6.12.2')
+
+
+class RunnerTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(i, 'installer module must implement the guarded installer')
+
+    def test_sensitive_failure_never_includes_stderr_or_stdin(self):
+        with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 1, '', 'private-password')):
+            with self.assertRaises(i.InstallError) as caught:
+                i.Runner().run(['chroot', '/target', 'chpasswd'], input='user:private-password\n', sensitive=True)
+        self.assertNotIn('private-password', str(caught.exception))
+
+    def test_subprocess_uses_argv_without_shell_and_checked_returncode(self):
+        with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 0, 'ok', '')) as call:
+            self.assertEqual(i.Runner().run(['tool', '/tmp/path;echo bad']), 'ok')
+        self.assertEqual(call.call_args.args[0], ['tool', '/tmp/path;echo bad'])
+        self.assertFalse(call.call_args.kwargs.get('shell', False))
+
+
+class ProbeTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(i, 'installer module required')
+
+    def test_device_mapper_alias_and_dev_root_resolve_by_identity(self):
+        topo = topology(disk('/dev/vda', children=[disk('/dev/dm-0', kind='lvm')]))
+        mount = {'source': '/dev/root', 'target': '/', 'fstype': 'ext4', 'maj:min': topo.nodes['/dev/vda']['maj:min'], 'options': 'rw'}
+        probe = i.SystemProbe()
+        self.assertEqual(probe._block_source(mount, topo), '/dev/vda')
+        self.assertIn('/dev/vda', topo.protected)
+
+    def test_parent_reference_cycles_rejected(self):
+        data = json.dumps({'blockdevices': [disk('/dev/a', pkname='/dev/b'), disk('/dev/b', pkname='/dev/a')]})
+        with self.assertRaises(i.InstallError):
+            i.Topology.from_json(data)
+
+    def test_malformed_findmnt_rejected(self):
+        for data in ('{}', '{', '{"filesystems":[]}', '{"filesystems":[{"source":"overlay"}]}'):
+            with self.subTest(data=data), self.assertRaises(i.InstallError):
+                i.SystemProbe._mounts(data)
+
+    def test_unknown_mounted_block_source_refused(self):
+        mount = {'source': '/dev/missing', 'target': '/', 'fstype': 'ext4', 'maj:min': '100:1', 'options': 'rw'}
+        with self.assertRaises(i.InstallError):
+            i.SystemProbe()._block_source(mount, topology(disk()))
+
+    def test_payload_symlink_never_resolves_into_host_from_target(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / 'usr/lib/modules/6.12.1').mkdir(parents=True)
+            (root / 'lib').symlink_to('/usr/lib')
+            (root / 'boot').mkdir()
+            (root / 'boot/vmlinuz-6.12.1').touch()
+            self.assertEqual(i.select_kernel(root, None), '6.12.1')
+            with self.assertRaises(i.InstallError):
+                i.target_path(root, '/lib/modules')
+
+    def test_leaf_symlink_removal_does_not_follow_link(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp) / 'root'
+            outside = Path(tmp) / 'keep'
+            outside.write_text('keep')
+            (root / 'etc/hydra/live-mode').mkdir(parents=True)
+            (root / 'etc/sudoers.d').mkdir()
+            (root / 'etc/hydra/live-mode/active').touch()
+            (root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
+            (root / 'etc/sudoers.d/90-hydra-live').symlink_to(outside)
+            i.remove_live_files(root)
+            self.assertEqual(outside.read_text(), 'keep')
+            self.assertFalse((root / 'etc/sudoers.d/90-hydra-live').is_symlink())
+
+    def test_read_only_cli_invalid_target_and_confirmation_never_prompt_password(self):
+        with patch.object(i, 'getpass') as password, contextlib.redirect_stderr(io.StringIO()):
+            self.assertEqual(i.main(['--disk', '/dev/hydra-this-device-does-not-exist', '--user', 'emil']), 1)
+            self.assertEqual(i.main(['--execute']), 1)
+        password.getpass.assert_not_called()
+
+    def test_preflight_refuses_container_before_execution(self):
+        with patch('os.geteuid', return_value=0), patch.object(i.sys.stdin, 'isatty', return_value=True), \
+             patch.object(i.Path, 'exists', return_value=True):
+            with self.assertRaisesRegex(i.InstallError, 'container'):
+                i.preflight(i.Request('/dev/vda', 'emil'), FakeRunner(Path('/unused')))
+
+
+class AccountTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(i)
+
+    def test_live_account_removed_without_touching_other_home(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / 'etc').mkdir()
+            (root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\nhydra-live:x:1000:1000:Live:/home/hydra-live:/bin/bash\n')
+            (root / 'home/hydra-live').mkdir(parents=True)
+            (root / 'home/hydra-live/config').touch()
+            (root / 'home/keep').mkdir()
+            runner = FakeRunner(root)
+            i.remove_live_account(root, runner.run)
+            self.assertIn(['chroot', str(root), 'userdel', 'hydra-live'], runner.calls)
+            self.assertFalse((root / 'home/hydra-live').exists())
+            self.assertTrue((root / 'home/keep').exists())
+
+    def test_live_account_unexpected_home_or_root_uid_refused(self):
+        for row in ('hydra-live:x:1000:1000:Live:/home/other:/bin/bash',
+                    'hydra-live:x:0:0:Live:/home/hydra-live:/bin/bash'):
+            with self.subTest(row=row), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp)
+                (root / 'etc').mkdir()
+                (root / 'etc/passwd').write_text(row + '\n')
+                runner = FakeRunner(root)
+                with self.assertRaises(i.InstallError):
+                    i.remove_live_account(root, runner.run)
+                self.assertEqual(runner.calls, [])
+
+    def test_live_manifest_nested_unit_paths_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            live = root / 'etc/hydra/live-mode'
+            live.mkdir(parents=True)
+            (live / 'active').touch()
+            (live / 'remove-on-install.json').write_text(json.dumps(['/etc/systemd/system/getty@foo/bar.service.d/90-hydra-live.conf']))
+            with self.assertRaises(i.InstallError):
+                i.remove_live_files(root)
+
+
+class AdditionalFailureTests(unittest.TestCase):
+    setUp = PipelineTests.setUp
+    install = PipelineTests.install
+    def test_later_chroot_failure_aborts_bootloader_and_unmounts_all(self):
+        runner = FakeRunner(self.root, 'dracut')
+        with self.assertRaises(i.InstallError):
+            self.install(runner)
+        self.assertFalse(any('grub-install' in c for c in runner.calls))
+        self.assertEqual([c for c in runner.calls if c[0] == 'umount'], [
+            ['umount', '--recursive', str(self.root / 'sys')],
+            ['umount', str(self.root / 'proc')],
+            ['umount', '--recursive', str(self.root / 'dev')],
+            ['umount', str(self.root / 'boot/efi')],
+            ['umount', str(self.root)],
+        ])
+
+    def test_revalidation_detects_mounted_disk_before_partitioning(self):
+        class ChangedProbe(FakeProbe):
+            def snapshot(self, payload):
+                self.calls += 1
+                return topology(disk(mounts=['/mounted'])) if self.calls > 1 else self.topo
+        self.probe = ChangedProbe()
+        runner = FakeRunner(self.root)
+        with self.assertRaises(i.InstallError):
+            self.install(runner)
+        self.assertFalse(any(c[0] == 'parted' for c in runner.calls))
+
+    def test_unknown_multidevice_filesystem_refused(self):
+        topo = topology(disk())
+        for kind in ('zfs', 'btrfs'):
+            mount = {'source': 'pool' if kind == 'zfs' else '/dev/vda',
+                     'target': '/', 'fstype': kind, 'maj:min': '0:22', 'options': 'rw'}
+            with self.subTest(kind=kind), self.assertRaises(i.InstallError):
+                i.SystemProbe()._block_source(mount, topo)
+
+    def test_prompt_requires_controlling_tty(self):
+        with patch('builtins.open', side_effect=OSError('no tty')):
+            with self.assertRaises(i.InstallError):
+                i.prompt_password()
+
+
+class SnapshotTests(unittest.TestCase):
+    def test_snapshot_connects_live_loop_to_its_physical_backing(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            payload = Path(tmp) / 'squashfs.img'
+            payload.touch()
+            data = {'blockdevices': [disk(), disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', mounts=['/run/initramfs/live'])]), disk('/dev/loop0', kind='loop', mounts=['/'])]}
+            media = {'source': '/dev/vdb1', 'target': '/run/initramfs/live', 'fstype': 'iso9660', 'maj:min': data['blockdevices'][1]['children'][0]['maj:min'], 'options': 'ro'}
+            loop = {'source': '/dev/loop0', 'target': '/', 'fstype': 'squashfs', 'maj:min': data['blockdevices'][2]['maj:min'], 'options': 'ro'}
+            class ReadOnlyRunner:
+                def run(self, args):
+                    if args[0] == 'lsblk':
+                        return json.dumps(data)
+                    return json.dumps({'filesystems': [media] if '--target' in args else [media, loop]})
+            real_read = Path.read_text
+            def read(path, *args, **kw):
+                if str(path).startswith('/sys/dev/block/'):
+                    return str(payload)
+                return real_read(path, *args, **kw)
+            with patch.object(Path, 'read_text', read):
+                topo = i.SystemProbe(ReadOnlyRunner()).snapshot(str(payload))
+            self.assertFalse(topo.component('/dev/vda') & topo.protected)
+            self.assertIn('/dev/loop0', topo.component('/dev/vdb'))
+            self.assertTrue(topo.component('/dev/vdb') & topo.protected)
+
+    def test_failed_extraction_retains_diagnostic_on_target(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp) / 'target'
+            root.mkdir()
+            request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
+            runner = FakeRunner(root, 'unsquashfs-extract')
+            with self.assertRaises(i.InstallError):
+                i.install(request, runner, FakeProbe(), root, 'secret', io.StringIO())
+            diagnostic = root / 'var/log/hydra-install.log'
+            self.assertTrue(diagnostic.is_file())
+            self.assertIn('unsquashfs', diagnostic.read_text())
+            self.assertNotIn('secret', diagnostic.read_text())
+
+
+class TargetPathTests(unittest.TestCase):
+    def test_live_mail_parent_symlink_is_resolved_within_target(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / 'etc').mkdir()
+            (root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
+            (root / 'var/spool').mkdir(parents=True)
+            (root / 'var/mail').mkdir()
+            (root / 'var/spool/mail').symlink_to('/var/mail')
+            i.remove_live_account(root, FakeRunner(root).run)
+
+    def test_kernel_version_symlink_is_not_followed_into_host(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp) / 'root'
+            external = Path(tmp) / 'external'
+            external.mkdir()
+            (root / 'lib/modules').mkdir(parents=True)
+            (root / 'lib/modules/6.12.1').symlink_to(external)
+            (root / 'boot').mkdir()
+            (root / 'boot/vmlinuz-6.12.1').touch()
+            with self.assertRaises(i.InstallError):
+                i.select_kernel(root, None)
+
+
+class IdentityTests(unittest.TestCase):
+    def test_conflicting_device_identities_rejected(self):
+        a, b = disk(), disk('/dev/vdb')
+        b['maj:min'] = a['maj:min']
+        with self.assertRaises(i.InstallError):
+            i.Topology.from_json(json.dumps({'blockdevices': [a, b]}))
+
+
+class TtyTests(unittest.TestCase):
+    def test_password_stream_opens_real_nonseekable_tty(self):
+        master, slave = pty.openpty()
+        try:
+            slave_name = os.ttyname(slave)
+            real_open = open
+            def tty_open(path, mode):
+                self.assertEqual(path, '/dev/tty')
+                return real_open(slave_name, mode, opener=lambda path, flags: os.open(path, flags | os.O_NOCTTY))
+            with patch('builtins.open', tty_open), patch.object(i.getpass, 'getpass', return_value='test-only-secret'):
+                self.assertEqual(i.prompt_password(), 'test-only-secret')
+        finally:
+            os.close(slave)
+            os.close(master)
+
+
+class ReviewRegressionTests(unittest.TestCase):
+    def request(self):
+        return i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
+
+    def test_parentless_protected_stacked_root_refuses_all_targets(self):
+        for kind in ('lvm', 'crypt', 'dm', 'mpath', 'md', 'raid0', 'raid1', 'raid4', 'raid5', 'raid6', 'raid10'):
+            with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
+                topo = topology(disk(), disk('/dev/dm-0', kind=kind, mounts=['/']))
+                i.validate_request(self.request(), FakeProbe(topo))
+
+    def test_child_relationship_is_not_evidence_of_backing_parent(self):
+        for kind in ('part', 'lvm', 'crypt', 'raid1'):
+            node = disk('/dev/missing-backing', kind=kind,
+                        children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
+            with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
+                topo = topology(disk(), node)
+                i.validate_request(self.request(), FakeProbe(topo))
+
+    def test_complete_stack_on_other_disk_preserves_eligible_target(self):
+        stack = disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', children=[
+            disk('/dev/dm-crypt', kind='crypt', children=[disk('/dev/dm-lvm', kind='lvm', mounts=['/'])])])])
+        topo = topology(disk(), stack)
+        self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
+        self.assertTrue(topo.component('/dev/vdb') & topo.protected)
+
+    def test_empty_optical_drive_does_not_block_eligible_target(self):
+        optical = disk('/dev/sr0', kind='rom')
+        optical['size'] = 0
+        topo = topology(disk(), optical)
+        self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
+        self.assertFalse(topo.component('/dev/sr0') & topo.protected)
+
+    def test_zero_size_disk_cannot_be_selected(self):
+        target = disk()
+        target['size'] = 0
+        with self.assertRaises(i.InstallError):
+            topo = topology(target)
+            i.validate_request(self.request(), FakeProbe(topo))
+
+    def test_zero_size_mounted_optical_drive_is_not_treated_as_empty(self):
+        optical = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
+        optical['size'] = 0
+        with self.assertRaises(i.InstallError):
+            topology(disk(), optical)
+
+    def test_zero_size_optical_with_separately_listed_dependent_refused(self):
+        optical = disk('/dev/sr0', kind='rom')
+        optical['size'] = 0
+        dependent = disk('/dev/dm-0', kind='crypt', pkname='/dev/sr0', mounts=['/'])
+        with self.assertRaises(i.InstallError):
+            topology(disk(), optical, dependent)
+
+    def test_zero_size_optical_cannot_be_protected_by_findmnt(self):
+        optical = disk('/dev/sr0', kind='rom')
+        optical['size'] = 0
+        topo = topology(disk(), optical)
+        with self.assertRaises(i.InstallError):
+            topo.protect('/dev/sr0')
+
+    def test_zero_size_optical_with_dependents_is_not_irrelevant(self):
+        optical = disk('/dev/sr0', kind='rom', children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
+        optical['size'] = 0
+        with self.assertRaises(i.InstallError):
+            topology(disk(), optical)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_iso_smoke.py

@@ -0,0 +1,634 @@
+"""Regression tests for disposable-VM isolation and trustworthy serial evidence."""
+import importlib.util
+from pathlib import Path
+import socket
+import select
+import subprocess
+import sys
+import time
+import tempfile
+import threading
+import unittest
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'dev/iso-smoke.py'
+
+
+class SmokeTests(unittest.TestCase):
+    def setUp(self):
+        self.assertTrue(SCRIPT.is_file(), 'disposable ISO acceptance harness is missing')
+        spec = importlib.util.spec_from_file_location('iso_smoke', SCRIPT)
+        self.mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.mod)
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.code = self.root / 'code.fd'
+        self.variables = self.root / 'vars.fd'
+        self.code.write_bytes(b'firmware')
+        self.variables.write_bytes(b'fresh vars')
+
+    def test_existing_disk_or_symlink_cannot_be_a_test_drive(self):
+        disk = self.root / 'disk.qcow2'
+        disk.write_bytes(b'original')
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.reserve_disk(disk)
+        link = self.root / 'link.qcow2'
+        link.symlink_to(disk)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.reserve_disk(link)
+        self.assertEqual(disk.read_bytes(), b'original')
+
+    def test_qemu_only_accepts_owned_regular_disk_and_readonly_iso(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        iso = self.root / 'trusted.iso'
+        iso.write_bytes(b'ISO')
+        argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                                  self.variables, iso, 2048)
+        drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
+        self.assertTrue(any('format=qcow2' in item and str(disk) in item for item in drives))
+        self.assertTrue(any('media=cdrom' in item and 'readonly=on' in item for item in drives))
+        self.assertIn('accel=kvm', ' '.join(argv))
+        disk.unlink()
+        disk.symlink_to(iso)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, iso, 2048)
+
+    def test_installed_reboot_has_no_optical_medium(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                                  self.variables, None, 2048)
+        self.assertNotIn('media=cdrom', ' '.join(argv))
+        self.assertNotIn('-cdrom', argv)
+
+    def test_hardlinked_disk_is_refused(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        (self.root / 'alias').hardlink_to(disk)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, None, 2048)
+
+    def test_result_frame_ignores_command_echo_and_preserves_failure(self):
+        command = self.mod.frame_command('printf sentinel; false', 'testtoken')
+        result = subprocess.run(['/bin/sh', '-c', command], capture_output=True)
+        echo = command.encode() + b'\r\n'
+        self.assertIsNone(self.mod.parse_frame(echo, 'testtoken'))
+        self.assertEqual(self.mod.parse_frame(echo + result.stdout, 'testtoken'),
+                         (1, b'sentinel'))
+
+    def test_console_redacts_password_even_when_split_across_reads(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        log = self.root / 'serial.log'
+        console = self.mod.Console(one, log, ['split-secret'])
+        self.addCleanup(console.close)
+        two.sendall(b'hello split-')
+        console.read(0.1)
+        two.sendall(b'secret goodbye')
+        console.read(0.1)
+        console.close()
+        self.assertNotIn(b'split-secret', log.read_bytes())
+        self.assertIn(b'[REDACTED]', log.read_bytes())
+
+    def test_console_timeout_is_bounded_and_partial_frame_not_success(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'serial.log', [])
+        self.addCleanup(console.close)
+        two.sendall(b'\\036token:END:0\\037')
+        with self.assertRaises(self.mod.SmokeError):
+            console.command('true', timeout=0.05)
+
+    def installer_console(self):
+        one, two = socket.socketpair()
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'installer.log', ['test-secret'])
+        self.addCleanup(console.close)
+        return console, two
+
+    def test_captured_installer_refusal_fails_before_password_timeout(self):
+        console, guest = self.installer_console()
+        guest.sendall(b'\x1b]3008;type=command\x1b\\hydra-install: Command failed (1): '
+                      b'findmnt --json --target /LiveOS/squashfs.img '
+                      b'--output SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS: \r\n')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*findmnt'):
+            console.expect_installer(rb'Installed account password: ', 0.05, 'testtoken')
+
+    def test_installer_shell_return_without_completion_is_failure_even_zero(self):
+        for status in (0, 1, 127):
+            with self.subTest(status=status):
+                console, guest = self.installer_console()
+                guest.sendall(b'\x1etesttoken:INSTALL_EXIT:' + str(status).encode() + b'\x1f')
+                with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*INSTALL_EXIT'):
+                    console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                             0.05, 'testtoken')
+
+    def test_installer_echo_cannot_fake_exit_and_complete_prompt_still_required(self):
+        console, guest = self.installer_console()
+        guest.sendall(self.mod.installer_command('testtoken').encode() + b'\r\n'
+                      b'Installed account password: ')
+        self.assertEqual(console.expect_installer(rb'Installed account password: ',
+                         0.05, 'testtoken').group(), b'Installed account password: ')
+        guest.sendall(b'Installation completed. Diagnostic')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Timed out'):
+            console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                     0.05, 'testtoken')
+        guest.sendall(b' log: /var/log/hydra-install.log\r\n')
+        self.assertIn(b'/var/log/hydra-install.log', console.expect_installer(
+            rb'Installation completed\. Diagnostic log: [^\r\n]+', 0.05, 'testtoken').group())
+
+    def test_installer_wrapper_marks_actual_shell_return_and_redacts_error(self):
+        console, guest = self.installer_console()
+        fake_sudo = self.root / 'sudo'
+        fake_sudo.write_text('#!/bin/sh\nprintf "hydra-install: test-secret refused\\n"\nexit 1\n')
+        fake_sudo.chmod(0o700)
+        result = subprocess.run(['/bin/sh', '-c', self.mod.installer_command('testtoken')],
+                                env={'PATH': str(self.root)}, capture_output=True, check=True)
+        self.assertIn(b'\x1etesttoken:INSTALL_EXIT:1\x1f', result.stdout)
+        guest.sendall(result.stdout)
+        with self.assertRaisesRegex(self.mod.SmokeError, r'Installer.*\[REDACTED\] refused') as caught:
+            console.expect_installer(rb'Repeat password: ', 0.05, 'testtoken')
+        self.assertNotIn('test-secret', str(caught.exception))
+
+    def test_installer_followup_waits_for_exit_and_new_shell_prompt(self):
+        console, guest = self.installer_console()
+        premature = []
+        def respond():
+            guest.sendall(b'Installation completed. Diagnostic log: /run/installer.log\r\n')
+            premature.append(bool(select.select([guest], [], [], 0.05)[0]))
+            guest.sendall(b'\x1etesttoken:INSTALL_EXIT:0\x1f')
+            premature.append(bool(select.select([guest], [], [], 0.05)[0]))
+            guest.sendall(b'\x1b[?2004hhydra-live@hydra-os ~ $ ')
+            data = b''
+            while not data.endswith(b'\n'):
+                data += guest.recv(65536)
+            result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
+            guest.sendall(result.stdout)
+        worker = threading.Thread(target=respond, daemon=True)
+        worker.start()
+        console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                 1, 'testtoken')
+        console.finish_installer('testtoken', timeout=1)
+        self.assertEqual(console.command('printf ready', timeout=1), b'ready')
+        worker.join(timeout=1)
+        self.assertEqual(premature, [False, False], 'command reached installer before shell was ready')
+
+    def test_installer_completion_does_not_override_nonzero_return(self):
+        console, guest = self.installer_console()
+        guest.sendall(b'\x1etesttoken:INSTALL_EXIT:1\x1fhydra-live@hydra-os ~ $ ')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*status 1'):
+            console.finish_installer('testtoken', timeout=0.05)
+
+    def shutdown_vm(self, program):
+        one, two = socket.socketpair()
+        vm = self.mod.VM.__new__(self.mod.VM)
+        vm.console = self.mod.Console(one, self.root / 'shutdown.log', ['split-secret'])
+        self.addCleanup(vm.console.close)
+        code = ('import socket,sys,time; s=socket.socket(fileno=int(sys.argv[1])); ' + program)
+        vm.process = subprocess.Popen([sys.executable, '-c', code, str(two.fileno())],
+                                      pass_fds=(two.fileno(),), stdin=subprocess.DEVNULL)
+        two.close()
+        def reap():
+            if vm.process.poll() is None:
+                vm.process.terminate()
+            vm.process.wait(timeout=2)
+        self.addCleanup(reap)
+        return vm
+
+    def test_poweroff_drains_and_redacts_serial_until_owned_process_exits(self):
+        vm = self.shutdown_vm("s.sendall(b'Stopping split-'); time.sleep(.02); "
+                              "s.sendall(b'secret service\\nPower down\\n'); s.close()")
+        vm.wait_poweroff(1)
+        output = (self.root / 'shutdown.log').read_bytes()
+        self.assertIn(b'Stopping [REDACTED] service\nPower down\n', output)
+        self.assertNotIn(b'split-secret', output)
+
+    def test_poweroff_preserves_failure_cause_and_original_timeout(self):
+        for program, error in (("s.sendall(b'Stop job stuck\\n'); s.close(); time.sleep(1)",
+                                'Guest failed to power off in time'),
+                               ("s.sendall(b'QEMU shutdown failed\\n'); s.close(); sys.exit(7)",
+                                'Test QEMU failed during poweroff')):
+            with self.subTest(program=program):
+                vm = self.shutdown_vm(program)
+                started = time.monotonic()
+                with self.assertRaisesRegex(self.mod.SmokeError, error):
+                    vm.wait_poweroff(.15)
+                self.assertLess(time.monotonic() - started, .5)
+                output = (self.root / 'shutdown.log').read_bytes()
+                self.assertIn(b'Stop job stuck' if 'time.sleep' in program else b'QEMU shutdown failed', output)
+                if 'time.sleep' in program:
+                    self.assertIsNone(vm.process.poll(), 'timeout must not kill or falsely pass the owned process')
+
+    def test_frame_real_shell_roundtrip_does_not_accept_sentinel_in_echo(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'serial.log', [])
+        self.addCleanup(console.close)
+        def guest():
+            data = b''
+            while not data.endswith(b'\n'):
+                data += two.recv(65536)
+            two.sendall(data)
+            result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
+            two.sendall(result.stdout)
+        thread = threading.Thread(target=guest)
+        thread.start()
+        self.assertEqual(console.command("printf 'persisted-data'", timeout=2), b'persisted-data')
+        thread.join(timeout=2)
+
+    def test_firmware_vars_must_be_private_copy_in_scratch(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        self.variables.unlink()
+        self.variables.symlink_to(self.code)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, None, 2048)
+
+    def test_drive_outside_scratch_and_ambiguous_qemu_paths_are_refused(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        other = self.root / 'other'
+        other.mkdir()
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', other, disk, self.code,
+                               self.variables, None, 2048)
+        comma = self.root / 'disk,cache=unsafe.qcow2'
+        self.mod.reserve_disk(comma)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, comma, self.code,
+                               self.variables, None, 2048)
+
+    def probe_fixture(self, case, live=False):
+        # Run the production shell against ordinary scratch files. Commands that
+        # query guest state are boundary fixtures; grep/find/test are real tools.
+        import os
+        root = self.root / 'guest'
+        for directory in ('etc/hydra', 'etc/greetd', 'etc/sudoers.d',
+                          'etc/systemd/system/getty@tty2.service.d',
+                          'etc/systemd/system/serial-getty@ttyS0.service.d',
+                          'usr/share/firmware', 'proc', 'sys/firmware/efi',
+                          'run/initramfs/live/LiveOS'):
+            (root / directory).mkdir(parents=True, exist_ok=True)
+        (root / 'etc/greetd/config.toml').write_text('normal authentication')
+        (root / 'usr/share/firmware/OVMF_CODE.fd').write_text('firmware')
+        (root / 'proc/cmdline').write_text('root=UUID=disk ro' if case != 'live-cmdline' else 'root=live:CDLABEL=HYDRA_OS rd.live.image')
+        if live or case == 'live-marker':
+            (root / 'etc/hydra/live-mode').mkdir(exist_ok=True)
+            (root / 'etc/hydra/live-mode/active').touch()
+        if live:
+            (root / 'run/initramfs/live/LiveOS/squashfs.img').touch()
+        files = {
+            'live-dropin': ('etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf', 'live'),
+            'live-sudoers': ('etc/sudoers.d/90-hydra-live', 'live'),
+            'autologin': ('etc/systemd/system/serial-getty@ttyS0.service.d/other.conf', 'ExecStart=agetty --autologin hydratest'),
+            'nopasswd': ('etc/sudoers.d/other', 'hydratest ALL=(ALL) NOPASSWD: ALL'),
+            'live-greetd': ('etc/greetd/live.toml', 'live'),
+        }
+        if case in files:
+            path, text = files[case]
+            (root / path).write_text(text)
+        if case == 'config-read-error':
+            unreadable = root / 'etc/sudoers.d/unreadable'
+            unreadable.write_text('configuration')
+            unreadable.chmod(0)
+        if case == 'cmdline-read-error':
+            (root / 'proc/cmdline').unlink()
+        commands = self.root / 'fixture-bin'
+        commands.mkdir(exist_ok=True)
+        dispatcher = commands / 'fixture'
+        dispatcher.write_text('''#!/usr/bin/env python3
+import os, pathlib, sys
+name = pathlib.Path(sys.argv[0]).name
+args = sys.argv[1:]
+case = os.environ['SMOKE_CASE']
+if name == 'id': print('hydra-live')
+elif name == 'findmnt': print('/dev/vda2' if 'SOURCE' in args else ('overlay' if os.environ['SMOKE_LIVE']=='1' else 'ext4'))
+elif name == 'systemctl':
+    if args[0] == 'is-active':
+        states = [unit != case for unit in args[1:]]
+        print('\\n'.join('active' if state else 'failed' for state in states))
+        sys.exit(0 if any(states) else 3)
+    print('success' if 'Result' in args else '0')
+elif name == 'getent':
+    if args[:2] == ['passwd','hydra-live']:
+        if case == 'account-lookup-error': sys.exit(3)
+        if case != 'live-account': sys.exit(2)
+        print('hydra-live:x:1000:1000::/home/hydra-live:/bin/bash')
+    elif args[0] == 'shadow': print(args[1] + ':' + ('!' if args[1]=='root' else '$6$hash') + ':1::::::')
+    else: print('127.0.0.1 STREAM example')
+elif name == 'ip': print('10.203.0.1/24' if 'addr' in args else 'default via 10.0.2.2')
+elif name == 'nmcli': print('connected')
+else: print('fixture tool')
+''')
+        dispatcher.chmod(0o755)
+        for command in ('id', 'findmnt', 'systemctl', 'getent', 'ip', 'nft',
+                        'nmcli', 'sudo', 'hydra', 'hydra-install',
+                        'qemu-system-x86_64', 'qemu-img', 'remote-viewer', 'foot', 'labwc'):
+            path = commands / command
+            if not path.exists():
+                path.symlink_to(dispatcher)
+        probe = self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK
+        for path in ('/etc/', '/usr/share', '/proc/cmdline', '/sys/firmware/efi', '/run/initramfs/live/LiveOS/squashfs.img'):
+            probe = probe.replace(path, str(root) + path)
+        import shutil
+        for command in ('python3', 'sh', 'grep', 'find', 'cut', 'readlink', 'base64'):
+            target = shutil.which(command)
+            self.assertIsNotNone(target)
+            (commands / command).symlink_to(target) if not (commands / command).exists() else None
+        env = dict(os.environ, PATH=str(commands),
+                   SMOKE_CASE=case, SMOKE_LIVE='1' if live else '0')
+        result = subprocess.run(['/bin/sh', '-c', self.mod.frame_command(probe, 'probecheck')],
+                                env=env, capture_output=True, timeout=10)
+        parsed = self.mod.parse_frame(result.stdout, 'probecheck')
+        self.assertIsNotNone(parsed, result.stderr)
+        return parsed[0]
+
+    def test_installed_probe_rejects_each_retained_live_artifact(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        for case in ('live-cmdline', 'live-account', 'live-dropin', 'live-sudoers',
+                     'autologin', 'nopasswd', 'live-marker', 'live-greetd'):
+            with self.subTest(case=case):
+                # Give each negative case an independent clean filesystem.
+                import shutil
+                shutil.rmtree(self.root / 'guest')
+                self.assertNotEqual(self.probe_fixture(case), 0)
+
+    def test_installed_probe_does_not_treat_account_query_error_as_absence(self):
+        self.assertNotEqual(self.probe_fixture('account-lookup-error'), 0)
+
+    def test_installed_probe_rejects_unreadable_configuration(self):
+        self.assertNotEqual(self.probe_fixture('config-read-error'), 0)
+
+    def test_installed_probe_rejects_unreadable_boot_arguments(self):
+        self.assertNotEqual(self.probe_fixture('cmdline-read-error'), 0)
+
+    def test_installed_probe_rejects_each_failed_required_service(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        for service in ('NetworkManager', 'greetd', 'hydra-network', 'hydra-dnsmasq'):
+            with self.subTest(service=service):
+                self.assertNotEqual(self.probe_fixture(service), 0)
+
+    def test_live_probe_rejects_each_failed_required_service(self):
+        self.assertEqual(self.probe_fixture('clean', live=True), 0)
+        for service in ('NetworkManager', 'hydra-network', 'hydra-dnsmasq'):
+            with self.subTest(service=service):
+                self.assertNotEqual(self.probe_fixture(service, live=True), 0)
+
+    def tcp_probe(self, fragments):
+        # Execute the exact Python client embedded in the guest network probe,
+        # redirecting only its address to a fresh loopback socket.
+        import ast
+        import time
+        source = self.mod.NETWORK_CHECK.split("<<'NETWORK'\n", 1)[1].rsplit('\nNETWORK\n', 1)[0]
+        assignments = (node for node in ast.walk(ast.parse(source)) if isinstance(node, ast.Assign))
+        value = next(node.value for node in assignments
+                     if any(isinstance(target, ast.Name) and target.id == 'client_code' for target in node.targets))
+        client = ast.parse(ast.literal_eval(value))
+        listener = socket.socket()
+        self.addCleanup(listener.close)
+        listener.bind(('127.0.0.1', 0))
+        listener.listen(1)
+        listener.settimeout(3)
+        host, port = listener.getsockname()
+        class Redirect(ast.NodeTransformer):
+            def visit_Tuple(self, node):
+                if isinstance(node.elts[0], ast.Constant) and node.elts[0].value == '10.203.0.250':
+                    return ast.Tuple(elts=[ast.Constant(host), ast.Constant(port)], ctx=ast.Load())
+                return self.generic_visit(node)
+        client = ast.fix_missing_locations(Redirect().visit(client))
+        def server():
+            connection, _ = listener.accept()
+            with connection:
+                for fragment in fragments:
+                    try:
+                        connection.sendall(fragment)
+                    except BrokenPipeError:
+                        break
+                    time.sleep(.05)
+        thread = threading.Thread(target=server)
+        thread.start()
+        result = subprocess.run(['python3', '-c', ast.unparse(client)], capture_output=True, timeout=5)
+        thread.join(timeout=3)
+        self.assertFalse(thread.is_alive())
+        return result
+
+    def test_tcp_probe_accepts_complete_marker_across_fragmented_reads(self):
+        result = self.tcp_probe([b'HYDRA_', b'LAN_TO_', b'VM_OK'])
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_tcp_probe_rejects_truncated_marker(self):
+        self.assertNotEqual(self.tcp_probe([b'HYDRA_LAN_TO_']).returncode, 0)
+
+    def lifecycle_fixture(self):
+        import json
+        import os
+        import uuid
+        name = 'logout-test'
+        runtime = self.root / 'private'
+        run = runtime / name
+        run.mkdir(parents=True, mode=0o700)
+        runtime.chmod(0o700)
+        audio = self.root / 'user-runtime'
+        (audio / 'pulse').mkdir(parents=True)
+        context = {'name': name, 'uuid': str(uuid.uuid4()), 'runtime': str(runtime),
+                   'user_runtime': str(audio), 'uid': os.getuid()}
+        process = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
+                                    '-name', name, '-uuid', context['uuid']])
+        self.addCleanup(lambda: process.wait(timeout=3))
+        self.addCleanup(lambda: process.poll() is None and process.kill())
+        (run / 'qemu.pid').write_text(str(process.pid))
+        (run / 'identity.json').write_text(json.dumps({'name': name, 'uuid': context['uuid']}))
+        listeners = []
+        for path in (run / 'qmp.sock', run / 'spice.sock', audio / 'pipewire-0', audio / 'pulse/native'):
+            listener = socket.socket(socket.AF_UNIX)
+            listener.bind(str(path))
+            listener.listen(5)
+            listener.settimeout(.1)
+            listeners.append(listener)
+            self.addCleanup(listener.close)
+        stopped = threading.Event()
+        def qmp_server():
+            while not stopped.is_set():
+                try:
+                    connection, _ = listeners[0].accept()
+                except socket.timeout:
+                    continue
+                except OSError:
+                    break
+                with connection, connection.makefile('rwb', buffering=0) as stream:
+                    stream.write(b'{"QMP": {}}\n')
+                    for line in stream:
+                        request = json.loads(line)
+                        replies = {'query-name': {'name': context.get('qmp_name', name)},
+                                   'query-uuid': {'UUID': context['uuid']},
+                                   'query-status': {'status': 'running'},
+                                   'query-kvm': {'enabled': True, 'present': True}}
+                        if request['execute'] == 'query-kvm' and 'on_snapshot' in context:
+                            context.pop('on_snapshot')()
+                        stream.write(json.dumps({'return': replies.get(request['execute'], {}),
+                                                 'id': request['id']}).encode() + b'\n')
+        thread = threading.Thread(target=qmp_server)
+        thread.start()
+        self.addCleanup(lambda: thread.join(timeout=3))
+        self.addCleanup(stopped.set)
+        return context, process, run
+
+    def test_logout_probe_detects_lost_private_socket(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        self.assertEqual(self.mod.lifecycle_snapshot(context), before)
+        (run / 'spice.sock').unlink()
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_rejects_stale_metadata_and_foreign_qmp_identity(self):
+        import json
+        context, process, run = self.lifecycle_fixture()
+        self.mod.lifecycle_snapshot(context)
+        (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': 'stale'}))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+        (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': context['uuid']}))
+        context['qmp_name'] = 'foreign-process'
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_rejects_dead_process_and_changed_runtime_socket(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        audio = Path(context['user_runtime']) / 'pipewire-0'
+        audio.unlink()
+        replacement = socket.socket(socket.AF_UNIX)
+        self.addCleanup(replacement.close)
+        replacement.bind(str(audio))
+        replacement.listen(1)
+        after = self.mod.lifecycle_snapshot(context)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.verify_lifecycle_survival(before, after)
+        process.terminate()
+        process.wait(timeout=3)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_accepts_verified_same_runtime_and_rejects_process_restart(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        self.mod.verify_lifecycle_survival(before, self.mod.lifecycle_snapshot(context))
+        process.terminate()
+        process.wait(timeout=3)
+        replacement = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
+                                        '-name', context['name'], '-uuid', context['uuid']])
+        self.addCleanup(lambda: replacement.wait(timeout=3))
+        self.addCleanup(lambda: replacement.poll() is None and replacement.kill())
+        (run / 'qemu.pid').write_text(str(replacement.pid))
+        changed = self.mod.lifecycle_snapshot(context)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.verify_lifecycle_survival(before, changed)
+
+    def test_large_guest_probe_survives_serial_tty_canonical_line_limit(self):
+        import os
+        import pty
+        import termios
+        master, slave = pty.openpty()
+        self.addCleanup(os.close, master)
+        self.addCleanup(os.close, slave)
+        options = termios.tcgetattr(slave)
+        options[3] &= ~termios.ECHO
+        termios.tcsetattr(slave, termios.TCSANOW, options)
+        process = subprocess.Popen(['/bin/sh'], stdin=slave, stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        self.addCleanup(process.stdout.close)
+        self.addCleanup(process.stderr.close)
+        self.addCleanup(lambda: process.wait(timeout=3))
+        self.addCleanup(lambda: process.poll() is None and process.kill())
+        payload = 'guest-proof-' * 1200
+        command = self.mod.frame_command('printf %s ' + payload, 'longprobe')
+        os.write(master, (command + '\nexit\n').encode())
+        output, error = process.communicate(timeout=3)
+        self.assertEqual(self.mod.parse_frame(output, 'longprobe'), (0, payload.encode()), error)
+
+    def test_generated_guest_lifecycle_probe_compiles(self):
+        compile(self.mod.lifecycle_guest_script(), '<guest-lifecycle-probe>', 'exec')
+
+    def session_fixture(self, records):
+        import json
+        import os
+        from unittest.mock import patch
+        source = self.root / 'session-records.json'
+        source.write_text(json.dumps(records))
+        commands = self.root / 'session-fixture-bin'
+        commands.mkdir()
+        command = commands / 'loginctl'
+        command.write_text('''#!/usr/bin/env python3
+import json, os, sys
+records = json.load(open(os.environ['SMOKE_SESSION_RECORDS']))
+args = sys.argv[1:]
+if args[0] == 'show-user':
+    print(' '.join(record['Id'] for record in records if str(record['User']) == args[1]))
+elif args[0] == 'show-session':
+    record = next((record for record in records if record['Id'] == args[1]), None)
+    if record is None: sys.exit(1)
+    for key, value in record.items(): print(key + '=' + str(value))
+else: sys.exit(1)
+''')
+        command.chmod(0o755)
+        environment = patch.dict(os.environ, {'PATH': str(commands) + ':' + os.environ['PATH'],
+                                              'SMOKE_SESSION_RECORDS': str(source)})
+        environment.start()
+        self.addCleanup(environment.stop)
+        return source
+
+    def session_record(self, ident='serial1', state='active', cls='user', leader=1234):
+        import os
+        return {'Id': ident, 'User': os.getuid(), 'State': state, 'Class': cls,
+                'Leader': leader, 'TTY': 'ttyS0' if cls == 'user' else ''}
+
+    def test_logout_sessions_accept_closing_scope_and_manager_with_surviving_vm(self):
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        self.session_fixture([dict(original, State='closing'),
+                              self.session_record('manager1', 'active', 'manager')])
+        self.assertTrue(self.mod.no_open_logins(context))
+        self.assertEqual(self.mod.capture_logout_snapshot(context), context['before'])
+
+    def test_logout_sessions_initial_login_ignores_manager_but_rejects_open_replacement(self):
+        import json
+        import os
+        original = self.session_record()
+        context = {'uid': os.getuid(), 'login_session': original}
+        source = self.session_fixture([original, self.session_record('manager1', 'active', 'manager')])
+        self.assertEqual(self.mod.require_initial_login(os.getuid(), 'serial1'), original)
+        for state in ('online', 'active'):
+            with self.subTest(state=state):
+                source.write_text(json.dumps([dict(original, State='closing'),
+                                             self.session_record('new-login', state)]))
+                self.assertFalse(self.mod.no_open_logins(context))
+        source.write_text(json.dumps([dict(original, State='closing', Leader=9999)]))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.no_open_logins(context)
+
+    def test_logout_sessions_reject_login_appearing_during_snapshot(self):
+        import json
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        source = self.session_fixture([dict(original, State='closing')])
+        context['on_snapshot'] = lambda: source.write_text(json.dumps([
+            dict(original, State='closing'), self.session_record('relogin')]))
+        self.assertTrue(self.mod.no_open_logins(context))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.capture_logout_snapshot(context)
+
+    def test_guest_tools_accept_gentoo_remote_viewer_without_virt_viewer(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        self.assertEqual(self.probe_fixture('clean', live=True), 0)

tests/test_profiles.py

@@ -0,0 +1,54 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from hydra.profiles import Profile, load_profile, save_profile
+
+
+class ProfilesTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix='hydra-test-')
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.iso = self.root / 'installer with spaces,comma.iso'
+        self.iso.touch()
+
+    def test_roundtrip_paths_and_defaults(self):
+        profile = Profile('guest', self.iso)
+        save_profile(self.root / 'guest.toml', profile)
+        loaded = load_profile(self.root / 'guest.toml')
+        self.assertEqual(loaded.iso, self.iso)
+        self.assertEqual((loaded.memory, loaded.cpus, loaded.disk_size), (4096, 2, '40G'))
+        self.assertEqual((loaded.graphics, loaded.network, loaded.bridge), ('auto', 'user', 'hydrabr0'))
+
+    def test_invalid_names_resources_and_modes(self):
+        for name in ('', '../guest', 'Guest', '-guest', 'a' * 33, 'a\n'):
+            with self.subTest(name=name), self.assertRaises(ValueError):
+                Profile(name, self.iso)
+        for field, values in {'memory': (0, -1, True, '4096'), 'cpus': (0, -1, True, 257),
+                              'disk_size': ('0G', '-1G', '40G,evil', '40'),
+                              'graphics': ('gl',), 'network': ('public',),
+                              'bridge': ('../bad', 'a,evil')}.items():
+            for value in values:
+                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
+                    Profile('guest', self.iso, **{field: value})
+
+    def test_missing_iso_and_duplicate_profile_preserve_existing(self):
+        with self.assertRaises(ValueError):
+            Profile('guest', self.root / 'missing.iso')
+        dest = self.root / 'guest.toml'
+        save_profile(dest, Profile('guest', self.iso))
+        original = dest.read_bytes()
+        with self.assertRaises(FileExistsError):
+            save_profile(dest, Profile('guest', self.iso, memory=8192))
+        self.assertEqual(dest.read_bytes(), original)
+
+    def test_profile_name_must_match_filename(self):
+        path = self.root / 'guest.toml'
+        path.write_text(f'name = "other"\niso = "{self.iso}"\n')
+        with self.assertRaises(ValueError):
+            load_profile(path)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_qmp.py

@@ -0,0 +1,141 @@
+import json
+import socket
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+
+from hydra.qmp import QMPClient, QMPError
+
+
+class QMPServer:
+    def __init__(self, path, handler):
+        self.socket = socket.socket(socket.AF_UNIX)
+        self.socket.bind(str(path))
+        self.socket.listen(4)
+        self.socket.settimeout(0.1)
+        self.handler = handler
+        self.stop = threading.Event()
+        self.errors = []
+        self.thread = threading.Thread(target=self.run, daemon=True)
+        self.thread.start()
+
+    def run(self):
+        while not self.stop.is_set():
+            try:
+                conn, _ = self.socket.accept()
+            except socket.timeout:
+                continue
+            except OSError:
+                break
+            try:
+                with conn:
+                    conn.settimeout(1)
+                    self.handler(conn)
+            except (BrokenPipeError, ConnectionResetError, socket.timeout):
+                pass
+            except Exception as exc:
+                self.errors.append(exc)
+
+    def close(self):
+        self.stop.set()
+        self.socket.close()
+        self.thread.join(2)
+
+
+def send(conn, message):
+    conn.sendall(json.dumps(message).encode() + b'\r\n')
+
+
+def greeting(conn):
+    send(conn, {'QMP': {'version': {'qemu': {'major': 10, 'minor': 0, 'micro': 0},
+                                   'package': ''}, 'capabilities': []}})
+
+
+class QMPTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix='hq-')
+        self.addCleanup(self.tmp.cleanup)
+        self.path = Path(self.tmp.name) / 'qmp'
+
+    def server(self, handler):
+        srv = QMPServer(self.path, handler)
+        self.addCleanup(srv.close)
+        return srv
+
+    def test_handshake_ids_events_and_return(self):
+        commands = []
+
+        def handler(conn):
+            greeting(conn)
+            stream = conn.makefile('rb')
+            for index in range(2):
+                request = json.loads(stream.readline())
+                commands.append(request)
+                send(conn, {'event': 'STOP', 'data': {}})
+                send(conn, {'return': {'ignored': True}, 'id': 'unrelated'})
+                send(conn, {'return': {} if index == 0 else {'status': 'running'}, 'id': request['id']})
+        self.server(handler)
+        with QMPClient(self.path, timeout=0.5) as client:
+            self.assertEqual(client.command('query-status'), {'status': 'running'})
+        self.assertEqual([cmd['execute'] for cmd in commands], ['qmp_capabilities', 'query-status'])
+        self.assertNotEqual(commands[0]['id'], commands[1]['id'])
+
+    def test_command_error(self):
+        def handler(conn):
+            greeting(conn)
+            stream = conn.makefile('rb')
+            for index in range(2):
+                request = json.loads(stream.readline())
+                if index == 0:
+                    send(conn, {'return': {}, 'id': request['id']})
+                else:
+                    send(conn, {'error': {'class': 'CommandNotFound', 'desc': 'unsupported'}, 'id': request['id']})
+        self.server(handler)
+        with QMPClient(self.path, timeout=0.5) as client, self.assertRaises(QMPError):
+            client.command('missing')
+
+    def test_greeting_timeout_is_bounded(self):
+        self.server(lambda conn: time.sleep(0.3))
+        started = time.monotonic()
+        with self.assertRaises(QMPError):
+            QMPClient(self.path, timeout=0.08)
+        self.assertLess(time.monotonic() - started, 0.25)
+
+    def test_event_stream_cannot_extend_command_deadline(self):
+        def handler(conn):
+            greeting(conn)
+            request = json.loads(conn.makefile('rb').readline())
+            while True:
+                send(conn, {'event': 'STOP', 'data': {}})
+                time.sleep(0.01)
+        self.server(handler)
+        started = time.monotonic()
+        with self.assertRaises(QMPError):
+            QMPClient(self.path, timeout=0.08)
+        self.assertLess(time.monotonic() - started, 0.25)
+
+    def test_malformed_greeting_rejected(self):
+        self.server(lambda conn: send(conn, {'return': {}}))
+        with self.assertRaises(QMPError):
+            QMPClient(self.path)
+
+    def test_command_timeout_after_successful_handshake(self):
+        def handler(conn):
+            greeting(conn)
+            with conn.makefile('rb') as stream:
+                request = json.loads(stream.readline())
+                send(conn, {'return': {}, 'id': request['id']})
+                stream.readline()
+                time.sleep(0.3)
+        self.server(handler)
+        with QMPClient(self.path, timeout=0.08) as client:
+            started = time.monotonic()
+            with self.assertRaises(QMPError):
+                client.command('query-status')
+            self.assertLess(time.monotonic() - started, 0.25)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_runtime.py

@@ -0,0 +1,310 @@
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from hydra.profiles import Profile, save_profile
+from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv
+from test_qmp import QMPServer, greeting, send
+
+
+# External protocol fixtures isolate lifecycle tests from user VMs and real disks.
+FAKE_QEMU = r'''#!/usr/bin/env python3
+import json, os, socket, sys
+args = sys.argv[1:]
+def val(flag): return args[args.index(flag)+1]
+if os.environ.get('HYDRA_TEST_FAIL'):
+    print('intentional boot failure', file=sys.stderr)
+    sys.exit(1)
+path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',')
+pid = os.fork()
+if pid:
+    sys.exit(0)
+os.setsid()
+with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid()))
+s = socket.socket(socket.AF_UNIX)
+s.bind(path)
+s.listen(8)
+alive = True
+while alive:
+    c, _ = s.accept()
+    try:
+        with c:
+            c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n')
+            with c.makefile('rb') as stream:
+                for line in stream:
+                    cmd = json.loads(line)
+                    name = cmd['execute']
+                    result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')},
+                              'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {})
+                    c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n')
+                    if name == 'quit':
+                        alive = False
+                        break
+    except (BrokenPipeError, ConnectionResetError):
+        pass
+s.close()
+os.unlink(path)
+'''
+
+FAKE_IMG = r'''#!/usr/bin/env python3
+import sys
+with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture')
+'''
+
+
+class RuntimeTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix='hr-')
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma',
+                               self.root / 'run,comma', boot_timeout=1)
+        self.iso = self.root / 'installer space,comma.iso'
+        self.iso.touch()
+        self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M')
+        self.runtime.add(self.profile)
+        self.qemu = self.root / 'fake qemu'
+        self.qemu.write_text(FAKE_QEMU)
+        self.qemu.chmod(0o700)
+        self.img = self.root / 'fake img'
+        self.img.write_text(FAKE_IMG)
+        self.img.chmod(0o700)
+        env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)})
+        env.start()
+        self.addCleanup(env.stop)
+        self.addCleanup(self.cleanup_guest)
+
+    def cleanup_guest(self):
+        try:
+            if self.runtime.status('guest')['status'] == 'running':
+                self.runtime.stop('guest', force=True)
+        except VMError:
+            pass
+
+    def test_argv_is_private_and_comma_safe(self):
+        paths = self.runtime.paths('guest')
+        argv = build_qemu_argv(self.profile, paths)
+        self.assertEqual(argv[0], str(self.qemu))
+        blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
+        self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs])
+        self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs])
+        spice = argv[argv.index('-spice')+1]
+        self.assertIn('unix=on', spice)
+        self.assertIn('disable-copy-paste=on', spice)
+        self.assertIn('disable-agent-file-xfer=on', spice)
+        self.assertIn('run,,comma', spice)
+        self.assertIn('virtio-blk-pci,drive=disk', argv)
+        self.assertIn('virtio-net-pci,netdev=net', argv)
+        self.assertNotIn('-vnc', argv)
+        self.assertNotIn('-virtfs', argv)
+
+    def test_socket_removed_during_stat_is_safely_not_listening(self):
+        paths = self.runtime.paths('guest')
+        paths.qmp.parent.mkdir(parents=True)
+        with socket.socket(socket.AF_UNIX) as fixture:
+            fixture.bind(str(paths.qmp))
+            original_lstat = Path.lstat
+            def disappearing_socket(path):
+                if path == paths.qmp:
+                    path.unlink(missing_ok=True)
+                return original_lstat(path)
+            with patch.object(Path, 'lstat', disappearing_socket):
+                self.assertFalse(self.runtime._listening(paths.qmp))
+
+    def test_argv_provides_audio_and_manual_usb_channels(self):
+        argv = build_qemu_argv(self.profile, self.runtime.paths('guest'))
+        self.assertIn('spice,id=audio', argv)
+        self.assertIn('hda-duplex,audiodev=audio', argv)
+        for channel in range(3):
+            self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv)
+            self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv)
+        self.assertFalse(any('usb-host' in argument for argument in argv))
+
+    def test_metadata_symlink_does_not_overwrite_user_data(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        valuable = self.root / 'valuable data'
+        valuable.write_text('preserve me')
+        paths.metadata.symlink_to(valuable)
+        self.runtime.start('guest')
+        self.assertEqual(valuable.read_text(), 'preserve me')
+        self.assertFalse(paths.metadata.is_symlink())
+
+    def test_disk_symlink_never_changes_its_target(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        valuable = self.root / 'valuable disk'
+        valuable.write_text('preserve me')
+        paths.disk.symlink_to(valuable)
+        with self.assertRaises(VMError):
+            self.runtime.start('guest')
+        self.assertEqual(valuable.read_text(), 'preserve me')
+
+    def test_malformed_identity_record_is_safe_unknown(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.metadata.write_text('[]')
+        server = QMPServer(paths.qmp, greeting)
+        self.addCleanup(server.close)
+        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
+
+    def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
+        commands = []
+        def handler(conn):
+            greeting(conn)
+            for line in conn.makefile('rb'):
+                req = json.loads(line)
+                commands.append(req['execute'])
+                result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
+                send(conn, {'return': result, 'id': req['id']})
+        server = QMPServer(paths.qmp, handler)
+        self.addCleanup(server.close)
+        with self.assertRaises(VMError):
+            self.runtime.stop('guest', force=True)
+        self.assertNotIn('quit', commands)
+
+    def test_reused_pid_without_qmp_never_controls_a_process(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.pid.write_text(str(os.getpid()))
+        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
+        with patch('os.kill', side_effect=AssertionError('must not signal any PID')):
+            with self.assertRaises(VMError):
+                self.runtime.stop('guest', force=True)
+
+    def test_malformed_query_result_is_safe_unknown(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
+        def handler(conn):
+            greeting(conn)
+            for line in conn.makefile('rb'):
+                req = json.loads(line)
+                send(conn, {'return': [], 'id': req['id']})
+        server = QMPServer(paths.qmp, handler)
+        self.addCleanup(server.close)
+        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
+
+    def test_cli_qmp_failure_reports_error_without_traceback(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
+        def handler(conn):
+            greeting(conn)
+            for line in conn.makefile('rb'):
+                req = json.loads(line)
+                result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
+                if req['execute'] == 'system_powerdown':
+                    send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']})
+                else:
+                    send(conn, {'return': result, 'id': req['id']})
+        server = QMPServer(paths.qmp, handler)
+        self.addCleanup(server.close)
+        run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config),
+                              '--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime),
+                              'shutdown', 'guest'], capture_output=True, text=True, timeout=4)
+        self.assertEqual(run.returncode, 1)
+        self.assertIn('cannot power down', run.stderr)
+        self.assertNotIn('Traceback', run.stderr)
+
+    def test_disk_and_profile_survive_shutdown_and_force(self):
+        paths = self.runtime.paths('guest')
+        paths.disk.parent.mkdir(parents=True, exist_ok=True)
+        paths.disk.write_bytes(b'existing valuable disk')
+        self.runtime.start('guest')
+        self.assertEqual(self.runtime.status('guest')['status'], 'running')
+        self.runtime.shutdown('guest')
+        self.assertEqual(self.runtime.status('guest')['status'], 'running')
+        with self.assertRaises(VMError):
+            self.runtime.stop('guest', force=False)
+        self.runtime.stop('guest', force=True)
+        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
+        self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk')
+
+    def test_simultaneous_start_only_launches_one_vm(self):
+        barrier = threading.Barrier(2)
+        results = []
+        def start():
+            barrier.wait()
+            try:
+                results.append(self.runtime.start('guest')['status'])
+            except VMError as exc:
+                results.append(str(exc))
+        threads = [threading.Thread(target=start) for _ in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(4)
+            self.assertFalse(thread.is_alive())
+        self.assertEqual(results.count('running'), 1)
+        self.assertEqual(len(results), 2)
+        self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
+
+    def test_stale_socket_is_replaced_only_when_no_listener(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        stale = socket.socket(socket.AF_UNIX)
+        stale.bind(str(paths.qmp))
+        stale.close()
+        self.runtime.start('guest')
+        self.assertEqual(self.runtime.status('guest')['status'], 'running')
+
+    def test_failed_boot_is_reported_and_disk_preserved(self):
+        with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError):
+            self.runtime.start('guest')
+        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
+        self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
+        self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text())
+
+    def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self):
+        paths = self.runtime.paths('guest')
+        self.runtime.prepare(paths)
+        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
+        paths.pid.write_text(str(os.getpid()))
+        commands = []
+        def handler(conn):
+            greeting(conn)
+            for line in conn.makefile('rb'):
+                req = json.loads(line)
+                commands.append(req['execute'])
+                result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {})
+                send(conn, {'return': result, 'id': req['id']})
+        server = QMPServer(paths.qmp, handler)
+        self.addCleanup(server.close)
+        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
+        for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True),
+                        lambda: self.runtime.start('guest')):
+            with self.assertRaises(VMError):
+                control()
+        self.assertNotIn('quit', commands)
+        self.assertNotIn('system_powerdown', commands)
+        self.assertFalse(paths.disk.exists())
+
+    def test_private_runtime_and_socket_length(self):
+        self.runtime.prepare(self.runtime.paths('guest'))
+        self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700)
+        with self.assertRaises(VMError):
+            Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest')
+
+    def test_cli_lifecycle_and_global_paths(self):
+        base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state),
+                '--runtime', str(self.runtime.runtime)]
+        for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'),
+                               (['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]:
+            run = subprocess.run(base + args, text=True, capture_output=True, timeout=4)
+            self.assertEqual(run.returncode, 0, run.stderr)
+            self.assertIn(expected, run.stdout)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_session.py

@@ -0,0 +1,256 @@
+import contextlib
+import fcntl
+import io
+import json
+import os
+import shlex
+import socket
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+from xml.etree import ElementTree as ET
+
+from hydra.profiles import Profile
+from hydra.runtime import Runtime
+from hydra import session
+from hydra.cli import main, parser
+
+
+class SessionTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix='hs-')
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.iso = self.root / 'scratch.iso'
+        self.iso.touch()
+        self.manager = Runtime(self.root / 'config', self.root / 'state', self.root / 'run')
+        self.manager.add(Profile('alpha', self.iso))
+        self.manager.add(Profile('beta', self.iso))
+
+    def test_config_has_exact_titles_workspaces_and_explicit_fullscreen(self):
+        tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot', 'a & b', "x'y"]))
+        self.assertEqual([n.text for n in tree.findall('./desktops/names/name')], ['alpha', 'beta'])
+        self.assertEqual(tree.findtext('./desktops/popupTime'), '0')
+        rules = tree.findall('./windowRules/windowRule')
+        self.assertEqual([r.get('title') for r in rules], ['HydraOS: alpha', 'HydraOS: beta'])
+        for index, rule in enumerate(rules, 1):
+            self.assertEqual(rule.get('serverDecoration'), 'no')
+            self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('to'), str(index))
+            self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('follow'), 'no')
+            # Fullscreen is requested by the viewer, never toggled by a rule.
+            self.assertIsNone(rule.find("./action[@name='ToggleFullscreen']"))
+        recovery = tree.find("./keyboard/keybind[@key='C-A-S-Escape']/action")
+        self.assertEqual(shlex.split(recovery.get('command')), ['foot', 'a & b', "x'y"])
+
+    def test_only_reserved_host_keys_and_inert_mouse_binding(self):
+        tree = ET.fromstring(session.build_labwc_config(['alpha'], ['foot']))
+        self.assertIsNone(tree.find('./keyboard/default'))
+        self.assertIsNone(tree.find('./mouse/default'))
+        bindings = tree.findall('./keyboard/keybind')
+        self.assertEqual({b.get('key') for b in bindings}, {'C-A-S-Left', 'C-A-S-Right', 'C-A-S-Escape', 'XF86MonBrightnessUp', 'XF86MonBrightnessDown'})
+        self.assertTrue(all(b.get('overrideInhibition') == 'yes' for b in bindings))
+        self.assertEqual(len(tree.findall('./mouse/context/mousebind')), 1)
+        self.assertEqual(tree.findall('./mouse/context/mousebind/action'), [])
+        self.assertEqual(tree.findtext('./libinput/device/tap'), 'yes')
+
+    def test_viewer_argv_is_private_safe_and_manual_usb_only(self):
+        paths = self.manager.paths('alpha')
+        argv = session.build_viewer_argv(paths, '/tool path/viewer')
+        self.assertEqual(argv[0], '/tool path/viewer')
+        self.assertIn('--full-screen', argv)
+        self.assertIn('--hide-toolbar', argv)
+        self.assertFalse(any(arg.startswith('--spice-shared-dir') for arg in argv))
+        self.assertFalse(any(arg.startswith('--spice-share-cd') for arg in argv))
+        self.assertIn('--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0', argv)
+        self.assertIn('--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0', argv)
+        self.assertIn('usb-device-select=ctrl+alt+shift+u', argv[argv.index('--hotkeys') + 1])
+        self.assertEqual(argv[-1], 'spice+unix://' + str(paths.spice))
+        windowed = session.build_viewer_argv(paths, 'remote-viewer', windowed=True)
+        self.assertNotIn('--full-screen', windowed)
+        self.assertNotIn('--hide-toolbar', windowed)
+        with self.assertRaises(ValueError):
+            session.viewer_title('bad*')
+
+    def test_missing_display_and_viewer_are_recoverable_cli_errors(self):
+        base = ['--config', str(self.manager.config), '--state', str(self.manager.state), '--runtime', str(self.manager.runtime)]
+        with patch.dict(os.environ, {}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error:
+            self.assertEqual(main(base + ['view', 'alpha']), 1)
+        self.assertIn('display', error.getvalue().lower())
+        with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': '/missing/remote-viewer'}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error:
+            self.assertEqual(main(base + ['view', 'alpha']), 1)
+        self.assertIn('viewer', error.getvalue().lower())
+
+    def test_windowed_viewer_exit_leaves_vm_control_untouched(self):
+        viewer = self.root / 'fake-viewer'
+        viewer.write_text('#!/bin/sh\ncase "$1" in --help) echo --hide-toolbar; exit 0;; esac\nexit 7\n')
+        viewer.chmod(0o700)
+        self.manager.paths('alpha').spice.parent.mkdir(parents=True, exist_ok=True)
+        sock = socket.socket(socket.AF_UNIX)
+        self.addCleanup(sock.close)
+        sock.bind(str(self.manager.paths('alpha').spice))
+        with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer)}), patch.object(self.manager, 'stop', side_effect=AssertionError('must not stop')):
+            self.assertEqual(session.view(self.manager, 'alpha', windowed=True), 7)
+        self.assertTrue(self.manager.paths('alpha').spice.exists())
+
+    def test_session_parser_and_default_physical_backend(self):
+        self.assertTrue(parser().parse_args(['session', '--nested']).nested)
+        physical = session.compositor_environment({'WAYLAND_DISPLAY': 'active', 'DISPLAY': ':0', 'WLR_BACKENDS': 'wayland'}, nested=False)
+        self.assertNotIn('WAYLAND_DISPLAY', physical)
+        self.assertNotIn('DISPLAY', physical)
+        self.assertEqual(physical['WLR_BACKENDS'], 'drm,libinput')
+        nested = session.compositor_environment({'WAYLAND_DISPLAY': 'active'}, nested=True)
+        self.assertEqual(nested['WLR_BACKENDS'], 'wayland')
+        with self.assertRaises(session.SessionError):
+            session.compositor_environment({}, nested=True)
+
+    def test_uri_escapes_percent_query_and_fragment(self):
+        from dataclasses import replace
+        paths = replace(self.manager.paths('alpha'), spice=self.root / 'a %?#.sock')
+        self.assertTrue(session.build_viewer_argv(paths, 'viewer')[-1].endswith('/a%20%25%3F%23.sock'))
+
+    def test_workspace_names_cannot_be_interpreted_as_relative_actions(self):
+        tree = ET.fromstring(session.build_labwc_config(['left', 'right', 'current'], ['foot']))
+        self.assertEqual([a.get('to') for a in tree.findall("./windowRules/windowRule/action[@name='SendToDesktop']")], ['1', '2', '3'])
+
+    def test_launch_writes_only_selected_private_runtime_and_serializes_paths(self):
+        capture = self.root / 'capture.json'
+        labwc = self.root / 'fake-labwc'
+        labwc.write_text('#!' + os.sys.executable + '\n' + '''import json, os, sys
+from pathlib import Path
+if '--version' in sys.argv:
+    print('labwc 0.20.2')
+else:
+    config = Path(sys.argv[sys.argv.index('-C') + 1])
+    manifest = json.loads((config.parent / 'session.json').read_text())
+    Path(os.environ['CAPTURE']).write_text(json.dumps({'manifest': manifest, 'config': str(config), 'xml': (config / 'rc.xml').read_text(), 'env': dict(os.environ), 'autostart': (config / 'autostart').read_text()}))
+''')
+        labwc.chmod(0o700)
+        old_home = self.root / 'untouched-home'
+        old_home.mkdir()
+        marker = old_home / 'keep'
+        marker.write_text('valuable')
+        with patch.dict(os.environ, {'HYDRA_LABWC': str(labwc), 'HYDRA_TERMINAL': str(labwc), 'HYDRA_BRIGHTNESS': str(labwc), 'CAPTURE': str(capture), 'HOME': str(old_home), 'DISPLAY': ':fake', 'WAYLAND_DISPLAY': 'active'}):
+            self.assertEqual(session.launch_session(self.manager), 0)
+        data = json.loads(capture.read_text())
+        self.assertTrue(Path(data['config']).is_relative_to(self.manager.runtime))
+        self.assertFalse(Path(data['config']).exists())
+        self.assertEqual(data['manifest']['config'], str(self.manager.config))
+        self.assertNotIn('DISPLAY', data['env'])
+        self.assertNotIn('WAYLAND_DISPLAY', data['env'])
+        self.assertEqual(list(old_home.iterdir()), [marker])
+        self.assertEqual(marker.read_text(), 'valuable')
+
+    def test_missing_socket_and_stock_viewer_do_not_create_disk(self):
+        viewer = self.root / 'viewer'
+        viewer.write_text('#!/bin/sh\necho --hide-toolbar\n')
+        viewer.chmod(0o700)
+        with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'fake', 'HYDRA_VIEWER': str(viewer)}):
+            with self.assertRaisesRegex(session.SessionError, 'no SPICE socket'):
+                session.view(self.manager, 'alpha')
+            viewer.write_text('#!/bin/sh\necho stock-viewer\n')
+            with self.assertRaisesRegex(session.SessionError, 'adaptation'):
+                session.view(self.manager, 'alpha')
+        self.assertFalse(self.manager.paths('alpha').disk.exists())
+
+    def test_viewer_settings_are_isolated_and_disable_clipboard(self):
+        env = session._viewer_environment(self.root / 'viewer-state')
+        self.assertEqual(Path(env['XDG_CONFIG_HOME']).joinpath('virt-viewer/settings').read_text(), '[virt-viewer]\nshare-clipboard=false\n')
+        self.assertEqual(env['GSETTINGS_BACKEND'], 'memory')
+
+    def test_initial_workspace_can_select_a_running_vm_after_stopped_profiles(self):
+        tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot'], initial='beta'))
+        self.assertEqual(tree.findtext('./desktops/initial'), '2')
+        with self.assertRaises(session.SessionError):
+            session.build_labwc_config(['alpha'], ['foot'], initial='missing')
+
+    def test_bad_helper_manifest_is_a_cli_error(self):
+        manifest = self.root / 'bad-session.json'
+        manifest.write_text('[]')
+        with contextlib.redirect_stderr(io.StringIO()) as error:
+            self.assertEqual(main(['_session-helper', 'recover', str(manifest)]), 1)
+        self.assertIn('manifest', error.getvalue().lower())
+
+    def test_no_running_vms_opens_recovery_without_start_or_stop(self):
+        terminal = self.root / 'terminal'
+        capture = self.root / 'recovery-argv.json'
+        terminal.write_text('#!' + os.sys.executable + '\nimport json, os, sys\nfrom pathlib import Path\nPath(os.environ["CAPTURE"]).write_text(json.dumps(sys.argv[1:]))\n')
+        terminal.chmod(0o700)
+        manifest = self.root / 'session.json'
+        manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha', 'beta'], 'terminal': str(terminal), 'directory': str(self.root)}))
+        with patch.dict(os.environ, {'CAPTURE': str(capture)}), patch.object(Runtime, 'start', side_effect=AssertionError('must not start')), patch.object(Runtime, 'stop', side_effect=AssertionError('must not stop')):
+            self.assertEqual(session.session_helper('clients', manifest), 0)
+        self.assertEqual(json.loads(capture.read_text())[0], '--title=HydraOS-recovery')
+        self.assertEqual(self.manager.status('alpha')['status'], 'stopped')
+        self.assertFalse(self.manager.paths('alpha').disk.exists())
+
+    def test_reconnect_does_not_duplicate_a_session_viewer(self):
+        manifest = self.root / 'session.json'
+        manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha'], 'terminal': '/bin/sh'}))
+        with (self.root / 'alpha.viewer.lock').open('a') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with patch.object(session, 'view', side_effect=AssertionError('must not duplicate')), contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(session.session_helper('reconnect:alpha', manifest), 0)
+
+    def _recovery_fixture(self):
+        for name in ('s', 'q'):
+            self.manager.add(Profile(name, self.iso))
+            paths = self.manager.paths(name)
+            paths.spice.parent.mkdir(parents=True)
+            fixture = socket.socket(socket.AF_UNIX)
+            self.addCleanup(fixture.close)
+            fixture.bind(str(paths.spice))
+        viewer = self.root / 'viewer'
+        viewer.write_text('#!' + os.sys.executable + '\n' + '''import os, sys
+from pathlib import Path
+from urllib.parse import unquote, urlsplit
+if '--help' in sys.argv:
+    print('--hide-toolbar')
+else:
+    name = Path(unquote(urlsplit(sys.argv[-1]).path)).parent.name
+    Path(os.environ['RECOVERY_CAPTURE'], name + '.connected').write_text(name)
+''')
+        viewer.chmod(0o700)
+        shell = self.root / 'shell'
+        shell.write_text('#!' + os.sys.executable + '\nimport os\nfrom pathlib import Path\nPath(os.environ["RECOVERY_CAPTURE"], "shell.opened").touch()\n')
+        shell.chmod(0o700)
+        manifest = self.root / 'session.json'
+        manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['s', 'q'], 'terminal': '/bin/sh'}))
+        return manifest, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer), 'RECOVERY_CAPTURE': str(self.root), 'SHELL': str(shell)}
+
+    def _check_recovery_vm_name(self, name):
+        manifest, env = self._recovery_fixture()
+        processes = []
+        original_popen = subprocess.Popen
+        def launch(*args, **kwargs):
+            process = original_popen(*args, **kwargs)
+            processes.append(process)
+            return process
+        with patch.dict(os.environ, env), patch('builtins.input', side_effect=[name, ':quit', EOFError()]), patch.object(subprocess, 'Popen', launch), contextlib.redirect_stdout(io.StringIO()):
+            self.assertEqual(session.session_helper('recover', manifest), 0)
+            for process in processes:
+                self.assertEqual(process.wait(timeout=5), 0)
+        connected = self.root / (name + '.connected')
+        self.assertTrue(connected.exists(), f'VM {name} was not reconnected')
+        self.assertEqual(connected.read_text(), name)
+        self.assertFalse((self.root / 'shell.opened').exists())
+        self.assertFalse(self.manager.paths(name).disk.exists())
+
+    def test_recovery_reconnects_vm_named_s(self):
+        self._check_recovery_vm_name('s')
+
+    def test_recovery_reconnects_vm_named_q(self):
+        self._check_recovery_vm_name('q')
+
+    def test_recovery_colon_commands_open_shell_and_close_terminal(self):
+        manifest, env = self._recovery_fixture()
+        with patch.dict(os.environ, env), patch('builtins.input', side_effect=[':shell', ':quit', EOFError()]), contextlib.redirect_stdout(io.StringIO()):
+            self.assertEqual(session.session_helper('recover', manifest), 0)
+        self.assertTrue((self.root / 'shell.opened').exists())
+        self.assertFalse((self.root / 's.connected').exists())
+        self.assertFalse((self.root / 'q.connected').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()