fix: preserve guest network identities and refresh session profiles

README.md

@@ -24,3 +24,32 @@ 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

@@ -1,11 +0,0 @@
-#!/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

@@ -0,0 +1,8 @@
+#!/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

@@ -0,0 +1,100 @@
+#!/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

@@ -0,0 +1,78 @@
+#!/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

@@ -0,0 +1,25 @@
+.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

@@ -0,0 +1,63 @@
+/* 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

@@ -1,911 +0,0 @@
-#!/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

@@ -0,0 +1,196 @@
+#!/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

@@ -0,0 +1,273 @@
+#!/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,6 +8,9 @@ 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";
@@ -17,7 +20,9 @@ let
     doCheck = true;
   });
 in pkgs.mkShell {
-  packages = [ pkgs.python3 pkgs.qemu viewer graphics.labwc pkgs.foot
+  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
                pkgs.brightnessctl pkgs.wayland-utils pkgs.wlr-randr ];
-  passthru = { inherit viewer; labwc = graphics.labwc; };
+  passthru = { inherit viewer qemu; labwc = graphics.labwc; };
 }

dev/virtual-keyboard-unstable-v1.xml

@@ -0,0 +1,113 @@
+<?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

@@ -1,85 +0,0 @@
-# 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,13 +72,17 @@ 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.
+**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.
 
 **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

@@ -0,0 +1,178 @@
+# 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

@@ -1,392 +0,0 @@
-"""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
+from .profiles import Profile, load_defaults
 from .runtime import Runtime, RuntimeError
 from .session import launch_session, session_helper, view
 
@@ -17,16 +17,26 @@ 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, required=True)
+    add.add_argument('--iso', type=Path)
     add.add_argument('--memory', type=int, default=4096)
     add.add_argument('--cpus', type=int, default=2)
     add.add_argument('--disk-size', default='40G')
     add.add_argument('--graphics', choices=('auto', 'software'), default='auto')
-    add.add_argument('--network', choices=('user', 'bridge'), default='user')
-    add.add_argument('--bridge', default='hydrabr0')
+    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')
     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='?')
@@ -57,12 +67,18 @@ 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, args.network, args.bridge)
+                              args.graphics, **preferences)
             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)} for profile in runtime.profiles()]
+            result = [{'name': profile.name, 'iso': str(profile.iso) if profile.iso else None} 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

@@ -1,721 +0,0 @@
-"""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,24 +16,40 @@ def validate_name(name: str) -> str:
 @dataclass(frozen=True)
 class Profile:
     name: str
-    iso: Path
+    iso: Path | None = None
     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)
-        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)
+        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())
         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'):
@@ -67,6 +83,22 @@ 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,6 +3,7 @@
 import contextlib
 import errno
 import fcntl
+import hashlib
 import json
 import os
 import socket
@@ -11,7 +12,7 @@ import subprocess
 import tempfile
 import time
 import uuid
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 
 from .profiles import Profile, load_profile, save_profile, validate_name
@@ -54,6 +55,14 @@ 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,
@@ -64,15 +73,21 @@ def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
             '-mon', 'chardev=qmp,mode=control']
     if accel == 'kvm':
         argv += ['-cpu', 'host']
-    for options in (
+    blockdevs = [
         {'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', '-device', 'ide-cd,drive=iso',
-             '-boot', 'order=dc', '-device', 'virtio-net-pci,netdev=net',
+    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)}',
              '-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']
@@ -83,12 +98,15 @@ 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']
+        argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl' + mode]
     else:
-        argv += ['-display', 'none', '-device', 'virtio-vga']
+        argv += ['-display', 'none', '-device', 'virtio-vga' + mode]
     return argv + ['-spice', spice]
 
 
@@ -125,9 +143,35 @@ 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 []
@@ -234,12 +278,13 @@ 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,9 +49,10 @@ 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))))
-    labels = ET.SubElement(desktops, 'names')
-    for name in names:
-        ET.SubElement(labels, 'name').text = name
+    if names:
+        labels = ET.SubElement(desktops, 'names')
+        for name in names:
+            ET.SubElement(labels, 'name').text = name
     if initial is not None:
         ET.SubElement(desktops, 'initial').text = str(names.index(initial) + 1)
     ET.SubElement(desktops, 'popupTime').text = '0'
@@ -173,25 +174,45 @@ 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)
-    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
+    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.
+
 
 
 def session_helper(command: str, manifest: Path) -> int:
@@ -203,14 +224,22 @@ 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') and (
+    if command not in ('clients', 'recover', 'refresh') 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, :quit to close this terminal.')
+              '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.')
         while True:
             try:
                 print(json.dumps([runtime.status(name) for name in data['names']], indent=2))
@@ -219,12 +248,14 @@ 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, or :quit.')
+                print('Enter a configured VM name, :shell, :refresh, 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

@@ -1,38 +0,0 @@
-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

@@ -1,30 +0,0 @@
-# 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

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

image/portage/make.conf

@@ -1,16 +0,0 @@
-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

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

image/portage/package.use/hydra

@@ -1,8 +0,0 @@
-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

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

image/sources.json

@@ -1,17 +0,0 @@
-{
-  "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

@@ -0,0 +1,74 @@
+--- 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;
-@@ -951,6 +952,13 @@
- {
+@@ -952,6 +953,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,3 +110,38 @@
      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

@@ -1,9 +0,0 @@
-#!/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

@@ -1,320 +0,0 @@
-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

@@ -0,0 +1,66 @@
+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

@@ -1,559 +0,0 @@
-"""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

@@ -1,634 +0,0 @@
-"""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,9 +33,12 @@ class ProfilesTests(unittest.TestCase):
                 with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                     Profile('guest', self.iso, **{field: value})
 
-    def test_missing_iso_and_duplicate_profile_preserve_existing(self):
-        with self.assertRaises(ValueError):
-            Profile('guest', self.root / 'missing.iso')
+    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):
         dest = self.root / 'guest.toml'
         save_profile(dest, Profile('guest', self.iso))
         original = dest.read_bytes()
@@ -43,6 +46,44 @@ 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,6 +88,90 @@ 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)
@@ -101,7 +185,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.assertIn('virtio-net-pci,netdev=net', argv)
+        self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
         self.assertNotIn('-vnc', argv)
         self.assertNotIn('-virtfs', argv)
 

tests/test_session.py

@@ -55,6 +55,11 @@ 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')
@@ -82,7 +87,8 @@ class SessionTests(unittest.TestCase):
             self.assertEqual(main(base + ['view', 'alpha']), 1)
         self.assertIn('viewer', error.getvalue().lower())
 
-    def test_windowed_viewer_exit_leaves_vm_control_untouched(self):
+    def test_windowed_viewer_exit_with_missing_iso_leaves_vm_control_untouched(self):
+        self.iso.unlink()
         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)
@@ -172,6 +178,45 @@ 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

@@ -0,0 +1,48 @@
+"""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')