Capture redacted guest serial while waiting for shutdown

README.md

@@ -24,32 +24,3 @@ a refresh-rate setting does not prove delivered FPS.
 - [Hardware target](docs/hardware/asus-fx517zc.md)
 
 Build outputs, VM disks, runtime state and credentials stay outside Git.
-
-## Run the prototype
-
-```sh
-python3 -m unittest discover -s tests -v
-nix-shell dev/tools.nix
-bin/hydra add guest --iso /absolute/path/installer.iso
-bin/hydra start guest
-bin/hydra view guest
-# From a dedicated graphical login; --nested uses the current graphical display.
-bin/hydra session --nested
-```
-
-After installation, shut the guest down and run `bin/hydra media guest --detach`
-to boot its existing disk without installation media. `add` also supports disk-only
-profiles, `--width`, `--height`, and `--refresh-millihz`. Distribution preferences
-in `/etc/hydra/defaults.toml` affect only new profiles; explicit CLI options win.
-
-`nix-shell dev/tools.nix --run 'python3 dev/smoke.py'` checks two disposable KVM
-counter guests and private headless viewers, including manual USB selection,
-reserved switching under inhibition, recovery and reconnect. See
-[verification commands and physical checks still pending](docs/verification.md).
-
-An empty session opens local recovery. Enter `:shell`, create and start guests
-with `hydra add` / `hydra start`, then leave the shell and enter `:refresh`.
-Refresh rebuilds workspaces and reconnects viewers for the current profiles;
-running VMs keep their processes and disks. Use it after adding or removing
-profiles. New profiles save an individual MAC address; legacy profiles obtain
-a stable address from their owner and disk path, including on the shared bridge.

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/build-input-probe.sh

@@ -1,8 +0,0 @@
-#!/bin/sh
-set -eu
-source_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
-build_dir=$(mktemp -d /tmp/hydra-input-build-XXXXXX)
-wayland-scanner client-header "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-client-protocol.h"
-wayland-scanner private-code "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-protocol.c"
-cc -Wall -Wextra -Werror "$source_dir/input-probe.c" "$build_dir/virtual-keyboard-protocol.c" -I"$build_dir" $(pkg-config --cflags --libs wayland-client xkbcommon) -o "$build_dir/input-probe"
-printf '%s\n' "$build_dir/input-probe"

dev/check-edid.py

@@ -1,100 +0,0 @@
-#!/usr/bin/env python3
-"""Compile actual QEMU EDID code and decode the guest-visible response.
-
-Catches loss of the configured refresh rate, a frontend overriding it, and
-monitor ranges excluding the generated preferred mode. No VM is touched.
-"""
-import os
-from pathlib import Path
-import subprocess
-import sys
-import tempfile
-import tarfile
-
-source_root = Path(sys.argv[1])
-# Accept the pinned QEMU source archive directly. Read only the four files
-# needed here, avoiding unrelated upstream absolute symlinks in rom sources.
-if source_root.is_file():
-    extracted = tempfile.TemporaryDirectory(prefix='hydra-edid-source-')
-    with tarfile.open(source_root) as archive:
-        members = {name: next(member for member in archive.getmembers()
-                              if member.name.endswith('/' + name)) for name in (
-            'hw/display/edid-generate.c', 'include/hw/display/edid.h',
-            'include/hw/virtio/virtio-gpu.h', 'hw/display/virtio-gpu-base.c')}
-        source_root = Path(extracted.name)
-        for name, member in members.items():
-            destination = source_root / name
-            destination.parent.mkdir(parents=True, exist_ok=True)
-            destination.write_bytes(archive.extractfile(member).read())
-    subprocess.run(['patch', '--batch', '-p1', '-i',
-                    str(Path(__file__).resolve().parents[1] / 'patches/qemu-virtio-refresh-rate.patch')],
-                   cwd=source_root, check=True)
-edid = (source_root / 'hw/display/edid-generate.c').read_text()
-edid = '\n'.join(line for line in edid.splitlines() if not line.startswith('#include'))
-header = (source_root / 'include/hw/display/edid.h').read_text()
-header = header[header.index('typedef struct qemu_edid_info'):header.index('void qemu_edid_region_io')]
-gpu = (source_root / 'hw/display/virtio-gpu-base.c').read_text()
-start = gpu.index('void\nvirtio_gpu_base_generate_edid')
-gpu = gpu[start:gpu.index('static void virtio_gpu_invalidate_display', start)]
-stubs = r'''
-#include <stdint.h>
-#include <stddef.h>
-#include <string.h>
-#include <stdlib.h>
-#include <stdio.h>
-#define ARRAY_SIZE(x) (sizeof(x)/sizeof((x)[0]))
-#define MAX(a,b) ((a) > (b) ? (a) : (b))
-#define MIN(a,b) ((a) < (b) ? (a) : (b))
-#define DIV_ROUND_UP(n,d) (((n) + (d) - 1) / (d))
-#define cpu_to_le32(x) (x)
-static void stw_le_p(void *p,uint16_t v) { uint8_t *b=p; b[0]=v; b[1]=v>>8; }
-static void stw_be_p(void *p,uint16_t v) { uint8_t *b=p; b[1]=v; b[0]=v>>8; }
-static void stl_le_p(void *p,uint32_t v) { uint8_t *b=p; for(int i=0;i<4;i++) b[i]=v>>(8*i); }
-typedef struct VirtIOGPUOutput { const char *name; } VirtIOGPUOutput;
-typedef struct VirtIOGPUOutputList { struct VirtIOGPUOutputList *next; VirtIOGPUOutput *value; } VirtIOGPUOutputList;
-struct virtio_gpu_requested_state { uint16_t width_mm,height_mm; uint32_t width,height,refresh_rate; int x,y; };
-typedef struct VirtIOGPUBase { struct { uint32_t refresh_rate; VirtIOGPUOutputList *outputs; } conf; struct virtio_gpu_requested_state req_state[1]; } VirtIOGPUBase;
-struct virtio_gpu_resp_edid { uint32_t size; uint8_t edid[1024]; };
-'''
-main = r'''
-int main(int argc,char **argv) {
-    VirtIOGPUBase gpu = {0}; struct virtio_gpu_resp_edid out = {0};
-    gpu.conf.refresh_rate=atoi(argv[1]); gpu.req_state[0].refresh_rate=atoi(argv[2]);
-    gpu.req_state[0].width=atoi(argv[3]); gpu.req_state[0].height=atoi(argv[4]);
-    virtio_gpu_base_generate_edid(&gpu,0,&out);
-    fwrite(out.edid,1,qemu_edid_size(out.edid),stdout); return 0;
-}
-'''
-compiler = os.environ.get('CC', 'cc')
-with tempfile.TemporaryDirectory(prefix='qemu-edid-probe-') as temporary:
-    directory = Path(temporary)
-    (directory / 'probe.c').write_text(stubs + header + edid + gpu + main)
-    subprocess.run([compiler, '-O2', str(directory / 'probe.c'), '-o', str(directory / 'probe')], check=True)
-    for configured, frontend, width, height, expected in [
-        (0, 0, 1920, 1080, 75),
-        (0, 60000, 1920, 1080, 60),
-        (144000, 0, 1920, 1080, 144),
-        (144000, 60000, 1920, 1080, 144),
-        (144000, 0, 1600, 900, 144),
-        (0, 144000, 1920, 1080, 144),
-        (50000, 0, 1920, 1080, 50),
-        (255000, 0, 1280, 800, 255),
-    ]:
-        data = subprocess.check_output([str(directory / 'probe'), *map(str, (configured, frontend, width, height))])
-        assert all(sum(data[offset:offset + 128]) % 256 == 0 for offset in range(0, len(data), 128)), 'invalid EDID checksum'
-        timing = data[54:72]
-        clock = int.from_bytes(timing[:2], 'little') * 10000
-        horizontal = timing[2] + ((timing[4] >> 4) << 8)
-        hblank = timing[3] + ((timing[4] & 15) << 8)
-        vertical = timing[5] + ((timing[7] >> 4) << 8)
-        vblank = timing[6] + ((timing[7] & 15) << 8)
-        actual = clock / (horizontal + hblank) / (vertical + vblank)
-        assert (horizontal, vertical) == (width, height)
-        assert abs(actual - expected) < 0.01, f'configured={configured}, frontend={frontend}: wanted {expected} Hz, got {actual}'
-        ranges = next(data[offset:offset + 18] for offset in (54, 72, 90, 108) if data[offset:offset + 5] == b'\0\0\0\xfd\0')
-        assert ranges[5] <= actual <= ranges[6], f'vertical range {ranges[5:7]} excludes {actual}'
-        horizontal_khz = clock / (horizontal + hblank) / 1000
-        assert ranges[7] <= horizontal_khz <= ranges[8], f'horizontal range {ranges[7:9]} excludes {horizontal_khz}'
-        if configured == 0 and expected <= 75:
-            assert ranges[5:9] == bytes((50, 125, 30, 160)), 'legacy monitor range changed'
-        print(f'PASS configured={configured}, frontend={frontend}: {width}x{height}@{actual:.6f}, ranges={list(ranges[5:9])}')

dev/check-network.py

@@ -1,78 +0,0 @@
-#!/usr/bin/env python3
-"""Check real NIC identities on a bridge isolated in user/net/mount namespaces."""
-import argparse
-import json
-from pathlib import Path
-import subprocess
-import sys
-import tempfile
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from hydra.profiles import Profile
-from hydra.runtime import Runtime
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('--namespace-root', type=Path, help=argparse.SUPPRESS)
-    args = parser.parse_args()
-    if args.namespace_root is None:
-        root = Path(tempfile.mkdtemp(prefix='hydra-network-', dir='/tmp'))
-        print('Evidence: ' + str(root), flush=True)
-        (root / 'parent-namespaces.json').write_text(json.dumps({name: str(Path('/proc/self/ns', name).readlink())
-                                                               for name in ('user', 'net', 'mnt')}))
-        # All bridge and /etc changes below occur after this explicit namespace
-        # boundary. The inherited host bridge configuration is never modified.
-        subprocess.run(['unshare', '-Urnm', '--', str(Path(sys.executable).resolve()),
-                        str(Path(__file__).resolve()), '--namespace-root', str(root)], check=True)
-        return
-    root = args.namespace_root
-    parent_namespaces = json.loads((root / 'parent-namespaces.json').read_text())
-    namespaces = {name: str(Path('/proc/self/ns', name).readlink()) for name in ('user', 'net', 'mnt')}
-    assert all(namespaces[name] != parent_namespaces[name] for name in namespaces), 'private namespaces required'
-    private_etc = root / 'etc'
-    (private_etc / 'qemu').mkdir(parents=True)
-    (private_etc / 'qemu/bridge.conf').write_text('allow hydrabr0\n')
-    subprocess.run(['mount', '--bind', str(private_etc), '/etc'], check=True)
-    subprocess.run(['ip', 'link', 'add', 'hydrabr0', 'type', 'bridge'], check=True)
-    subprocess.run(['ip', 'link', 'set', 'hydrabr0', 'up'], check=True)
-    manager = Runtime(root / 'config', root / 'state', root / 'run')
-    result = {'evidence': str(root), 'namespace': 'private user/net/mount',
-              'parent_namespaces': parent_namespaces, 'probe_namespaces': namespaces, 'names': {}}
-    try:
-        for name in ('alpha', 'beta'):
-            manager.add(Profile(name, memory=128, cpus=1, disk_size='8M',
-                                graphics='software', network='bridge'))
-            manager.start(name)
-
-        def actual_mac(name):
-            with manager._connect(manager.paths(name)) as qmp:
-                network = qmp.command('human-monitor-command', {'command-line': 'info network'})
-            mac = manager.profile(name).mac
-            assert 'macaddr=' + mac in network, network
-            assert int(mac.split(':')[0], 16) & 3 == 2
-            return mac, network
-
-        for name in ('alpha', 'beta'):
-            mac, network = actual_mac(name)
-            result['names'][name] = {'mac': mac, 'network': network}
-        assert result['names']['alpha']['mac'] != result['names']['beta']['mac']
-        links = json.loads(subprocess.check_output(['ip', '-j', 'link', 'show', 'master', 'hydrabr0'], text=True))
-        assert len(links) == 2, links
-        result['bridge_ports'] = [link['ifname'] for link in links]
-        manager.stop('alpha', force=True)
-        manager.start('alpha')
-        assert actual_mac('alpha')[0] == result['names']['alpha']['mac']
-        result['restart_preserved_mac'] = True
-        result['dhcp_and_guest_network_drivers_pending'] = True
-        result['passed'] = True
-        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
-        print(json.dumps(result, indent=2))
-    finally:
-        for profile in manager.profiles():
-            if manager.status(profile.name)['status'] == 'running':
-                manager.stop(profile.name, force=True)
-
-
-if __name__ == '__main__':
-    main()

dev/guest-probe.S

@@ -1,25 +0,0 @@
-.code16
-.global _start
-_start:
- cli
- xor %ax,%ax
- mov %ax,%ds
- mov %ax,%ss
- mov $0x7c00,%sp
- movl $0x48594452,0x508
- movl $0,0x500
- movw $0,0x504
-loop:
- incl 0x500
- in $0x64,%al
- test $1,%al
- jz loop
- in $0x60,%al
- mov 0x504,%bx
- mov %al,0x600(%bx)
- inc %bx
- and $255,%bx
- mov %bx,0x504
- jmp loop
-.org 510
-.word 0xaa55

dev/input-probe.c

@@ -1,63 +0,0 @@
-/* Private-display integration helper; stdin uses Linux evdev key codes.
- * Compile via dev/build-input-probe.sh inside the temporary tools shell. */
-#define _GNU_SOURCE
-#include <assert.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <sys/mman.h>
-#include <unistd.h>
-#include <wayland-client.h>
-#include <xkbcommon/xkbcommon.h>
-#include "virtual-keyboard-client-protocol.h"
-static struct wl_seat *seat;
-static struct zwp_virtual_keyboard_manager_v1 *manager;
-static void global(void *data, struct wl_registry *registry, uint32_t id,
-                   const char *interface, uint32_t version) {
-    (void)data; (void)version;
-    if (!strcmp(interface,"wl_seat"))
-        seat = wl_registry_bind(registry,id,&wl_seat_interface,1);
-    else if (!strcmp(interface,"zwp_virtual_keyboard_manager_v1"))
-        manager = wl_registry_bind(registry,id,&zwp_virtual_keyboard_manager_v1_interface,1);
-}
-static void removed(void *data,struct wl_registry *registry,uint32_t id) {
-    (void)data; (void)registry; (void)id;
-}
-int main(void) {
-    /* The caller supplies the separately created scratch Wayland runtime. */
-    assert(getenv("WAYLAND_DISPLAY") && getenv("XDG_RUNTIME_DIR"));
-    struct wl_display *display = wl_display_connect(NULL); assert(display);
-    struct wl_registry *registry = wl_display_get_registry(display);
-    struct wl_registry_listener listener={global,removed};
-    wl_registry_add_listener(registry,&listener,NULL); wl_display_roundtrip(display);
-    assert(seat && manager);
-    struct zwp_virtual_keyboard_v1 *keyboard = zwp_virtual_keyboard_manager_v1_create_virtual_keyboard(manager,seat);
-    struct xkb_context *context=xkb_context_new(XKB_CONTEXT_NO_FLAGS);
-    struct xkb_rule_names names={.rules="evdev",.model="pc105",.layout="us"};
-    struct xkb_keymap *map=xkb_keymap_new_from_names(context,&names,XKB_KEYMAP_COMPILE_NO_FLAGS); assert(map);
-    struct xkb_state *state=xkb_state_new(map); assert(state);
-    char *text=xkb_keymap_get_as_string(map,XKB_KEYMAP_FORMAT_TEXT_V1);
-    size_t size=strlen(text)+1;
-    int fd=memfd_create("hydra-private-keymap",0); assert(fd>=0);
-    assert(write(fd,text,size)==(ssize_t)size);
-    zwp_virtual_keyboard_v1_keymap(keyboard,WL_KEYBOARD_KEYMAP_FORMAT_XKB_V1,fd,size);
-    wl_display_roundtrip(display); close(fd); free(text);
-    char command[16]; unsigned key; unsigned tick=0;
-    while (scanf("%15s %u",command,&key)==2) {
-        if (!strcmp(command,"wait")) { usleep(key*1000); continue; }
-        unsigned down=!strcmp(command,"down");
-        assert(down || !strcmp(command,"up"));
-        xkb_state_update_key(state,key+8,down?XKB_KEY_DOWN:XKB_KEY_UP);
-        zwp_virtual_keyboard_v1_key(keyboard,++tick,key,down);
-        zwp_virtual_keyboard_v1_modifiers(keyboard,
-            xkb_state_serialize_mods(state,XKB_STATE_MODS_DEPRESSED),
-            xkb_state_serialize_mods(state,XKB_STATE_MODS_LATCHED),
-            xkb_state_serialize_mods(state,XKB_STATE_MODS_LOCKED),
-            xkb_state_serialize_layout(state,XKB_STATE_LAYOUT_EFFECTIVE));
-        assert(wl_display_roundtrip(display)>=0);
-    }
-    zwp_virtual_keyboard_v1_destroy(keyboard); wl_display_roundtrip(display);
-    xkb_state_unref(state); xkb_keymap_unref(map); xkb_context_unref(context);
-    wl_display_disconnect(display);
-    return 0;
-}

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/session-refresh-smoke.py

@@ -1,196 +0,0 @@
-#!/usr/bin/env python3
-"""Exercise the product's empty-session/recovery/refresh loop privately."""
-import argparse
-import json
-import os
-from pathlib import Path
-import shutil
-import signal
-import subprocess
-import sys
-import tempfile
-import time
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from hydra.runtime import Runtime
-from smoke import (wait_for, counter, focus_events, fresh_focus_transition,
-                   measure_hidden_interval, classify_compositor_errors)
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
-    parser.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
-    parser.add_argument('--keyboard', required=True, help='compiled dev/input-probe.c')
-    args = parser.parse_args()
-    for name in ('labwc', 'viewer', 'keyboard'):
-        executable = shutil.which(getattr(args, name))
-        if not executable:
-            parser.error('missing executable: ' + getattr(args, name))
-        setattr(args, name, executable)
-    if not os.access('/dev/kvm', os.R_OK | os.W_OK):
-        parser.error('KVM access required')
-    source = Path(__file__).resolve().parents[1]
-    root = Path(tempfile.mkdtemp(prefix='hydra-session-refresh-', dir='/tmp'))
-    print('Evidence: ' + str(root), flush=True)
-    xdg = root / 'xdg'; xdg.mkdir(mode=0o700)
-    manager = Runtime(root / 'config', root / 'state', root / 'run')
-    cli = [sys.executable, str(source / 'bin/hydra'), '--config', str(manager.config),
-           '--state', str(manager.state), '--runtime', str(manager.runtime)]
-    subprocess.run(['as', '--32', str(source / 'dev/guest-probe.S'), '-o', str(root / 'boot.o')], check=True)
-    subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(root / 'boot.o'), str(root / 'boot.bin')], check=True)
-    (root / 'boot.raw').write_bytes((root / 'boot.bin').read_bytes() + bytes(8 * 1024 * 1024 - 512))
-
-    def wrapper(name, code):
-        path = root / name
-        path.write_text('#!' + sys.executable + '\n' + code)
-        path.chmod(0o700)
-        return str(path)
-
-    common = 'import json, os, pathlib, subprocess, sys, time\nroot = pathlib.Path(' + repr(str(root)) + ')\n'
-    labwc = wrapper('labwc-private', common + '''
-if '--version' in sys.argv:
-    os.execv(@LABWC@, [@LABWC@] + sys.argv[1:])
-config = pathlib.Path(sys.argv[sys.argv.index('-C') + 1])
-number = len(list(root.glob('snapshot-*.json')))
-data = json.loads((config.parent / 'session.json').read_text())
-data.update(pid=os.getpid(), config_directory=str(config))
-(root / ('snapshot-%d.xml' % number)).write_text((config / 'rc.xml').read_text())
-(root / ('snapshot-%d.json' % number)).write_text(json.dumps(data))
-# Test-only wrapper replaces physical backends before the real compositor runs.
-os.environ.update(WLR_BACKENDS='headless', WLR_RENDERER='pixman',
-                  WLR_XWAYLAND='/bin/false', LABWC_UPDATE_ACTIVATION_ENV='0',
-                  XDG_DATA_DIRS='/nonexistent')
-os.execv(@LABWC@, [@LABWC@, '-d'] + sys.argv[1:])
-'''.replace('@LABWC@', repr(args.labwc)))
-    viewer = wrapper('viewer-private', common + '''
-if '--help' in sys.argv:
-    os.execv(VIEWER, [VIEWER] + sys.argv[1:])
-name = sys.argv[sys.argv.index('--title') + 1].split(': ', 1)[1]
-(root / (name + '-viewer.json')).write_text(json.dumps({
-    'argv': sys.argv[1:], 'config': os.environ['XDG_CONFIG_HOME'],
-    'gdk_backend': os.environ.get('GDK_BACKEND'), 'wayland_display': os.environ['WAYLAND_DISPLAY']}))
-log = os.open(root / (name + '.log'), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
-os.dup2(log, 1); os.dup2(log, 2)
-os.environ['WAYLAND_DEBUG'] = '1'
-os.execv(VIEWER, [VIEWER] + sys.argv[1:])
-'''.replace('VIEWER', repr(args.viewer)))
-    terminal = wrapper('terminal-private', common + '''
-manifest = pathlib.Path(sys.argv[-1])
-socket = pathlib.Path(os.environ['XDG_RUNTIME_DIR']) / os.environ['WAYLAND_DISPLAY']
-(root / 'recovery-ready').write_text(str(manifest))
-while not (root / 'recovery-input').exists():
-    if not socket.exists():
-        sys.exit(0)
-    time.sleep(.05)
-if not socket.exists():
-    sys.exit(0)
-commands = (root / 'recovery-input').read_text()
-# The actual product recovery helper executes :shell and :refresh.
-with (root / 'recovery.log').open('a') as log:
-    sys.exit(subprocess.run(sys.argv[2:], input=commands, text=True, stdout=log, stderr=log).returncode)
-''')
-    shell = wrapper('scratch-shell', common + 'sys.path.insert(0, ' + repr(str(source)) + ')\n' + '''
-from hydra.runtime import Runtime
-from smoke import counter, wait_for
-cli = CLI
-manager = Runtime(root / 'config', root / 'state', root / 'run')
-for name in ('alpha', 'beta'):
-    subprocess.run(cli + ['add', name, '--memory', '128', '--cpus', '1', '--disk-size', '8M',
-                         '--graphics', 'software', '--network', 'user'], check=True)
-    paths = manager.paths(name); manager.prepare(paths)
-    subprocess.run([os.environ['HYDRA_QEMU_IMG'], 'convert', '-f', 'raw', '-O', 'qcow2',
-                    str(root / 'boot.raw'), str(paths.disk)], check=True)
-    subprocess.run(cli + ['start', name], check=True)
-    time.sleep(.3)
-identities = {name: {'pid': manager.paths(name).pid.read_text().strip(),
-                    'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
-              for name in ('alpha', 'beta')}
-(root / 'before.json').write_text(json.dumps(identities))
-'''.replace('CLI', repr(cli)).replace('from smoke import', 'from dev.smoke import'))
-    env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), HYDRA_LABWC=labwc,
-               HYDRA_VIEWER=viewer, HYDRA_TERMINAL=terminal, HYDRA_BRIGHTNESS=shutil.which('true'),
-               SHELL=shell, GSETTINGS_BACKEND='memory')
-    for key in ('WAYLAND_DISPLAY', 'DISPLAY', 'LABWC_PID'):
-        env.pop(key, None)
-    log = (root / 'session.log').open('w')
-    launcher, injector = None, None
-    result = {'evidence': str(root), 'physical_input_pending': True,
-              'terminal': 'test-only command feeder; real recovery helper and scratch :shell',
-              'backend': 'test-only wrapper around actual product launch_session and actual labwc'}
-    try:
-        launcher = subprocess.Popen(cli + ['session'], env=env, stdout=log, stderr=log)
-        wait_for(lambda: (root / 'recovery-ready').exists(), 'empty session did not open recovery')
-        first = json.loads((root / 'snapshot-0.json').read_text())
-        assert first['names'] == [], 'session was not initially empty'
-        (root / 'recovery-input').write_text(':shell\n:refresh\n')
-        wait_for(lambda: (root / 'snapshot-1.json').exists(), 'product loop did not restart after :refresh', timeout=15)
-        second = json.loads((root / 'snapshot-1.json').read_text())
-        assert second['names'] == ['alpha', 'beta'], second
-        assert first['pid'] != second['pid'], 'compositor was not refreshed'
-        logs = {name: root / (name + '.log') for name in ('alpha', 'beta')}
-        for name, path in logs.items():
-            wait_for(lambda: path.exists() and 'set_title("HydraOS: ' + name + '")' in path.read_text(),
-                     name + ' product viewer did not connect')
-        viewers = {name: json.loads((root / (name + '-viewer.json')).read_text()) for name in logs}
-        assert len({data['config'] for data in viewers.values()}) == 2, 'viewer settings were shared'
-        for data in viewers.values():
-            assert data['gdk_backend'] == 'wayland'
-            assert data['config'].startswith(str(manager.runtime) + '/viewer-')
-            assert '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0' in data['argv']
-            assert '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0' in data['argv']
-        alog, blog = logs['alpha'], logs['beta']
-        display = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
-        keyboard_env = dict(env, WAYLAND_DISPLAY=display)
-        injector = subprocess.Popen([args.keyboard], env=keyboard_env, stdin=subprocess.PIPE, text=True)
-        time.sleep(.2)
-        wait_for(lambda: focus_events(alog) and focus_events(alog)[-1] == 'enter', 'alpha did not focus')
-
-        def switch(code, source_log, target_log):
-            before_source, before_target = focus_events(source_log), focus_events(target_log)
-            injector.stdin.write(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29\nwait 100\n')
-            injector.stdin.flush()
-            wait_for(lambda: fresh_focus_transition(before_source, before_target,
-                     focus_events(source_log), focus_events(target_log)), 'fresh reserved focus transition failed')
-
-        switch(106, alog, blog)
-        intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
-        switch(105, blog, alog)
-        intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
-        before = json.loads((root / 'before.json').read_text())
-        after = {name: {'pid': manager.paths(name).pid.read_text().strip(),
-                       'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
-                 for name in ('alpha', 'beta')}
-        for name in before:
-            assert after[name]['pid'] == before[name]['pid'] and after[name]['uuid'] == before[name]['uuid']
-            assert after[name]['counter'] != before[name]['counter']
-        # Only the PID recorded by our own wrapper, verified against its private
-        # config path, may be signalled. Never inherit a host compositor PID.
-        command = Path('/proc') / str(second['pid']) / 'cmdline'
-        assert second['config_directory'].encode() in command.read_bytes()
-        os.kill(second['pid'], signal.SIGTERM)
-        assert launcher.wait(timeout=10) == 0, 'product session exit failed'
-        for name in before:
-            assert manager.status(name)['status'] == 'running'
-            assert manager.paths(name).pid.read_text().strip() == before[name]['pid']
-        result.update(initial_names=first['names'], refreshed_names=second['names'],
-                      compositor_pids=[first['pid'], second['pid']], before=before, after=after,
-                      viewers=viewers,
-                      hidden_intervals=intervals, survived_session_exit=True,
-                      expected_xwayland_diagnostics=classify_compositor_errors((root / 'session.log').read_text()),
-                      passed=True)
-        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
-        print(json.dumps(result, indent=2))
-    finally:
-        if injector and injector.poll() is None:
-            injector.terminate(); injector.wait(timeout=5)
-        if launcher and launcher.poll() is None:
-            launcher.send_signal(signal.SIGINT); launcher.wait(timeout=12)
-        for profile in manager.profiles():
-            if manager.status(profile.name)['status'] == 'running':
-                manager.stop(profile.name, force=True)
-        log.close()
-
-
-if __name__ == '__main__':
-    main()

dev/smoke.py

@@ -1,273 +0,0 @@
-#!/usr/bin/env python3
-"""Two disposable KVM guests on a private headless display; never host input."""
-import argparse
-import json
-import os
-from pathlib import Path
-import re
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-from xml.etree import ElementTree as ET
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from hydra.profiles import Profile
-from hydra.runtime import Runtime
-from hydra.session import build_labwc_config, build_viewer_argv, _viewer_environment
-
-
-def wait_for(predicate, description, timeout=8):
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        if predicate():
-            return
-        time.sleep(.05)
-    raise AssertionError(description)
-
-
-def counter(manager, name):
-    with manager._connect(manager.paths(name)) as qmp:
-        reply = qmp.command('human-monitor-command', {'command-line': 'xp /3wx 0x500'})
-    values = [int(word, 16) for word in re.findall(r'0x([0-9a-fA-F]+)', reply)]
-    assert len(values) == 3 and values[2] == 0x48594452, 'scratch guest did not boot'
-    return values[0]
-
-
-def focus_events(log):
-    """Only keyboard focus, excluding pointer-enter and old window titles."""
-    return re.findall(r'wl_keyboard#\d+\.(enter|leave)\(', log.read_text())
-
-
-def fresh_focus_transition(source_before, target_before, source_after, target_after):
-    return (len(source_after) > len(source_before) and source_after[-1] == 'leave'
-            and len(target_after) > len(target_before) and target_after[-1] == 'enter')
-
-
-def measure_hidden_interval(read_counter, hidden_log, visible_log, duration=.5):
-    hidden_focus, visible_focus = focus_events(hidden_log), focus_events(visible_log)
-    assert hidden_focus and hidden_focus[-1] == 'leave', 'sampled guest is not hidden'
-    assert visible_focus and visible_focus[-1] == 'enter', 'other guest is not focused'
-
-    def unchanged():
-        assert (focus_events(hidden_log) == hidden_focus and
-                focus_events(visible_log) == visible_focus), 'focus changed during hidden interval'
-
-    start = time.monotonic()
-    before = read_counter()
-    unchanged()
-    while time.monotonic() - start < duration:
-        time.sleep(min(.05, duration))
-        unchanged()
-    after = read_counter()
-    unchanged()
-    assert after != before, 'hidden guest stopped working during continuously hidden interval'
-    return {'before': before, 'after': after,
-            'elapsed_seconds': time.monotonic() - start, 'focus_unchanged': True}
-
-
-def classify_compositor_errors(log):
-    expected = []
-    for line in log.splitlines():
-        if '[ERROR]' not in line:
-            continue
-        if re.search(r'\[xwayland/sockets\.c:\d+\] Failed to bind socket '
-                     r'@?/tmp/\.X11-unix/X\d+: Address already in use$', line):
-            expected.append(line)
-        else:
-            raise AssertionError('Unexpected compositor error: ' + line)
-    return expected
-
-
-def main():
-    cli = argparse.ArgumentParser(description=__doc__)
-    cli.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
-    cli.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
-    cli.add_argument('--keyboard', help='compiled dev/input-probe.c; built automatically when omitted')
-    args = cli.parse_args()
-    for name in ('viewer', 'labwc'):
-        value = shutil.which(getattr(args, name))
-        if not value:
-            cli.error(f'missing {name}: {getattr(args, name)}')
-        setattr(args, name, value)
-    source = Path(__file__).resolve().parent
-    if not args.keyboard:
-        args.keyboard = subprocess.check_output([str(source / 'build-input-probe.sh')], text=True).strip()
-    if not os.access('/dev/kvm', os.R_OK | os.W_OK):
-        cli.error('KVM access required for this integration check')
-    root = Path(tempfile.mkdtemp(prefix='hydra-smoke-', dir='/tmp'))
-    print(f'Evidence: {root}', flush=True)
-    xdg = root / 'xdg'
-    xdg.mkdir(mode=0o700)
-    config = root / 'labwc'
-    config.mkdir()
-    recovery = root / 'recovery'
-    unreserved = root / 'unreserved'
-    names = ['alpha', 'beta']
-    rc = ET.fromstring(build_labwc_config(names, [sys.executable, '-c',
-                           'from pathlib import Path; Path(' + repr(str(recovery)) + ').touch()'], initial='alpha'))
-    # Test-only controls prove native labwc inhibition is active; production
-    # config has no ToggleKeybinds or ordinary host shortcut.
-    keyboard = rc.find('keyboard')
-    binding = ET.SubElement(keyboard, 'keybind', key='C-A-S-i')
-    ET.SubElement(binding, 'action', name='ToggleKeybinds')
-    binding = ET.SubElement(keyboard, 'keybind', key='F12')
-    ET.SubElement(binding, 'action', name='Execute', command=shutil.which('touch') + ' ' + str(unreserved))
-    ET.ElementTree(rc).write(config / 'rc.xml')
-    (config / 'autostart').write_text('#!/bin/sh\n')
-    (config / 'environment').touch()
-    env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), WLR_BACKENDS='headless',
-               WLR_RENDERER='pixman', WLR_XWAYLAND='/bin/false', XDG_DATA_DIRS='/nonexistent',
-               GSETTINGS_BACKEND='memory', LABWC_UPDATE_ACTIVATION_ENV='0')
-    for key in ('DISPLAY', 'WAYLAND_DISPLAY'):
-        env.pop(key, None)
-    manager = Runtime(root / 'config', root / 'state', root / 'run')
-    boot_object, boot = root / 'boot.o', root / 'boot.bin'
-    subprocess.run(['as', '--32', str(source / 'guest-probe.S'), '-o', str(boot_object)], check=True)
-    subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(boot_object), str(boot)], check=True)
-    raw = root / 'boot.raw'
-    raw.write_bytes(boot.read_bytes() + bytes(8 * 1024 * 1024 - 512))
-    media = root / 'installation.iso'
-    media.touch()
-    for name in names:
-        manager.add(Profile(name, media if name == 'alpha' else None, memory=128, cpus=1, disk_size='8M', graphics='software',
-                            width=1920, height=1080, refresh_millihz=144000))
-        paths = manager.paths(name)
-        manager.prepare(paths)
-        subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'convert', '-f', 'raw',
-                        '-O', 'qcow2', str(raw), str(paths.disk)], check=True)
-    processes, streams = [], []
-    result = {'evidence': str(root), 'physical_checks_pending': True}
-    try:
-        cl = (root / 'labwc.log').open('w'); streams.append(cl)
-        compositor = subprocess.Popen([args.labwc, '-d', '-C', str(config)], env=env, stdout=cl, stderr=cl)
-        processes.append(compositor)
-        wait_for(lambda: any(not p.name.endswith('.lock') for p in xdg.glob('wayland-*')), 'private display did not start')
-        env['WAYLAND_DISPLAY'] = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
-        injector = subprocess.Popen([args.keyboard], env=env, stdin=subprocess.PIPE, text=True)
-        processes.append(injector)
-
-        def keys(commands):
-            injector.stdin.write(commands + '\nwait 100\n'); injector.stdin.flush()
-            time.sleep(.15)
-            assert injector.poll() is None, 'private input helper exited'
-
-        def chord(code):
-            keys(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29')
-
-        def switch(key, source_log, target_log):
-            source_before, target_before = focus_events(source_log), focus_events(target_log)
-            chord(key)
-            wait_for(lambda: fresh_focus_transition(source_before, target_before,
-                                                   focus_events(source_log), focus_events(target_log)),
-                     'reserved switch did not produce fresh keyboard leave/enter')
-            return {'from_events': len(focus_events(source_log)),
-                    'to_events': len(focus_events(target_log)), 'keycode': key}
-
-        def attach(name, number):
-            vl = (root / f'{name}-{number}.log').open('w'); streams.append(vl)
-            previous = dict(os.environ)
-            try:
-                os.environ.update(env)
-                viewer_env = _viewer_environment(root / f'{name}-{number}-settings')
-            finally:
-                os.environ.clear(); os.environ.update(previous)
-            viewer_env['WAYLAND_DEBUG'] = '1'
-            proc = subprocess.Popen(build_viewer_argv(manager.paths(name), args.viewer),
-                                    env=viewer_env, stdout=vl, stderr=vl)
-            processes.append(proc)
-            log = root / f'{name}-{number}.log'
-            wait_for(lambda: f'set_title("HydraOS: {name}")' in log.read_text(), f'{name} viewer did not connect')
-            return proc, log
-
-        for name in names:
-            manager.start(name)
-        media.unlink()  # Open QEMU media disappears; status/control/view still work.
-        time.sleep(1)
-        before = {name: counter(manager, name) for name in names}
-        with manager._connect(manager.paths('alpha')) as qmp:
-            children = qmp.command('qom-list', {'path': '/machine/peripheral-anon'})
-            video = next(child['name'] for child in children if child['type'] == 'child<virtio-vga>')
-            mode = {prop: qmp.command('qom-get', {'path': '/machine/peripheral-anon/' + video, 'property': prop})
-                    for prop in ('xres', 'yres', 'refresh_rate')}
-            assert mode == {'xres': 1920, 'yres': 1080, 'refresh_rate': 144000}, mode
-            usb_before = qmp.command('human-monitor-command', {'command-line': 'info usb'})
-        identities = {name: manager.paths(name).pid.read_text().strip() for name in names}
-        alpha, alog = attach('alpha', 1)
-        beta, blog = attach('beta', 1)
-        wait_for(lambda: '.enter(' in alog.read_text(), 'alpha did not receive focus')
-        # F12 must work before inhibition; then become suppressed in alpha.
-        keys('down 88\nup 88')
-        wait_for(unreserved.exists, 'ordinary test binding failed before inhibition')
-        unreserved.unlink()
-        chord(23)  # I: test-only ToggleKeybinds.
-        keys('down 88\nup 88')
-        assert not unreserved.exists(), 'keybinding inhibition did not suppress ordinary F12'
-        switch(106, alog, blog)  # Right: beta focused, alpha continuously hidden.
-        hidden_intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
-        switch(105, blog, alog)  # Left: alpha focused, beta continuously hidden.
-        hidden_intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
-        chord(1)  # Escape: production recovery binding still works.
-        wait_for(recovery.exists, 'reserved recovery failed with inhibition')
-        dialog = 'set_title("Select USB devices for redirection")'
-        assert dialog not in alog.read_text(), 'USB selector opened automatically'
-        # Hold modifiers through opening, then release inside the modal dialog.
-        keys('down 29\ndown 56\ndown 42\ndown 22')
-        wait_for(lambda: dialog in alog.read_text(), 'manual USB selector did not open', timeout=3)
-        keys('up 22\nup 42\nup 56\nup 29')
-        keys('down 1\nup 1')  # Close selector.
-        post_usb_switches = [switch(106, alog, blog), switch(105, blog, alog)]
-        assert alpha.poll() is None and beta.poll() is None
-        assert identities == {name: manager.paths(name).pid.read_text().strip() for name in names}, 'switch restarted a guest'
-        assert alog.read_text().count('set_title("HydraOS: alpha")') == 1, 'switch reconnected alpha'
-        assert blog.read_text().count('set_title("HydraOS: beta")') == 1, 'switch reconnected beta'
-        with manager._connect(manager.paths('alpha')) as qmp:
-            usb_after = qmp.command('human-monitor-command', {'command-line': 'info usb'})
-        assert usb_after == usb_before, 'opening selector attached an unselected USB device'
-        alpha.terminate(); alpha.wait(timeout=5)
-        assert manager.status('alpha')['status'] == 'running'
-        alpha, attached = attach('alpha', 2)
-        after = {name: counter(manager, name) for name in names}
-        assert all(after[name] != before[name] for name in names), 'background guest stopped working'
-        # Quit only beta; alpha's counter/lifecycle remains independent.
-        manager.stop('beta', force=True)
-        assert manager.status('alpha')['status'] == 'running'
-        assert counter(manager, 'alpha') != after['alpha']
-        compositor.terminate(); compositor.wait(timeout=5)
-        assert manager.status('alpha')['status'] == 'running'
-        manager.stop('alpha', force=True)
-        manager.media('alpha', None)
-        assert manager.profile('alpha').iso is None
-        manager.start('alpha')
-        time.sleep(1)
-        disk_boot_counter = counter(manager, 'alpha')
-        assert disk_boot_counter > 0
-        expected_compositor_noise = classify_compositor_errors((root / 'labwc.log').read_text())
-        result.update(hidden_guest_intervals=hidden_intervals, post_usb_switches=post_usb_switches,
-                      expected_compositor_noise=expected_compositor_noise,
-                      manual_usb_dialog=True, standard_us_keymap=True,
-                      native_labwc_inhibition=True, reserved_switch_and_recovery=True,
-                      qemu_survives_viewer_and_compositor=True, independent_lifecycle=True,
-                      guest_counters_before=before, guest_counters_after=after,
-                      guest_pids=identities, explicit_reattach=True, requested_qemu_mode=mode,
-                      usb_devices_unchanged_without_selection=True,
-                      missing_iso_does_not_block_views_or_control=True,
-                      restart_without_original_media=True, disk_boot_counter=disk_boot_counter,
-                      viewer_version=subprocess.check_output([args.viewer, '--version'], text=True).strip(),
-                      guest_input_and_physical_modifiers='not proven by this boot counter workload')
-        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
-        print(json.dumps(result, indent=2))
-    finally:
-        for proc in reversed(processes):
-            if proc.poll() is None:
-                proc.terminate(); proc.wait(timeout=5)
-        for stream in streams:
-            stream.close()
-        for name in names:
-            if manager.status(name)['status'] != 'stopped':
-                manager.stop(name, force=True)
-
-
-if __name__ == '__main__':
-    main()

dev/tools.nix

@@ -8,9 +8,6 @@ let
     url = "https://github.com/NixOS/nixpkgs/archive/7fd4004d0511991fff3a0524856a9be09e28c5f0.tar.gz";
     sha256 = "sha256-i5qk1kLiit98rRbh1AXmWBNT9QXC/y4l5Wk/u5EdpDQ=";
   }) {};
-  qemu = (pkgs.qemu.override { hostCpuOnly = true; }).overrideAttrs (old: {
-    patches = (old.patches or []) ++ [ ../patches/qemu-virtio-refresh-rate.patch ];
-  });
   viewer = pkgs.virt-viewer.overrideAttrs (old: {
     src = pkgs.fetchurl {
       url = "https://releases.pagure.org/virt-viewer/virt-viewer-11.0.tar.xz";
@@ -20,9 +17,7 @@ let
     doCheck = true;
   });
 in pkgs.mkShell {
-  nativeBuildInputs = [ pkgs.stdenv.cc pkgs.pkg-config pkgs.wayland-scanner ];
-  buildInputs = [ pkgs.wayland.dev pkgs.libxkbcommon.dev ];
-  packages = [ pkgs.python3 qemu viewer graphics.labwc pkgs.foot
+  packages = [ pkgs.python3 pkgs.qemu viewer graphics.labwc pkgs.foot
                pkgs.brightnessctl pkgs.wayland-utils pkgs.wlr-randr ];
-  passthru = { inherit viewer qemu; labwc = graphics.labwc; };
+  passthru = { inherit viewer; labwc = graphics.labwc; };
 }

dev/virtual-keyboard-unstable-v1.xml

@@ -1,113 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<protocol name="virtual_keyboard_unstable_v1">
-  <copyright>
-    Copyright © 2008-2011  Kristian Høgsberg
-    Copyright © 2010-2013  Intel Corporation
-    Copyright © 2012-2013  Collabora, Ltd.
-    Copyright © 2018       Purism SPC
-
-    Permission is hereby granted, free of charge, to any person obtaining a
-    copy of this software and associated documentation files (the "Software"),
-    to deal in the Software without restriction, including without limitation
-    the rights to use, copy, modify, merge, publish, distribute, sublicense,
-    and/or sell copies of the Software, and to permit persons to whom the
-    Software is furnished to do so, subject to the following conditions:
-
-    The above copyright notice and this permission notice (including the next
-    paragraph) shall be included in all copies or substantial portions of the
-    Software.
-
-    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
-    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
-    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-    DEALINGS IN THE SOFTWARE.
-  </copyright>
-
-  <interface name="zwp_virtual_keyboard_v1" version="1">
-    <description summary="virtual keyboard">
-      The virtual keyboard provides an application with requests which emulate
-      the behaviour of a physical keyboard.
-
-      This interface can be used by clients on its own to provide raw input
-      events, or it can accompany the input method protocol.
-    </description>
-
-    <request name="keymap">
-      <description summary="keyboard mapping">
-        Provide a file descriptor to the compositor which can be
-        memory-mapped to provide a keyboard mapping description.
-
-        Format carries a value from the keymap_format enumeration.
-      </description>
-      <arg name="format" type="uint" summary="keymap format"/>
-      <arg name="fd" type="fd" summary="keymap file descriptor"/>
-      <arg name="size" type="uint" summary="keymap size, in bytes"/>
-    </request>
-
-    <enum name="error">
-      <entry name="no_keymap" value="0" summary="No keymap was set"/>
-    </enum>
-
-    <request name="key">
-      <description summary="key event">
-        A key was pressed or released.
-        The time argument is a timestamp with millisecond granularity, with an
-        undefined base. All requests regarding a single object must share the
-        same clock.
-
-        Keymap must be set before issuing this request.
-
-        State carries a value from the key_state enumeration.
-      </description>
-      <arg name="time" type="uint" summary="timestamp with millisecond granularity"/>
-      <arg name="key" type="uint" summary="key that produced the event"/>
-      <arg name="state" type="uint" summary="physical state of the key"/>
-    </request>
-
-    <request name="modifiers">
-      <description summary="modifier and group state">
-        Notifies the compositor that the modifier and/or group state has
-        changed, and it should update state.
-
-        The client should use wl_keyboard.modifiers event to synchronize its
-        internal state with seat state.
-
-        Keymap must be set before issuing this request.
-      </description>
-      <arg name="mods_depressed" type="uint" summary="depressed modifiers"/>
-      <arg name="mods_latched" type="uint" summary="latched modifiers"/>
-      <arg name="mods_locked" type="uint" summary="locked modifiers"/>
-      <arg name="group" type="uint" summary="keyboard layout"/>
-    </request>
-
-    <request name="destroy" type="destructor" since="1">
-      <description summary="destroy the virtual keyboard keyboard object"/>
-    </request>
-  </interface>
-
-  <interface name="zwp_virtual_keyboard_manager_v1" version="1">
-    <description summary="virtual keyboard manager">
-      A virtual keyboard manager allows an application to provide keyboard
-      input events as if they came from a physical keyboard.
-    </description>
-
-    <enum name="error">
-      <entry name="unauthorized" value="0" summary="client not authorized to use the interface"/>
-    </enum>
-
-    <request name="create_virtual_keyboard">
-      <description summary="Create a new virtual keyboard">
-        Creates a new virtual keyboard associated to a seat.
-
-        If the compositor enables a keyboard to perform arbitrary actions, it
-        should present an error when an untrusted client requests a new
-        keyboard.
-      </description>
-      <arg name="seat" type="object" interface="wl_seat"/>
-      <arg name="id" type="new_id" interface="zwp_virtual_keyboard_v1"/>
-    </request>
-  </interface>
-</protocol>

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.

docs/superpowers/plans/2026-10-13-graphics-session.md

@@ -72,17 +72,13 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
 
 ### Task 3: Integration and recovery
 
-**Files:** Create `dev/smoke.py`, `docs/verification.md`; update README with verified commands. Amend runtime/profile/CLI/session tests and dependency patches for integration defects and the installed-guest workflow below.
+**Files:** Create `dev/smoke.py`, `docs/verification.md`; update README with verified commands.
 
 **Interfaces:** Consumes Tasks1/2, produces disposable-VM integration evidence.
 
 - [ ] Run two disposable KVM instances/viewers, never existing lab disks; verify private QMP and independent lifecycle.
 - [ ] Verify viewer/session exit preserves QEMU and reattach works; switching neither restarts nor reconnects guests.
 - [ ] Check input only in disposable session, without injecting into current host or user's guests; add regression tests for discovered defects.
-- [ ] Resolve Task 2's observed USB shortcut dispatch failure; demonstrate manual selector opening without automatic USB redirection on the private display. Check reserved switching/recovery actions with a standard keymap and keyboard inhibition.
-- [ ] Complete installed-guest boot workflow: optional installation media, explicit attach/detach while stopped, disk boot without the original ISO, and status/stop usable even if previously configured ISO disappears. Preserve exclusive disk creation and verified QMP control.
-- [ ] Add validated display dimensions/optional refresh and distribution defaults under `/etc/hydra/defaults.toml`. Defaults affect newly created profiles only; explicit CLI values win. Developer default remains user networking and stock-QEMU-compatible refresh; installed Hydra uses hydrabr0 and ASUS 1920x1080/144000 mHz preference.
-- [ ] Port the existing tested `qemu-virtio-refresh-rate.patch` from the old lab as a tracked dependency patch; validate the generated EDID and requested mode. Distinguish advertised refresh from achieved frame rate and physical display verification.
 - [ ] Record held-key, physical brightness and 144Hz checks needing dedicated TTY. Nested tests cannot prove them.
 - [ ] Run full suite and independent review; fix material findings with reproducing tests.
 - [ ] Commit evidence and push reviewed prototype branch to supplied repository.

docs/verification.md

@@ -1,178 +0,0 @@
-# Graphics and recovery verification
-
-Run from the checkout. Tools are temporary Nix development dependencies; no
-persistent host installation is required. The two-guest smoke test needs access
-to `/dev/kvm` and creates its own headless Wayland display, VM disks and sockets
-under a fresh `/tmp/hydra-smoke-*` directory. It never uses the inherited display,
-active TTY, existing guests or existing disks. Evidence remains in that directory;
-the script stops only its own verified QEMU instances on exit.
-
-```sh
-python3 -m unittest discover -s tests -v
-nix-build dev/tools.nix -A viewer --no-out-link
-nix-build dev/tools.nix -A qemu --no-out-link
-nix-shell dev/tools.nix --run 'python3 dev/smoke.py'
-nix-shell dev/tools.nix --run 'python3 dev/session-refresh-smoke.py --keyboard "$(dev/build-input-probe.sh)"'
-nix-shell dev/tools.nix --run 'python3 dev/check-network.py'
-```
-
-The smoke test builds a private input helper with an ordinary evdev/pc105 US
-keyboard map. It boots two tiny counter workloads from separate scratch qcow2
-disks, connects two fullscreen viewers, and checks reserved switching/recovery,
-manual USB selection, explicit reconnect and independent QMP lifecycles.
-After each confirmed workspace switch, the hidden guest is sampled across a
-half-second interval with no focus changes, switches or reconnects; the check is
-then repeated with the other guest hidden. `result.json` records both intervals.
-Guest memory counters must advance and QEMU PIDs remain unchanged through
-switching. Viewer protocol logs must show no reconnection during switching.
-A test-only `ToggleKeybinds` action and F12 marker establish native labwc
-keybinding inhibition before checking the production reserved bindings. Those
-extra controls are absent from the production session configuration. See the
-[labwc configuration manual](https://labwc.github.io/labwc-config.5.html) for
-`overrideInhibition` and the [actions manual](https://labwc.github.io/labwc-actions.5.html)
-for native `ToggleKeybinds` behavior.
-
-USB acceptance checks the actual selector window title, holds the modifiers
-through opening, releases them inside the modal dialog, then asserts fresh
-keyboard leave/enter transitions for both Right and Left switches.
-QEMU's USB device list must stay unchanged without a selection. Both viewer
-USB filters deny automatic hotplug and connection redirection. This check does
-not attach a physical USB device or prove guest driver behavior.
-
-The viewer adaptation handles only the configured manual USB accelerator before
-forwarding an event to the guest. Upstream keyboard grabs remove the GTK
-application accelerator group; without this adaptation, the configured USB
-shortcut reaches the viewer but does not open its selector.
-
-The pinned labwc 0.20.2 runtime has no documented runtime XWayland-disable
-option. This native-Wayland probe sets `WLR_XWAYLAND=/bin/false` so no X server
-is launched, and explicitly disables activation-environment updates. It can
-still log an occupied X socket while wlroots reserves a display number. Only
-`[xwayland/sockets.c] Failed to bind socket ... Address already in use` is
-classified as expected noise in `result.json`; any other compositor `[ERROR]`
-fails the probe. Full logs remain available. This classification does not
-suppress viewer errors or compiler/test diagnostics. The viewer build's upstream
-Meson boolean-check and compiler-warning-option guidance are build noise; the four-test
-result, compiler errors and process exit status remain the build acceptance gate.
-
-## Fresh-session recovery and profile changes
-
-A session with no profiles opens recovery with first-use guidance. Enter `:shell`
-to run `hydra add NAME --iso PATH` and `hydra start NAME`; leave the shell and enter
-`:refresh`. After adding or removing profiles, refresh reloads the profile list,
-rebuilds workspace rules, restarts only the session's own compositor, and attaches
-new isolated viewers to running guests. QEMU processes and workloads survive.
-The recovery prompt can then reconnect the new names; reserved Left/Right chords
-switch between their workspaces. `:quit` closes the recovery prompt.
-
-`dev/session-refresh-smoke.py` exercises the actual product launch loop, recovery
-helper, `:shell` and `:refresh`, starting with an empty profile directory. A
-test-only labwc wrapper selects private headless backends; a terminal command
-feeder drives recovery and a scratch shell adds and starts two counter guests.
-Actual labwc and native Wayland viewers then demonstrate fresh keyboard focus
-transitions in both directions, hidden workload continuity, unchanged QEMU
-PIDs/UUIDs through refresh, and guest survival after normal compositor exit.
-This checks the recovery command flow, not physical terminal rendering/usability.
-It leaves snapshots, recovery output, protocol logs and `result.json` in
-`/tmp/hydra-session-refresh-*` and stops only its verified scratch QEMU processes.
-
-## Bridge identities
-
-New profiles persist a locally administered unicast `mac` in their TOML file.
-It remains stable across stop/start, media changes and runtime socket-directory
-changes. The generated address incorporates the VM disk path and user identity,
-so separate profiles/users avoid QEMU's identical default MAC on a shared bridge.
-Existing profiles without `mac` use the same deterministic calculation without
-rewriting their file. Relocating a legacy disk or changing its owner changes
-that fallback; explicitly persisting its current address retains the identity.
-Explicit TOML addresses must also be locally administered unicast, and users
-copying profiles must assign distinct addresses to concurrently used copies.
-
-`dev/check-network.py` requires unprivileged user/network/mount namespaces. It
-asserts all three namespace identities changed before creating a private
-`hydrabr0` and mounting a private QEMU bridge-helper ACL over `/etc` within that
-namespace. Two actual scratch QEMUs must report different configured NIC MACs,
-both bridge TAP ports must exist, and restart must preserve the first guest's
-address. Evidence remains under `/tmp/hydra-network-*`; host bridges, host `/etc`
-and existing guests are unchanged. This does not prove guest network drivers,
-DHCP leases or external connectivity; those need the installed-guest harness.
-
-## Installed guest workflow
-
-`--iso` is optional. Attach or detach media only while the VM is stopped:
-
-```sh
-bin/hydra add guest --iso /absolute/path/installer.iso
-bin/hydra start guest
-bin/hydra view guest
-# Shut down from inside the guest, then check it is stopped.
-bin/hydra status guest
-bin/hydra media guest --detach
-bin/hydra start guest
-```
-
-For an existing stopped guest, `bin/hydra media guest --iso PATH` attaches
-installation/rescue media. A missing configured ISO prevents a new start with
-an actionable error; it does not prevent listing profiles, status, stopping a
-running guest, reconnecting a viewer or detaching the missing media. Media
-updates share the VM lifecycle lock and preserve its qcow2 disk. Abrupt stopping
-still requires `bin/hydra stop guest --force` and verified QMP identity.
-
-## Display preferences and EDID
-
-New developer profiles use user networking, 1280×800 and `refresh_millihz=0`.
-Zero omits the patched QEMU refresh property and remains compatible with stock
-QEMU. Hydra's installed defaults can use this flat `/etc/hydra/defaults.toml`:
-
-```toml
-network = "bridge"
-bridge = "hydrabr0"
-width = 1920
-height = 1080
-refresh_millihz = 144000
-```
-
-Only `add` reads distribution defaults. Explicit `--network`, `--bridge`,
-`--width`, `--height` and `--refresh-millihz` options win; existing profiles do
-not change. `--defaults PATH` selects another preferences file for development.
-Missing defaults use developer preferences; unknown or invalid fields fail
-before creating a profile. Dimensions are integers from 1 to 16384, refresh is
-zero for automatic refresh or an integer from 50000 to 255000 mHz, matching
-the vertical range descriptor limits. This bound does not establish every
-resolution/refresh combination; pixel-clock and horizontal timing limits are
-mode-dependent and the verified target remains 1920×1080/144000 mHz. Nonzero
-refresh requires the tracked patched QEMU provided by `dev/tools.nix` and distribution packaging.
-
-The patch keeps the requested refresh through frontend resizing and extends
-EDID monitor ranges to include the generated preferred mode. To verify the
-actual patched QEMU generator with the pinned source archive:
-
-```sh
-qemu_source=$(nix-build dev/tools.nix -A qemu.src --no-out-link)
-nix-shell dev/tools.nix --run "python3 dev/check-edid.py '$qemu_source'"
-```
-
-The probe applies the tracked patch to four relevant source files in a temporary
-directory, compiles the actual EDID and virtio GPU generator functions with a
-small dependency harness, and decodes the result. It checks 60/75 Hz fallback,
-144 Hz configuration, a conflicting 60 Hz frontend request, resized dimensions,
-checksums and monitor ranges. It also checks the supported vertical-refresh
-boundaries: 1920×1080/50000 mHz and 1280×800/255000 mHz. Range endpoints
-use floor/ceiling of the encoded timing after pixel-clock rounding, so the
-49.998895 Hz lower-boundary timing is included by a 49 Hz minimum. Normal
-automatic 60/75 Hz cases retain the legacy 50–125 Hz range. At 1920×1080/144000 mHz the preferred timing is
-143.997231 Hz after EDID clock rounding. This is an advertised guest mode,
-not measured frame delivery or proof of a 144 Hz physical output.
-
-## Still requiring a dedicated physical session
-
-- Physical reserved chords, brightness/backlight permissions and touchpad input.
-- Physical held keys/modifiers across guest switching, grabs and modal dialogs;
-  guest key-up delivery, pointer grab release and absence of stuck modifiers.
-- Physical USB selection/redirection and guest drivers; no automatic redirection.
-- Toolbar behavior at physical screen edges and actual recovery shell usability.
-- ASUS output selection at 1920×1080/144 Hz, guest-reported display mode, delivered
-  frame rate, hidden guest graphics continuity, tearing and switching latency.
-
-Headless checks establish CPU workload continuity, protocol focus transitions
-and the exercised compositor actions. They cannot prove physical input or scanout.

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

@@ -7,7 +7,7 @@ import subprocess
 import sys
 from pathlib import Path
 
-from .profiles import Profile, load_defaults
+from .profiles import Profile
 from .runtime import Runtime, RuntimeError
 from .session import launch_session, session_helper, view
 
@@ -17,26 +17,16 @@ def parser() -> argparse.ArgumentParser:
     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')
-    cli.add_argument('--defaults', type=Path, default=Path('/etc/hydra/defaults.toml'),
-                     help='Distribution preferences for newly created profiles')
     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)
+    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'))
-    add.add_argument('--bridge')
-    add.add_argument('--width', type=int)
-    add.add_argument('--height', type=int)
-    add.add_argument('--refresh-millihz', type=int)
-    media = commands.add_parser('media', help='Attach or detach installation media while stopped')
-    media.add_argument('name')
-    selection = media.add_mutually_exclusive_group(required=True)
-    selection.add_argument('--iso', type=Path)
-    selection.add_argument('--detach', action='store_true')
+    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='?')
@@ -67,18 +57,12 @@ def main(argv: list[str] | None = None) -> int:
         elif args.command == '_session-helper':
             return session_helper(args.operation, args.manifest)
         elif args.command == 'add':
-            preferences = load_defaults(args.defaults)
-            preferences.update({key: getattr(args, key) for key in
-                                ('network', 'bridge', 'width', 'height', 'refresh_millihz')
-                                if getattr(args, key) is not None})
             profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size,
-                              args.graphics, **preferences)
+                              args.graphics, args.network, args.bridge)
             runtime.add(profile)
             result = {'name': profile.name, 'status': 'added'}
-        elif args.command == 'media':
-            result = runtime.media(args.name, None if args.detach else args.iso)
         elif args.command == 'list':
-            result = [{'name': profile.name, 'iso': str(profile.iso) if profile.iso else None} for profile in runtime.profiles()]
+            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':

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

@@ -16,40 +16,24 @@ def validate_name(name: str) -> str:
 @dataclass(frozen=True)
 class Profile:
     name: str
-    iso: Path | None = None
+    iso: Path
     memory: int = 4096
     cpus: int = 2
     disk_size: str = '40G'
     graphics: str = 'auto'
     network: str = 'user'
     bridge: str = 'hydrabr0'
-    width: int = 1280
-    height: int = 800
-    refresh_millihz: int = 0
-    mac: str | None = None
 
     def __post_init__(self):
         validate_name(self.name)
-        if self.iso is not None:
-            if not isinstance(self.iso, (str, Path)) or not str(self.iso):
-                raise ValueError('ISO must be a nonempty path or omitted')
-            object.__setattr__(self, 'iso', Path(self.iso).expanduser().resolve())
-        if self.mac is not None:
-            if (not isinstance(self.mac, str) or
-                not re.fullmatch(r'[0-9a-fA-F]{2}(:[0-9a-fA-F]{2}){5}', self.mac) or
-                int(self.mac[:2], 16) & 3 != 2):
-                raise ValueError('mac must be a locally administered unicast Ethernet address')
-            object.__setattr__(self, 'mac', self.mac.lower())
+        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}')
-        for key in ('width', 'height'):
-            if type(getattr(self, key)) is not int or not 1 <= getattr(self, key) <= 16384:
-                raise ValueError(f'{key} must be an integer between 1 and 16384')
-        if type(self.refresh_millihz) is not int or not (self.refresh_millihz == 0 or
-                                                       50000 <= self.refresh_millihz <= 255000):
-            raise ValueError('refresh_millihz must be 0 (automatic) or an integer between 50000 and 255000')
         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'):
@@ -83,22 +67,6 @@ def save_profile(path: Path, profile: Profile) -> None:
     lines = []
     for field in Profile.__dataclass_fields__:
         value = getattr(profile, field)
-        if value is None:
-            continue
         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))
-
-
-def load_defaults(path: Path) -> dict:
-    """Distribution preferences apply only when creating a new profile."""
-    try:
-        with Path(path).open('rb') as stream:
-            data = tomllib.load(stream)
-    except FileNotFoundError:
-        return {}
-    unknown = data.keys() - {'network', 'bridge', 'width', 'height', 'refresh_millihz'}
-    if unknown:
-        raise ValueError(f'Unknown defaults fields: {", ".join(sorted(unknown))}')
-    Profile('defaults-check', **data)
-    return data

hydra/runtime.py

@@ -3,7 +3,6 @@
 import contextlib
 import errno
 import fcntl
-import hashlib
 import json
 import os
 import socket
@@ -12,7 +11,7 @@ import subprocess
 import tempfile
 import time
 import uuid
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from pathlib import Path
 
 from .profiles import Profile, load_profile, save_profile, validate_name
@@ -55,14 +54,6 @@ def _intel_render_node() -> Path | None:
     return None
 
 
-def _stable_mac(paths: VMPaths) -> str:
-    # Disk/state identity survives restarts and runtime socket-directory changes.
-    # UID distinguishes owners even when a shared bridge sees the same names.
-    digest = bytearray(hashlib.sha256(f'{os.getuid()}:{paths.disk}'.encode()).digest()[:6])
-    digest[0] = (digest[0] & 0xfc) | 0x02
-    return ':'.join(f'{octet:02x}' for octet in digest)
-
-
 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,
@@ -73,21 +64,15 @@ def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
             '-mon', 'chardev=qmp,mode=control']
     if accel == 'kvm':
         argv += ['-cpu', 'host']
-    blockdevs = [
+    for options in (
         {'driver': 'file', 'filename': str(paths.disk), 'node-name': 'disk-file'},
         {'driver': 'qcow2', 'file': 'disk-file', 'node-name': 'disk'},
-    ]
-    if profile.iso is not None:
-        blockdevs += [
         {'driver': 'file', 'filename': str(profile.iso), 'node-name': 'iso-file', 'read-only': True},
         {'driver': 'raw', 'file': 'iso-file', 'node-name': 'iso', 'read-only': True},
-        ]
-    for options in blockdevs:
+    ):
         argv += ['-blockdev', json.dumps(options)]
-    argv += ['-device', 'virtio-blk-pci,drive=disk']
-    if profile.iso is not None:
-        argv += ['-device', 'ide-cd,drive=iso']
-    argv += ['-boot', 'order=dc' if profile.iso is not None else 'order=c', '-device', f'virtio-net-pci,netdev=net,mac={profile.mac or _stable_mac(paths)}',
+    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']
@@ -98,15 +83,12 @@ def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
                  '-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')
-    mode = f',xres={profile.width},yres={profile.height}'
-    if profile.refresh_millihz:
-        mode += f',refresh_rate={profile.refresh_millihz}'
     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' + mode]
+        argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl']
     else:
-        argv += ['-display', 'none', '-device', 'virtio-vga' + mode]
+        argv += ['-display', 'none', '-device', 'virtio-vga']
     return argv + ['-spice', spice]
 
 
@@ -143,35 +125,9 @@ class Runtime:
             _private_dir(directory)
 
     def add(self, profile: Profile) -> None:
-        self._validate_media(profile.iso)
         _private_dir(self.config)
-        if profile.mac is None:
-            profile = replace(profile, mac=_stable_mac(self.paths(profile.name)))
         save_profile(self.config / f'{profile.name}.toml', profile)
 
-    @staticmethod
-    def _validate_media(iso: Path | None) -> None:
-        if iso is not None and not iso.is_file():
-            raise RuntimeError(f'ISO does not exist or is not a file: {iso}; attach valid media or detach it while stopped')
-
-    def media(self, name: str, iso: Path | None) -> dict:
-        paths = self.paths(name)
-        with self._locked(paths):
-            current = self.status(name)
-            if current['status'] != 'stopped':
-                raise RuntimeError(f'Cannot change media for {name}: {current["status"]}')
-            profile = replace(self.profile(name), iso=iso)
-            self._validate_media(profile.iso)
-            destination = self.config / f'{name}.toml'
-            if destination.is_symlink():
-                raise RuntimeError('Refusing to replace a symlink profile')
-            with tempfile.TemporaryDirectory(prefix=f'.{name}-', dir=self.config) as tmp:
-                temp = Path(tmp) / f'{name}.toml'
-                save_profile(temp, profile)
-                temp.chmod(0o600)
-                os.replace(temp, destination)
-        return {'name': name, 'iso': str(profile.iso) if profile.iso else None, 'status': 'media-updated'}
-
     def profiles(self) -> list[Profile]:
         if not self.config.exists():
             return []
@@ -278,13 +234,12 @@ class Runtime:
             temp.unlink(missing_ok=True)
 
     def start(self, name: str) -> dict:
+        profile = self.profile(name)
         paths = self.paths(name)
         with self._locked(paths):
-            profile = self.profile(name)
             current = self.status(name)
             if current['status'] != 'stopped':
                 raise RuntimeError(f'Cannot start {name}: {current["status"]}')
-            self._validate_media(profile.iso)
             for path in (paths.qmp, paths.spice):
                 if self._listening(path):
                     raise RuntimeError(f'Runtime socket is already in use: {path}')

hydra/session.py

@@ -49,10 +49,9 @@ def build_labwc_config(names: list[str], recovery_argv: list[str],
     core = ET.SubElement(root, 'core')
     ET.SubElement(core, 'primarySelection').text = 'no'
     desktops = ET.SubElement(root, 'desktops', number=str(max(1, len(names))))
-    if names:
-        labels = ET.SubElement(desktops, 'names')
-        for name in names:
-            ET.SubElement(labels, 'name').text = name
+    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'
@@ -174,45 +173,25 @@ def launch_session(runtime: Runtime, *, nested=False) -> int:
     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)
-    while True:
-        profiles = runtime.profiles()
-        initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None)
-        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')
-            (config / 'environment').touch()
-            env['PYTHONUNBUFFERED'] = '1'
-            compositor = subprocess.Popen([labwc, '-C', str(config)], env=env)
-            refresh = False
-            try:
-                while True:
-                    if (directory / 'refresh.request').exists():
-                        refresh = True
-                        break
-                    if compositor.poll() is not None:
-                        # A requesting recovery terminal can exit before polling.
-                        refresh = (directory / 'refresh.request').exists()
-                        break
-                    time.sleep(.1)
-            finally:
-                if compositor.poll() is None:
-                    compositor.terminate()
-                compositor.wait(timeout=10)
-            if not refresh:
-                return compositor.returncode
-            # Only our compositor was stopped. Its viewers disconnect; detached
-            # QEMU instances retain their identity and running workloads.
-
+    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:
@@ -224,22 +203,14 @@ def session_helper(command: str, manifest: Path) -> int:
         raise SessionError('Invalid session manifest')
     for name in data['names']:
         validate_name(name)
-    if command not in ('clients', 'recover', 'refresh') and (
+    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 == 'refresh':
-        fd = os.open(manifest.parent / 'refresh.request', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
-        os.close(fd)
-        print('Refreshing workspaces and viewers; VMs keep running.')
-        return 0
     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, :refresh to rebuild workspaces/viewers, or :quit.')
-        if not data['names']:
-            print('No VM profiles yet. Use :shell to run hydra add NAME --iso PATH and hydra start NAME.\n'
-                  'After adding or removing profiles, leave the shell and enter :refresh; running VMs are preserved.')
+              '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))
@@ -248,14 +219,12 @@ def session_helper(command: str, manifest: Path) -> int:
                 return 0
             if choice == ':quit':
                 return 0
-            if choice == ':refresh':
-                return session_helper('refresh', manifest)
             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, :refresh, or :quit.')
+                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

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/qemu-virtio-refresh-rate.patch

@@ -1,74 +0,0 @@
---- a/include/hw/virtio/virtio-gpu.h
-+++ b/include/hw/virtio/virtio-gpu.h
-@@ -128,6 +128,7 @@
-     uint32_t flags;
-     uint32_t xres;
-     uint32_t yres;
-+    uint32_t refresh_rate;
-     uint64_t hostmem;
-     VirtIOGPUOutputList *outputs;
- };
-@@ -173,7 +174,8 @@
-     DEFINE_PROP_BIT("edid", _state, _conf.flags, \
-                     VIRTIO_GPU_FLAG_EDID_ENABLED, true), \
-     DEFINE_PROP_UINT32("xres", _state, _conf.xres, 1280), \
--    DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800)
-+    DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800), \
-+    DEFINE_PROP_UINT32("refresh_rate", _state, _conf.refresh_rate, 0)
- 
- typedef struct VGPUDMABuf {
-     QemuDmaBuf *buf;
---- a/hw/display/virtio-gpu-base.c
-+++ b/hw/display/virtio-gpu-base.c
-@@ -64,7 +64,9 @@
-         .height_mm = g->req_state[scanout].height_mm,
-         .prefx = g->req_state[scanout].width,
-         .prefy = g->req_state[scanout].height,
--        .refresh_rate = g->req_state[scanout].refresh_rate,
-+        /* An explicit monitor refresh rate, in mHz, survives UI resizes. */
-+        .refresh_rate = g->conf.refresh_rate ? g->conf.refresh_rate :
-+                        g->req_state[scanout].refresh_rate,
-     };
- 
-     for (output_idx = 0, node = g->conf.outputs;
---- a/hw/display/edid-generate.c
-+++ b/hw/display/edid-generate.c
-@@ -220,17 +220,20 @@
-     desc[5 + len] = '\n';
- }
- 
--static void edid_desc_ranges(uint8_t *desc)
-+static void edid_desc_ranges(uint8_t *desc, uint64_t clock,
-+                             uint32_t htotal, uint32_t vtotal)
- {
-     edid_desc_type(desc, 0xfd);
- 
--    /* vertical (50 -> 125 Hz) */
--    desc[5] =  50;
--    desc[6] = 125;
--
--    /* horizontal (30 -> 160 kHz) */
--    desc[7] =  30;
--    desc[8] = 160;
-+    /* Enclose the encoded timing after its 10 kHz pixel-clock rounding,
-+     * while retaining legacy ranges whenever they already contain it. */
-+    uint64_t vperiod = (uint64_t)htotal * vtotal;
-+    uint64_t hperiod = (uint64_t)htotal * 1000;
-+    desc[5] = MIN(50, clock / vperiod);
-+    desc[6] = MIN(255, MAX(125, DIV_ROUND_UP(clock, vperiod)));
-+
-+    desc[7] = MIN(30, clock / hperiod);
-+    desc[8] = MIN(255, MAX(160, DIV_ROUND_UP(clock, hperiod)));
- 
-     /* max dot clock (2550 MHz) */
-     desc[9] = 2550 / 10;
-@@ -513,7 +516,8 @@
-      * so dta descriptor offsets don't move any more.
-      */
- 
--    edid_desc_ranges(desc);
-+    edid_desc_ranges(desc, (uint64_t)timings.clock * 10000,
-+                     info->prefx + timings.xblank, info->prefy + timings.yblank);
-     desc = edid_desc_next(edid, dta, desc);
- 
-     if (desc && info->name) {

patches/virt-viewer-hydra-session.patch

@@ -8,20 +8,20 @@
      gboolean attach;
      gboolean shared;
      gboolean quitting;
-@@ -952,6 +953,13 @@
+@@ -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);
-         gchar *d = strstr(title, "%d");
 @@ -2134,6 +2142,7 @@
  static gboolean opt_verbose = FALSE;
  static gboolean opt_debug = FALSE;
@@ -110,38 +110,3 @@
      gtk_window_set_title(GTK_WINDOW(self->window), title);
      if (self->subtitle) {
          gtk_header_bar_set_title(GTK_HEADER_BAR(header), self->subtitle);
-@@ -1467,6 +1475,34 @@
-     display = self->display;
-     event = (GdkEventKey *)ev;
- 
-+    /* Keyboard grabs remove GTK's application accelerator group so normal
-+     * shortcuts reach the guest. Keep only the explicitly configured manual
-+     * USB selector available before forwarding this event to the display. */
-+    if (!self->kiosk) {
-+        gchar **accels = gtk_application_get_accels_for_action(
-+            GTK_APPLICATION(self->app), "win.usb-device-select");
-+        gboolean matched = FALSE;
-+        for (guint i = 0; accels[i]; i++) {
-+            guint key;
-+            GdkModifierType mods;
-+            gtk_accelerator_parse(accels[i], &key, &mods);
-+            if (key && gdk_keyval_to_lower(event->keyval) == key &&
-+                (event->state & gtk_accelerator_get_default_mod_mask()) == mods) {
-+                matched = TRUE;
-+                break;
-+            }
-+        }
-+        g_strfreev(accels);
-+        if (matched) {
-+            GAction *action = g_action_map_lookup_action(
-+                G_ACTION_MAP(self->window), "usb-device-select");
-+            if (action && g_action_get_enabled(action)) {
-+                g_action_activate(action, NULL);
-+                return TRUE;
-+            }
-+        }
-+    }
-+
-     gtk_widget_grab_focus(GTK_WIDGET(display));
- 
-     // Look through keymaps - if set for mappings and intercept

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())

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_cli.py

@@ -1,66 +0,0 @@
-import json
-import subprocess
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-from hydra.profiles import load_profile
-
-
-class CLITests(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory(prefix='hc-')
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.defaults = self.root / 'defaults.toml'
-        self.base = [sys.executable, 'bin/hydra', '--config', str(self.root / 'config'),
-                     '--state', str(self.root / 'state'), '--runtime', str(self.root / 'run'),
-                     '--defaults', str(self.defaults)]
-
-    def run_cli(self, *args):
-        return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5)
-
-    def test_defaults_affect_new_profiles_only_and_explicit_options_win(self):
-        self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n')
-        added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0')
-        self.assertEqual(added.returncode, 0, added.stderr)
-        first = load_profile(self.root / 'config/first.toml')
-        self.assertEqual((first.network, first.width, first.height, first.refresh_millihz), ('user', 1600, 1080, 0))
-        added = self.run_cli('add', 'second')
-        self.assertEqual(added.returncode, 0, added.stderr)
-        second = load_profile(self.root / 'config/second.toml')
-        self.assertEqual((second.network, second.width, second.height, second.refresh_millihz), ('bridge', 1920, 1080, 144000))
-        self.defaults.write_text('width=1024\nnetwork="user"\n')
-        self.assertEqual(load_profile(self.root / 'config/second.toml'), second)
-        self.defaults.write_text('width="bad"\n')
-        listed = self.run_cli('list')
-        self.assertEqual(listed.returncode, 0, listed.stderr)
-        self.assertEqual(len(json.loads(listed.stdout)), 2)
-        self.assertEqual(self.run_cli('add', 'third').returncode, 1)
-        self.assertFalse((self.root / 'config/third.toml').exists())
-
-    def test_missing_defaults_use_developer_preferences_and_media_cli(self):
-        added = self.run_cli('add', 'guest')
-        self.assertEqual(added.returncode, 0, added.stderr)
-        profile = load_profile(self.root / 'config/guest.toml')
-        self.assertEqual((profile.network, profile.width, profile.height, profile.refresh_millihz), ('user', 1280, 800, 0))
-        iso = self.root / 'installer.iso'
-        iso.touch()
-        attached = self.run_cli('media', 'guest', '--iso', str(iso))
-        self.assertEqual(attached.returncode, 0, attached.stderr)
-        iso.unlink()
-        listed = self.run_cli('list')
-        self.assertEqual(listed.returncode, 0, listed.stderr)
-        self.assertEqual(self.run_cli('status', 'guest').returncode, 0)
-        detached = self.run_cli('media', 'guest', '--detach')
-        self.assertEqual(detached.returncode, 0, detached.stderr)
-        self.assertIsNone(load_profile(self.root / 'config/guest.toml').iso)
-
-    def test_unknown_or_invalid_distribution_defaults_are_rejected(self):
-        for content in ('iso="/private.iso"\n', 'network="public"\n', 'refresh_millihz=true\n', '[display]\nwidth=1920\n'):
-            self.defaults.write_text(content)
-            result = self.run_cli('add', 'guest')
-            self.assertEqual(result.returncode, 1, result.stderr)
-            self.assertNotIn('Traceback', result.stderr)
-            self.assertFalse((self.root / 'config/guest.toml').exists())

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

@@ -33,12 +33,9 @@ class ProfilesTests(unittest.TestCase):
                 with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                     Profile('guest', self.iso, **{field: value})
 
-    def test_invalid_media_path_types(self):
-        for iso in (True, 42, ''):
-            with self.subTest(iso=iso), self.assertRaises(ValueError):
-                Profile('guest', iso)
-
-    def test_duplicate_profile_preserves_existing(self):
+    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()
@@ -46,44 +43,6 @@ class ProfilesTests(unittest.TestCase):
             save_profile(dest, Profile('guest', self.iso, memory=8192))
         self.assertEqual(dest.read_bytes(), original)
 
-    def test_disk_only_profile_roundtrips_without_media(self):
-        dest = self.root / 'guest.toml'
-        save_profile(dest, Profile('guest'))
-        self.assertIsNone(load_profile(dest).iso)
-        self.assertNotIn('iso =', dest.read_text())
-
-    def test_missing_media_does_not_block_loading_profile(self):
-        dest = self.root / 'guest.toml'
-        save_profile(dest, Profile('guest', self.iso))
-        self.iso.unlink()
-        self.assertEqual(load_profile(dest).iso, self.iso)
-
-    def test_display_bounds_and_types(self):
-        for field, values in {'width': (0, True, 16385), 'height': (0, '1080', 16385),
-                              'refresh_millihz': (-1, True, 1000001)}.items():
-            for value in values:
-                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
-                    Profile('guest', self.iso, **{field: value})
-
-    def test_explicit_mac_roundtrip_and_validation(self):
-        dest = self.root / 'guest.toml'
-        save_profile(dest, Profile('guest', mac='02:ab:cd:ef:01:23'))
-        self.assertEqual(load_profile(dest).mac, '02:ab:cd:ef:01:23')
-        for value in ('ff:00:00:00:00:01', '00:00:00:00:00:01',
-                      '02:00:00:00:00:zz', '02:00:00:00:00', True):
-            with self.subTest(mac=value), self.assertRaises(ValueError):
-                Profile('guest', mac=value)
-
-    def test_refresh_vertical_descriptor_boundaries(self):
-        for value in (0, 50000, 255000):
-            path = self.root / 'guest.toml'
-            save_profile(path, Profile('guest', refresh_millihz=value))
-            self.assertEqual(load_profile(path).refresh_millihz, value)
-            path.unlink()
-        for value in (1, 30000, 49999, 255001, 1000000):
-            with self.subTest(refresh=value), self.assertRaises(ValueError):
-                Profile('guest', refresh_millihz=value)
-
     def test_profile_name_must_match_filename(self):
         path = self.root / 'guest.toml'
         path.write_text(f'name = "other"\niso = "{self.iso}"\n')

tests/test_runtime.py

@@ -88,90 +88,6 @@ class RuntimeTests(unittest.TestCase):
         except VMError:
             pass
 
-    def test_mac_persisted_distinct_and_stable_across_restarts(self):
-        self.runtime.add(Profile('other', network='bridge'))
-        other = self.runtime.profile('other')
-        guest = self.runtime.profile('guest')
-        self.assertIsNotNone(guest.mac)
-        self.assertNotEqual(guest.mac, other.mac)
-        octet = int(guest.mac.split(':')[0], 16)
-        self.assertEqual(octet & 3, 2)  # locally administered, unicast
-        def nic(profile, runtime):
-            return next(arg for arg in build_qemu_argv(profile, runtime.paths(profile.name))
-                        if arg.startswith('virtio-net-pci,'))
-        expected = nic(guest, self.runtime)
-        self.runtime.start('guest')
-        self.runtime.stop('guest', force=True)
-        self.runtime.start('guest')
-        self.runtime.stop('guest', force=True)
-        self.assertEqual(nic(self.runtime.profile('guest'), self.runtime), expected)
-        changed_runtime = Runtime(self.runtime.config, self.runtime.state, self.root / 'different-run')
-        self.assertEqual(nic(changed_runtime.profile('guest'), changed_runtime), expected)
-        self.runtime.media('guest', None)
-        self.assertEqual(self.runtime.profile('guest').mac, guest.mac)
-
-    def test_legacy_profiles_gain_stable_owner_specific_argv_mac(self):
-        file = self.runtime.config / 'legacy.toml'
-        save_profile(file, Profile('legacy', network='bridge'))
-        original = file.read_bytes()
-        legacy = self.runtime.profile('legacy')
-        self.assertIsNone(legacy.mac)
-        paths = self.runtime.paths('legacy')
-        def mac():
-            return next(arg for arg in build_qemu_argv(legacy, paths) if arg.startswith('virtio-net-pci,'))
-        with patch('os.getuid', return_value=1000):
-            first = mac()
-            self.assertEqual(mac(), first)
-            self.assertNotEqual(first, next(arg for arg in build_qemu_argv(
-                legacy, self.runtime.paths('other')) if arg.startswith('virtio-net-pci,')))
-        with patch('os.getuid', return_value=1001):
-            self.assertNotEqual(mac(), first)
-        self.assertEqual(file.read_bytes(), original)
-
-    def test_disk_boot_omits_media_and_requests_display(self):
-        profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
-                          graphics='software')
-        argv = build_qemu_argv(profile, self.runtime.paths('guest'))
-        self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
-        self.assertNotIn('ide-cd,drive=iso', argv)
-        self.assertIn('order=c', argv)
-        devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
-        self.assertEqual(len(devices), 2)
-        stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
-        self.assertFalse(any('refresh_rate=' in arg for arg in stock))
-
-    def test_missing_iso_is_error_at_start_without_creating_disk(self):
-        self.iso.unlink()
-        with self.assertRaisesRegex(VMError, 'ISO'):
-            self.runtime.start('guest')
-        self.assertFalse(self.runtime.paths('guest').disk.exists())
-        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
-
-    def test_media_changes_only_while_stopped_and_preserves_disk(self):
-        paths = self.runtime.paths('guest')
-        self.runtime.start('guest')
-        disk = paths.disk.read_bytes()
-        config = (self.runtime.config / 'guest.toml').read_bytes()
-        for iso in (None, self.iso):
-            with self.assertRaises(VMError):
-                self.runtime.media('guest', iso)
-        self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
-        self.iso.unlink()
-        self.assertEqual(self.runtime.profile('guest').iso, self.iso)
-        self.runtime.stop('guest', force=True)
-        self.runtime.media('guest', None)
-        self.assertIsNone(self.runtime.profile('guest').iso)
-        self.runtime.start('guest')
-        self.runtime.stop('guest', force=True)
-        self.assertEqual(paths.disk.read_bytes(), disk)
-        replacement = self.root / 'replacement.iso'
-        replacement.touch()
-        self.runtime.media('guest', replacement)
-        self.assertEqual(self.runtime.profile('guest').iso, replacement)
-        with self.assertRaisesRegex(VMError, 'ISO'):
-            self.runtime.media('guest', self.iso)
-        self.assertEqual(self.runtime.profile('guest').iso, replacement)
-
     def test_argv_is_private_and_comma_safe(self):
         paths = self.runtime.paths('guest')
         argv = build_qemu_argv(self.profile, paths)
@@ -185,7 +101,7 @@ class RuntimeTests(unittest.TestCase):
         self.assertIn('disable-agent-file-xfer=on', spice)
         self.assertIn('run,,comma', spice)
         self.assertIn('virtio-blk-pci,drive=disk', argv)
-        self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
+        self.assertIn('virtio-net-pci,netdev=net', argv)
         self.assertNotIn('-vnc', argv)
         self.assertNotIn('-virtfs', argv)
 

tests/test_session.py

@@ -55,11 +55,6 @@ class SessionTests(unittest.TestCase):
         self.assertEqual(tree.findall('./mouse/context/mousebind/action'), [])
         self.assertEqual(tree.findtext('./libinput/device/tap'), 'yes')
 
-    def test_empty_session_has_one_workspace_without_empty_names(self):
-        tree = ET.fromstring(session.build_labwc_config([], ['foot']))
-        self.assertEqual(tree.find('./desktops').get('number'), '1')
-        self.assertIsNone(tree.find('./desktops/names'))
-
     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')
@@ -87,8 +82,7 @@ class SessionTests(unittest.TestCase):
             self.assertEqual(main(base + ['view', 'alpha']), 1)
         self.assertIn('viewer', error.getvalue().lower())
 
-    def test_windowed_viewer_exit_with_missing_iso_leaves_vm_control_untouched(self):
-        self.iso.unlink()
+    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)
@@ -178,45 +172,6 @@ else:
             self.assertEqual(main(['_session-helper', 'recover', str(manifest)]), 1)
         self.assertIn('manifest', error.getvalue().lower())
 
-    def test_empty_launch_refresh_rebuilds_profile_snapshot(self):
-        manager = Runtime(self.root / 'empty-config', self.root / 'empty-state', self.root / 'empty-run')
-        capture = self.root / 'refresh-capture.json'
-        compositor = self.root / 'refresh-labwc'
-        compositor.write_text('#!' + os.sys.executable + '\n' + r'''import json, os, subprocess, sys
-from pathlib import Path
-if '--version' in sys.argv:
-    print('labwc 0.20.2')
-    sys.exit(0)
-config = Path(sys.argv[sys.argv.index('-C') + 1])
-manifest = config.parent / 'session.json'
-data = json.loads(manifest.read_text())
-capture = Path(os.environ['CAPTURE'])
-seen = json.loads(capture.read_text()) if capture.exists() else []
-seen.append({'names': data['names'], 'xml': (config / 'rc.xml').read_text()})
-capture.write_text(json.dumps(seen))
-if len(seen) == 1:
-    launcher = os.environ['HYDRA_SOURCE_LAUNCHER']
-    for name in ('first', 'second'):
-        subprocess.run([sys.executable, launcher, '--config', data['config'], '--state', data['state'],
-                        '--runtime', data['runtime'], 'add', name], check=True)
-    subprocess.run([sys.executable, launcher, '_session-helper', 'recover', str(manifest)],
-                   input=':refresh\n', text=True, check=True)
-    sys.exit(67)
-sys.exit(0)
-''')
-        compositor.chmod(0o700)
-        with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
-                                     'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture),
-                                     'HYDRA_SOURCE_LAUNCHER': str(Path('bin/hydra').resolve())}), contextlib.redirect_stdout(io.StringIO()):
-            self.assertEqual(session.launch_session(manager), 0)
-        seen = json.loads(capture.read_text())
-        self.assertEqual([entry['names'] for entry in seen], [[], ['first', 'second']])
-        tree = ET.fromstring(seen[1]['xml'])
-        self.assertEqual(tree.find('desktops').get('number'), '2')
-        self.assertEqual([rule.get('title') for rule in tree.findall('./windowRules/windowRule')],
-                         ['HydraOS: first', 'HydraOS: second'])
-        self.assertFalse(manager.paths('first').disk.exists())
-
     def test_no_running_vms_opens_recovery_without_start_or_stop(self):
         terminal = self.root / 'terminal'
         capture = self.root / 'recovery-argv.json'

tests/test_smoke.py

@@ -1,48 +0,0 @@
-"""Evidence gates used by the real private-display integration probe."""
-import tempfile
-import unittest
-from pathlib import Path
-
-from dev import smoke
-
-
-class SmokeEvidenceTests(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory(prefix='hydra-smoke-test-')
-        self.addCleanup(self.tmp.cleanup)
-        root = Path(self.tmp.name)
-        self.hidden = root / 'hidden.log'
-        self.visible = root / 'visible.log'
-        self.hidden.write_text('wl_keyboard#21.enter(1)\nwl_keyboard#21.leave(2)\n')
-        self.visible.write_text('wl_keyboard#21.enter(3)\n')
-
-    def test_focus_parser_ignores_pointer_events(self):
-        self.hidden.write_text(self.hidden.read_text() + 'wl_pointer#22.enter(4)\n')
-        self.assertEqual(smoke.focus_events(self.hidden), ['enter', 'leave'])
-
-    def test_old_focus_events_cannot_prove_a_new_switch(self):
-        self.assertFalse(smoke.fresh_focus_transition(['enter', 'leave'], ['enter'],
-                                                     ['enter', 'leave'], ['enter']))
-        self.assertTrue(smoke.fresh_focus_transition(['enter'], ['leave'],
-                                                    ['enter', 'leave'], ['leave', 'enter']))
-
-    def test_hidden_guest_must_advance_during_hidden_interval(self):
-        with self.assertRaisesRegex(AssertionError, 'hidden guest stopped'):
-            smoke.measure_hidden_interval(lambda: 10, self.hidden, self.visible, duration=.01)
-        readings = iter((10, 20))
-        result = smoke.measure_hidden_interval(lambda: next(readings), self.hidden, self.visible, duration=.01)
-        self.assertEqual((result['before'], result['after']), (10, 20))
-        self.assertGreaterEqual(result['elapsed_seconds'], .01)
-
-    def test_focus_change_during_sampling_invalidates_hidden_evidence(self):
-        def reading():
-            self.hidden.write_text(self.hidden.read_text() + 'wl_keyboard#21.enter(4)\n')
-            return 10
-        with self.assertRaisesRegex(AssertionError, 'focus changed'):
-            smoke.measure_hidden_interval(reading, self.hidden, self.visible, duration=.01)
-
-    def test_compositor_classifier_does_not_hide_other_errors(self):
-        occupied = '[ERROR] [xwayland/sockets.c:47] Failed to bind socket @/tmp/.X11-unix/X2: Address already in use'
-        self.assertEqual(smoke.classify_compositor_errors(occupied), [occupied])
-        with self.assertRaisesRegex(AssertionError, 'Unexpected compositor'):
-            smoke.classify_compositor_errors(occupied + '\n[ERROR] [config.c:1] invalid configuration')