Capture redacted guest serial while waiting for shutdown

Wait for installer exit and live shell before followup probes
Fix ISO tool checks and fail promptly on installer refusal
2026-10-13 05:47:48 +03:00 · 2026-10-13 05:39:59 +03:00 · 2026-10-13 05:33:27 +03:00 · 2026-10-13 04:55:37 +03:00 · 2026-10-13 04:49:54 +03:00 · 2026-10-13 04:35:05 +03:00
73 changed files with 138 additions and 10097 deletions
@@ -6,12 +6,9 @@ desktops may use the same independent VM manager and viewers.
 ## Status
-The first installable UEFI preview passed a complete disposable QEMU/KVM run:
+Under active development, not yet an installation-ready distribution. The target
-installation, boot without the ISO, password login, persistent data, networking,
+is an installable UEFI ISO, first validated on ASUS TUF Dash F15 FX517ZC. Existing
-nested KVM and VM survival across user logout. See the
+NixOS and VM disks are not installation or test targets.
 [build and acceptance record](docs/releases/2026-10-13/acceptance.json).
 The first physical target is ASUS TUF Dash F15 FX517ZC; laptop hardware checks
 remain pending. Existing NixOS and VM disks were not installation or test targets.
 1. Independent QEMU/KVM lifecycle with private QMP control.
 2. Minimal labwc session and fullscreen SPICE viewers.
@@ -27,87 +24,3 @@ a refresh-rate setting does not prove delivered FPS.
 - [Hardware target](docs/hardware/asus-fx517zc.md)
 Build outputs, VM disks, runtime state and credentials stay outside Git.
 ## First use
 Follow the [installation guide](docs/install.md) to install from the live ISO.
 The initial empty graphical session opens a recovery terminal. Enter `:shell`
 for an ordinary host shell; `sudo nmtui` configures host network connections.
 Guest operating systems are supplied by you:
 ```sh
 hydra add work --iso /absolute/path/ubuntu.iso --memory 8192 --cpus 4
 hydra start work
 hydra add experiments --iso /absolute/path/gentoo.iso
 hydra start experiments
 exit
 ```
 Back at `hydra recovery>`, enter `:refresh` to show the newly created guests.
 Guests remain running while you switch, reconnect viewers or leave the host
 graphical session. Shut a guest down from inside its OS, or request an orderly
 shutdown with `hydra shutdown NAME`. Check `hydra status NAME` before detaching
 its installer with `hydra media NAME --detach` and starting it again.
 | Shortcut | Action |
 | --- | --- |
 | Ctrl+Alt+Shift+Left / Right | Switch guest workspaces |
 | Ctrl+Alt+Shift+Escape | Open local recovery |
 | Ctrl+Alt+Shift+R | Release viewer input capture |
 | Ctrl+Alt+Shift+U | Choose a USB device to redirect |
 | Brightness keys | Adjust host backlight |
 New guests share `hydrabr0` (`10.203.0.0/24`) and have distinct persistent MAC
 addresses. Services can communicate between guests without per-port forwarding;
 access from your LAN requires a router route through the HydraOS host. Each
 guest controls its own services and firewall. Host Internet access is separate
 from this local guest network.
 Profiles live in `~/.config/hydra`, disks in `~/.local/state/hydra`; system defaults
 for new profiles are in `/etc/hydra/defaults.toml`. Use `hydra --help` for overrides.
 Other host desktops can use `hydra start NAME` and `hydra view NAME --windowed`
 without entering the minimal labwc session. Ordinary Gentoo administration and
 Portage remain available through the recovery shell.
 ## Build the ISO
 Follow the [isolated Gentoo build instructions](image/README.md). Outputs are
 `build/artifacts/hydra-os.iso` and `hydra-os.iso.sha256`; binary images stay out
 of Git. [Recorded source inputs and package versions](docs/releases/2026-10-13/rootfs-manifest.json)
 identify the tested image. The builder uses Gentoo systemd for login/device
 management, labwc for the minimal graphical session, and QEMU/KVM for guests.
 Portage and the tracked QEMU/viewer patches remain available for ordinary updates.
 Resume a build from the same source checkout and work directory used to create
 its builder. Moving from a worktree to another checkout requires a fresh
 `--work` directory under the project, or using the original worktree; provenance
 checks deliberately reject an existing container mounted to a different source.
 ## 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.
@@ -1,8 +0,0 @@
 #!/bin/sh
 set -eu
 source_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
 build_dir=$(mktemp -d /tmp/hydra-input-build-XXXXXX)
 wayland-scanner client-header "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-client-protocol.h"
 wayland-scanner private-code "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-protocol.c"
 cc -Wall -Wextra -Werror "$source_dir/input-probe.c" "$build_dir/virtual-keyboard-protocol.c" -I"$build_dir" $(pkg-config --cflags --libs wayland-client xkbcommon) -o "$build_dir/input-probe"
 printf '%s\n' "$build_dir/input-probe"
@@ -1,100 +0,0 @@
 #!/usr/bin/env python3
 """Compile actual QEMU EDID code and decode the guest-visible response.
 Catches loss of the configured refresh rate, a frontend overriding it, and
 monitor ranges excluding the generated preferred mode. No VM is touched.
 """
 import os
 from pathlib import Path
 import subprocess
 import sys
 import tempfile
 import tarfile
 source_root = Path(sys.argv[1])
 # Accept the pinned QEMU source archive directly. Read only the four files
 # needed here, avoiding unrelated upstream absolute symlinks in rom sources.
 if source_root.is_file():
    extracted = tempfile.TemporaryDirectory(prefix='hydra-edid-source-')
    with tarfile.open(source_root) as archive:
        members = {name: next(member for member in archive.getmembers()
                              if member.name.endswith('/' + name)) for name in (
            'hw/display/edid-generate.c', 'include/hw/display/edid.h',
            'include/hw/virtio/virtio-gpu.h', 'hw/display/virtio-gpu-base.c')}
        source_root = Path(extracted.name)
        for name, member in members.items():
            destination = source_root / name
            destination.parent.mkdir(parents=True, exist_ok=True)
            destination.write_bytes(archive.extractfile(member).read())
    subprocess.run(['patch', '--batch', '-p1', '-i',
                    str(Path(__file__).resolve().parents[1] / 'patches/qemu-virtio-refresh-rate.patch')],
                   cwd=source_root, check=True)
 edid = (source_root / 'hw/display/edid-generate.c').read_text()
 edid = '\n'.join(line for line in edid.splitlines() if not line.startswith('#include'))
 header = (source_root / 'include/hw/display/edid.h').read_text()
 header = header[header.index('typedef struct qemu_edid_info'):header.index('void qemu_edid_region_io')]
 gpu = (source_root / 'hw/display/virtio-gpu-base.c').read_text()
 start = gpu.index('void\nvirtio_gpu_base_generate_edid')
 gpu = gpu[start:gpu.index('static void virtio_gpu_invalidate_display', start)]
 stubs = r'''
 #include <stdint.h>
 #include <stddef.h>
 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
 #define ARRAY_SIZE(x) (sizeof(x)/sizeof((x)[0]))
 #define MAX(a,b) ((a) > (b) ? (a) : (b))
 #define MIN(a,b) ((a) < (b) ? (a) : (b))
 #define DIV_ROUND_UP(n,d) (((n) + (d) - 1) / (d))
 #define cpu_to_le32(x) (x)
 static void stw_le_p(void *p,uint16_t v) { uint8_t *b=p; b[0]=v; b[1]=v>>8; }
 static void stw_be_p(void *p,uint16_t v) { uint8_t *b=p; b[1]=v; b[0]=v>>8; }
 static void stl_le_p(void *p,uint32_t v) { uint8_t *b=p; for(int i=0;i<4;i++) b[i]=v>>(8*i); }
 typedef struct VirtIOGPUOutput { const char *name; } VirtIOGPUOutput;
 typedef struct VirtIOGPUOutputList { struct VirtIOGPUOutputList *next; VirtIOGPUOutput *value; } VirtIOGPUOutputList;
 struct virtio_gpu_requested_state { uint16_t width_mm,height_mm; uint32_t width,height,refresh_rate; int x,y; };
 typedef struct VirtIOGPUBase { struct { uint32_t refresh_rate; VirtIOGPUOutputList *outputs; } conf; struct virtio_gpu_requested_state req_state[1]; } VirtIOGPUBase;
 struct virtio_gpu_resp_edid { uint32_t size; uint8_t edid[1024]; };
 '''
 main = r'''
 int main(int argc,char **argv) {
    VirtIOGPUBase gpu = {0}; struct virtio_gpu_resp_edid out = {0};
    gpu.conf.refresh_rate=atoi(argv[1]); gpu.req_state[0].refresh_rate=atoi(argv[2]);
    gpu.req_state[0].width=atoi(argv[3]); gpu.req_state[0].height=atoi(argv[4]);
    virtio_gpu_base_generate_edid(&gpu,0,&out);
    fwrite(out.edid,1,qemu_edid_size(out.edid),stdout); return 0;
 }
 '''
 compiler = os.environ.get('CC', 'cc')
 with tempfile.TemporaryDirectory(prefix='qemu-edid-probe-') as temporary:
    directory = Path(temporary)
    (directory / 'probe.c').write_text(stubs + header + edid + gpu + main)
    subprocess.run([compiler, '-O2', str(directory / 'probe.c'), '-o', str(directory / 'probe')], check=True)
    for configured, frontend, width, height, expected in [
        (0, 0, 1920, 1080, 75),
        (0, 60000, 1920, 1080, 60),
        (144000, 0, 1920, 1080, 144),
        (144000, 60000, 1920, 1080, 144),
        (144000, 0, 1600, 900, 144),
        (0, 144000, 1920, 1080, 144),
        (50000, 0, 1920, 1080, 50),
        (255000, 0, 1280, 800, 255),
    ]:
        data = subprocess.check_output([str(directory / 'probe'), *map(str, (configured, frontend, width, height))])
        assert all(sum(data[offset:offset + 128]) % 256 == 0 for offset in range(0, len(data), 128)), 'invalid EDID checksum'
        timing = data[54:72]
        clock = int.from_bytes(timing[:2], 'little') * 10000
        horizontal = timing[2] + ((timing[4] >> 4) << 8)
        hblank = timing[3] + ((timing[4] & 15) << 8)
        vertical = timing[5] + ((timing[7] >> 4) << 8)
        vblank = timing[6] + ((timing[7] & 15) << 8)
        actual = clock / (horizontal + hblank) / (vertical + vblank)
        assert (horizontal, vertical) == (width, height)
        assert abs(actual - expected) < 0.01, f'configured={configured}, frontend={frontend}: wanted {expected} Hz, got {actual}'
        ranges = next(data[offset:offset + 18] for offset in (54, 72, 90, 108) if data[offset:offset + 5] == b'\0\0\0\xfd\0')
        assert ranges[5] <= actual <= ranges[6], f'vertical range {ranges[5:7]} excludes {actual}'
        horizontal_khz = clock / (horizontal + hblank) / 1000
        assert ranges[7] <= horizontal_khz <= ranges[8], f'horizontal range {ranges[7:9]} excludes {horizontal_khz}'
        if configured == 0 and expected <= 75:
            assert ranges[5:9] == bytes((50, 125, 30, 160)), 'legacy monitor range changed'
        print(f'PASS configured={configured}, frontend={frontend}: {width}x{height}@{actual:.6f}, ranges={list(ranges[5:9])}')
@@ -1,78 +0,0 @@
 #!/usr/bin/env python3
 """Check real NIC identities on a bridge isolated in user/net/mount namespaces."""
 import argparse
 import json
 from pathlib import Path
 import subprocess
 import sys
 import tempfile
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from hydra.profiles import Profile
 from hydra.runtime import Runtime
 def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--namespace-root', type=Path, help=argparse.SUPPRESS)
    args = parser.parse_args()
    if args.namespace_root is None:
        root = Path(tempfile.mkdtemp(prefix='hydra-network-', dir='/tmp'))
        print('Evidence: ' + str(root), flush=True)
        (root / 'parent-namespaces.json').write_text(json.dumps({name: str(Path('/proc/self/ns', name).readlink())
                                                               for name in ('user', 'net', 'mnt')}))
        # All bridge and /etc changes below occur after this explicit namespace
        # boundary. The inherited host bridge configuration is never modified.
        subprocess.run(['unshare', '-Urnm', '--', str(Path(sys.executable).resolve()),
                        str(Path(__file__).resolve()), '--namespace-root', str(root)], check=True)
        return
    root = args.namespace_root
    parent_namespaces = json.loads((root / 'parent-namespaces.json').read_text())
    namespaces = {name: str(Path('/proc/self/ns', name).readlink()) for name in ('user', 'net', 'mnt')}
    assert all(namespaces[name] != parent_namespaces[name] for name in namespaces), 'private namespaces required'
    private_etc = root / 'etc'
    (private_etc / 'qemu').mkdir(parents=True)
    (private_etc / 'qemu/bridge.conf').write_text('allow hydrabr0\n')
    subprocess.run(['mount', '--bind', str(private_etc), '/etc'], check=True)
    subprocess.run(['ip', 'link', 'add', 'hydrabr0', 'type', 'bridge'], check=True)
    subprocess.run(['ip', 'link', 'set', 'hydrabr0', 'up'], check=True)
    manager = Runtime(root / 'config', root / 'state', root / 'run')
    result = {'evidence': str(root), 'namespace': 'private user/net/mount',
              'parent_namespaces': parent_namespaces, 'probe_namespaces': namespaces, 'names': {}}
    try:
        for name in ('alpha', 'beta'):
            manager.add(Profile(name, memory=128, cpus=1, disk_size='8M',
                                graphics='software', network='bridge'))
            manager.start(name)
        def actual_mac(name):
            with manager._connect(manager.paths(name)) as qmp:
                network = qmp.command('human-monitor-command', {'command-line': 'info network'})
            mac = manager.profile(name).mac
            assert 'macaddr=' + mac in network, network
            assert int(mac.split(':')[0], 16) & 3 == 2
            return mac, network
        for name in ('alpha', 'beta'):
            mac, network = actual_mac(name)
            result['names'][name] = {'mac': mac, 'network': network}
        assert result['names']['alpha']['mac'] != result['names']['beta']['mac']
        links = json.loads(subprocess.check_output(['ip', '-j', 'link', 'show', 'master', 'hydrabr0'], text=True))
        assert len(links) == 2, links
        result['bridge_ports'] = [link['ifname'] for link in links]
        manager.stop('alpha', force=True)
        manager.start('alpha')
        assert actual_mac('alpha')[0] == result['names']['alpha']['mac']
        result['restart_preserved_mac'] = True
        result['dhcp_and_guest_network_drivers_pending'] = True
        result['passed'] = True
        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
        print(json.dumps(result, indent=2))
    finally:
        for profile in manager.profiles():
            if manager.status(profile.name)['status'] == 'running':
                manager.stop(profile.name, force=True)
 if __name__ == '__main__':
    main()
@@ -1,25 +0,0 @@
 .code16
 .global _start
 _start:
 cli
 xor %ax,%ax
 mov %ax,%ds
 mov %ax,%ss
 mov $0x7c00,%sp
 movl $0x48594452,0x508
 movl $0,0x500
 movw $0,0x504
 loop:
 incl 0x500
 in $0x64,%al
 test $1,%al
 jz loop
 in $0x60,%al
 mov 0x504,%bx
 mov %al,0x600(%bx)
 inc %bx
 and $255,%bx
 mov %bx,0x504
 jmp loop
 .org 510
 .word 0xaa55
@@ -1,63 +0,0 @@
 /* Private-display integration helper; stdin uses Linux evdev key codes.
 * Compile via dev/build-input-probe.sh inside the temporary tools shell. */
 #define _GNU_SOURCE
 #include <assert.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/mman.h>
 #include <unistd.h>
 #include <wayland-client.h>
 #include <xkbcommon/xkbcommon.h>
 #include "virtual-keyboard-client-protocol.h"
 static struct wl_seat *seat;
 static struct zwp_virtual_keyboard_manager_v1 *manager;
 static void global(void *data, struct wl_registry *registry, uint32_t id,
                   const char *interface, uint32_t version) {
    (void)data; (void)version;
    if (!strcmp(interface,"wl_seat"))
        seat = wl_registry_bind(registry,id,&wl_seat_interface,1);
    else if (!strcmp(interface,"zwp_virtual_keyboard_manager_v1"))
        manager = wl_registry_bind(registry,id,&zwp_virtual_keyboard_manager_v1_interface,1);
 }
 static void removed(void *data,struct wl_registry *registry,uint32_t id) {
    (void)data; (void)registry; (void)id;
 }
 int main(void) {
    /* The caller supplies the separately created scratch Wayland runtime. */
    assert(getenv("WAYLAND_DISPLAY") && getenv("XDG_RUNTIME_DIR"));
    struct wl_display *display = wl_display_connect(NULL); assert(display);
    struct wl_registry *registry = wl_display_get_registry(display);
    struct wl_registry_listener listener={global,removed};
    wl_registry_add_listener(registry,&listener,NULL); wl_display_roundtrip(display);
    assert(seat && manager);
    struct zwp_virtual_keyboard_v1 *keyboard = zwp_virtual_keyboard_manager_v1_create_virtual_keyboard(manager,seat);
    struct xkb_context *context=xkb_context_new(XKB_CONTEXT_NO_FLAGS);
    struct xkb_rule_names names={.rules="evdev",.model="pc105",.layout="us"};
    struct xkb_keymap *map=xkb_keymap_new_from_names(context,&names,XKB_KEYMAP_COMPILE_NO_FLAGS); assert(map);
    struct xkb_state *state=xkb_state_new(map); assert(state);
    char *text=xkb_keymap_get_as_string(map,XKB_KEYMAP_FORMAT_TEXT_V1);
    size_t size=strlen(text)+1;
    int fd=memfd_create("hydra-private-keymap",0); assert(fd>=0);
    assert(write(fd,text,size)==(ssize_t)size);
    zwp_virtual_keyboard_v1_keymap(keyboard,WL_KEYBOARD_KEYMAP_FORMAT_XKB_V1,fd,size);
    wl_display_roundtrip(display); close(fd); free(text);
    char command[16]; unsigned key; unsigned tick=0;
    while (scanf("%15s %u",command,&key)==2) {
        if (!strcmp(command,"wait")) { usleep(key*1000); continue; }
        unsigned down=!strcmp(command,"down");
        assert(down || !strcmp(command,"up"));
        xkb_state_update_key(state,key+8,down?XKB_KEY_DOWN:XKB_KEY_UP);
        zwp_virtual_keyboard_v1_key(keyboard,++tick,key,down);
        zwp_virtual_keyboard_v1_modifiers(keyboard,
            xkb_state_serialize_mods(state,XKB_STATE_MODS_DEPRESSED),
            xkb_state_serialize_mods(state,XKB_STATE_MODS_LATCHED),
            xkb_state_serialize_mods(state,XKB_STATE_MODS_LOCKED),
            xkb_state_serialize_layout(state,XKB_STATE_LAYOUT_EFFECTIVE));
        assert(wl_display_roundtrip(display)>=0);
    }
    zwp_virtual_keyboard_v1_destroy(keyboard); wl_display_roundtrip(display);
    xkb_state_unref(state); xkb_keymap_unref(map); xkb_context_unref(context);
    wl_display_disconnect(display);
    return 0;
 }
@@ -1,196 +0,0 @@
 #!/usr/bin/env python3
 """Exercise the product's empty-session/recovery/refresh loop privately."""
 import argparse
 import json
 import os
 from pathlib import Path
 import shutil
 import signal
 import subprocess
 import sys
 import tempfile
 import time
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from hydra.runtime import Runtime
 from smoke import (wait_for, counter, focus_events, fresh_focus_transition,
                   measure_hidden_interval, classify_compositor_errors)
 def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
    parser.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
    parser.add_argument('--keyboard', required=True, help='compiled dev/input-probe.c')
    args = parser.parse_args()
    for name in ('labwc', 'viewer', 'keyboard'):
        executable = shutil.which(getattr(args, name))
        if not executable:
            parser.error('missing executable: ' + getattr(args, name))
        setattr(args, name, executable)
    if not os.access('/dev/kvm', os.R_OK | os.W_OK):
        parser.error('KVM access required')
    source = Path(__file__).resolve().parents[1]
    root = Path(tempfile.mkdtemp(prefix='hydra-session-refresh-', dir='/tmp'))
    print('Evidence: ' + str(root), flush=True)
    xdg = root / 'xdg'; xdg.mkdir(mode=0o700)
    manager = Runtime(root / 'config', root / 'state', root / 'run')
    cli = [sys.executable, str(source / 'bin/hydra'), '--config', str(manager.config),
           '--state', str(manager.state), '--runtime', str(manager.runtime)]
    subprocess.run(['as', '--32', str(source / 'dev/guest-probe.S'), '-o', str(root / 'boot.o')], check=True)
    subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(root / 'boot.o'), str(root / 'boot.bin')], check=True)
    (root / 'boot.raw').write_bytes((root / 'boot.bin').read_bytes() + bytes(8 * 1024 * 1024 - 512))
    def wrapper(name, code):
        path = root / name
        path.write_text('#!' + sys.executable + '\n' + code)
        path.chmod(0o700)
        return str(path)
    common = 'import json, os, pathlib, subprocess, sys, time\nroot = pathlib.Path(' + repr(str(root)) + ')\n'
    labwc = wrapper('labwc-private', common + '''
 if '--version' in sys.argv:
    os.execv(@LABWC@, [@LABWC@] + sys.argv[1:])
 config = pathlib.Path(sys.argv[sys.argv.index('-C') + 1])
 number = len(list(root.glob('snapshot-*.json')))
 data = json.loads((config.parent / 'session.json').read_text())
 data.update(pid=os.getpid(), config_directory=str(config))
 (root / ('snapshot-%d.xml' % number)).write_text((config / 'rc.xml').read_text())
 (root / ('snapshot-%d.json' % number)).write_text(json.dumps(data))
 # Test-only wrapper replaces physical backends before the real compositor runs.
 os.environ.update(WLR_BACKENDS='headless', WLR_RENDERER='pixman',
                  WLR_XWAYLAND='/bin/false', LABWC_UPDATE_ACTIVATION_ENV='0',
                  XDG_DATA_DIRS='/nonexistent')
 os.execv(@LABWC@, [@LABWC@, '-d'] + sys.argv[1:])
 '''.replace('@LABWC@', repr(args.labwc)))
    viewer = wrapper('viewer-private', common + '''
 if '--help' in sys.argv:
    os.execv(VIEWER, [VIEWER] + sys.argv[1:])
 name = sys.argv[sys.argv.index('--title') + 1].split(': ', 1)[1]
 (root / (name + '-viewer.json')).write_text(json.dumps({
    'argv': sys.argv[1:], 'config': os.environ['XDG_CONFIG_HOME'],
    'gdk_backend': os.environ.get('GDK_BACKEND'), 'wayland_display': os.environ['WAYLAND_DISPLAY']}))
 log = os.open(root / (name + '.log'), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
 os.dup2(log, 1); os.dup2(log, 2)
 os.environ['WAYLAND_DEBUG'] = '1'
 os.execv(VIEWER, [VIEWER] + sys.argv[1:])
 '''.replace('VIEWER', repr(args.viewer)))
    terminal = wrapper('terminal-private', common + '''
 manifest = pathlib.Path(sys.argv[-1])
 socket = pathlib.Path(os.environ['XDG_RUNTIME_DIR']) / os.environ['WAYLAND_DISPLAY']
 (root / 'recovery-ready').write_text(str(manifest))
 while not (root / 'recovery-input').exists():
    if not socket.exists():
        sys.exit(0)
    time.sleep(.05)
 if not socket.exists():
    sys.exit(0)
 commands = (root / 'recovery-input').read_text()
 # The actual product recovery helper executes :shell and :refresh.
 with (root / 'recovery.log').open('a') as log:
    sys.exit(subprocess.run(sys.argv[2:], input=commands, text=True, stdout=log, stderr=log).returncode)
 ''')
    shell = wrapper('scratch-shell', common + 'sys.path.insert(0, ' + repr(str(source)) + ')\n' + '''
 from hydra.runtime import Runtime
 from smoke import counter, wait_for
 cli = CLI
 manager = Runtime(root / 'config', root / 'state', root / 'run')
 for name in ('alpha', 'beta'):
    subprocess.run(cli + ['add', name, '--memory', '128', '--cpus', '1', '--disk-size', '8M',
                         '--graphics', 'software', '--network', 'user'], check=True)
    paths = manager.paths(name); manager.prepare(paths)
    subprocess.run([os.environ['HYDRA_QEMU_IMG'], 'convert', '-f', 'raw', '-O', 'qcow2',
                    str(root / 'boot.raw'), str(paths.disk)], check=True)
    subprocess.run(cli + ['start', name], check=True)
    time.sleep(.3)
 identities = {name: {'pid': manager.paths(name).pid.read_text().strip(),
                    'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
              for name in ('alpha', 'beta')}
 (root / 'before.json').write_text(json.dumps(identities))
 '''.replace('CLI', repr(cli)).replace('from smoke import', 'from dev.smoke import'))
    env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), HYDRA_LABWC=labwc,
               HYDRA_VIEWER=viewer, HYDRA_TERMINAL=terminal, HYDRA_BRIGHTNESS=shutil.which('true'),
               SHELL=shell, GSETTINGS_BACKEND='memory')
    for key in ('WAYLAND_DISPLAY', 'DISPLAY', 'LABWC_PID'):
        env.pop(key, None)
    log = (root / 'session.log').open('w')
    launcher, injector = None, None
    result = {'evidence': str(root), 'physical_input_pending': True,
              'terminal': 'test-only command feeder; real recovery helper and scratch :shell',
              'backend': 'test-only wrapper around actual product launch_session and actual labwc'}
    try:
        launcher = subprocess.Popen(cli + ['session'], env=env, stdout=log, stderr=log)
        wait_for(lambda: (root / 'recovery-ready').exists(), 'empty session did not open recovery')
        first = json.loads((root / 'snapshot-0.json').read_text())
        assert first['names'] == [], 'session was not initially empty'
        (root / 'recovery-input').write_text(':shell\n:refresh\n')
        wait_for(lambda: (root / 'snapshot-1.json').exists(), 'product loop did not restart after :refresh', timeout=15)
        second = json.loads((root / 'snapshot-1.json').read_text())
        assert second['names'] == ['alpha', 'beta'], second
        assert first['pid'] != second['pid'], 'compositor was not refreshed'
        logs = {name: root / (name + '.log') for name in ('alpha', 'beta')}
        for name, path in logs.items():
            wait_for(lambda: path.exists() and 'set_title("HydraOS: ' + name + '")' in path.read_text(),
                     name + ' product viewer did not connect')
        viewers = {name: json.loads((root / (name + '-viewer.json')).read_text()) for name in logs}
        assert len({data['config'] for data in viewers.values()}) == 2, 'viewer settings were shared'
        for data in viewers.values():
            assert data['gdk_backend'] == 'wayland'
            assert data['config'].startswith(str(manager.runtime) + '/viewer-')
            assert '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0' in data['argv']
            assert '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0' in data['argv']
        alog, blog = logs['alpha'], logs['beta']
        display = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
        keyboard_env = dict(env, WAYLAND_DISPLAY=display)
        injector = subprocess.Popen([args.keyboard], env=keyboard_env, stdin=subprocess.PIPE, text=True)
        time.sleep(.2)
        wait_for(lambda: focus_events(alog) and focus_events(alog)[-1] == 'enter', 'alpha did not focus')
        def switch(code, source_log, target_log):
            before_source, before_target = focus_events(source_log), focus_events(target_log)
            injector.stdin.write(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29\nwait 100\n')
            injector.stdin.flush()
            wait_for(lambda: fresh_focus_transition(before_source, before_target,
                     focus_events(source_log), focus_events(target_log)), 'fresh reserved focus transition failed')
        switch(106, alog, blog)
        intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
        switch(105, blog, alog)
        intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
        before = json.loads((root / 'before.json').read_text())
        after = {name: {'pid': manager.paths(name).pid.read_text().strip(),
                       'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
                 for name in ('alpha', 'beta')}
        for name in before:
            assert after[name]['pid'] == before[name]['pid'] and after[name]['uuid'] == before[name]['uuid']
            assert after[name]['counter'] != before[name]['counter']
        # Only the PID recorded by our own wrapper, verified against its private
        # config path, may be signalled. Never inherit a host compositor PID.
        command = Path('/proc') / str(second['pid']) / 'cmdline'
        assert second['config_directory'].encode() in command.read_bytes()
        os.kill(second['pid'], signal.SIGTERM)
        assert launcher.wait(timeout=10) == 0, 'product session exit failed'
        for name in before:
            assert manager.status(name)['status'] == 'running'
            assert manager.paths(name).pid.read_text().strip() == before[name]['pid']
        result.update(initial_names=first['names'], refreshed_names=second['names'],
                      compositor_pids=[first['pid'], second['pid']], before=before, after=after,
                      viewers=viewers,
                      hidden_intervals=intervals, survived_session_exit=True,
                      expected_xwayland_diagnostics=classify_compositor_errors((root / 'session.log').read_text()),
                      passed=True)
        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
        print(json.dumps(result, indent=2))
    finally:
        if injector and injector.poll() is None:
            injector.terminate(); injector.wait(timeout=5)
        if launcher and launcher.poll() is None:
            launcher.send_signal(signal.SIGINT); launcher.wait(timeout=12)
        for profile in manager.profiles():
            if manager.status(profile.name)['status'] == 'running':
                manager.stop(profile.name, force=True)
        log.close()
 if __name__ == '__main__':
    main()
@@ -1,273 +0,0 @@
 #!/usr/bin/env python3
 """Two disposable KVM guests on a private headless display; never host input."""
 import argparse
 import json
 import os
 from pathlib import Path
 import re
 import shutil
 import subprocess
 import sys
 import tempfile
 import time
 from xml.etree import ElementTree as ET
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from hydra.profiles import Profile
 from hydra.runtime import Runtime
 from hydra.session import build_labwc_config, build_viewer_argv, _viewer_environment
 def wait_for(predicate, description, timeout=8):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if predicate():
            return
        time.sleep(.05)
    raise AssertionError(description)
 def counter(manager, name):
    with manager._connect(manager.paths(name)) as qmp:
        reply = qmp.command('human-monitor-command', {'command-line': 'xp /3wx 0x500'})
    values = [int(word, 16) for word in re.findall(r'0x([0-9a-fA-F]+)', reply)]
    assert len(values) == 3 and values[2] == 0x48594452, 'scratch guest did not boot'
    return values[0]
 def focus_events(log):
    """Only keyboard focus, excluding pointer-enter and old window titles."""
    return re.findall(r'wl_keyboard#\d+\.(enter|leave)\(', log.read_text())
 def fresh_focus_transition(source_before, target_before, source_after, target_after):
    return (len(source_after) > len(source_before) and source_after[-1] == 'leave'
            and len(target_after) > len(target_before) and target_after[-1] == 'enter')
 def measure_hidden_interval(read_counter, hidden_log, visible_log, duration=.5):
    hidden_focus, visible_focus = focus_events(hidden_log), focus_events(visible_log)
    assert hidden_focus and hidden_focus[-1] == 'leave', 'sampled guest is not hidden'
    assert visible_focus and visible_focus[-1] == 'enter', 'other guest is not focused'
    def unchanged():
        assert (focus_events(hidden_log) == hidden_focus and
                focus_events(visible_log) == visible_focus), 'focus changed during hidden interval'
    start = time.monotonic()
    before = read_counter()
    unchanged()
    while time.monotonic() - start < duration:
        time.sleep(min(.05, duration))
        unchanged()
    after = read_counter()
    unchanged()
    assert after != before, 'hidden guest stopped working during continuously hidden interval'
    return {'before': before, 'after': after,
            'elapsed_seconds': time.monotonic() - start, 'focus_unchanged': True}
 def classify_compositor_errors(log):
    expected = []
    for line in log.splitlines():
        if '[ERROR]' not in line:
            continue
        if re.search(r'\[xwayland/sockets\.c:\d+\] Failed to bind socket '
                     r'@?/tmp/\.X11-unix/X\d+: Address already in use$', line):
            expected.append(line)
        else:
            raise AssertionError('Unexpected compositor error: ' + line)
    return expected
 def main():
    cli = argparse.ArgumentParser(description=__doc__)
    cli.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
    cli.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
    cli.add_argument('--keyboard', help='compiled dev/input-probe.c; built automatically when omitted')
    args = cli.parse_args()
    for name in ('viewer', 'labwc'):
        value = shutil.which(getattr(args, name))
        if not value:
            cli.error(f'missing {name}: {getattr(args, name)}')
        setattr(args, name, value)
    source = Path(__file__).resolve().parent
    if not args.keyboard:
        args.keyboard = subprocess.check_output([str(source / 'build-input-probe.sh')], text=True).strip()
    if not os.access('/dev/kvm', os.R_OK | os.W_OK):
        cli.error('KVM access required for this integration check')
    root = Path(tempfile.mkdtemp(prefix='hydra-smoke-', dir='/tmp'))
    print(f'Evidence: {root}', flush=True)
    xdg = root / 'xdg'
    xdg.mkdir(mode=0o700)
    config = root / 'labwc'
    config.mkdir()
    recovery = root / 'recovery'
    unreserved = root / 'unreserved'
    names = ['alpha', 'beta']
    rc = ET.fromstring(build_labwc_config(names, [sys.executable, '-c',
                           'from pathlib import Path; Path(' + repr(str(recovery)) + ').touch()'], initial='alpha'))
    # Test-only controls prove native labwc inhibition is active; production
    # config has no ToggleKeybinds or ordinary host shortcut.
    keyboard = rc.find('keyboard')
    binding = ET.SubElement(keyboard, 'keybind', key='C-A-S-i')
    ET.SubElement(binding, 'action', name='ToggleKeybinds')
    binding = ET.SubElement(keyboard, 'keybind', key='F12')
    ET.SubElement(binding, 'action', name='Execute', command=shutil.which('touch') + ' ' + str(unreserved))
    ET.ElementTree(rc).write(config / 'rc.xml')
    (config / 'autostart').write_text('#!/bin/sh\n')
    (config / 'environment').touch()
    env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), WLR_BACKENDS='headless',
               WLR_RENDERER='pixman', WLR_XWAYLAND='/bin/false', XDG_DATA_DIRS='/nonexistent',
               GSETTINGS_BACKEND='memory', LABWC_UPDATE_ACTIVATION_ENV='0')
    for key in ('DISPLAY', 'WAYLAND_DISPLAY'):
        env.pop(key, None)
    manager = Runtime(root / 'config', root / 'state', root / 'run')
    boot_object, boot = root / 'boot.o', root / 'boot.bin'
    subprocess.run(['as', '--32', str(source / 'guest-probe.S'), '-o', str(boot_object)], check=True)
    subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(boot_object), str(boot)], check=True)
    raw = root / 'boot.raw'
    raw.write_bytes(boot.read_bytes() + bytes(8 * 1024 * 1024 - 512))
    media = root / 'installation.iso'
    media.touch()
    for name in names:
        manager.add(Profile(name, media if name == 'alpha' else None, memory=128, cpus=1, disk_size='8M', graphics='software',
                            width=1920, height=1080, refresh_millihz=144000))
        paths = manager.paths(name)
        manager.prepare(paths)
        subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'convert', '-f', 'raw',
                        '-O', 'qcow2', str(raw), str(paths.disk)], check=True)
    processes, streams = [], []
    result = {'evidence': str(root), 'physical_checks_pending': True}
    try:
        cl = (root / 'labwc.log').open('w'); streams.append(cl)
        compositor = subprocess.Popen([args.labwc, '-d', '-C', str(config)], env=env, stdout=cl, stderr=cl)
        processes.append(compositor)
        wait_for(lambda: any(not p.name.endswith('.lock') for p in xdg.glob('wayland-*')), 'private display did not start')
        env['WAYLAND_DISPLAY'] = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
        injector = subprocess.Popen([args.keyboard], env=env, stdin=subprocess.PIPE, text=True)
        processes.append(injector)
        def keys(commands):
            injector.stdin.write(commands + '\nwait 100\n'); injector.stdin.flush()
            time.sleep(.15)
            assert injector.poll() is None, 'private input helper exited'
        def chord(code):
            keys(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29')
        def switch(key, source_log, target_log):
            source_before, target_before = focus_events(source_log), focus_events(target_log)
            chord(key)
            wait_for(lambda: fresh_focus_transition(source_before, target_before,
                                                   focus_events(source_log), focus_events(target_log)),
                     'reserved switch did not produce fresh keyboard leave/enter')
            return {'from_events': len(focus_events(source_log)),
                    'to_events': len(focus_events(target_log)), 'keycode': key}
        def attach(name, number):
            vl = (root / f'{name}-{number}.log').open('w'); streams.append(vl)
            previous = dict(os.environ)
            try:
                os.environ.update(env)
                viewer_env = _viewer_environment(root / f'{name}-{number}-settings')
            finally:
                os.environ.clear(); os.environ.update(previous)
            viewer_env['WAYLAND_DEBUG'] = '1'
            proc = subprocess.Popen(build_viewer_argv(manager.paths(name), args.viewer),
                                    env=viewer_env, stdout=vl, stderr=vl)
            processes.append(proc)
            log = root / f'{name}-{number}.log'
            wait_for(lambda: f'set_title("HydraOS: {name}")' in log.read_text(), f'{name} viewer did not connect')
            return proc, log
        for name in names:
            manager.start(name)
        media.unlink()  # Open QEMU media disappears; status/control/view still work.
        time.sleep(1)
        before = {name: counter(manager, name) for name in names}
        with manager._connect(manager.paths('alpha')) as qmp:
            children = qmp.command('qom-list', {'path': '/machine/peripheral-anon'})
            video = next(child['name'] for child in children if child['type'] == 'child<virtio-vga>')
            mode = {prop: qmp.command('qom-get', {'path': '/machine/peripheral-anon/' + video, 'property': prop})
                    for prop in ('xres', 'yres', 'refresh_rate')}
            assert mode == {'xres': 1920, 'yres': 1080, 'refresh_rate': 144000}, mode
            usb_before = qmp.command('human-monitor-command', {'command-line': 'info usb'})
        identities = {name: manager.paths(name).pid.read_text().strip() for name in names}
        alpha, alog = attach('alpha', 1)
        beta, blog = attach('beta', 1)
        wait_for(lambda: '.enter(' in alog.read_text(), 'alpha did not receive focus')
        # F12 must work before inhibition; then become suppressed in alpha.
        keys('down 88\nup 88')
        wait_for(unreserved.exists, 'ordinary test binding failed before inhibition')
        unreserved.unlink()
        chord(23)  # I: test-only ToggleKeybinds.
        keys('down 88\nup 88')
        assert not unreserved.exists(), 'keybinding inhibition did not suppress ordinary F12'
        switch(106, alog, blog)  # Right: beta focused, alpha continuously hidden.
        hidden_intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
        switch(105, blog, alog)  # Left: alpha focused, beta continuously hidden.
        hidden_intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
        chord(1)  # Escape: production recovery binding still works.
        wait_for(recovery.exists, 'reserved recovery failed with inhibition')
        dialog = 'set_title("Select USB devices for redirection")'
        assert dialog not in alog.read_text(), 'USB selector opened automatically'
        # Hold modifiers through opening, then release inside the modal dialog.
        keys('down 29\ndown 56\ndown 42\ndown 22')
        wait_for(lambda: dialog in alog.read_text(), 'manual USB selector did not open', timeout=3)
        keys('up 22\nup 42\nup 56\nup 29')
        keys('down 1\nup 1')  # Close selector.
        post_usb_switches = [switch(106, alog, blog), switch(105, blog, alog)]
        assert alpha.poll() is None and beta.poll() is None
        assert identities == {name: manager.paths(name).pid.read_text().strip() for name in names}, 'switch restarted a guest'
        assert alog.read_text().count('set_title("HydraOS: alpha")') == 1, 'switch reconnected alpha'
        assert blog.read_text().count('set_title("HydraOS: beta")') == 1, 'switch reconnected beta'
        with manager._connect(manager.paths('alpha')) as qmp:
            usb_after = qmp.command('human-monitor-command', {'command-line': 'info usb'})
        assert usb_after == usb_before, 'opening selector attached an unselected USB device'
        alpha.terminate(); alpha.wait(timeout=5)
        assert manager.status('alpha')['status'] == 'running'
        alpha, attached = attach('alpha', 2)
        after = {name: counter(manager, name) for name in names}
        assert all(after[name] != before[name] for name in names), 'background guest stopped working'
        # Quit only beta; alpha's counter/lifecycle remains independent.
        manager.stop('beta', force=True)
        assert manager.status('alpha')['status'] == 'running'
        assert counter(manager, 'alpha') != after['alpha']
        compositor.terminate(); compositor.wait(timeout=5)
        assert manager.status('alpha')['status'] == 'running'
        manager.stop('alpha', force=True)
        manager.media('alpha', None)
        assert manager.profile('alpha').iso is None
        manager.start('alpha')
        time.sleep(1)
        disk_boot_counter = counter(manager, 'alpha')
        assert disk_boot_counter > 0
        expected_compositor_noise = classify_compositor_errors((root / 'labwc.log').read_text())
        result.update(hidden_guest_intervals=hidden_intervals, post_usb_switches=post_usb_switches,
                      expected_compositor_noise=expected_compositor_noise,
                      manual_usb_dialog=True, standard_us_keymap=True,
                      native_labwc_inhibition=True, reserved_switch_and_recovery=True,
                      qemu_survives_viewer_and_compositor=True, independent_lifecycle=True,
                      guest_counters_before=before, guest_counters_after=after,
                      guest_pids=identities, explicit_reattach=True, requested_qemu_mode=mode,
                      usb_devices_unchanged_without_selection=True,
                      missing_iso_does_not_block_views_or_control=True,
                      restart_without_original_media=True, disk_boot_counter=disk_boot_counter,
                      viewer_version=subprocess.check_output([args.viewer, '--version'], text=True).strip(),
                      guest_input_and_physical_modifiers='not proven by this boot counter workload')
        (root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
        print(json.dumps(result, indent=2))
    finally:
        for proc in reversed(processes):
            if proc.poll() is None:
                proc.terminate(); proc.wait(timeout=5)
        for stream in streams:
            stream.close()
        for name in names:
            if manager.status(name)['status'] != 'stopped':
                manager.stop(name, force=True)
 if __name__ == '__main__':
    main()
@@ -8,9 +8,6 @@ let
    url = "https://github.com/NixOS/nixpkgs/archive/7fd4004d0511991fff3a0524856a9be09e28c5f0.tar.gz";
    sha256 = "sha256-i5qk1kLiit98rRbh1AXmWBNT9QXC/y4l5Wk/u5EdpDQ=";
  }) {};
  qemu = (pkgs.qemu.override { hostCpuOnly = true; }).overrideAttrs (old: {
    patches = (old.patches or []) ++ [ ../patches/qemu-virtio-refresh-rate.patch ];
  });
  viewer = pkgs.virt-viewer.overrideAttrs (old: {
    src = pkgs.fetchurl {
      url = "https://releases.pagure.org/virt-viewer/virt-viewer-11.0.tar.xz";
@@ -20,9 +17,7 @@ let
    doCheck = true;
  });
 in pkgs.mkShell {
-  nativeBuildInputs = [ pkgs.stdenv.cc pkgs.pkg-config pkgs.wayland-scanner ];
+  packages = [ pkgs.python3 pkgs.qemu viewer graphics.labwc pkgs.foot
  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 qemu; labwc = graphics.labwc; };
+  passthru = { inherit viewer; labwc = graphics.labwc; };
 }
@@ -1,113 +0,0 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <protocol name="virtual_keyboard_unstable_v1">
  <copyright>
    Copyright © 2008-2011  Kristian Høgsberg
    Copyright © 2010-2013  Intel Corporation
    Copyright © 2012-2013  Collabora, Ltd.
    Copyright © 2018       Purism SPC
    Permission is hereby granted, free of charge, to any person obtaining a
    copy of this software and associated documentation files (the "Software"),
    to deal in the Software without restriction, including without limitation
    the rights to use, copy, modify, merge, publish, distribute, sublicense,
    and/or sell copies of the Software, and to permit persons to whom the
    Software is furnished to do so, subject to the following conditions:
    The above copyright notice and this permission notice (including the next
    paragraph) shall be included in all copies or substantial portions of the
    Software.
    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
    DEALINGS IN THE SOFTWARE.
  </copyright>
  <interface name="zwp_virtual_keyboard_v1" version="1">
    <description summary="virtual keyboard">
      The virtual keyboard provides an application with requests which emulate
      the behaviour of a physical keyboard.
      This interface can be used by clients on its own to provide raw input
      events, or it can accompany the input method protocol.
    </description>
    <request name="keymap">
      <description summary="keyboard mapping">
        Provide a file descriptor to the compositor which can be
        memory-mapped to provide a keyboard mapping description.
        Format carries a value from the keymap_format enumeration.
      </description>
      <arg name="format" type="uint" summary="keymap format"/>
      <arg name="fd" type="fd" summary="keymap file descriptor"/>
      <arg name="size" type="uint" summary="keymap size, in bytes"/>
    </request>
    <enum name="error">
      <entry name="no_keymap" value="0" summary="No keymap was set"/>
    </enum>
    <request name="key">
      <description summary="key event">
        A key was pressed or released.
        The time argument is a timestamp with millisecond granularity, with an
        undefined base. All requests regarding a single object must share the
        same clock.
        Keymap must be set before issuing this request.
        State carries a value from the key_state enumeration.
      </description>
      <arg name="time" type="uint" summary="timestamp with millisecond granularity"/>
      <arg name="key" type="uint" summary="key that produced the event"/>
      <arg name="state" type="uint" summary="physical state of the key"/>
    </request>
    <request name="modifiers">
      <description summary="modifier and group state">
        Notifies the compositor that the modifier and/or group state has
        changed, and it should update state.
        The client should use wl_keyboard.modifiers event to synchronize its
        internal state with seat state.
        Keymap must be set before issuing this request.
      </description>
      <arg name="mods_depressed" type="uint" summary="depressed modifiers"/>
      <arg name="mods_latched" type="uint" summary="latched modifiers"/>
      <arg name="mods_locked" type="uint" summary="locked modifiers"/>
      <arg name="group" type="uint" summary="keyboard layout"/>
    </request>
    <request name="destroy" type="destructor" since="1">
      <description summary="destroy the virtual keyboard keyboard object"/>
    </request>
  </interface>
  <interface name="zwp_virtual_keyboard_manager_v1" version="1">
    <description summary="virtual keyboard manager">
      A virtual keyboard manager allows an application to provide keyboard
      input events as if they came from a physical keyboard.
    </description>
    <enum name="error">
      <entry name="unauthorized" value="0" summary="client not authorized to use the interface"/>
    </enum>
    <request name="create_virtual_keyboard">
      <description summary="Create a new virtual keyboard">
        Creates a new virtual keyboard associated to a seat.
        If the compositor enables a keyboard to perform arbitrary actions, it
        should present an error when an untrusted client requests a new
        keyboard.
      </description>
      <arg name="seat" type="object" interface="wl_seat"/>
      <arg name="id" type="new_id" interface="zwp_virtual_keyboard_v1"/>
    </request>
  </interface>
 </protocol>
@@ -10,7 +10,7 @@ for a plan using the whole disk you intend to erase:
 ```sh
 hydra-install
-sudo hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER
+hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER
 ```
 Replace the placeholders with an actual whole disk and your desired account
@@ -42,27 +42,21 @@ 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 a generic
+Run the repository harness with QEMU/KVM, `qemu-img`, Python 3, and an OVMF
-x86_64 QEMU OVMF CODE/VARS pair. A Xen-specific OVMF build is unsuitable: the
+CODE/VARS pair. Use trusted image and firmware files. For example:
 first validation attempt with that variant produced no guest boot output.
 Use trusted image and matching firmware files. For example:
 ```sh
 install -d -m 700 "$HOME/.cache/hydra-tests"
 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 \
+  --ovmf-vars /absolute/path/to/OVMF_VARS.fd
  --scratch-parent "$HOME/.cache/hydra-tests"
 ```
-The harness creates a private directory under `--scratch-parent` (`/tmp` by default), a new 32 GiB sparse
+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.
 Prefer a disk-backed scratch directory: an installed disk stored on a tmpfs
 such as `/tmp` consumes host RAM. The example keeps test data in the user's cache.
 The test boots the live image, runs the real installer through the guest's
 controlling terminal using a generated password, writes a persistence sentinel,
@@ -73,9 +67,6 @@ 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.
 It also logs the installed account out and back in, checking that the same VM,
 private sockets and user audio services survive a measured interval without an
 open login session. This establishes process/socket continuity, not audio playback.
 `report.json` records each observed check, ISO/firmware SHA256, failures, and
 skipped nested virtualization. Serial/QEMU logs and the qcow2 remain in the
@@ -1,57 +0,0 @@
 {
  "format": 1,
  "status": "passed",
  "completed_utc": "2026-10-13T02:49:50Z",
  "artifact": {
    "file": "hydra-os.iso",
    "bytes": 2345109504,
    "sha256": "bdff621037ce9220767b7869499823c9b840b76c0b38268bc7f56a7b54390ce0",
    "source_commit": "903e7cf35339e50783a1bb3790b06e3204efab4d",
    "source_inputs_sha256": "cfe9ddf4549bdedcd91888adc8c4b00639e370b6296c361898d3898daf76d4de",
    "kernel": "6.18.54-gentoo-dist-bin",
    "package_count": 559,
    "rootfs_manifest_sha256": "c53ce549e98e0b5a56091d7960a2737534882e3f64a2497a12e3b2a27e522c48"
  },
  "validation": {
    "harness_commit": "4a636924df4e630788dcbbe8d0002261fa9a7df9",
    "harness_sha256": "64863d2f856cd3cee1e0c28af5aac2db2706b6403bf5deb32f64f57c33df1a53",
    "ovmf_code_sha256": "f3ff7e73448ed2845ee15356f394882f5618eb5dab92c9a30ec6ee0e1468553a",
    "ovmf_vars_input_sha256": "6ed987af3a3c155be71665f510eae3e007eda9b8b94afd59d45e91c4a11565cc",
    "tools": "Ubuntu QEMU8.2.2, Python3.12.3",
    "disk": "new 32 GiB sparse qcow2, only this disposable VM",
    "iso_unmodified": true,
    "manual_guest_changes": false,
    "full_private_report_sha256": "accfc85840681da69eeacabf82c9007dceb66952dac907e0e545b1061a9dee10",
    "checks": {
      "live_qmp_kvm": "passed",
      "live_boot": "passed",
      "real_installer": "passed",
      "live_poweroff": "passed",
      "installed_qmp_kvm": "passed",
      "installed_password_login": "passed",
      "disk_persistence": "passed",
      "installed_contract": "passed",
      "lan_to_vm_tcp": "passed",
      "nested_kvm": "passed",
      "installed_user_logout": "passed",
      "installed_poweroff": "passed"
    }
  },
  "scope": {
    "nested_virtualization": "Normal-user KVM VM launch, bridge helper and QMP; no nested guest OS boot",
    "audio": "Socket/service continuity only; signal/playback untested",
    "logout": "Actual getty logout, no open login for measured 15-second survival interval, relogin; unchanged VM process/start/UUID, runtime directories, private sockets and user audio service identities",
    "graphics": "Private graphical live recovery observed; actual two-guest labwc/viewer and profile-refresh probes recorded separately; no physical scanout assertion"
  },
  "pending_hardware": [
    "ASUS live boot",
    "Wi-Fi and Ethernet on laptop",
    "touchpad and physical input capture",
    "brightness",
    "physical audio playback",
    "suspend/resume",
    "USB device attachment and guest drivers",
    "physical display refresh and delivered FPS"
  ],
  "publication": "Source, patches and these records are in Git. ISO, VM disks, raw logs and private runtime identities remain local."
 }
@@ -1 +0,0 @@
 bdff621037ce9220767b7869499823c9b840b76c0b38268bc7f56a7b54390ce0  hydra-os.iso
@@ -34,40 +34,40 @@
 **Files:** `image/sources.json`, `image/packages`, `image/portage/`, `scripts/hydra-build`, `hydra/build.py`, `tests/test_build.py`.
 **Interfaces:** `hydra-build fetch|prepare|packages|rootfs|iso|status`, `--work PATH`, `--cache PATH`, optional `--podman PATH`. Each step resumable and validates prerequisite manifests. Outputs WORK/rootfs payload or dedicated builder container, WORK/artifacts/, build manifest. No host root required.
- [x] Test pinned download hash mismatch, required signaturefingerprint, safe workpath/refusal of root, resume state vs missingactualartifact, scoped Podman argv.
+- [ ] Test pinned download hash mismatch, required signaturefingerprint, safe workpath/refusal of root, resume state vs missingactualartifact, scoped Podman argv.
- [x] Observe RED; implement verifiedfetch and isolatedimport/container naming. Use stage3 20261011T170049Z SHA256 50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b and Portage4696089e782c20b5f0c04952f6698d6a8cc1acb4 archiveSHA256 d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca.
+- [ ] Observe RED; implement verifiedfetch and isolatedimport/container naming. Use stage3 20261011T170049Z SHA256 50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b and Portage4696089e782c20b5f0c04952f6698d6a8cc1acb4 archiveSHA256 d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca.
- [x] Configure officialbinhost signaturechecking, selective ~amd64 for labwc/wlroots, QEMU x86_64 only, required SPICE/GL/USB flags. Preserve licenses and installed packageDB.
+- [ ] Configure officialbinhost signaturechecking, selective ~amd64 for labwc/wlroots, QEMU x86_64 only, required SPICE/GL/USB flags. Preserve licenses and installed packageDB.
- [x] Run actual dependency resolution and package install; record exact versions/USE and any selective sourcebuild decisions. No claim of bit-identical build.
+- [ ] Run actual dependency resolution and package install; record exact versions/USE and any selective sourcebuild decisions. No claim of bit-identical build.
- [x] Run tests and commit builder inputs/scripts.
+- [ ] Run tests and commit builder inputs/scripts.
 ### Task 2: Host rootfs and bootable image
 **Files:** `image/rootfs/`, `image/grub.cfg`, `image/dracut.conf`, `image/configure-rootfs.sh`, `tests/test_image.py`.
 **Interfaces:** Consumes builder and tested CLI/session from graphics plan. Produces labelled HYDRA_OS ISO with boot/vmlinuz, boot/initramfs-live.img, LiveOS/squashfs.img and package/buildmanifest.
- [x] Test service/config generation, live-only overrides separation, exact kernelversion used for dracut, image label/cmdline agreement, networksubnet conflict handling.
+- [ ] Test service/config generation, live-only overrides separation, exact kernelversion used for dracut, image label/cmdline agreement, networksubnet conflict handling.
- [x] Implement rootfs overlay: greetd/minimalsession, recovery, localliveadministration, installedauth baseline, audio, NetworkManager, hydrabr0/DHCP/NAT and narrowlyscoped bridgehelper. Include laptopfirmware/bootdrivers and OVMF.
+- [ ] Implement rootfs overlay: greetd/minimalsession, recovery, localliveadministration, installedauth baseline, audio, NetworkManager, hydrabr0/DHCP/NAT and narrowlyscoped bridgehelper. Include laptopfirmware/bootdrivers and OVMF.
- [x] Build genericliveinitramfs with dmsquash-live and OverlayFS; squashfspayload retains kernel/packageDB. Assemble UEFIISO with GRUB; save SHA256 and manifest.
+- [ ] Build genericliveinitramfs with dmsquash-live and OverlayFS; squashfspayload retains kernel/packageDB. Assemble UEFIISO with GRUB; save SHA256 and manifest.
- [x] Run image checks and disposable OVMF/KVM liveboot; verify rootfilesystem ready and expected tools. Commit rootfs/imageconfig.
+- [ ] Run image checks and disposable OVMF/KVM liveboot; verify rootfilesystem ready and expected tools. Commit rootfs/imageconfig.
 ### Task 3: Explicit guarded installer
 **Files:** `bin/hydra-install`, `hydra/installer.py`, `tests/test_installer.py`.
 **Interfaces:** `hydra-install --disk DEVICE --user USER [--execute --confirm-erase DEVICE]`; without execute prints exactplan. Environment/default rootfs path points to live squashfs, never running root. Password via getpass/TTY, never CLI/log. Separate nonsecret dryrun from execution.
- [x] Test missing/mismatched confirmation, disk mounted directly or through children/ancestors, root/live source backingdisk, invaliddevice, symlink resolution, invalidusername, failedcommands abort subsequentsteps.
+- [ ] Test missing/mismatched confirmation, disk mounted directly or through children/ancestors, root/live source backingdisk, invaliddevice, symlink resolution, invalidusername, failedcommands abort subsequentsteps.
- [x] Observe RED; implement sysfs/lsblk topology validation, GPT1GiBESP/ext4, payload extraction, fstabUUIDs, user/password setup and liveoverridesremoval. Explicit target, no automaticdiscoveryselection.
+- [ ] Observe RED; implement sysfs/lsblk topology validation, GPT1GiBESP/ext4, payload extraction, fstabUUIDs, user/password setup and liveoverridesremoval. Explicit target, no automaticdiscoveryselection.
- [x] Install regularinitramfs and GRUBfallback --removable --no-nvram. Checked subprocessargv; handle failure diagnostics and mounts cleanup without deletingtargetdata.
+- [ ] Install regularinitramfs and GRUBfallback --removable --no-nvram. Checked subprocessargv; handle failure diagnostics and mounts cleanup without deletingtargetdata.
- [x] Test fakecommandfailure paths then complete actualinstallation inside fresh VM using disposable qcow2. Rootprivilege exists only in testguest.
+- [ ] Test fakecommandfailure paths then complete actualinstallation inside fresh VM using disposable qcow2. Rootprivilege exists only in testguest.
- [x] Commit installer and negative tests.
+- [ ] Commit installer and negative tests.
 ### Task 4: Installed-system validation and publication
 **Files:** `dev/iso-smoke.py`, `docs/install.md`, `docs/verification.md`, `README.md`.
 **Interfaces:** Consumes ISO and installer. Produces evidence from boot/install/reboot and releaseartifact checksum, plus pendinghardwarechecklist.
- [x] Boot ISO under OVMF/KVM with fresh disk; install only there; write persistence sentinel.
+- [ ] Boot ISO under OVMF/KVM with fresh disk; install only there; write persistence sentinel.
- [x] Remove ISO and boot installeddisk; verify sentinel, services, QEMUtools, network and shell/recovery. Check no liveoverlay/autologinadminoverride remains.
+- [ ] Remove ISO and boot installeddisk; verify sentinel, services, QEMUtools, network and shell/recovery. Check no liveoverlay/autologinadminoverride remains.
- [x] Test nested guest VM if KVM nesting available; distinguish virtualgraphics from real laptopresults.
+- [ ] Test nested guest VM if KVM nesting available; distinguish virtualgraphics from real laptopresults.
- [x] Run full test suite and independent review. Correct material defects; document remaining physicalchecks without claiming them passed.
+- [ ] Run full test suite and independent review. Correct material defects; document remaining physicalchecks without claiming them passed.
- [x] Save ISO locally with checksum, push source/patches/manifests to suppliedrepo. BinaryISO only uploaded when suitable repository release/storage path exists; never commit ISO to Git.
+- [ ] Save ISO locally with checksum, push source/patches/manifests to suppliedrepo. BinaryISO only uploaded when suitable repository release/storage path exists; never commit ISO to Git.
@@ -50,12 +50,12 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
 - `load_profile(path: Path) -> Profile`, `build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]`, `QMPClient.command(name: str, arguments: dict | None = None) -> dict`.
 - `VMPaths` provides disk, log, lock and private QMP/SPICE sockets. Session consumes stable name and socket paths only.
- [x] Write validation/QMP fake-server tests: invalid names/resources, missing ISO, paths containing spaces, greeting/capabilities, asynchronous events, IDs, timeouts.
+- [ ] Write validation/QMP fake-server tests: invalid names/resources, missing ISO, paths containing spaces, greeting/capabilities, asynchronous events, IDs, timeouts.
- [x] Run `python3 -m unittest discover -s tests -v`; observe failure before implementation.
+- [ ] Run `python3 -m unittest discover -s tests -v`; observe failure before implementation.
- [x] Implement validation and bounded QMP; subprocess argv arrays, named environment tool overrides for tests.
+- [ ] Implement validation and bounded QMP; subprocess argv arrays, named environment tool overrides for tests.
- [x] Add lifecycle tests: simultaneous start, existing-disk preservation, stale socket, failed boot, shutdown vs force, QMP VM identity.
+- [ ] Add lifecycle tests: simultaneous start, existing-disk preservation, stale socket, failed boot, shutdown vs force, QMP VM identity.
- [x] Implement locked safe disk creation, detached QEMU, virtio disk/network, SPICE no clipboard/file transfer, optional Intel GL. No display dependency in manager and no PID-only stop.
+- [ ] Implement locked safe disk creation, detached QEMU, virtio disk/network, SPICE no clipboard/file transfer, optional Intel GL. No display dependency in manager and no PID-only stop.
- [x] Run complete suite, help, scratch QEMU lifecycle probe. Commit `feat: add independent VM manager`.
+- [ ] Run complete suite, help, scratch QEMU lifecycle probe. Commit `feat: add independent VM manager`.
 ### Task 2: Minimal fullscreen session
@@ -63,12 +63,12 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
 **Interfaces:** Consumes Task1 Profile/VMPaths. Produces `hydra view NAME [--windowed]`, `hydra session [--nested]`, labwc config only in selected runtime. Viewer title is deterministic from validated name; each VM maps to one workspace. Recovery opens host terminal; session never shuts down a VM.
- [x] Probe actual labwc version/overrideInhibition and remote-viewer kiosk USB behavior on disposable state; record versions.
+- [ ] Probe actual labwc version/overrideInhibition and remote-viewer kiosk USB behavior on disposable state; record versions.
- [x] Resolve manual USB dialog with a small upstream viewer adaptation if necessary; never auto-redirect USB.
+- [ ] Resolve manual USB dialog with a small upstream viewer adaptation if necessary; never auto-redirect USB.
- [x] Write tests for valid escaped XML, exact window/workspace/fullscreen rules, no default keyboard/mouse bindings, overrideInhibition on host keys, missing viewer/display errors.
+- [ ] Write tests for valid escaped XML, exact window/workspace/fullscreen rules, no default keyboard/mouse bindings, overrideInhibition on host keys, missing viewer/display errors.
- [x] Run session tests and observe missing implementation failure.
+- [ ] Run session tests and observe missing implementation failure.
- [x] Implement session/viewer commands and pinned temporary tool environment; no host installation.
+- [ ] Implement session/viewer commands and pinned temporary tool environment; no host installation.
- [x] Run full tests and labwc config parsing. Commit `feat: add fullscreen HydraOS session`.
+- [ ] Run full tests and labwc config parsing. Commit `feat: add fullscreen HydraOS session`.
 ### Task 3: Integration and recovery
@@ -76,12 +76,12 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
 **Interfaces:** Consumes Tasks1/2, produces disposable-VM integration evidence.
- [x] Run two disposable KVM instances/viewers, never existing lab disks; verify private QMP and independent lifecycle.
+- [ ] Run two disposable KVM instances/viewers, never existing lab disks; verify private QMP and independent lifecycle.
- [x] Verify viewer/session exit preserves QEMU and reattach works; switching neither restarts nor reconnects guests.
+- [ ] Verify viewer/session exit preserves QEMU and reattach works; switching neither restarts nor reconnects guests.
- [x] Check input only in disposable session, without injecting into current host or user's guests; add regression tests for discovered defects.
+- [ ] Check input only in disposable session, without injecting into current host or user's guests; add regression tests for discovered defects.
- [x] Record held-key, physical brightness and 144Hz checks needing dedicated TTY. Nested tests cannot prove them.
+- [ ] Record held-key, physical brightness and 144Hz checks needing dedicated TTY. Nested tests cannot prove them.
- [x] Run full suite and independent review; fix material findings with reproducing tests.
+- [ ] Run full suite and independent review; fix material findings with reproducing tests.
- [x] Commit evidence and push reviewed prototype branch to supplied repository.
+- [ ] Commit evidence and push reviewed prototype branch to supplied repository.
 ## Next plan
@@ -1,193 +0,0 @@
 # Graphics and recovery verification
 The [2026-10-13 preview record](releases/2026-10-13/acceptance.json) identifies
 the exact ISO, firmware and harness used for the passing twelve-check acceptance
 run. The unmodified ISO installed onto a fresh 32 GiB qcow2 disk and booted from
 that disk with no ISO attached. Password login, persistent data, removed live
 overrides, installed services, DNS lookup, routed LAN-to-VM TCP, normal-user
 nested KVM/bridge access and normal shutdown passed. An actual logout followed
 by 15 seconds without an open login and a new login preserved the same VM,
 QMP/SPICE sockets and user audio services. Audio playback was not measured.
 Live graphical recovery was also observed through a private QEMU screenshot;
 the earlier two-guest compositor and profile-refresh probes cover fullscreen
 switching and hidden workload continuity. These virtual checks leave the physical
 ASUS checks listed below pending. No current host configuration or existing user
 VM disk was changed for this validation.
 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.
@@ -9,7 +9,6 @@ import os
 from pathlib import Path
 import re
 import shutil
 import shlex
 import subprocess
 import sys
 import tempfile
@@ -58,21 +57,6 @@ sys.exit(code)
 '''
 def rootfs_retry_guard_script():
    """Refuse deletion of a still-mounted export, including nested mountpoints."""
    return r"""mounts=$(findmnt --raw --noheadings --output TARGET)
 while IFS= read -r mountpoint; do
  case "$mountpoint" in
    "$root"|"$root/"*) echo 'Refusing to remove a mounted rootfs export' >&2; exit 1 ;;
  esac
 done <<< "$mounts"
 if [ -e "$root" ] || [ -L "$root" ]; then
  [ ! -L "$root" ] && [ -f "$root/.hydra-export-owned" ] || { echo 'Unowned rootfs output' >&2; exit 1; }
  rm -rf -- "$root"
 fi
 """
 def verify_hash(path, expected):
    digest = hashlib.sha256()
    with Path(path).open('rb') as stream:
@@ -206,8 +190,7 @@ if path.exists():
    def input_id(self):
        digest = hashlib.sha256()
-        for directory in ('image', 'patches'):
+        for path in sorted((self.source_dir / 'image').rglob('*')):
            for path in sorted((self.source_dir / directory).rglob('*')):
            if path.is_file():
                digest.update(str(path.relative_to(self.source_dir)).encode())
                digest.update(path.read_bytes())
@@ -226,17 +209,8 @@ if path.exists():
        entry = self.manifest().get(step, {})
        if not entry.get('complete') or entry.get('inputs') != self.input_id():
            return False
        if step in ('rootfs', 'iso'):
            from hydra import image
            if entry.get('source_inputs') != image.source_identity(self.source_dir):
                return False
            if not image.artifacts_valid(self.work / 'artifacts', entry):
                return False
            return self.complete('packages' if step == 'rootfs' else 'rootfs')
        if step == 'packages':
-            try:
+            if not (self.work / 'artifacts/packages.json').is_file():
                self.package_inventory()
            except (BuildError, OSError):
                return False
            return self.complete('prepare')
        if step == 'prepare':
@@ -252,7 +226,8 @@ if path.exists():
        if step == 'fetch':
            inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
            try:
-                for item in [inputs['stage3'], inputs['portage'], *inputs.get('local_packages', [])]:
+                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):
@@ -260,16 +235,6 @@ if path.exists():
            return True
        return False
    def package_inventory(self):
        from hydra import image
        checksum = self.manifest().get('packages', {}).get('inventory_sha256')
        if not isinstance(checksum, str) or not re.fullmatch('[a-f0-9]{64}', checksum):
            raise BuildError('Package inventory has no recorded checksum')
        try:
            return image.read_package_inventory(self.work / 'artifacts/packages.json', checksum)
        except (image.ImageError, OSError) as error:
            raise BuildError(f'Package inventory validation failed: {error}') from error
    def download(self, url, filename=None):
        parsed = urllib.parse.urlparse(url)
        if parsed.scheme != 'https':
@@ -290,8 +255,6 @@ if path.exists():
        verify_hash(archive, stage['sha256'])
        snapshot = self.download(inputs['portage']['url'], inputs['portage']['filename'])
        verify_hash(snapshot, inputs['portage']['sha256'])
        for item in inputs.get('local_packages', []):
            verify_hash(self.download(item['url'], item['filename']), item['sha256'])
        signature = self.download(stage['signature_url'])
        keyring = self.download(inputs['keyring_url'])
        verify_hash(keyring, inputs['keyring_sha256'])
@@ -332,12 +295,6 @@ if [ ! -f /var/db/repos/gentoo/.hydra-snapshot ]; then
  printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot
 fi
 cp -a /hydra-source/image/portage/. /etc/portage/
 mkdir -p /var/db/repos/hydra /var/cache/distfiles
 cp -a /hydra-source/image/overlay/. /var/db/repos/hydra/
 cp /hydra-cache/brightnessctl-0.5.1.tar.gz /var/cache/distfiles/brightnessctl-0.5.1.tar.gz
 mkdir -p /etc/portage/patches/app-emulation/qemu /etc/portage/patches/app-emulation/virt-viewer
 cp /hydra-source/patches/qemu-virtio-refresh-rate.patch /etc/portage/patches/app-emulation/qemu/10-hydra-refresh.patch
 cp /hydra-source/patches/virt-viewer-hydra-session.patch /etc/portage/patches/app-emulation/virt-viewer/10-hydra-session.patch
 if [ -d /etc/portage/gnupg ] && [ ! -s /etc/portage/gnupg/mykeyid ]; then
  mv /etc/portage/gnupg /etc/portage/gnupg.hydra-untrusted
 fi
@@ -354,145 +311,19 @@ chmod a+r /etc/portage/gnupg/pubring.kbx /etc/portage/gnupg/trustdb.gpg
        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',
+        command = ['emerge', '--verbose', '--update', '--newuse', '--deep', '--backtrack=100']
                   '--usepkg-exclude=app-emulation/qemu app-emulation/virt-viewer']
        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')
-        patches = hashlib.sha256()
+        script = package_job_script([*command, *atoms], inputs=inputs)
        for name in ('qemu-virtio-refresh-rate.patch', 'virt-viewer-hydra-session.patch'):
            patches.update((self.source_dir / 'patches' / name).read_bytes())
        patch_id = patches.hexdigest()
        forced = ['emerge', '--verbose', '--oneshot', '--usepkg=n', '--getbinpkg=n',
                  'app-emulation/qemu', 'app-emulation/virt-viewer']
        package_commands = shlex.join([*command, *atoms]) + '\n' + f'''if [ "$(cat /var/lib/hydra-build/patched-source.sha256 2>/dev/null || true)" != {shlex.quote(patch_id)} ]; then
  {shlex.join(forced)}
  printf '%s\\n' {shlex.quote(patch_id)} > /var/lib/hydra-build/patched-source.sha256
 fi'''
        script = package_job_script(['/bin/bash', '-euc', package_commands], 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
-        from hydra import image
+        (self.work / 'artifacts/packages.json').write_text(inventory)
-        try:
+        self.record('packages', inventory='artifacts/packages.json')
            image.validate_package_inventory(json.loads(inventory))
        except (ValueError, image.ImageError) as error:
            raise BuildError(f'Package inventory is malformed: {error}') from error
        target = self.work / 'artifacts/packages.json'
        temporary = target.with_suffix('.tmp')
        temporary.write_text(inventory)
        temporary.replace(target)
        self.record('packages', inventory='artifacts/packages.json', patched_source=patch_id,
                    inventory_sha256=image.digest(target))
    def assembly(self, snapshot_digest=None):
        """A separate rootless container processes an exported copy, never the builder."""
        if snapshot_digest is None:
            state = self.manifest().get('rootfs', {})
            snapshot_digest = state.get('snapshot_sha256')
        if not isinstance(snapshot_digest, str) or not re.fullmatch('[a-f0-9]{64}', snapshot_digest):
            raise BuildError('Verified exported snapshot is required')
        name = self.name + '-image-' + snapshot_digest[:12]
        image = 'localhost/hydra-image:' + snapshot_digest
        exists = self.pod('container', 'exists', name, capture=True, check=False).returncode == 0
        if not exists:
            self.pod('import', str(self.work / 'artifacts/builder-rootfs.tar'), image)
            self.pod('run', '--detach', '--name', name, '--hostname', 'hydra-image-assembler',
                     '--network', 'none', '--cap-add', 'SYS_ADMIN', '--security-opt', 'seccomp=unconfined',
                     '--volume', f'{self.source_dir}:/hydra-source:ro',
                     '--volume', f'{self.work}:/hydra-output:rw', image, '/bin/sleep', 'infinity')
        container = self.inspect('container', name)
        for destination, source, writable in (('/hydra-source', self.source_dir, False),
                                               ('/hydra-output', self.work, True)):
            mounts = [mount for mount in container.get('Mounts', []) if mount.get('Destination') == destination]
            if len(mounts) != 1 or mounts[0].get('Source') != str(source) or mounts[0].get('RW') is not writable:
                raise BuildError('Assembly mount provenance differs; refusing to write')
        provenance = self.inspect('image', container['Image'])
        if provenance.get('RootFS', {}).get('Layers') != ['sha256:' + snapshot_digest]:
            raise BuildError('Assembly image does not match the exported builder snapshot')
        if exists:
            self.pod('start', name)
        return name
    def rootfs(self):
        from hydra import image
        if self.active_package_job():
            raise BuildError('A contained package job is running; wait before export')
        if not self.complete('packages'):
            raise BuildError('Verified completed packages are required before rootfs export')
        if self.complete('rootfs'):
            return
        source_inputs = image.source_identity(self.source_dir)
        self.record('rootfs', complete=False)
        original = self.work / 'artifacts/builder-rootfs.tar'
        if original.is_symlink() or not original.parent.resolve().is_relative_to(self.work):
            raise BuildError('Raw export output must remain inside the dedicated work directory')
        original.touch(mode=0o600, exist_ok=True)
        original.chmod(0o600)  # The unsanitized intermediate can contain builder-private trust.
        self.pod('export', '--output', str(original), self.name)
        snapshot = image.digest(original)
        assembler = self.assembly(snapshot)
        commit = self.run(['git', '-C', str(self.source_dir), 'rev-parse', 'HEAD'], capture=True).stdout.strip()
        script = 'root=/hydra-output/rootfs\n' + rootfs_retry_guard_script() + r"""mkdir -p "$root"
 touch "$root/.hydra-export-owned"
 tar --extract --preserve-permissions --numeric-owner --file /hydra-output/artifacts/builder-rootfs.tar --directory "$root"
 /hydra-source/image/configure-rootfs.sh "$root"
 cp "$(python3 /hydra-source/hydra/image.py kernel-image "$root")" /hydra-output/artifacts/vmlinuz
 cp "$root/boot/initramfs-live.img" /hydra-output/artifacts/initramfs-live.img
 cp "$root/usr/share/hydra/build-manifest.json" /hydra-output/artifacts/rootfs-manifest.json
 # Keep ownership for safe retries outside the shipped payload.
 mksquashfs "$root" /hydra-output/artifacts/squashfs.img -noappend -comp zstd -processors 4 -e boot/initramfs-live.img .hydra-export-owned
 chmod 0644 /hydra-output/artifacts/vmlinuz /hydra-output/artifacts/initramfs-live.img /hydra-output/artifacts/squashfs.img /hydra-output/artifacts/rootfs-manifest.json
 """
        self.pod('exec', '--env', f'HYDRA_SOURCE_COMMIT={commit}', assembler, '/bin/bash', '-euc', script)
        if image.source_identity(self.source_dir) != source_inputs:
            raise BuildError('Source inputs changed during rootfs assembly; completion not recorded')
        manifest = json.loads((self.work / 'artifacts/rootfs-manifest.json').read_text())
        if manifest.get('packages') != self.package_inventory():
            raise BuildError('Payload package inventory differs from the validated package checkpoint')
        if manifest.get('source_files') != image.source_files(self.source_dir):
            raise BuildError('Payload source manifest is incomplete or stale')
        artifacts = {name: image.digest(self.work / 'artifacts' / name) for name in
                     ('vmlinuz', 'initramfs-live.img', 'squashfs.img', 'rootfs-manifest.json', 'packages.json')}
        self.record('rootfs', source_inputs=source_inputs, snapshot_sha256=snapshot,
                    assembler=assembler, kernel=manifest['kernel'], artifacts=artifacts)
    def iso(self):
        from hydra import image
        if not self.complete('rootfs'):
            raise BuildError('Verified rootfs artifacts are required before ISO assembly')
        if self.complete('iso'):
            return
        source_inputs = image.source_identity(self.source_dir)
        assembler = self.assembly()
        self.record('iso', complete=False)
        script = r"""tree=/hydra-output/iso-tree
 [ ! -L "$tree" ] || { echo 'ISO output cannot be a symlink' >&2; exit 1; }
 mkdir -p "$tree/boot/grub" "$tree/LiveOS"
 cp /hydra-output/artifacts/vmlinuz "$tree/boot/vmlinuz"
 cp /hydra-output/artifacts/initramfs-live.img "$tree/boot/initramfs-live.img"
 cp /hydra-output/artifacts/squashfs.img "$tree/LiveOS/squashfs.img"
 cp /hydra-source/image/grub.cfg "$tree/boot/grub/grub.cfg"
 cp /hydra-output/artifacts/rootfs-manifest.json "$tree/build-manifest.json"
 cp /hydra-output/artifacts/packages.json "$tree/packages.json"
 grub-mkrescue --directory=/usr/lib/grub/x86_64-efi --output=/hydra-output/artifacts/hydra-os.iso "$tree" -- -volid HYDRA_OS
 xorriso -indev /hydra-output/artifacts/hydra-os.iso -report_el_torito plain -pvd_info > /hydra-output/artifacts/iso-inspection.txt 2>&1
 chmod 0644 /hydra-output/artifacts/hydra-os.iso /hydra-output/artifacts/iso-inspection.txt
 """
        self.pod('exec', assembler, '/bin/bash', '-euc', script)
        inspection = (self.work / 'artifacts/iso-inspection.txt').read_text()
        if 'HYDRA_OS' not in inspection or not re.search(r'El Torito boot img.*UEFI', inspection):
            raise BuildError('ISO does not report the expected volume label and UEFI boot image')
        if image.source_identity(self.source_dir) != source_inputs:
            raise BuildError('Source inputs changed during ISO assembly; completion not recorded')
        checksum = image.digest(self.work / 'artifacts/hydra-os.iso')
        (self.work / 'artifacts/hydra-os.iso.sha256').write_text(checksum + '  hydra-os.iso\n')
        artifacts = {name: image.digest(self.work / 'artifacts' / name) for name in
                     ('hydra-os.iso', 'hydra-os.iso.sha256', 'iso-inspection.txt', 'packages.json')}
        self.record('iso', source_inputs=source_inputs, label=image.LABEL, artifacts=artifacts)
    def status(self):
        data = self.manifest()
@@ -549,6 +380,8 @@ def main(argv=None):
            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:
@@ -7,7 +7,7 @@ import subprocess
 import sys
 from pathlib import Path
-from .profiles import Profile, load_defaults
+from .profiles import Profile
 from .runtime import Runtime, RuntimeError
 from .session import launch_session, session_helper, view
@@ -17,26 +17,16 @@ def parser() -> argparse.ArgumentParser:
    cli.add_argument('--config', type=Path, default=Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'hydra')
    cli.add_argument('--state', type=Path, default=Path(os.environ.get('XDG_STATE_HOME', Path.home() / '.local/state')) / 'hydra')
    cli.add_argument('--runtime', type=Path, default=Path(os.environ.get('XDG_RUNTIME_DIR', f'/tmp/hydra-{os.getuid()}')) / 'hydra')
    cli.add_argument('--defaults', type=Path, default=Path('/etc/hydra/defaults.toml'),
                     help='Distribution preferences for newly created profiles')
    commands = cli.add_subparsers(dest='command', required=True)
    add = commands.add_parser('add', help='Create a VM profile without starting it')
    add.add_argument('name')
-    add.add_argument('--iso', type=Path)
+    add.add_argument('--iso', type=Path, required=True)
    add.add_argument('--memory', type=int, default=4096)
    add.add_argument('--cpus', type=int, default=2)
    add.add_argument('--disk-size', default='40G')
    add.add_argument('--graphics', choices=('auto', 'software'), default='auto')
-    add.add_argument('--network', choices=('user', 'bridge'))
+    add.add_argument('--network', choices=('user', 'bridge'), default='user')
-    add.add_argument('--bridge')
+    add.add_argument('--bridge', default='hydrabr0')
    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='?')
@@ -67,18 +57,12 @@ def main(argv: list[str] | None = None) -> int:
        elif args.command == '_session-helper':
            return session_helper(args.operation, args.manifest)
        elif args.command == 'add':
            preferences = load_defaults(args.defaults)
            preferences.update({key: getattr(args, key) for key in
                                ('network', 'bridge', 'width', 'height', 'refresh_millihz')
                                if getattr(args, key) is not None})
            profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size,
-                              args.graphics, **preferences)
+                              args.graphics, args.network, args.bridge)
            runtime.add(profile)
            result = {'name': profile.name, 'status': 'added'}
        elif args.command == 'media':
            result = runtime.media(args.name, None if args.detach else args.iso)
        elif args.command == 'list':
-            result = [{'name': profile.name, 'iso': str(profile.iso) if profile.iso else None} for profile in runtime.profiles()]
+            result = [{'name': profile.name, 'iso': str(profile.iso)} for profile in runtime.profiles()]
        elif args.command == 'status':
            result = runtime.status(args.name) if args.name else [runtime.status(p.name) for p in runtime.profiles()]
        elif args.command == 'stop':
@@ -1,404 +0,0 @@
 """Prepare only exported Gentoo payloads and configure the dedicated guest network."""
 import argparse
 import bz2
 import io
 import runpy
 import hashlib
 import ipaddress
 import json
 import os
 from pathlib import Path
 import re
 import shutil
 import subprocess
 import sys
 LABEL = 'HYDRA_OS'
 BRIDGE = 'hydrabr0'
 SUBNET = ipaddress.ip_network('10.203.0.0/24')
 CONNECTION_UUID = 'c84844bd-2955-4a61-9463-5e540140f9a2'
 class ImageError(RuntimeError):
    pass
 def root_path(root, relative, follow_leaf=True):
    root = Path(root).resolve()
    if root == Path('/') or not root.is_dir():
        raise ImageError('Expected an isolated exported rootfs directory')
    relative = Path(relative)
    if relative.is_absolute() or '..' in relative.parts:
        raise ImageError('Payload path must be relative and remain inside the export')
    path = root / relative
    checked = path if follow_leaf else path.parent
    if not checked.resolve().is_relative_to(root):
        raise ImageError(f'Payload symlink escapes the export: {relative}')
    return path
 def remove(root, relative):
    path = root_path(root, relative, follow_leaf=False)
    if path.is_symlink() or path.is_file():
        path.unlink()
    elif path.is_dir():
        shutil.rmtree(path)
 def write(root, relative, content, mode=0o644):
    path = root_path(root, relative)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    path.chmod(mode)
    return path
 def sanitize_saved_environments(root):
    """Use Portage's parser to scrub exported declarations without evaluating them."""
    database = root_path(root, 'var/db/pkg')
    paths = sorted(database.glob('*/*/environment.bz2'))
    result = {'files_scanned': len(paths), 'filtered_files': 0, 'variable_names': []}
    if not paths:
        return result
    scripts = sorted(root_path(root, 'usr/lib/portage').glob('*/filter-bash-environment.py'))
    if not scripts:
        raise ImageError('Installed Portage environment filter is required for export')
    script = root_path(root, scripts[0].relative_to(root))
    api = runpy.run_path(str(script))
    pattern = re.compile(rb'(?i:(?:HTTP|HTTPS|FTP|ALL|NO)_PROXY|.*(?:PASSWORD|PASSWD|SECRET|TOKEN|API_KEY|ACCESS_KEY|CREDENTIAL|AUTHORIZATION).*)')
    declarations = re.compile(rb'^(?:declare|export|readonly)\s+(?:-[^ ]+\s+)*([A-Za-z_][A-Za-z0-9_]*)=', re.M)
    names = set()
    for path in paths:
        path = root_path(root, path.relative_to(root))
        try:
            original = bz2.decompress(path.read_bytes())
        except (OSError, ValueError) as error:
            raise ImageError('Cannot read an exported package environment') from error
        filtered = io.BytesIO()
        api['filter_bash_environment'](pattern, io.BytesIO(original), filtered)
        clean = filtered.getvalue()
        if clean != original:
            names.update(name.decode('ascii') for name in declarations.findall(original)
                         if pattern.fullmatch(name))
            path.write_bytes(bz2.compress(clean))
            result['filtered_files'] += 1
    result['variable_names'] = sorted(names)
    return result
 def sanitize_rootfs(root):
    """Erase builder identity/secrets in the exported copy, retaining Gentoo updates."""
    for relative in ('etc/portage/gnupg', 'etc/portage/gnupg.hydra-untrusted',
                     'var/lib/hydra-build', 'root', 'home', 'var/cache', 'var/tmp',
                     'tmp', 'run', 'var/log', 'var/lib/systemd/random-seed',
                     'var/lib/NetworkManager', 'etc/NetworkManager/system-connections',
                     'hydra-source', 'hydra-cache', 'hydra-output'):
        remove(root, relative)
    for name in root_path(root, 'etc/ssh').glob('ssh_host_*'):
        remove(root, str(name.relative_to(root)))
    for relative, mode in [('root', 0o700), ('home', 0o755), ('tmp', 0o1777),
                           ('var/tmp', 0o1777), ('run', 0o755), ('var/log', 0o755),
                           ('var/cache', 0o755), ('dev', 0o755), ('proc', 0o755), ('sys', 0o755)]:
        path = root_path(root, relative)
        path.mkdir(parents=True, exist_ok=True)
        path.chmod(mode)
    remove(root, 'etc/machine-id')
    write(root, 'etc/machine-id', '')
    remove(root, 'var/lib/dbus/machine-id')
    path = root_path(root, 'var/lib/dbus/machine-id', follow_leaf=False)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.symlink_to('/etc/machine-id')
    write(root, 'etc/hostname', 'hydra-os\n')
    write(root, 'etc/hosts', '127.0.0.1 localhost\n::1 localhost\n127.0.1.1 hydra-os\n')
    remove(root, 'etc/resolv.conf')
    root_path(root, 'etc/resolv.conf', follow_leaf=False).symlink_to('/run/NetworkManager/resolv.conf')
    audit = sanitize_saved_environments(root)
    write(root, 'usr/share/hydra/environment-sanitization.json', json.dumps(audit, indent=2) + '\n')
 def copy_overlay(root, overlay):
    for source in sorted(Path(overlay).rglob('*')):
        destination = root_path(root, source.relative_to(overlay))
        if source.is_symlink():
            raise ImageError('Source overlay must contain ordinary directories/files')
        if source.is_dir():
            destination.mkdir(parents=True, exist_ok=True)
        elif source.is_file():
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(source, destination)
            destination.chmod(source.stat().st_mode & 0o777)
 def _kernel_version(root):
    modules = root_path(root, 'lib/modules')
    versions = sorted(path.name for path in modules.iterdir() if path.is_dir()) if modules.is_dir() else []
    if len(versions) != 1 or not re.fullmatch(r'[A-Za-z0-9_.+-]+', versions[0]):
        raise ImageError('Exactly one installed payload kernel is required')
    return versions[0]
 def kernel_image(root):
    version = _kernel_version(root)
    candidates = [root_path(root, 'boot/' + prefix + version)
                  for prefix in ('kernel-', 'vmlinuz-')]
    images = [path for path in candidates if path.is_file()]
    if len(images) == 2 and images[0].samefile(images[1]):
        return images[1]  # Stable installer/GRUB vmlinuz alias for Gentoo kernel-bin.
    if len(images) != 1:
        raise ImageError('Installed modules require exactly one matching boot kernel image')
    return images[0]
 def kernel_version(root):
    kernel_image(root)
    return _kernel_version(root)
 def live_kernel_args():
    return [f'root=live:CDLABEL={LABEL}', 'rd.live.image', 'rd.live.overlay.overlayfs=1',
            'ro', 'console=tty0', 'console=ttyS0,115200n8']
 def source_files(source):
    source = Path(source)
    result = {}
    for directory in ('image', 'hydra', 'bin', 'session', 'patches'):
        for path in sorted((source / directory).rglob('*')):
            if path.is_file() and '__pycache__' not in path.parts and path.suffix != '.pyc':
                result[str(path.relative_to(source))] = digest(path)
    return result
 def source_identity(source):
    return hashlib.sha256(json.dumps(source_files(source), sort_keys=True).encode()).hexdigest()
 def payload_manifest(root, source):
    packages = []
    database = root_path(root, 'var/db/pkg')
    for use in sorted(database.glob('*/*/USE')):
        package = use.parent
        built = package / 'BUILD_TIME'
        packages.append({'package': str(package.relative_to(database)), 'use': use.read_text().split(),
                         'build_time': built.read_text().strip() if built.is_file() else None})
    commit = os.environ.get('HYDRA_SOURCE_COMMIT')
    if not commit:
        git = subprocess.run(['git', '-C', str(source), 'rev-parse', 'HEAD'], text=True, capture_output=True)
        commit = git.stdout.strip() if git.returncode == 0 else None
    return {'format': 1, 'label': LABEL, 'kernel': kernel_version(root),
            'live_kernel_args': live_kernel_args(), 'source_commit': commit,
            'source_files': source_files(source), 'sources': json.loads((Path(source) / 'image/sources.json').read_text()),
            'packages': packages}
 def artifacts_valid(directory, state):
    artifacts = state.get('artifacts')
    if not isinstance(artifacts, dict) or not artifacts:
        return False
    for name, expected in artifacts.items():
        if not isinstance(name, str) or Path(name).name != name or not isinstance(expected, str):
            return False
        path = Path(directory) / name
        try:
            if path.is_symlink() or not path.is_file() or digest(path) != expected:
                return False
        except OSError:
            return False
    return True
 def validate_package_inventory(data):
    if not isinstance(data, list) or not data:
        raise ImageError('Package inventory must be a nonempty array')
    seen = set()
    for record in data:
        if not isinstance(record, dict) or set(record) != {'package', 'use', 'build_time'}:
            raise ImageError('Malformed package inventory entry')
        package = record['package']
        if (not isinstance(package, str) or
            not re.fullmatch(r'[A-Za-z0-9+_.-]+/[A-Za-z0-9+_.-]+-[0-9][A-Za-z0-9+_.-]*', package) or
            package in seen):
            raise ImageError('Invalid or duplicate package inventory identity')
        seen.add(package)
        flags = record['use']
        if (not isinstance(flags, list) or any(not isinstance(flag, str) or
            not re.fullmatch(r'[A-Za-z0-9+_.@-]+', flag) for flag in flags) or len(set(flags)) != len(flags)):
            raise ImageError('Invalid package inventory USE flags')
        built = record['build_time']
        if built is not None and (not isinstance(built, str) or not re.fullmatch(r'[0-9]+', built)):
            raise ImageError('Invalid package inventory build time')
    return data
 def read_package_inventory(path, expected_digest=None):
    path = Path(path)
    if path.is_symlink() or not path.is_file():
        raise ImageError('Package inventory must be an ordinary file')
    if expected_digest is not None and digest(path) != expected_digest:
        raise ImageError('Package inventory checksum differs from checkpoint')
    try:
        return validate_package_inventory(json.loads(path.read_text()))
    except (ValueError, UnicodeError) as error:
        raise ImageError('Package inventory is not valid JSON') from error
 def enable_user_audio(root):
    root_path(root, 'etc/systemd/user')
    result = subprocess.run(['systemctl', '--root=' + str(root), '--global', 'enable',
                             'pipewire.socket', 'pipewire-pulse.socket', 'wireplumber.service'],
                            text=True, capture_output=True)
    if result.returncode:
        raise ImageError('Cannot enable user audio sockets: ' + result.stderr.strip())
 def configure_rootfs(root, source):
    root, source = Path(root), Path(source)
    # Refuse incomplete integrations before changing the exported copy.
    for name in ('bin/hydra', 'bin/hydra-install', 'hydra/session.py', 'hydra/installer.py'):
        if not (source / name).is_file():
            raise ImageError(f'Required committed runtime/installer asset is missing: {name}')
    version = kernel_version(root)
    kernel = kernel_image(root)
    alias = root_path(root, 'boot/vmlinuz-' + version)
    if not alias.exists():
        alias.symlink_to(kernel.name)
    sanitize_rootfs(root)
    copy_overlay(root, source / 'image/rootfs')
    write(root, 'etc/hydra/live-mode/dracut.conf', (source / 'image/dracut.conf').read_text())
    root_path(root, 'etc/hydra/live-mode/dracut.conf.d').mkdir()
    package = root_path(root, 'usr/lib/hydra/hydra')
    package.mkdir(parents=True, exist_ok=True)
    for module in sorted((source / 'hydra').glob('*.py')):
        shutil.copyfile(module, root_path(root, 'usr/lib/hydra/hydra/' + module.name))
    # Session helpers derive this checkout-compatible launcher from __file__.
    for name in ('hydra', 'hydra-install'):
        write(root, 'usr/lib/hydra/bin/' + name, (source / 'bin' / name).read_text(), 0o755)
    for name, module in (('hydra', 'cli'), ('hydra-install', 'installer')):
        write(root, 'usr/bin/' + name,
              '#!/usr/bin/python3\nimport sys\nsys.path.insert(0, "/usr/lib/hydra")\n'
              f'from hydra.{module} import main\nraise SystemExit(main())\n', 0o755)
    write(root, 'usr/bin/hydra-session', '#!/bin/sh\nexec /usr/bin/hydra session "$@"\n', 0o755)
    write(root, 'usr/bin/hydra-network', '#!/usr/bin/python3\nimport sys\nsys.path.insert(0, "/usr/lib/hydra")\n'
          'from hydra.image import network_main\nraise SystemExit(network_main())\n', 0o755)
    write(root, 'usr/share/hydra/build-manifest.json', json.dumps(payload_manifest(root, source), indent=2) + '\n')
    make = root_path(root, 'etc/portage/make.conf')
    if make.is_file():
        settings = make.read_text()
        settings = re.sub(r'(?<=[\s"\'])-(?:usersandbox|sandbox|network-sandbox|ipc-sandbox|pid-sandbox)(?=[\s"\']|$)', '', settings)
        make.write_text(settings)
    for path in root_path(root, 'etc/sudoers.d').iterdir():
        path.chmod(0o440)
    return version
 def check_network_conflicts(routes, addresses):
    try:
        if not isinstance(routes, list) or not isinstance(addresses, list):
            raise ValueError('Expected JSON arrays')
        for route in routes:
            dst = route.get('dst')
            if dst in ('default', None):
                continue
            network = ipaddress.ip_network(dst, strict=False)
            if network.version != 4:
                raise ValueError('Unexpected non-IPv4 route')
            if network.overlaps(SUBNET) and not (route.get('dev') == BRIDGE and network.subnet_of(SUBNET)):
                raise ImageError(f'Guest subnet {SUBNET} conflicts with route {dst} on {route.get("dev")}')
        for interface in addresses:
            for addr in interface['addr_info']:
                if addr.get('family') != 'inet':
                    continue
                network = ipaddress.ip_network(f'{addr["local"]}/{addr["prefixlen"]}', strict=False)
                owned = interface['ifname'] == BRIDGE and addr['local'] == '10.203.0.1' and addr['prefixlen'] == 24
                if network.overlaps(SUBNET) and not owned:
                    raise ImageError(f'Guest subnet {SUBNET} conflicts with address on {interface["ifname"]}')
    except (ValueError, KeyError, TypeError, AttributeError) as error:
        raise ImageError('Cannot safely interpret IPv4 routes/addresses') from error
 def configure_network(run=subprocess.run):
    def command(argv, **kwargs):
        result = run(argv, text=True, capture_output=True, **kwargs)
        if result.returncode:
            raise ImageError(f'Network command failed: {argv[0]}: {result.stderr.strip()}')
        return result.stdout
    routes = json.loads(command(['ip', '-json', '-4', 'route', 'show', 'table', 'all']))
    addresses = json.loads(command(['ip', '-json', '-4', 'address', 'show']))
    check_network_conflicts(routes, addresses)
    probe = run(['nmcli', '--get-values', 'connection.uuid,connection.type,connection.interface-name,ipv4.method,ipv4.addresses',
                 'connection', 'show', 'uuid', CONNECTION_UUID], text=True, capture_output=True)
    exists = probe.returncode == 0
    if exists and probe.stdout.strip().splitlines() != [CONNECTION_UUID, 'bridge', BRIDGE, 'manual', '10.203.0.1/24']:
        raise ImageError('Hydra bridge connection exists with unexpected configuration')
    if not exists and probe.returncode != 10:
        raise ImageError('Cannot query NetworkManager connections')
    links = json.loads(command(['ip', '-json', '-details', 'link', 'show']))
    for link in links:
        if link.get('ifname') == BRIDGE and (not exists or link.get('linkinfo', {}).get('info_kind') != 'bridge'):
            raise ImageError('Refusing to replace an existing interface named hydrabr0')
    if not exists:
        command(['nmcli', 'connection', 'add', 'type', 'bridge', 'con-name', 'hydra-guest',
                 'ifname', BRIDGE, 'connection.uuid', CONNECTION_UUID,
                 'connection.autoconnect', 'no', 'bridge.stp', 'no',
                 'ipv4.method', 'manual', 'ipv4.addresses', '10.203.0.1/24',
                 'ipv4.never-default', 'yes', 'ipv6.method', 'disabled'])
    # Activation is always gated by the fresh route/address check, including restarts.
    command(['nmcli', 'connection', 'up', 'uuid', CONNECTION_UUID])
    command(['sysctl', '-w', 'net.ipv4.ip_forward=1'])
    table = run(['nft', 'list', 'table', 'ip', 'hydra_nat'], text=True, capture_output=True)
    rules = ('delete table ip hydra_nat\n' if table.returncode == 0 else '') + '''table ip hydra_nat {
  chain postrouting {
    type nat hook postrouting priority srcnat; policy accept;
    ip saddr 10.203.0.0/24 oifname != "hydrabr0" masquerade
  }
 }
 '''
    command(['nft', '-f', '-'], input=rules)
 def network_main():
    try:
        configure_network()
    except (ImageError, OSError, ValueError) as error:
        print(f'hydra-network: {error}', file=sys.stderr)
        return 1
    return 0
 def digest(path):
    value = hashlib.sha256()
    with Path(path).open('rb') as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b''):
            value.update(block)
    return value.hexdigest()
 def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('step', choices=('configure', 'kernel', 'kernel-image', 'network', 'audio'))
    parser.add_argument('root', nargs='?', type=Path)
    parser.add_argument('--source', type=Path)
    options = parser.parse_args(argv)
    try:
        if options.step == 'network':
            return network_main()
        if options.root is None:
            raise ImageError('An isolated rootfs directory is required')
        if options.step == 'kernel':
            print(kernel_version(options.root))
        elif options.step == 'kernel-image':
            print(kernel_image(options.root))
        elif options.step == 'audio':
            enable_user_audio(options.root)
        else:
            if options.source is None:
                raise ImageError('Source checkout is required')
            configure_rootfs(options.root, options.source)
    except (ImageError, OSError, ValueError) as error:
        print(f'hydra-image: {error}', file=sys.stderr)
        return 1
    return 0
 if __name__ == '__main__':
    raise SystemExit(main())
@@ -10,7 +10,6 @@ from __future__ import annotations
 import argparse
 from dataclasses import dataclass
 import getpass
 import fcntl
 import json
 import os
 from pathlib import Path, PurePosixPath
@@ -18,7 +17,6 @@ import re
 import shlex
 import shutil
 import stat
 import struct
 import subprocess
 import sys
 import tempfile
@@ -74,7 +72,6 @@ class Topology:
    def __init__(self):
        self.nodes = {}
        self.edges = {}
        self.parents = {}
        self.protected = set()
    @classmethod
@@ -113,7 +110,6 @@ class Topology:
                raise InstallError('Inconsistent duplicate lsblk node')
            result.nodes[path] = node
            result.edges.setdefault(path, set())
            result.parents.setdefault(path, set())
            if any(m for m in mounts):
                result.protected.add(path)
            pkname = node.get('pkname')
@@ -170,17 +166,6 @@ class Topology:
    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')
        # Backing links discovered after lsblk must obey the same directed
        # ancestry rule. An undirected shared disk is not itself a cycle.
        seen, pending = set(), [parent]
        while pending:
            current = pending.pop()
            if current == child:
                raise InstallError('Cyclic block-device backing ancestry')
            if current not in seen:
                seen.add(current)
                pending.extend(self.parents[current] - seen)
        self.parents[child].add(parent)
        self.edges[child].add(parent)
        self.edges[parent].add(child)
@@ -253,6 +238,21 @@ class SystemProbe:
            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)
@@ -260,52 +260,8 @@ class SystemProbe:
            # 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)
        # Loops can hide further loops, including ancestors of a mounted
        # partition or device mapper. Resolve protected components to a fixed
        # point, independent of lsblk ordering and payload discovery order.
        inspected = set()
        while True:
            protected_devices = set()
            for path in topo.protected:
                protected_devices.update(topo.component(path))
            pending = [path for path in topo.nodes if path in protected_devices
                       and topo.nodes[path]['type'] == 'loop' and path not in inspected]
            if not pending:
                break
            for path in pending:
                identity = self._loop_backing_identity(path, topo.nodes[path]['maj:min'])
                sources = [p for p, n in topo.nodes.items() if n['maj:min'] == identity]
                if len(sources) != 1:
                    raise InstallError(f'Unknown physical backing for mounted loop: {path} ({identity})')
                source = sources[0]
                topo.link(path, source)
                topo.protect(source)
                inspected.add(path)
        return topo
    @staticmethod
    def _loop_backing_identity(path, expected_identity):
        # LOOP_GET_STATUS64 reports the backing inode's filesystem device,
        # independent of names which dracut switch_root may have relocated.
        # lo_rdevice is nonzero when the backing inode is itself a block device.
        try:
            descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC)
            try:
                info = os.fstat(descriptor)
                identity = f'{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}'
                if not stat.S_ISBLK(info.st_mode) or identity != expected_identity:
                    raise InstallError(f'Mounted loop device identity changed: {path}')
                status = fcntl.ioctl(descriptor, 0x4C05, bytes(232))
                if not isinstance(status, bytes) or len(status) != 232:
                    raise InstallError(f'Malformed loop status: {path}')
                filesystem_device, _, block_device = struct.unpack_from('=QQQ', status)
            finally:
                os.close(descriptor)
        except (OSError, struct.error) as exc:
            raise InstallError(f'Cannot inspect mounted loop backing identity: {path}') from exc
        device = block_device or filesystem_device
        return f'{os.major(device)}:{os.minor(device)}'
    @staticmethod
    def _mounts(raw):
        try:
@@ -16,40 +16,24 @@ def validate_name(name: str) -> str:
@dataclass(frozen=True)
 class Profile:
    name: str
-    iso: Path | None = None
+    iso: Path
    memory: int = 4096
    cpus: int = 2
    disk_size: str = '40G'
    graphics: str = 'auto'
    network: str = 'user'
    bridge: str = 'hydrabr0'
    width: int = 1280
    height: int = 800
    refresh_millihz: int = 0
    mac: str | None = None
    def __post_init__(self):
        validate_name(self.name)
-        if self.iso is not None:
+        iso = Path(self.iso).expanduser().resolve()
-            if not isinstance(self.iso, (str, Path)) or not str(self.iso):
+        if not iso.is_file():
-                raise ValueError('ISO must be a nonempty path or omitted')
+            raise ValueError(f'ISO does not exist or is not a file: {iso}')
-            object.__setattr__(self, 'iso', Path(self.iso).expanduser().resolve())
+        object.__setattr__(self, 'iso', iso)
        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'):
@@ -83,22 +67,6 @@ def save_profile(path: Path, profile: Profile) -> None:
    lines = []
    for field in Profile.__dataclass_fields__:
        value = getattr(profile, field)
        if value is None:
            continue
        lines.append(f'{field} = {value if type(value) is int else json.dumps(str(value), ensure_ascii=False)}\n')
    with path.open('x', encoding='utf-8') as stream:
        stream.write(''.join(lines))
 def load_defaults(path: Path) -> dict:
    """Distribution preferences apply only when creating a new profile."""
    try:
        with Path(path).open('rb') as stream:
            data = tomllib.load(stream)
    except FileNotFoundError:
        return {}
    unknown = data.keys() - {'network', 'bridge', 'width', 'height', 'refresh_millihz'}
    if unknown:
        raise ValueError(f'Unknown defaults fields: {", ".join(sorted(unknown))}')
    Profile('defaults-check', **data)
    return data
@@ -3,7 +3,6 @@
 import contextlib
 import errno
 import fcntl
 import hashlib
 import json
 import os
 import socket
@@ -12,7 +11,7 @@ import subprocess
 import tempfile
 import time
 import uuid
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from pathlib import Path
 from .profiles import Profile, load_profile, save_profile, validate_name
@@ -55,14 +54,6 @@ def _intel_render_node() -> Path | None:
    return None
 def _stable_mac(paths: VMPaths) -> str:
    # Disk/state identity survives restarts and runtime socket-directory changes.
    # UID distinguishes owners even when a shared bridge sees the same names.
    digest = bytearray(hashlib.sha256(f'{os.getuid()}:{paths.disk}'.encode()).digest()[:6])
    digest[0] = (digest[0] & 0xfc) | 0x02
    return ':'.join(f'{octet:02x}' for octet in digest)
 def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
    accel = 'kvm' if os.access('/dev/kvm', os.R_OK | os.W_OK) else 'tcg'
    argv = [os.environ.get('HYDRA_QEMU', 'qemu-system-x86_64'), '-name', profile.name,
@@ -73,21 +64,15 @@ def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
            '-mon', 'chardev=qmp,mode=control']
    if accel == 'kvm':
        argv += ['-cpu', 'host']
-    blockdevs = [
+    for options in (
        {'driver': 'file', 'filename': str(paths.disk), 'node-name': 'disk-file'},
        {'driver': 'qcow2', 'file': 'disk-file', 'node-name': 'disk'},
    ]
    if profile.iso is not None:
        blockdevs += [
        {'driver': 'file', 'filename': str(profile.iso), 'node-name': 'iso-file', 'read-only': True},
        {'driver': 'raw', 'file': 'iso-file', 'node-name': 'iso', 'read-only': True},
-        ]
+    ):
    for options in blockdevs:
        argv += ['-blockdev', json.dumps(options)]
-    argv += ['-device', 'virtio-blk-pci,drive=disk']
+    argv += ['-device', 'virtio-blk-pci,drive=disk', '-device', 'ide-cd,drive=iso',
-    if profile.iso is not None:
+             '-boot', 'order=dc', '-device', 'virtio-net-pci,netdev=net',
        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']
@@ -98,15 +83,12 @@ def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
                 '-device', f'usb-redir,chardev=usbredir{channel}']
    spice = (f'unix=on,addr={_qemu_value(paths.spice)},disable-ticketing=on,'
             'disable-copy-paste=on,disable-agent-file-xfer=on')
    mode = f',xres={profile.width},yres={profile.height}'
    if profile.refresh_millihz:
        mode += f',refresh_rate={profile.refresh_millihz}'
    render = _intel_render_node() if profile.graphics == 'auto' else None
    if render:
        spice += f',gl=on,rendernode={_qemu_value(render)}'
-        argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl' + mode]
+        argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl']
    else:
-        argv += ['-display', 'none', '-device', 'virtio-vga' + mode]
+        argv += ['-display', 'none', '-device', 'virtio-vga']
    return argv + ['-spice', spice]
@@ -143,35 +125,9 @@ class Runtime:
            _private_dir(directory)
    def add(self, profile: Profile) -> None:
        self._validate_media(profile.iso)
        _private_dir(self.config)
        if profile.mac is None:
            profile = replace(profile, mac=_stable_mac(self.paths(profile.name)))
        save_profile(self.config / f'{profile.name}.toml', profile)
    @staticmethod
    def _validate_media(iso: Path | None) -> None:
        if iso is not None and not iso.is_file():
            raise RuntimeError(f'ISO does not exist or is not a file: {iso}; attach valid media or detach it while stopped')
    def media(self, name: str, iso: Path | None) -> dict:
        paths = self.paths(name)
        with self._locked(paths):
            current = self.status(name)
            if current['status'] != 'stopped':
                raise RuntimeError(f'Cannot change media for {name}: {current["status"]}')
            profile = replace(self.profile(name), iso=iso)
            self._validate_media(profile.iso)
            destination = self.config / f'{name}.toml'
            if destination.is_symlink():
                raise RuntimeError('Refusing to replace a symlink profile')
            with tempfile.TemporaryDirectory(prefix=f'.{name}-', dir=self.config) as tmp:
                temp = Path(tmp) / f'{name}.toml'
                save_profile(temp, profile)
                temp.chmod(0o600)
                os.replace(temp, destination)
        return {'name': name, 'iso': str(profile.iso) if profile.iso else None, 'status': 'media-updated'}
    def profiles(self) -> list[Profile]:
        if not self.config.exists():
            return []
@@ -278,13 +234,12 @@ class Runtime:
            temp.unlink(missing_ok=True)
    def start(self, name: str) -> dict:
        profile = self.profile(name)
        paths = self.paths(name)
        with self._locked(paths):
            profile = self.profile(name)
            current = self.status(name)
            if current['status'] != 'stopped':
                raise RuntimeError(f'Cannot start {name}: {current["status"]}')
            self._validate_media(profile.iso)
            for path in (paths.qmp, paths.spice):
                if self._listening(path):
                    raise RuntimeError(f'Runtime socket is already in use: {path}')
@@ -49,7 +49,6 @@ def build_labwc_config(names: list[str], recovery_argv: list[str],
    core = ET.SubElement(root, 'core')
    ET.SubElement(core, 'primarySelection').text = 'no'
    desktops = ET.SubElement(root, 'desktops', number=str(max(1, len(names))))
    if names:
    labels = ET.SubElement(desktops, 'names')
    for name in names:
        ET.SubElement(labels, 'name').text = name
@@ -174,10 +173,9 @@ 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')
    _private_dir(runtime.runtime)
    while True:
    profiles = runtime.profiles()
    initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None)
    _private_dir(runtime.runtime)
    with tempfile.TemporaryDirectory(prefix='session-', dir=runtime.runtime) as tmp:
        directory = Path(tmp)
        config = directory / 'labwc'
@@ -190,29 +188,10 @@ def launch_session(runtime: Runtime, *, nested=False) -> int:
                                    _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'
-            compositor = subprocess.Popen([labwc, '-C', str(config)], env=env)
+        return subprocess.run([labwc, '-C', str(config)], env=env).returncode
            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:
@@ -224,22 +203,14 @@ def session_helper(command: str, manifest: Path) -> int:
        raise SessionError('Invalid session manifest')
    for name in data['names']:
        validate_name(name)
-    if command not in ('clients', 'recover', 'refresh') and (
+    if command not in ('clients', 'recover') and (
        not command.startswith('reconnect:') or command.split(':', 1)[1] not in data['names']):
        raise SessionError('Invalid session helper operation')
    runtime = Runtime(Path(data['config']), Path(data['state']), Path(data['runtime']))
    if command == 'refresh':
        fd = os.open(manifest.parent / 'refresh.request', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        os.close(fd)
        print('Refreshing workspaces and viewers; VMs keep running.')
        return 0
    if command == 'recover':
        print('HydraOS local recovery. VMs keep running independently.\n'
              'Ctrl+Alt+Shift+Left/Right switches workspaces; Ctrl+Alt+Shift+R releases input; Ctrl+Alt+Shift+U opens USB.\n'
-              'Enter a VM name to reconnect, :shell for a host shell, :refresh to rebuild workspaces/viewers, or :quit.')
+              'Enter a VM name to reconnect, :shell for a host shell, :quit to close this terminal.')
        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))
@@ -248,14 +219,12 @@ def session_helper(command: str, manifest: Path) -> int:
                return 0
            if choice == ':quit':
                return 0
            if choice == ':refresh':
                return session_helper('refresh', manifest)
            if choice == ':shell':
                subprocess.run([os.environ.get('SHELL', '/bin/sh')])
            elif choice in data['names']:
                subprocess.Popen(_module_argv('reconnect:' + choice, manifest), start_new_session=True)
            else:
-                print('Enter a configured VM name, :shell, :refresh, or :quit.')
+                print('Enter a configured VM name, :shell, or :quit.')
    if command.startswith('reconnect:'):
        name = command.split(':', 1)[1]
        # One viewer per VM inside this session; independent host view commands
@@ -8,8 +8,6 @@ scripts/hydra-build fetch
 scripts/hydra-build prepare
 scripts/hydra-build packages --pretend
 scripts/hydra-build packages
 scripts/hydra-build rootfs
 scripts/hydra-build iso
 scripts/hydra-build status
 ```
@@ -36,12 +34,5 @@ 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` exports the dedicated builder into a private intermediate and assembles
+`rootfs` and `iso` are reserved assembly commands and currently return an
-an isolated sanitized payload, generic live initramfs and squashfs. `iso` checks
+explicit error until the live-image assembly implementation is integrated.
 those artifacts and emits the UEFI `HYDRA_OS` ISO, SHA256 and build manifest in
 `build/artifacts/`. The working builder is retained for package retries. The
 installed Gentoo repository uses the standard rsync backend with signed
 metamanifest verification; the builder continues to use its pinned snapshot.
 QEMU and remote-viewer retain local patches and are excluded from upstream
 binary packages on installed updates. Disposable boot/install validation is
 required before treating an emitted ISO as release ready.
@@ -1,50 +0,0 @@
 #!/bin/bash
 set -euo pipefail
 if [[ $# != 1 || "$1" != /hydra-output/rootfs ]]; then
    echo "configure-rootfs: expected the isolated /hydra-output/rootfs export" >&2
    exit 1
 fi
 root=$1
 python3 /hydra-source/hydra/image.py configure "$root" --source /hydra-source
 # This is a separate rootfs inside the assembly container, never the host root.
 mounted=()
 cleanup_mounts() {
    status=$?
    trap - EXIT
    cleanup_failed=0
    for ((index=${#mounted[@]}-1; index>=0; index--)); do
        umount "${mounted[index]}" || cleanup_failed=1
    done
    if ((cleanup_failed)); then
        echo 'configure-rootfs: temporary mount cleanup failed; refusing success' >&2
        status=1
    fi
    exit "$status"
 }
 trap cleanup_mounts EXIT
 mount --bind /dev "$root/dev"
 mounted+=("$root/dev")
 mount -t proc proc "$root/proc"
 mounted+=("$root/proc")
 mount --bind /sys "$root/sys"
 mounted+=("$root/sys")
 for group in wheel kvm video render audio; do
    chroot "$root" getent group "$group" >/dev/null || chroot "$root" groupadd -r "$group"
 done
 if ! chroot "$root" getent passwd hydra-live >/dev/null; then
    chroot "$root" useradd --uid 1000 --create-home --home-dir /home/hydra-live --shell /bin/bash --groups wheel,kvm,video,render,audio hydra-live
 fi
 chroot "$root" usermod --lock hydra-live
 chroot "$root" usermod --lock root
 chroot "$root" locale-gen
 ln -snf /usr/share/zoneinfo/Europe/Moscow "$root/etc/localtime"
 # Enable the ordinary per-user sound services without starting a user manager.
 python3 /hydra-source/hydra/image.py audio "$root"
 # Set rootfs permissions now; capabilities are regenerated in the boot namespace.
 chroot "$root" chown root:kvm /usr/libexec/qemu-bridge-helper
 chroot "$root" chmod 0750 /usr/libexec/qemu-bridge-helper
 chroot "$root" visudo --check
 systemctl --root="$root" enable NetworkManager.service greetd.service hydra-network.service hydra-dnsmasq.service hydra-bridge-permissions.service getty@tty2.service serial-getty@ttyS0.service
 systemctl --root="$root" set-default graphical.target
 kernel=$(python3 /hydra-source/hydra/image.py kernel "$root")
 chroot "$root" dracut --force --no-hostonly --no-hostonly-cmdline --conf /etc/hydra/live-mode/dracut.conf --confdir /etc/hydra/live-mode/dracut.conf.d --add dmsquash-live /boot/initramfs-live.img "$kernel"
@@ -1,6 +0,0 @@
 hostonly="no"
 hostonly_cmdline="no"
 add_dracutmodules+=" dmsquash-live "
 filesystems+=" squashfs overlay iso9660 vfat ext4 "
 add_drivers+=" vmd nvme xhci_pci thunderbolt usb_storage uas sd_mod virtio_pci virtio_blk virtio_scsi virtio_net virtio_gpu i915 loop squashfs overlay sr_mod ahci "
 compress="zstd"
@@ -1,12 +0,0 @@
 set default=0
 set timeout=5
 insmod all_video
 insmod serial
 serial --unit=0 --speed=115200
 terminal_input console serial
 terminal_output console serial
 menuentry "HydraOS live (volatile host root)" {
    search --no-floppy --label HYDRA_OS --set=root
    linux /boot/vmlinuz root=live:CDLABEL=HYDRA_OS rd.live.image rd.live.overlay.overlayfs=1 ro console=tty0 console=ttyS0,115200n8
    initrd /boot/initramfs-live.img
 }
@@ -1,3 +0,0 @@
 masters = gentoo
 thin-manifests = true
 sign-manifests = false
@@ -1 +0,0 @@
 hydra
@@ -1 +0,0 @@
 DIST brightnessctl-0.5.1.tar.gz 8284 BLAKE2B 7d2226aae10fddc606c5ce71ed4a04eda583acf245ea1532d8730c73dc02ea7aeed47a0e25ca7f9d40e7b0170bb08d87617e9f4aa082495e78894982870ca7cc SHA512 41ac86357b3b5a3d54a043140a6bdb1fb68d5080a16e86692059ad58f4b9134a8532432a3f664c70c665a4ad73ea8a94cc9b88fa84abc28f0356b8b397aaebc7
@@ -1,18 +0,0 @@
 # HydraOS local package for the pinned Gentoo snapshot.
 EAPI=8
 inherit toolchain-funcs udev
 DESCRIPTION="Read and control backlight brightness without privileged launchers"
 HOMEPAGE="https://github.com/Hummer12007/brightnessctl"
 SRC_URI="https://github.com/Hummer12007/brightnessctl/archive/refs/tags/${PV}.tar.gz -> ${P}.tar.gz"
 LICENSE="MIT"
 SLOT="0"
 KEYWORDS="amd64"
 RDEPEND="virtual/udev"
 src_compile() {
    emake CC="$(tc-getCC)" VERSION="${PV}"
 }
 src_install() {
    # Upstream's udev mode uses 0755, never its alternate setuid mode.
    emake DESTDIR="${D}" PREFIX="${EPREFIX}/usr" UDEVDIR="${EPREFIX}$(get_udevdir)/rules.d" INSTALL_UDEV_RULES=1 install
    dodoc README.md LICENSE
 }
@@ -28,4 +28,3 @@ dev-libs/libisoburn
 dev-lang/python
 sec-keys/openpgp-keys-gentoo-release
 sec-keys/openpgp-keys-gentoo-auth
 sys-apps/brightnessctl
@@ -4,10 +4,10 @@ CXXFLAGS="${COMMON_FLAGS}"
 FCFLAGS="${COMMON_FLAGS}"
 FFLAGS="${COMMON_FLAGS}"
 MAKEOPTS="-j4"
-EMERGE_DEFAULT_OPTS="--jobs=1 --getbinpkg --binpkg-respect-use=y --autounmask=n --usepkg-exclude='app-emulation/qemu app-emulation/virt-viewer'"
+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 virgl"
+VIDEO_CARDS="intel"
 USE="X dbus wayland opengl alsa"
 QEMU_SOFTMMU_TARGETS="x86_64"
 QEMU_USER_TARGETS=""
@@ -6,5 +6,3 @@ net-misc/networkmanager wifi
 sys-boot/grub mount
 media-libs/freetype harfbuzz
 sys-kernel/installkernel dracut
 sys-fs/squashfs-tools zstd
@@ -1,4 +0,0 @@
 [hydra]
 location = /var/db/repos/hydra
 masters = gentoo
 auto-sync = no
@@ -1,3 +0,0 @@
 GRUB_DISTRIBUTOR="HydraOS"
 GRUB_TIMEOUT=5
 GRUB_CMDLINE_LINUX_DEFAULT="console=tty0 console=ttyS0,115200n8"
@@ -1,3 +0,0 @@
 hostonly="no"
 hostonly_cmdline="no"
 omit_dracutmodules+=" dmsquash-live "
@@ -1,5 +0,0 @@
 [terminal]
 vt = 1
 [default_session]
 command = "/usr/bin/agreety --cmd /usr/bin/hydra-session"
 user = "greetd"
@@ -1,8 +0,0 @@
 [terminal]
 vt = 1
 [default_session]
 command = "/usr/bin/agreety --cmd /usr/bin/hydra-session"
 user = "greetd"
 [initial_session]
 command = "/usr/bin/hydra-session"
 user = "hydra-live"
@@ -1,5 +0,0 @@
 network = "bridge"
 bridge = "hydrabr0"
 width = 1920
 height = 1080
 refresh_millihz = 144000
@@ -1,11 +0,0 @@
 interface=hydrabr0
 bind-dynamic
 except-interface=lo
 listen-address=10.203.0.1
 dhcp-range=10.203.0.20,10.203.0.254,255.255.255.0,12h
 dhcp-option=option:router,10.203.0.1
 dhcp-option=option:dns-server,10.203.0.1
 dhcp-leasefile=/var/lib/hydra/dnsmasq.leases
 resolv-file=/run/NetworkManager/resolv.conf
 domain-needed
 bogus-priv
@@ -1 +0,0 @@
 HydraOS live medium: changes to the host root are not persistent.
@@ -1,7 +0,0 @@
 [
  "/etc/greetd/live.toml",
  "/etc/sudoers.d/90-hydra-live",
  "/etc/systemd/system/greetd.service.d/90-hydra-live.conf",
  "/etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf",
  "/etc/systemd/system/serial-getty@ttyS0.service.d/90-hydra-live.conf"
 ]
@@ -1 +0,0 @@
 LANG=en_US.UTF-8
@@ -1,2 +0,0 @@
 en_US.UTF-8 UTF-8
 ru_RU.UTF-8 UTF-8
@@ -1,5 +0,0 @@
 HydraOS: fullscreen VMs with an ordinary Gentoo host underneath.
 Live host changes are volatile. Use sudo hydra-install for the guarded installer.
 Use hydra --help, nmcli, or nmtui to configure VMs and network connections.
 Ctrl+Alt+F2 opens local recovery; return to the graphical session on Ctrl+Alt+F1.
 Guest bridge: 10.203.0.0/24. Routed LAN access requires a route on your LAN router.
@@ -1,6 +0,0 @@
 # Preserve the scoped bridge-helper permission after ordinary QEMU updates.
 post_pkg_postinst() {
    if [[ ${CATEGORY}/${PN} == app-emulation/qemu ]]; then
        "${EROOT}usr/libexec/hydra-bridge-permissions" || die "Hydra bridge permissions failed"
    fi
 }
@@ -1,11 +0,0 @@
 [gentoo]
 location = /var/db/repos/gentoo
 sync-type = rsync
 sync-uri = rsync://rsync.gentoo.org/gentoo-portage
 auto-sync = yes
 sync-rsync-verify-metamanifest = yes
 sync-rsync-verify-max-age = 3
 sync-openpgp-key-package = sec-keys/openpgp-keys-gentoo-release
 sync-openpgp-key-path = /usr/share/openpgp-keys/gentoo-release.asc
 sync-openpgp-keyserver = hkps://keys.gentoo.org
 sync-webrsync-verify-signature = yes
@@ -1 +0,0 @@
 allow hydrabr0
@@ -1 +0,0 @@
 %wheel ALL=(ALL:ALL) ALL
@@ -1 +0,0 @@
 hydra-live ALL=(ALL:ALL) NOPASSWD: ALL
@@ -1 +0,0 @@
 net.ipv4.ip_forward = 1
@@ -1,3 +0,0 @@
 [Login]
 KillUserProcesses=no
 UserStopDelaySec=infinity
@@ -1,3 +0,0 @@
 [Service]
 ExecStart=
 ExecStart=-/sbin/agetty --autologin hydra-live --noclear %I $TERM
@@ -1,3 +0,0 @@
 [Service]
 ExecStart=
 ExecStart=/usr/bin/greetd --config /etc/greetd/live.toml
@@ -1,8 +0,0 @@
 [Unit]
 Description=Restrict QEMU bridge helper to HydraOS VM users
 Before=greetd.service
 [Service]
 Type=oneshot
 ExecStart=/usr/libexec/hydra-bridge-permissions
 [Install]
 WantedBy=multi-user.target
@@ -1,12 +0,0 @@
 [Unit]
 Description=DHCP and DNS for HydraOS guests
 Requires=hydra-network.service
 PartOf=hydra-network.service
 After=hydra-network.service NetworkManager.service
 [Service]
 Type=simple
 StateDirectory=hydra
 ExecStart=/usr/sbin/dnsmasq --keep-in-foreground --conf-file=/etc/hydra/dnsmasq.conf
 Restart=on-failure
 [Install]
 WantedBy=multi-user.target
@@ -1,12 +0,0 @@
 [Unit]
 Description=HydraOS guest bridge and Internet NAT
 Requires=NetworkManager.service
 PartOf=NetworkManager.service
 After=NetworkManager.service
 Before=hydra-dnsmasq.service
 [Service]
 Type=oneshot
 RemainAfterExit=yes
 ExecStart=/usr/bin/hydra-network
 [Install]
 WantedBy=multi-user.target
@@ -1,3 +0,0 @@
 [Service]
 ExecStart=
 ExecStart=-/sbin/agetty --autologin hydra-live --noclear %I $TERM
@@ -1,7 +0,0 @@
 #!/bin/sh
 set -eu
 helper=/usr/libexec/qemu-bridge-helper
 [ -f "$helper" ] && [ ! -L "$helper" ]
 chown root:kvm "$helper"
 chmod 0750 "$helper"
 setcap cap_net_admin+ep "$helper"
@@ -13,15 +13,5 @@
  },
  "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/",
+  "binhost": "https://distfiles.gentoo.org/releases/amd64/binpackages/23.0/x86-64/"
  "local_packages": [
    {
      "atom": "sys-apps/brightnessctl",
      "version": "0.5.1",
      "url": "https://github.com/Hummer12007/brightnessctl/archive/refs/tags/0.5.1.tar.gz",
      "filename": "brightnessctl-0.5.1.tar.gz",
      "sha256": "a68869e23f56ac4f2e28f1783002810ddbf10f95e1af9b48b2912fb169f46994",
      "license": "MIT"
    }
  ]
 }
@@ -1,74 +0,0 @@
 --- a/include/hw/virtio/virtio-gpu.h
 +++ b/include/hw/virtio/virtio-gpu.h
@@ -128,6 +128,7 @@
     uint32_t flags;
     uint32_t xres;
     uint32_t yres;
 +    uint32_t refresh_rate;
     uint64_t hostmem;
     VirtIOGPUOutputList *outputs;
 };
@@ -173,7 +174,8 @@
     DEFINE_PROP_BIT("edid", _state, _conf.flags, \
                     VIRTIO_GPU_FLAG_EDID_ENABLED, true), \
     DEFINE_PROP_UINT32("xres", _state, _conf.xres, 1280), \
 -    DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800)
 +    DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800), \
 +    DEFINE_PROP_UINT32("refresh_rate", _state, _conf.refresh_rate, 0)
 typedef struct VGPUDMABuf {
     QemuDmaBuf *buf;
 --- a/hw/display/virtio-gpu-base.c
 +++ b/hw/display/virtio-gpu-base.c
@@ -64,7 +64,9 @@
         .height_mm = g->req_state[scanout].height_mm,
         .prefx = g->req_state[scanout].width,
         .prefy = g->req_state[scanout].height,
 -        .refresh_rate = g->req_state[scanout].refresh_rate,
 +        /* An explicit monitor refresh rate, in mHz, survives UI resizes. */
 +        .refresh_rate = g->conf.refresh_rate ? g->conf.refresh_rate :
 +                        g->req_state[scanout].refresh_rate,
     };
     for (output_idx = 0, node = g->conf.outputs;
 --- a/hw/display/edid-generate.c
 +++ b/hw/display/edid-generate.c
@@ -220,17 +220,20 @@
     desc[5 + len] = '\n';
 }
 -static void edid_desc_ranges(uint8_t *desc)
 +static void edid_desc_ranges(uint8_t *desc, uint64_t clock,
 +                             uint32_t htotal, uint32_t vtotal)
 {
     edid_desc_type(desc, 0xfd);
 -    /* vertical (50 -> 125 Hz) */
 -    desc[5] =  50;
 -    desc[6] = 125;
 -
 -    /* horizontal (30 -> 160 kHz) */
 -    desc[7] =  30;
 -    desc[8] = 160;
 +    /* Enclose the encoded timing after its 10 kHz pixel-clock rounding,
 +     * while retaining legacy ranges whenever they already contain it. */
 +    uint64_t vperiod = (uint64_t)htotal * vtotal;
 +    uint64_t hperiod = (uint64_t)htotal * 1000;
 +    desc[5] = MIN(50, clock / vperiod);
 +    desc[6] = MIN(255, MAX(125, DIV_ROUND_UP(clock, vperiod)));
 +
 +    desc[7] = MIN(30, clock / hperiod);
 +    desc[8] = MIN(255, MAX(160, DIV_ROUND_UP(clock, hperiod)));
     /* max dot clock (2550 MHz) */
     desc[9] = 2550 / 10;
@@ -513,7 +516,8 @@
      * so dta descriptor offsets don't move any more.
      */
 -    edid_desc_ranges(desc);
 +    edid_desc_ranges(desc, (uint64_t)timings.clock * 10000,
 +                     info->prefx + timings.xblank, info->prefy + timings.yblank);
     desc = edid_desc_next(edid, dta, desc);
     if (desc && info->name) {
@@ -8,20 +8,20 @@
     gboolean attach;
     gboolean shared;
     gboolean quitting;
-@@ -952,6 +953,13 @@
+@@ -951,6 +952,13 @@
 {
     gchar *subtitle = NULL;
     const gchar *title = virt_viewer_app_get_title(app);
- 
+
 +    /* Hidden fullscreen windows use the requested title as stable identity. */
 +    if (title != NULL && virt_viewer_app_get_hide_toolbar(app) &&
 +        virt_viewer_app_get_fullscreen(app)) {
 +        g_object_set(window, "subtitle", title, NULL);
 +        return;
 +    }
-+
+ 
     if (title != NULL) {
         VirtViewerDisplay *display = virt_viewer_window_get_display(window);
         gchar *d = strstr(title, "%d");
@@ -2134,6 +2142,7 @@
 static gboolean opt_verbose = FALSE;
 static gboolean opt_debug = FALSE;
@@ -110,38 +110,3 @@
     gtk_window_set_title(GTK_WINDOW(self->window), title);
     if (self->subtitle) {
         gtk_header_bar_set_title(GTK_HEADER_BAR(header), self->subtitle);
@@ -1467,6 +1475,34 @@
     display = self->display;
     event = (GdkEventKey *)ev;
 +    /* Keyboard grabs remove GTK's application accelerator group so normal
 +     * shortcuts reach the guest. Keep only the explicitly configured manual
 +     * USB selector available before forwarding this event to the display. */
 +    if (!self->kiosk) {
 +        gchar **accels = gtk_application_get_accels_for_action(
 +            GTK_APPLICATION(self->app), "win.usb-device-select");
 +        gboolean matched = FALSE;
 +        for (guint i = 0; accels[i]; i++) {
 +            guint key;
 +            GdkModifierType mods;
 +            gtk_accelerator_parse(accels[i], &key, &mods);
 +            if (key && gdk_keyval_to_lower(event->keyval) == key &&
 +                (event->state & gtk_accelerator_get_default_mod_mask()) == mods) {
 +                matched = TRUE;
 +                break;
 +            }
 +        }
 +        g_strfreev(accels);
 +        if (matched) {
 +            GAction *action = g_action_map_lookup_action(
 +                G_ACTION_MAP(self->window), "usb-device-select");
 +            if (action && g_action_get_enabled(action)) {
 +                g_action_activate(action, NULL);
 +                return TRUE;
 +            }
 +        }
 +    }
 +
     gtk_widget_grab_focus(GTK_WIDGET(display));
     // Look through keymaps - if set for mappings and intercept
@@ -1,205 +0,0 @@
 {
  "filesystems": [
    {
      "source": "LiveOS_rootfs",
      "target": "/",
      "fstype": "overlay",
      "maj:min": "0:31",
      "options": "rw,relatime,lowerdir=/run/rootfsbase,upperdir=/run/overlayfs,workdir=/run/ovlwork,uuid=on",
      "children": [
        {
          "source": "tmpfs",
          "target": "/tmp",
          "fstype": "tmpfs",
          "maj:min": "0:40",
          "options": "rw,nosuid,nodev,nr_inodes=1048576,inode64,usrquota"
        },
        {
          "source": "devtmpfs",
          "target": "/dev",
          "fstype": "devtmpfs",
          "maj:min": "0:6",
          "options": "rw,nosuid,size=1930828k,nr_inodes=482707,mode=755,inode64",
          "children": [
            {
              "source": "hugetlbfs",
              "target": "/dev/hugepages",
              "fstype": "hugetlbfs",
              "maj:min": "0:36",
              "options": "rw,nosuid,nodev,relatime,pagesize=2M"
            },
            {
              "source": "mqueue",
              "target": "/dev/mqueue",
              "fstype": "mqueue",
              "maj:min": "0:21",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "tmpfs",
              "target": "/dev/shm",
              "fstype": "tmpfs",
              "maj:min": "0:24",
              "options": "rw,nosuid,nodev,inode64,usrquota"
            },
            {
              "source": "devpts",
              "target": "/dev/pts",
              "fstype": "devpts",
              "maj:min": "0:25",
              "options": "rw,nosuid,noexec,relatime,gid=5,mode=600,ptmxmode=000"
            }
          ]
        },
        {
          "source": "sysfs",
          "target": "/sys",
          "fstype": "sysfs",
          "maj:min": "0:23",
          "options": "rw,nosuid,nodev,noexec,relatime",
          "children": [
            {
              "source": "fusectl",
              "target": "/sys/fs/fuse/connections",
              "fstype": "fusectl",
              "maj:min": "0:37",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "tracefs",
              "target": "/sys/kernel/tracing",
              "fstype": "tracefs",
              "maj:min": "0:13",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "debugfs",
              "target": "/sys/kernel/debug",
              "fstype": "debugfs",
              "maj:min": "0:8",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "securityfs",
              "target": "/sys/kernel/security",
              "fstype": "securityfs",
              "maj:min": "0:7",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "cgroup2",
              "target": "/sys/fs/cgroup",
              "fstype": "cgroup2",
              "maj:min": "0:27",
              "options": "rw,nosuid,nodev,noexec,relatime,nsdelegate,memory_recursiveprot,memory_hugetlb_accounting"
            },
            {
              "source": "none",
              "target": "/sys/fs/pstore",
              "fstype": "pstore",
              "maj:min": "0:28",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "efivarfs",
              "target": "/sys/firmware/efi/efivars",
              "fstype": "efivarfs",
              "maj:min": "0:29",
              "options": "rw,nosuid,nodev,noexec,relatime"
            },
            {
              "source": "bpf",
              "target": "/sys/fs/bpf",
              "fstype": "bpf",
              "maj:min": "0:30",
              "options": "rw,nosuid,nodev,noexec,relatime,mode=700"
            },
            {
              "source": "configfs",
              "target": "/sys/kernel/config",
              "fstype": "configfs",
              "maj:min": "0:19",
              "options": "rw,nosuid,nodev,noexec,relatime"
            }
          ]
        },
        {
          "source": "proc",
          "target": "/proc",
          "fstype": "proc",
          "maj:min": "0:22",
          "options": "rw,nosuid,nodev,noexec,relatime",
          "children": [
            {
              "source": "systemd-1",
              "target": "/proc/sys/fs/binfmt_misc",
              "fstype": "autofs",
              "maj:min": "0:33",
              "options": "rw,relatime,fd=38,pgrp=1,timeout=0,minproto=5,maxproto=5,direct,pipe_ino=5489",
              "children": [
                {
                  "source": "binfmt_misc",
                  "target": "/proc/sys/fs/binfmt_misc",
                  "fstype": "binfmt_misc",
                  "maj:min": "0:42",
                  "options": "rw,nosuid,nodev,noexec,relatime"
                }
              ]
            }
          ]
        },
        {
          "source": "tmpfs",
          "target": "/run",
          "fstype": "tmpfs",
          "maj:min": "0:26",
          "options": "rw,nosuid,nodev,size=797732k,nr_inodes=819200,mode=755,inode64",
          "children": [
            {
              "source": "none",
              "target": "/run/credentials/systemd-journald.service",
              "fstype": "tmpfs",
              "maj:min": "0:35",
              "options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
            },
            {
              "source": "tmpfs",
              "target": "/run/user/1000",
              "fstype": "tmpfs",
              "maj:min": "0:46",
              "options": "rw,nosuid,nodev,relatime,size=398864k,nr_inodes=99716,mode=700,uid=1000,gid=1000,inode64"
            },
            {
              "source": "/dev/sr0",
              "target": "/run/initramfs/live",
              "fstype": "iso9660",
              "maj:min": "11:0",
              "options": "ro,relatime,nojoliet,check=s,map=n,blocksize=2048,iocharset=utf8"
            },
            {
              "source": "/dev/loop0",
              "target": "/run/rootfsbase",
              "fstype": "squashfs",
              "maj:min": "7:0",
              "options": "ro,relatime,errors=continue"
            },
            {
              "source": "none",
              "target": "/run/credentials/getty@tty2.service",
              "fstype": "tmpfs",
              "maj:min": "0:47",
              "options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
            },
            {
              "source": "none",
              "target": "/run/credentials/serial-getty@ttyS0.service",
              "fstype": "tmpfs",
              "maj:min": "0:48",
              "options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
            }
          ]
        }
      ]
    }
  ]
 }
@@ -1,40 +0,0 @@
 {
  "blockdevices": [
    {
      "path": "/dev/loop0",
      "type": "loop",
      "size": 2220142592,
      "maj:min": "7:0",
      "pkname": null,
      "mountpoints": [
        "/run/rootfsbase"
      ]
    },
    {
      "path": "/dev/sr0",
      "type": "rom",
      "size": 2345107456,
      "maj:min": "11:0",
      "pkname": null,
      "mountpoints": [
        "/run/initramfs/live"
      ]
    },
    {
      "path": "/dev/sr1",
      "type": "rom",
      "size": 1073741312,
      "maj:min": "11:1",
      "pkname": null,
      "mountpoints": []
    },
    {
      "path": "/dev/vda",
      "type": "disk",
      "size": 34359738368,
      "maj:min": "253:0",
      "pkname": null,
      "mountpoints": []
    }
  ]
 }
@@ -1,66 +0,0 @@
 import json
 import subprocess
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 from hydra.profiles import load_profile
 class CLITests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='hc-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.defaults = self.root / 'defaults.toml'
        self.base = [sys.executable, 'bin/hydra', '--config', str(self.root / 'config'),
                     '--state', str(self.root / 'state'), '--runtime', str(self.root / 'run'),
                     '--defaults', str(self.defaults)]
    def run_cli(self, *args):
        return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5)
    def test_defaults_affect_new_profiles_only_and_explicit_options_win(self):
        self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n')
        added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0')
        self.assertEqual(added.returncode, 0, added.stderr)
        first = load_profile(self.root / 'config/first.toml')
        self.assertEqual((first.network, first.width, first.height, first.refresh_millihz), ('user', 1600, 1080, 0))
        added = self.run_cli('add', 'second')
        self.assertEqual(added.returncode, 0, added.stderr)
        second = load_profile(self.root / 'config/second.toml')
        self.assertEqual((second.network, second.width, second.height, second.refresh_millihz), ('bridge', 1920, 1080, 144000))
        self.defaults.write_text('width=1024\nnetwork="user"\n')
        self.assertEqual(load_profile(self.root / 'config/second.toml'), second)
        self.defaults.write_text('width="bad"\n')
        listed = self.run_cli('list')
        self.assertEqual(listed.returncode, 0, listed.stderr)
        self.assertEqual(len(json.loads(listed.stdout)), 2)
        self.assertEqual(self.run_cli('add', 'third').returncode, 1)
        self.assertFalse((self.root / 'config/third.toml').exists())
    def test_missing_defaults_use_developer_preferences_and_media_cli(self):
        added = self.run_cli('add', 'guest')
        self.assertEqual(added.returncode, 0, added.stderr)
        profile = load_profile(self.root / 'config/guest.toml')
        self.assertEqual((profile.network, profile.width, profile.height, profile.refresh_millihz), ('user', 1280, 800, 0))
        iso = self.root / 'installer.iso'
        iso.touch()
        attached = self.run_cli('media', 'guest', '--iso', str(iso))
        self.assertEqual(attached.returncode, 0, attached.stderr)
        iso.unlink()
        listed = self.run_cli('list')
        self.assertEqual(listed.returncode, 0, listed.stderr)
        self.assertEqual(self.run_cli('status', 'guest').returncode, 0)
        detached = self.run_cli('media', 'guest', '--detach')
        self.assertEqual(detached.returncode, 0, detached.stderr)
        self.assertIsNone(load_profile(self.root / 'config/guest.toml').iso)
    def test_unknown_or_invalid_distribution_defaults_are_rejected(self):
        for content in ('iso="/private.iso"\n', 'network="public"\n', 'refresh_millihz=true\n', '[display]\nwidth=1920\n'):
            self.defaults.write_text(content)
            result = self.run_cli('add', 'guest')
            self.assertEqual(result.returncode, 1, result.stderr)
            self.assertNotIn('Traceback', result.stderr)
            self.assertFalse((self.root / 'config/guest.toml').exists())
@@ -1,579 +0,0 @@
 import importlib.util
 import bz2
 import json
 import os
 import shlex
 import sys
 from pathlib import Path
 import subprocess
 import tempfile
 import tomllib
 import unittest
 try:
    from hydra import image
 except ImportError:
    image = None
 class ImageTests(unittest.TestCase):
    def setUp(self):
        self.assertIsNotNone(image, 'image pipeline is missing')
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name) / 'rootfs'
        self.root.mkdir()
    def file(self, relative, content='preserved'):
        path = self.root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content)
        return path
    def kernel(self, version='6.12.58-gentoo-dist'):
        (self.root / 'lib/modules' / version).mkdir(parents=True)
        self.file('boot/vmlinuz-' + version, 'kernel')
        return version
    def test_kernel_from_payload_not_running_host(self):
        self.kernel()
        self.assertEqual(image.kernel_version(self.root), '6.12.58-gentoo-dist')
    def test_real_gentoo_kernel_bin_layout(self):
        version = '6.18.54-gentoo-dist-bin'
        (self.root / 'lib/modules' / version).mkdir(parents=True)
        kernel = self.file('boot/kernel-' + version, 'real Gentoo kernel')
        self.assertEqual(image.kernel_version(self.root), version)
        self.assertEqual(image.kernel_image(self.root), kernel)
    def test_configured_real_kernel_layout_is_accepted_by_installer(self):
        from hydra.installer import select_kernel
        version = '6.18.54-gentoo-dist-bin'
        (self.root / 'lib/modules' / version).mkdir(parents=True)
        kernel = self.file('boot/kernel-' + version, 'real Gentoo kernel')
        image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
        self.assertEqual(select_kernel(self.root, None), version)
        alias = self.root / ('boot/vmlinuz-' + version)
        self.assertTrue(alias.samefile(kernel))
        self.assertEqual(image.kernel_image(self.root), alias)
    def test_exported_repository_sync_works_with_extracted_tree(self):
        import configparser
        self.kernel()
        self.file('etc/portage/repos.conf/gentoo.conf',
                  '[gentoo]\nlocation=/var/db/repos/gentoo\nsync-type=git\nauto-sync=no\n')
        self.file('var/db/repos/gentoo/profiles/repo_name', 'gentoo')
        image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
        config = configparser.ConfigParser()
        config.read(self.root / 'etc/portage/repos.conf/gentoo.conf')
        repo = config['gentoo']
        self.assertEqual(repo['sync-type'], 'rsync')
        self.assertEqual(repo['sync-uri'], 'rsync://rsync.gentoo.org/gentoo-portage')
        self.assertEqual(repo['auto-sync'], 'yes')
        self.assertEqual(repo['sync-rsync-verify-metamanifest'], 'yes')
        self.assertEqual(repo['sync-openpgp-key-path'], '/usr/share/openpgp-keys/gentoo-release.asc')
        self.assertFalse((self.root / 'var/db/repos/gentoo/.git').exists())
        source = Path(__file__).resolve().parents[1]
        builder_config = (source / 'image/portage/repos.conf/gentoo.conf').read_text()
        self.assertIn('auto-sync = no', builder_config)
    def test_conflicting_kernel_images_refused(self):
        version = self.kernel()
        self.file('boot/kernel-' + version, 'different kernel')
        with self.assertRaises(image.ImageError):
            image.kernel_version(self.root)
    def test_ambiguous_or_missing_kernel_refused(self):
        with self.assertRaises(image.ImageError):
            image.kernel_version(self.root)
        self.kernel('6.1-one')
        self.kernel('6.1-two')
        with self.assertRaises(image.ImageError):
            image.kernel_version(self.root)
    def test_module_directory_without_kernel_refused(self):
        self.kernel()
        (self.root / 'boot/vmlinuz-6.12.58-gentoo-dist').unlink()
        with self.assertRaises(image.ImageError):
            image.kernel_version(self.root)
    def test_mount_faults_clean_only_successful_bindings(self):
        source = Path(__file__).resolve().parents[1]
        tools = Path(self.tmp.name) / 'tools'
        tools.mkdir()
        for name in ('python3', 'chroot', 'ln', 'systemctl'):
            script = tools / name
            script.write_text('#!/bin/sh\nexit 0\n')
            script.chmod(0o755)
        mount = tools / 'mount'
        mount.write_text('#!/bin/sh\nn=$(cat "$STATE/count" 2>/dev/null || echo 0)\n'
                         'n=$((n+1)); echo "$n" > "$STATE/count"\n'
                         'test "$n" != "$FAIL_MOUNT"\n')
        mount.chmod(0o755)
        unmount = tools / 'umount'
        unmount.write_text('#!/bin/sh\nprintf "%s\\n" "$@" >> "$STATE/unmounts"\n'
                           'test "$FAIL_UNMOUNT" != 1\n')
        unmount.chmod(0o755)
        for failed, cleanup_failed in ((1, 0), (2, 0), (3, 0), (0, 0), (0, 1)):
            with self.subTest(failed=failed, cleanup_failed=cleanup_failed):
                state = Path(self.tmp.name) / f'state-{failed}-{cleanup_failed}'
                state.mkdir()
                env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'],
                           STATE=str(state), FAIL_MOUNT=str(failed), FAIL_UNMOUNT=str(cleanup_failed))
                result = subprocess.run(['bash', str(source / 'image/configure-rootfs.sh'), '/hydra-output/rootfs'],
                                        env=env, capture_output=True, text=True)
                expected = ['/hydra-output/rootfs/dev', '/hydra-output/rootfs/proc', '/hydra-output/rootfs/sys']
                expected = list(reversed(expected[:failed - 1 if failed else 3]))
                actual = (state / 'unmounts').read_text().splitlines() if (state / 'unmounts').exists() else []
                self.assertEqual(actual, expected)
                self.assertEqual(result.returncode == 0, not failed and not cleanup_failed)
    def test_retry_guard_uses_valid_real_findmnt_options(self):
        from hydra.build import rootfs_retry_guard_script
        self.file('.hydra-export-owned')
        script = 'set -eu\nroot=' + shlex.quote(str(self.root)) + '\n' + rootfs_retry_guard_script()
        result = subprocess.run(['bash', '-c', script], capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(self.root.exists())
    def test_retry_refuses_removing_owned_rootfs_with_stale_child_mount(self):
        from hydra.build import rootfs_retry_guard_script
        marker = self.file('.hydra-export-owned')
        preserved = self.file('dev/guest-data')
        tools = Path(self.tmp.name) / 'guard-tools'
        tools.mkdir()
        findmnt = tools / 'findmnt'
        findmnt.write_text('#!/bin/sh\nprintf "%s\\n" / "$STALE_MOUNT"\n')
        findmnt.chmod(0o755)
        env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'], STALE_MOUNT=str(self.root / 'dev'))
        script = 'set -eu\nroot=' + shlex.quote(str(self.root)) + '\n' + rootfs_retry_guard_script()
        result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertTrue(marker.is_file())
        self.assertTrue(preserved.is_file())
        self.assertIn('mounted', result.stderr)
        env['STALE_MOUNT'] = str(Path(self.tmp.name) / 'different-rootfs')
        result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(self.root.exists())
    def test_export_sanitization_preserves_update_inputs(self):
        preserved = ['var/db/pkg/sys-apps/portage-3/USE', 'var/lib/portage/world',
                     'var/db/repos/gentoo/profiles/repo_name', 'usr/share/openpgp-keys/gentoo-release.asc',
                     'etc/portage/patches/app-emulation/qemu/patch.patch', 'etc/portage/make.conf']
        removed = ['etc/portage/gnupg/private-keys-v1.d/key', 'etc/portage/gnupg.hydra-untrusted/key',
                   'var/lib/hydra-build/packages.log', 'root/.ssh/id_ed25519', 'root/.bash_history',
                   'var/cache/binpkgs/package.tbz2', 'var/tmp/portage/build/log', 'run/container-state',
                   'var/log/emerge.log', 'var/lib/systemd/random-seed', 'hydra-source/private', 'hydra-cache/private']
        for name in preserved + removed:
            self.file(name)
        self.file('etc/machine-id', 'builder-id')
        self.file('etc/resolv.conf', 'nameserver 127.0.0.1')
        image.sanitize_rootfs(self.root)
        for name in preserved:
            self.assertEqual((self.root / name).read_text(), 'preserved', name)
        for name in removed:
            self.assertFalse((self.root / name).exists(), name)
        self.assertEqual((self.root / 'etc/machine-id').read_text(), '')
        self.assertEqual((self.root / 'etc/hostname').read_text(), 'hydra-os\n')
        self.assertTrue((self.root / 'etc/resolv.conf').is_symlink())
    def test_export_filters_saved_environment_proxies_but_retains_package_metadata(self):
        self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
        # Portable stand-in for the external Portage filter API, not the scrubber.
        helper = self.file('usr/lib/portage/python3.14/filter-bash-environment.py',
            "import re\ndef filter_bash_environment(pattern, source, output):\n"
            "    for line in source:\n"
            "        match = re.match(rb'declare -x ([A-Za-z_][A-Za-z0-9_]*)=', line)\n"
            "        if not match or not pattern.fullmatch(match.group(1)):\n"
            "            output.write(line)\n")
        path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
        path.parent.mkdir(parents=True)
        package_metadata = b'declare -x LICENSE="GPL-2"\ndeclare -x EAPI="8"\n'
        path.write_bytes(bz2.compress(package_metadata + b'declare -x HTTPS_PROXY="https://user:secret@builder.invalid"\n'
                                     b'declare -x HTTP_PROXY="http://127.0.0.1:1234"\n'
                                     b'declare -x NO_PROXY="localhost"\n'))
        result = image.sanitize_saved_environments(self.root)
        self.assertEqual(bz2.decompress(path.read_bytes()), package_metadata)
        self.assertEqual(result['filtered_files'], 1)
        self.assertEqual(result['variable_names'], ['HTTPS_PROXY', 'HTTP_PROXY', 'NO_PROXY'])
    def test_saved_environment_filter_refuses_missing_portage_helper(self):
        self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
        path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
        path.parent.mkdir(parents=True)
        original = bz2.compress(b'declare -x HTTPS_PROXY="private"\n')
        path.write_bytes(original)
        with self.assertRaises(image.ImageError):
            image.sanitize_saved_environments(self.root)
        self.assertEqual(path.read_bytes(), original)
    def test_sanitization_cannot_follow_exported_symlink_to_host(self):
        outside = Path(self.tmp.name) / 'host'
        outside.mkdir()
        secret = outside / 'machine-id'
        secret.write_text('host-id')
        (self.root / 'etc').symlink_to(outside)
        with self.assertRaises(image.ImageError):
            image.sanitize_rootfs(self.root)
        self.assertEqual(secret.read_text(), 'host-id')
    def test_installed_configuration_restores_package_sandbox_defaults(self):
        self.kernel()
        self.file('etc/portage/make.conf', 'FEATURES="getbinpkg binpkg-request-signature -usersandbox -sandbox -network-sandbox -ipc-sandbox -pid-sandbox"\n')
        image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
        features = (self.root / 'etc/portage/make.conf').read_text()
        for option in ('-usersandbox', '-sandbox', '-network-sandbox', '-ipc-sandbox', '-pid-sandbox'):
            self.assertNotIn(option, features)
        self.assertIn('binpkg-request-signature', features)
        self.assertEqual((self.root / 'usr/share/hydra/build-manifest.json').is_file(), True)
    def test_installed_helper_uses_relocated_payload_and_opens_empty_recovery(self):
        self.kernel()
        image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
        terminal = Path(self.tmp.name) / 'terminal'
        terminal.write_text('#!/bin/sh\nshift\nexec "$@"\n')
        terminal.chmod(0o755)
        manifest = Path(self.tmp.name) / 'session.json'
        manifest.write_text(json.dumps({'config': str(Path(self.tmp.name) / 'config'),
                                       'state': str(Path(self.tmp.name) / 'state'),
                                       'runtime': str(Path(self.tmp.name) / 'runtime'),
                                       'terminal': str(terminal), 'names': []}))
        script = r"""import json, pathlib, subprocess, sys
 sys.path.insert(0, sys.argv[1])
 from hydra.session import _module_argv
 helper = _module_argv('clients', pathlib.Path(sys.argv[2]))
 help_result = subprocess.run(helper[:2] + ['--help'], text=True, capture_output=True)
 print(help_result.stdout)
 print(help_result.stderr, file=sys.stderr)
 if help_result.returncode:
    sys.exit(help_result.returncode)
 result = subprocess.run(helper, input=':quit\n', text=True, capture_output=True, timeout=10)
 print(result.stdout)
 print(result.stderr, file=sys.stderr)
 sys.exit(result.returncode)
 """
        result = subprocess.run([sys.executable, '-I', '-c', script, str(self.root / 'usr/lib/hydra'), str(manifest)],
                                cwd=self.tmp.name, text=True, capture_output=True, timeout=20)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn('HydraOS local recovery', result.stdout)
        self.assertIn('add', result.stdout)
    def test_installed_updates_exclude_unadapted_qemu_and_viewer_binaries(self):
        self.kernel()
        source = Path(__file__).resolve().parents[1]
        self.file('etc/portage/make.conf', (source / 'image/portage/make.conf').read_text())
        image.configure_rootfs(self.root, source)
        settings = subprocess.run(['bash', '-euc', 'source "$1"; printf "%s" "$EMERGE_DEFAULT_OPTS"',
                                   '--', str(self.root / 'etc/portage/make.conf')], text=True, capture_output=True, check=True)
        defaults = shlex.split(settings.stdout)
        exclusions = [item.split('=', 1)[1].split() for item in defaults if item.startswith('--usepkg-exclude=')]
        self.assertEqual(exclusions, [['app-emulation/qemu', 'app-emulation/virt-viewer']])
        self.assertIn('--getbinpkg', defaults)
    def test_export_enables_pulse_socket_without_starting_user_services(self):
        source = Path(__file__).resolve().parents[1]
        self.assertTrue(callable(getattr(image, 'enable_user_audio', None)), 'user audio enable is missing')
        units = self.root / 'usr/lib/systemd/user'
        units.mkdir(parents=True)
        for name, section in [('pipewire.socket', 'Socket'), ('pipewire-pulse.socket', 'Socket'),
                              ('wireplumber.service', 'Service')]:
            target = 'sockets.target' if section == 'Socket' else 'default.target'
            (units / name).write_text(f'[{section}]\n[Install]\nWantedBy={target}\n')
        image.enable_user_audio(self.root)
        for target, name in [('sockets.target', 'pipewire.socket'), ('sockets.target', 'pipewire-pulse.socket'),
                             ('default.target', 'wireplumber.service')]:
            link = self.root / 'etc/systemd/user' / (target + '.wants') / name
            self.assertTrue(link.is_symlink(), name)
            self.assertEqual(os.readlink(link), '/usr/lib/systemd/user/' + name)
    def test_rootfs_configuration_separates_live_overrides_and_installed_auth(self):
        source = Path(__file__).resolve().parents[1]
        image.copy_overlay(self.root, source / 'image/rootfs')
        installed = tomllib.loads((self.root / 'etc/greetd/config.toml').read_text())
        live = tomllib.loads((self.root / 'etc/greetd/live.toml').read_text())
        self.assertNotIn('initial_session', installed)
        self.assertEqual(live['initial_session']['user'], 'hydra-live')
        manifest = json.loads((self.root / 'etc/hydra/live-mode/remove-on-install.json').read_text())
        for name in manifest:
            self.assertTrue((self.root / name.lstrip('/')).is_file(), name)
        self.assertIn('/etc/sudoers.d/90-hydra-live', manifest)
        self.assertIn('/etc/greetd/live.toml', manifest)
        self.assertNotIn('root=live:', (self.root / 'etc/dracut.conf.d/50-hydra.conf').read_text())
        defaults = tomllib.loads((self.root / 'etc/hydra/defaults.toml').read_text())
        self.assertEqual(defaults, dict(network='bridge', bridge='hydrabr0', width=1920, height=1080, refresh_millihz=144000))
    def test_overlay_cannot_write_through_host_symlink(self):
        outside = Path(self.tmp.name) / 'host'
        outside.mkdir()
        (self.root / 'etc').symlink_to(outside)
        with self.assertRaises(image.ImageError):
            image.copy_overlay(self.root, Path(__file__).resolve().parents[1] / 'image/rootfs')
        self.assertEqual(list(outside.iterdir()), [])
    def test_live_kernel_args_match_label_and_never_copy_payload_to_ram(self):
        args = image.live_kernel_args()
        self.assertIn('root=live:CDLABEL=HYDRA_OS', args)
        self.assertIn('rd.live.overlay.overlayfs=1', args)
        self.assertNotIn('rd.live.ram=1', args)
        cfg = (Path(__file__).resolve().parents[1] / 'image/grub.cfg').read_text()
        linux = next(line.strip() for line in cfg.splitlines() if line.strip().startswith('linux '))
        self.assertEqual(linux.split()[2:], args)
    def test_manifest_records_actual_kernel_packages_and_source_content(self):
        self.kernel()
        self.file('var/db/pkg/gui-wm/labwc-0.20.0/USE', 'X png')
        self.file('var/db/pkg/gui-wm/labwc-0.20.0/BUILD_TIME', '123')
        source = Path(__file__).resolve().parents[1]
        result = image.payload_manifest(self.root, source)
        self.assertEqual(result['kernel'], '6.12.58-gentoo-dist')
        self.assertEqual(result['packages'], [{'package': 'gui-wm/labwc-0.20.0', 'use': ['X', 'png'], 'build_time': '123'}])
        self.assertIn('hydra/session.py', result['source_files'])
        self.assertIn('image/rootfs/etc/hydra/defaults.toml', result['source_files'])
    def test_artifact_checkpoint_checks_bytes_not_flags(self):
        path = self.file('artifacts/squashfs.img', 'actual-payload')
        state = {'artifacts': {'squashfs.img': image.digest(path)}}
        self.assertTrue(image.artifacts_valid(path.parent, state))
        path.write_text('changed-payload')
        self.assertFalse(image.artifacts_valid(path.parent, state))
        path.unlink()
        self.assertFalse(image.artifacts_valid(path.parent, state))
        self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {}}))
        self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {'../escape': '0' * 64}}))
 class ImageBuildTests(unittest.TestCase):
    def setUp(self):
        from hydra.build import Builder
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.project = Path(self.tmp.name)
        self.builder = Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
    def inventory(self, records=None):
        records = records if records is not None else [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]
        self.builder.work.mkdir(exist_ok=True)
        artifacts = self.builder.work / 'artifacts'
        artifacts.mkdir(exist_ok=True)
        path = artifacts / 'packages.json'
        path.write_text(json.dumps(records))
        original = self.builder.complete
        self.builder.complete = lambda step: True if step == 'prepare' else original(step)
        self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
        return path
    def test_changed_inventory_invalidates_package_checkpoint(self):
        path = self.inventory()
        self.assertTrue(self.builder.complete('packages'))
        path.write_text('[{"package":"app-emulation/qemu-8.0", "use": [], "build_time": null}]')
        self.assertFalse(self.builder.complete('packages'))
    def test_malformed_inventory_with_matching_digest_is_refused(self):
        for data in ['{', '{}', '[]', '[{"package":"wrong"}]',
                     '[{"package":"app-emulation/qemu-9.2.4", "use":"spice", "build_time":"123"}]',
                     '[{"package":"app-emulation/qemu-9.2.4", "use":[], "build_time":42}]']:
            with self.subTest(data=data):
                path = self.inventory()
                path.write_text(data)
                self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
                self.assertFalse(self.builder.complete('packages'))
    def test_changed_inventory_invalidates_completed_rootfs_and_iso(self):
        path = self.inventory()
        artifacts = path.parent
        self.builder.active_package_job = lambda: False
        self.builder.assembly = lambda snapshot=None: 'isolated-assembler'
        self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
        def pod(*args, **kwargs):
            if args[0] == 'export':
                snapshot = artifacts / 'builder-rootfs.tar'
                self.assertTrue(snapshot.is_file(), 'raw snapshot must be created privately before export')
                self.assertEqual(snapshot.stat().st_mode & 0o777, 0o600)
                snapshot.write_bytes(b'exported fixture')
            elif args[0] == 'exec' and 'configure-rootfs.sh' in args[-1]:
                for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
                    (artifacts / name).write_bytes(b'artifact')
                (artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
                    'packages': [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]}))
            elif args[0] == 'exec' and 'grub-mkrescue' in args[-1]:
                (artifacts / 'hydra-os.iso').write_bytes(b'ISO fixture')
                (artifacts / 'iso-inspection.txt').write_text('HYDRA_OS\nEl Torito boot img : 1 UEFI\n')
            else:
                self.fail('Unexpected Podman operation: ' + repr(args))
        self.builder.pod = pod
        self.builder.rootfs()
        self.builder.iso()
        self.assertTrue(self.builder.complete('rootfs'))
        self.assertTrue(self.builder.complete('iso'))
        path.write_text('[{"package":"app-emulation/qemu-9.2.5", "use":[], "build_time":null}]')
        self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
        self.assertTrue(self.builder.complete('packages'))
        self.assertFalse(self.builder.complete('rootfs'))
        self.assertFalse(self.builder.complete('iso'))
    def test_export_inventory_must_match_validated_package_checkpoint(self):
        from hydra.build import BuildError
        path = self.inventory()
        self.builder.complete = lambda step: step == 'packages'
        self.builder.active_package_job = lambda: False
        self.builder.assembly = lambda snapshot: 'isolated-assembler'
        self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
        def pod(*args, **kwargs):
            artifacts = self.builder.work / 'artifacts'
            if args[0] == 'export':
                (artifacts / 'builder-rootfs.tar').write_bytes(b'exported fixture')
            elif args[0] == 'exec':
                for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
                    (artifacts / name).write_bytes(b'artifact')
                # Actual exported package differs from the checkpoint inventory.
                (artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
                    'packages': [{'package': 'app-emulation/qemu-9.2.5', 'use': ['spice'], 'build_time': '456'}]}))
            else:
                self.fail('Unexpected Podman operation: ' + repr(args))
        self.builder.pod = pod
        with self.assertRaisesRegex(BuildError, 'inventory'):
            self.builder.rootfs()
        self.assertFalse(self.builder.manifest()['rootfs']['complete'])
    def test_corrupt_local_package_archive_invalidates_fetch_checkpoint(self):
        import hashlib
        source = self.project / 'image'
        source.mkdir()
        (source / 'sources.json').write_text(json.dumps({
            'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'stage').hexdigest()},
            'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()},
            'local_packages': [{'url': 'https://example.test/brightnessctl.tar.gz',
                                'filename': 'brightnessctl.tar.gz', 'sha256': hashlib.sha256(b'original-source').hexdigest()}]}))
        with self.builder.locked():
            (self.builder.cache / 'stage.tar').write_bytes(b'stage')
            (self.builder.cache / 'repo.tar').write_bytes(b'repo')
            archive = self.builder.cache / 'brightnessctl.tar.gz'
            archive.write_bytes(b'original-source')
            self.builder.record('fetch')
            self.assertTrue(self.builder.complete('fetch'))
            archive.write_bytes(b'corrupted-source')
            self.assertFalse(self.builder.complete('fetch'))
    def test_changed_canonical_patch_invalidates_package_input_identity(self):
        patches = self.project / 'patches'
        patches.mkdir()
        patch = patches / 'qemu-virtio-refresh-rate.patch'
        patch.write_text('old patch')
        before = self.builder.input_id()
        patch.write_text('new patch')
        self.assertNotEqual(self.builder.input_id(), before)
    def test_running_package_job_blocks_export(self):
        from hydra.build import BuildError
        self.builder.active_package_job = lambda: True
        self.builder.pod = lambda *args, **kw: self.fail('Running builder was modified or exported')
        self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing')
        with self.assertRaisesRegex(BuildError, 'running'):
            self.builder.rootfs()
    def test_unverified_packages_block_export(self):
        from hydra.build import BuildError
        self.builder.active_package_job = lambda: False
        self.builder.pod = lambda *args, **kw: self.fail('Unverified builder was exported')
        self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing')
        with self.assertRaises(BuildError):
            self.builder.rootfs()
    def test_iso_requires_checked_rootfs_artifacts(self):
        from hydra.build import BuildError
        self.builder.pod = lambda *args, **kw: self.fail('ISO build started without checked rootfs')
        self.assertTrue(callable(getattr(self.builder, 'iso', None)), 'iso pipeline is missing')
        with self.assertRaises(BuildError):
            self.builder.iso()
 class NetworkTests(unittest.TestCase):
    def setUp(self):
        self.assertIsNotNone(image, 'network preparation is missing')
    def test_default_route_does_not_conflict(self):
        image.check_network_conflicts([{'dst': 'default', 'gateway': '192.168.1.1', 'dev': 'eth0'},
                                      {'dst': '192.168.1.0/24', 'dev': 'eth0'}], [])
    def test_overlapping_lan_vpn_address_or_route_fails_closed(self):
        for routes, addresses in [([{'dst': '10.203.0.0/16', 'dev': 'tun0'}], []),
                                  ([], [{'ifname': 'wlan0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.77', 'prefixlen': 24}]}]),
                                  ([{'dst': '10.203.0.8/32', 'dev': 'eth0'}], [])]:
            with self.subTest(routes=routes, addresses=addresses), self.assertRaises(image.ImageError):
                image.check_network_conflicts(routes, addresses)
    def test_existing_owned_bridge_route_allowed(self):
        image.check_network_conflicts([{'dst': '10.203.0.0/24', 'dev': 'hydrabr0'}],
                                      [{'ifname': 'hydrabr0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.1', 'prefixlen': 24}]}])
    def test_malformed_data_not_silently_accepted(self):
        for routes, addresses in [('bad', []), ([{'dst': 'nonsense', 'dev': 'eth0'}], []),
                                  ([], [{'ifname': 'eth0', 'addr_info': [{'family': 'inet', 'local': 'bad', 'prefixlen': 24}]}])]:
            with self.subTest(routes=routes), self.assertRaises(image.ImageError):
                image.check_network_conflicts(routes, addresses)
    def test_network_conflict_stops_before_any_mutation(self):
        calls = []
        def run(argv, **kwargs):
            calls.append(argv)
            if argv == ['ip', '-json', '-4', 'route', 'show', 'table', 'all']:
                return subprocess.CompletedProcess(argv, 0, json.dumps([{'dst': '10.203.0.0/24', 'dev': 'eth0'}]), '')
            if argv == ['ip', '-json', '-4', 'address', 'show']:
                return subprocess.CompletedProcess(argv, 0, '[]', '')
            self.fail('Unexpected operation before conflict validation: ' + repr(argv))
        with self.assertRaises(image.ImageError):
            image.configure_network(run)
        self.assertEqual(len(calls), 2)
    def test_manual_bridge_has_nat_without_forward_filter(self):
        changes = []
        def run(argv, **kwargs):
            responses = {
                ('ip', '-json', '-4', 'route', 'show', 'table', 'all'): '[]',
                ('ip', '-json', '-4', 'address', 'show'): '[]',
                ('ip', '-json', '-details', 'link', 'show'): '[]',
            }
            if tuple(argv) in responses:
                return subprocess.CompletedProcess(argv, 0, responses[tuple(argv)], '')
            if argv[0] == 'nmcli' and argv[1] == '--get-values':
                return subprocess.CompletedProcess(argv, 10, '', 'not found')
            if argv == ['nft', 'list', 'table', 'ip', 'hydra_nat']:
                return subprocess.CompletedProcess(argv, 1, '', 'not found')
            changes.append((argv, kwargs))
            return subprocess.CompletedProcess(argv, 0, '', '')
        image.configure_network(run)
        addition = next(argv for argv, kwargs in changes if argv[:3] == ['nmcli', 'connection', 'add'])
        self.assertEqual(addition[addition.index('ipv4.method') + 1], 'manual')
        rules = next(kwargs['input'] for argv, kwargs in changes if argv == ['nft', '-f', '-'])
        self.assertIn('masquerade', rules)
        self.assertNotIn('hook forward', rules)
        self.assertNotIn('reject', rules)
        self.assertNotIn('flush ruleset', rules)
    def test_foreign_bridge_name_refused(self):
        calls = []
        def run(argv, **kwargs):
            calls.append(argv)
            if argv[0] == 'ip':
                output = '[{"ifname": "hydrabr0", "linkinfo": {"info_kind": "bridge"}}]' if 'link' in argv else '[]'
                return subprocess.CompletedProcess(argv, 0, output, '')
            if argv[0] == 'nmcli' and argv[1] == '--get-values':
                return subprocess.CompletedProcess(argv, 10, '', 'not found')
            self.fail('Foreign interface triggered a network mutation: ' + repr(argv))
        with self.assertRaises(image.ImageError):
            image.configure_network(run)
    def test_changed_owned_connection_refused(self):
        def run(argv, **kwargs):
            if argv[0] == 'ip':
                return subprocess.CompletedProcess(argv, 0, '[]', '')
            if argv[0] == 'nmcli' and argv[1] == '--get-values':
                return subprocess.CompletedProcess(argv, 0, image.CONNECTION_UUID + '\nbridge\nhydrabr0\nshared\n10.203.0.1/24\n', '')
            self.fail('Changed connection triggered a network mutation: ' + repr(argv))
        with self.assertRaises(image.ImageError):
            image.configure_network(run)
 if __name__ == '__main__':
    unittest.main()
@@ -4,9 +4,6 @@ import io
 import json
 import os
 import pty
 import stat
 import struct
 from types import SimpleNamespace
 from pathlib import Path
 import subprocess
 import tempfile
@@ -406,163 +403,6 @@ class AdditionalFailureTests(unittest.TestCase):
                i.prompt_password()
 class RelocatedLiveLoopTests(unittest.TestCase):
    def probe(self, *, backing_device=None, backing_rdevice=0, ioctl_error=None,
              opened_device=None, opened_mode=stat.S_IFBLK):
        fixtures = Path(__file__).parent / 'fixtures'
        data = json.loads((fixtures / 'installer-dracut-lsblk.json').read_text())
        mounts = json.loads((fixtures / 'installer-dracut-findmnt.json').read_text())
        media = next(m for m in i.SystemProbe._mounts(json.dumps(mounts)) if m['source'] == '/dev/sr0')
        calls = []
        class ReadOnlyRunner:
            def run(self, args):
                calls.append(args)
                if args[0] == 'lsblk':
                    return json.dumps(data)
                if '--target' not in args:
                    return json.dumps(mounts)
                if args[args.index('--target') + 1] != i.DEFAULT_PAYLOAD:
                    raise i.InstallError('Real dracut stale backing path has no current mount')
                return json.dumps({'filesystems': [media]})
        probe = i.SystemProbe(ReadOnlyRunner())
        probe.physical_disk = lambda path: path == '/dev/vda'
        packet = struct.pack('=QQQ', os.makedev(11, 0) if backing_device is None else backing_device,
                             3462, backing_rdevice) + bytes(208)
        with contextlib.ExitStack() as stack:
            stack.enter_context(patch.object(Path, 'is_file', return_value=True))
            stack.enter_context(patch.object(Path, 'read_text', return_value='/LiveOS/squashfs.img'))
            opened = stack.enter_context(patch('os.open', return_value=91))
            closed = stack.enter_context(patch('os.close'))
            stack.enter_context(patch('os.fstat', return_value=SimpleNamespace(st_mode=opened_mode,
                st_rdev=os.makedev(7, 0) if opened_device is None else opened_device)))
            # Patch the syscall even before the installer imports fcntl.
            syscall = stack.enter_context(patch('fcntl.ioctl', side_effect=ioctl_error,
                                                return_value=packet))
            topo = probe.snapshot(i.DEFAULT_PAYLOAD)
            opened.assert_called_once_with('/dev/loop0', os.O_RDONLY | os.O_CLOEXEC)
            closed.assert_called_once_with(91)
        return probe, topo, calls, syscall
    def test_actual_switched_root_topology_resolves_loop_by_kernel_identity(self):
        probe, topo, calls, syscall = self.probe()
        self.assertFalse(topo.component('/dev/vda') & topo.protected)
        self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
        self.assertTrue(topo.component('/dev/sr0') & topo.protected)
        self.assertEqual(syscall.call_args.args, (91, 0x4C05, bytes(232)))
        self.assertFalse(any('/LiveOS/squashfs.img' in args for args in calls))
        probe.snapshot = lambda payload: topo
        self.assertEqual(i.validate_request(i.Request('/dev/vda', 'hydratest'), probe), '/dev/vda')
        probe.physical_disk = lambda path: path == '/dev/sr0'
        with self.assertRaises(i.InstallError):
            i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
        # A mounted loop backed by a normal disk still forbids that whole disk.
        topo.nodes['/dev/sr0']['type'] = 'disk'
        topo.nodes['/dev/sr0']['size'] = 32 * 1024 ** 3
        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
            i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
    def test_unknown_loop_filesystem_identity_fails_closed(self):
        for identity in (os.makedev(8, 1), os.makedev(0, 99)):
            with self.subTest(identity=identity), self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
                self.probe(backing_device=identity)
    def test_unreadable_or_changed_loop_device_fails_closed(self):
        cases = [{'ioctl_error': OSError('cannot read loop status')},
                 {'opened_device': os.makedev(7, 1)}, {'opened_mode': stat.S_IFREG}]
        for case in cases:
            with self.subTest(case=case), self.assertRaises(i.InstallError):
                self.probe(**case)
    def test_truncated_loop_status_fails_closed_and_closes_descriptor(self):
        with patch('os.open', return_value=91), patch('os.close') as closed, \
             patch('os.fstat', return_value=SimpleNamespace(st_mode=stat.S_IFBLK, st_rdev=os.makedev(7, 0))), \
             patch('fcntl.ioctl', return_value=bytes(16)):
            with self.assertRaisesRegex(i.InstallError, 'Malformed loop status'):
                i.SystemProbe._loop_backing_identity('/dev/loop0', '7:0')
            closed.assert_called_once_with(91)
    def test_direct_block_backing_uses_kernel_rdevice(self):
        _, topo, _, _ = self.probe(backing_device=os.makedev(0, 6), backing_rdevice=os.makedev(11, 0))
        self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
 class NestedLoopAncestryTests(unittest.TestCase):
    def probe(self, backings=None, *, indirect=False, payload_loop=False, shared=False):
        loop0 = disk('/dev/loop0', kind='loop', mounts=['/mnt/also'] if shared else [])
        loop0['maj:min'] = '7:0'
        loop1 = disk('/dev/loop1', kind='loop', mounts=[] if payload_loop else ['/mnt/valuable'])
        loop1['maj:min'] = '7:1'
        target, unrelated = disk('/dev/vda'), disk('/dev/vdb')
        target['maj:min'], unrelated['maj:min'] = '253:0', '253:1'
        media = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
        media['maj:min'] = '11:0'
        nodes = [loop0, loop1, target, unrelated, media]
        if indirect:
            partition = disk('/dev/loop0p1', kind='part', pkname='/dev/loop0')
            partition['maj:min'] = '259:0'
            nodes.insert(1, partition)
        identities = backings or {'/dev/loop0': '253:0',
                                 '/dev/loop1': '253:0' if shared else '259:0' if indirect else '7:0'}
        mount = {'source': '/dev/loop1' if payload_loop else '/dev/sr0',
                 'target': '/payload' if payload_loop else '/run/initramfs/live',
                 'fstype': 'ext4' if payload_loop else 'iso9660',
                 'maj:min': '7:1' if payload_loop else '11:0', 'options': 'ro'}
        class ReadOnlyRunner:
            def run(self, args):
                if args[0] == 'lsblk':
                    return json.dumps({'blockdevices': nodes})
                full_mount = {'source': '/dev/sr0', 'target': '/run/initramfs/live',
                              'fstype': 'iso9660', 'maj:min': '11:0', 'options': 'ro'}
                return json.dumps({'filesystems': [mount] if '--target' in args else [full_mount]})
        probe = i.SystemProbe(ReadOnlyRunner())
        probe.physical_disk = lambda path: path in {'/dev/vda', '/dev/vdb'}
        with patch.object(Path, 'is_file', return_value=True), \
             patch.object(i.SystemProbe, '_loop_backing_identity',
                          side_effect=lambda path, expected: identities[path]):
            topo = probe.snapshot(i.DEFAULT_PAYLOAD)
        probe.snapshot = lambda payload: topo
        return probe, topo
    def test_reverse_order_nested_loop_refuses_physical_backing(self):
        probe, _ = self.probe()
        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
    def test_reverse_order_nested_loop_allows_unrelated_disk(self):
        probe, topo = self.probe()
        self.assertIn('/dev/vda', topo.component('/dev/loop1'))
        self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
    def test_unknown_nested_backing_refuses_unrelated_installation(self):
        with self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
            self.probe({'/dev/loop1': '7:0', '/dev/loop0': '0:99'})
    def test_nested_backing_cycles_fail_closed(self):
        for backings in ({'/dev/loop1': '7:0', '/dev/loop0': '7:1'},
                         {'/dev/loop1': '7:1', '/dev/loop0': '253:0'}):
            with self.subTest(backings=backings), self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
                self.probe(backings)
    def test_nested_backing_through_loop_partition_refuses_disk(self):
        probe, _ = self.probe(indirect=True)
        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
    def test_cycle_through_loop_partition_fails_closed(self):
        with self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
            self.probe({'/dev/loop1': '259:0', '/dev/loop0': '7:1'}, indirect=True)
    def test_unmounted_payload_loop_ancestors_refuse_disk(self):
        probe, _ = self.probe(payload_loop=True)
        with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
            i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
    def test_two_loops_on_same_backing_are_not_a_directed_cycle(self):
        probe, topo = self.probe(shared=True)
        self.assertIn('/dev/loop0', topo.component('/dev/loop1'))
        self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
 class SnapshotTests(unittest.TestCase):
    def test_snapshot_connects_live_loop_to_its_physical_backing(self):
        with tempfile.TemporaryDirectory() as tmp:
@@ -576,8 +416,12 @@ class SnapshotTests(unittest.TestCase):
                    if args[0] == 'lsblk':
                        return json.dumps(data)
                    return json.dumps({'filesystems': [media] if '--target' in args else [media, loop]})
-            with patch.object(i.SystemProbe, '_loop_backing_identity',
+            real_read = Path.read_text
-                              return_value=media['maj:min']):
+            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'))
@@ -33,12 +33,9 @@ class ProfilesTests(unittest.TestCase):
                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                    Profile('guest', self.iso, **{field: value})
-    def test_invalid_media_path_types(self):
+    def test_missing_iso_and_duplicate_profile_preserve_existing(self):
-        for iso in (True, 42, ''):
+        with self.assertRaises(ValueError):
-            with self.subTest(iso=iso), self.assertRaises(ValueError):
+            Profile('guest', self.root / 'missing.iso')
                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()
@@ -46,44 +43,6 @@ class ProfilesTests(unittest.TestCase):
            save_profile(dest, Profile('guest', self.iso, memory=8192))
        self.assertEqual(dest.read_bytes(), original)
    def test_disk_only_profile_roundtrips_without_media(self):
        dest = self.root / 'guest.toml'
        save_profile(dest, Profile('guest'))
        self.assertIsNone(load_profile(dest).iso)
        self.assertNotIn('iso =', dest.read_text())
    def test_missing_media_does_not_block_loading_profile(self):
        dest = self.root / 'guest.toml'
        save_profile(dest, Profile('guest', self.iso))
        self.iso.unlink()
        self.assertEqual(load_profile(dest).iso, self.iso)
    def test_display_bounds_and_types(self):
        for field, values in {'width': (0, True, 16385), 'height': (0, '1080', 16385),
                              'refresh_millihz': (-1, True, 1000001)}.items():
            for value in values:
                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                    Profile('guest', self.iso, **{field: value})
    def test_explicit_mac_roundtrip_and_validation(self):
        dest = self.root / 'guest.toml'
        save_profile(dest, Profile('guest', mac='02:ab:cd:ef:01:23'))
        self.assertEqual(load_profile(dest).mac, '02:ab:cd:ef:01:23')
        for value in ('ff:00:00:00:00:01', '00:00:00:00:00:01',
                      '02:00:00:00:00:zz', '02:00:00:00:00', True):
            with self.subTest(mac=value), self.assertRaises(ValueError):
                Profile('guest', mac=value)
    def test_refresh_vertical_descriptor_boundaries(self):
        for value in (0, 50000, 255000):
            path = self.root / 'guest.toml'
            save_profile(path, Profile('guest', refresh_millihz=value))
            self.assertEqual(load_profile(path).refresh_millihz, value)
            path.unlink()
        for value in (1, 30000, 49999, 255001, 1000000):
            with self.subTest(refresh=value), self.assertRaises(ValueError):
                Profile('guest', refresh_millihz=value)
    def test_profile_name_must_match_filename(self):
        path = self.root / 'guest.toml'
        path.write_text(f'name = "other"\niso = "{self.iso}"\n')
@@ -88,90 +88,6 @@ class RuntimeTests(unittest.TestCase):
        except VMError:
            pass
    def test_mac_persisted_distinct_and_stable_across_restarts(self):
        self.runtime.add(Profile('other', network='bridge'))
        other = self.runtime.profile('other')
        guest = self.runtime.profile('guest')
        self.assertIsNotNone(guest.mac)
        self.assertNotEqual(guest.mac, other.mac)
        octet = int(guest.mac.split(':')[0], 16)
        self.assertEqual(octet & 3, 2)  # locally administered, unicast
        def nic(profile, runtime):
            return next(arg for arg in build_qemu_argv(profile, runtime.paths(profile.name))
                        if arg.startswith('virtio-net-pci,'))
        expected = nic(guest, self.runtime)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.assertEqual(nic(self.runtime.profile('guest'), self.runtime), expected)
        changed_runtime = Runtime(self.runtime.config, self.runtime.state, self.root / 'different-run')
        self.assertEqual(nic(changed_runtime.profile('guest'), changed_runtime), expected)
        self.runtime.media('guest', None)
        self.assertEqual(self.runtime.profile('guest').mac, guest.mac)
    def test_legacy_profiles_gain_stable_owner_specific_argv_mac(self):
        file = self.runtime.config / 'legacy.toml'
        save_profile(file, Profile('legacy', network='bridge'))
        original = file.read_bytes()
        legacy = self.runtime.profile('legacy')
        self.assertIsNone(legacy.mac)
        paths = self.runtime.paths('legacy')
        def mac():
            return next(arg for arg in build_qemu_argv(legacy, paths) if arg.startswith('virtio-net-pci,'))
        with patch('os.getuid', return_value=1000):
            first = mac()
            self.assertEqual(mac(), first)
            self.assertNotEqual(first, next(arg for arg in build_qemu_argv(
                legacy, self.runtime.paths('other')) if arg.startswith('virtio-net-pci,')))
        with patch('os.getuid', return_value=1001):
            self.assertNotEqual(mac(), first)
        self.assertEqual(file.read_bytes(), original)
    def test_disk_boot_omits_media_and_requests_display(self):
        profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
                          graphics='software')
        argv = build_qemu_argv(profile, self.runtime.paths('guest'))
        self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
        self.assertNotIn('ide-cd,drive=iso', argv)
        self.assertIn('order=c', argv)
        devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
        self.assertEqual(len(devices), 2)
        stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
        self.assertFalse(any('refresh_rate=' in arg for arg in stock))
    def test_missing_iso_is_error_at_start_without_creating_disk(self):
        self.iso.unlink()
        with self.assertRaisesRegex(VMError, 'ISO'):
            self.runtime.start('guest')
        self.assertFalse(self.runtime.paths('guest').disk.exists())
        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
    def test_media_changes_only_while_stopped_and_preserves_disk(self):
        paths = self.runtime.paths('guest')
        self.runtime.start('guest')
        disk = paths.disk.read_bytes()
        config = (self.runtime.config / 'guest.toml').read_bytes()
        for iso in (None, self.iso):
            with self.assertRaises(VMError):
                self.runtime.media('guest', iso)
        self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
        self.iso.unlink()
        self.assertEqual(self.runtime.profile('guest').iso, self.iso)
        self.runtime.stop('guest', force=True)
        self.runtime.media('guest', None)
        self.assertIsNone(self.runtime.profile('guest').iso)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.assertEqual(paths.disk.read_bytes(), disk)
        replacement = self.root / 'replacement.iso'
        replacement.touch()
        self.runtime.media('guest', replacement)
        self.assertEqual(self.runtime.profile('guest').iso, replacement)
        with self.assertRaisesRegex(VMError, 'ISO'):
            self.runtime.media('guest', self.iso)
        self.assertEqual(self.runtime.profile('guest').iso, replacement)
    def test_argv_is_private_and_comma_safe(self):
        paths = self.runtime.paths('guest')
        argv = build_qemu_argv(self.profile, paths)
@@ -185,7 +101,7 @@ class RuntimeTests(unittest.TestCase):
        self.assertIn('disable-agent-file-xfer=on', spice)
        self.assertIn('run,,comma', spice)
        self.assertIn('virtio-blk-pci,drive=disk', argv)
-        self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
+        self.assertIn('virtio-net-pci,netdev=net', argv)
        self.assertNotIn('-vnc', argv)
        self.assertNotIn('-virtfs', argv)
@@ -55,11 +55,6 @@ class SessionTests(unittest.TestCase):
        self.assertEqual(tree.findall('./mouse/context/mousebind/action'), [])
        self.assertEqual(tree.findtext('./libinput/device/tap'), 'yes')
    def test_empty_session_has_one_workspace_without_empty_names(self):
        tree = ET.fromstring(session.build_labwc_config([], ['foot']))
        self.assertEqual(tree.find('./desktops').get('number'), '1')
        self.assertIsNone(tree.find('./desktops/names'))
    def test_viewer_argv_is_private_safe_and_manual_usb_only(self):
        paths = self.manager.paths('alpha')
        argv = session.build_viewer_argv(paths, '/tool path/viewer')
@@ -87,8 +82,7 @@ class SessionTests(unittest.TestCase):
            self.assertEqual(main(base + ['view', 'alpha']), 1)
        self.assertIn('viewer', error.getvalue().lower())
-    def test_windowed_viewer_exit_with_missing_iso_leaves_vm_control_untouched(self):
+    def test_windowed_viewer_exit_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)
@@ -178,45 +172,6 @@ else:
            self.assertEqual(main(['_session-helper', 'recover', str(manifest)]), 1)
        self.assertIn('manifest', error.getvalue().lower())
    def test_empty_launch_refresh_rebuilds_profile_snapshot(self):
        manager = Runtime(self.root / 'empty-config', self.root / 'empty-state', self.root / 'empty-run')
        capture = self.root / 'refresh-capture.json'
        compositor = self.root / 'refresh-labwc'
        compositor.write_text('#!' + os.sys.executable + '\n' + r'''import json, os, subprocess, sys
 from pathlib import Path
 if '--version' in sys.argv:
    print('labwc 0.20.2')
    sys.exit(0)
 config = Path(sys.argv[sys.argv.index('-C') + 1])
 manifest = config.parent / 'session.json'
 data = json.loads(manifest.read_text())
 capture = Path(os.environ['CAPTURE'])
 seen = json.loads(capture.read_text()) if capture.exists() else []
 seen.append({'names': data['names'], 'xml': (config / 'rc.xml').read_text()})
 capture.write_text(json.dumps(seen))
 if len(seen) == 1:
    launcher = os.environ['HYDRA_SOURCE_LAUNCHER']
    for name in ('first', 'second'):
        subprocess.run([sys.executable, launcher, '--config', data['config'], '--state', data['state'],
                        '--runtime', data['runtime'], 'add', name], check=True)
    subprocess.run([sys.executable, launcher, '_session-helper', 'recover', str(manifest)],
                   input=':refresh\n', text=True, check=True)
    sys.exit(67)
 sys.exit(0)
 ''')
        compositor.chmod(0o700)
        with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
                                     'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture),
                                     'HYDRA_SOURCE_LAUNCHER': str(Path('bin/hydra').resolve())}), contextlib.redirect_stdout(io.StringIO()):
            self.assertEqual(session.launch_session(manager), 0)
        seen = json.loads(capture.read_text())
        self.assertEqual([entry['names'] for entry in seen], [[], ['first', 'second']])
        tree = ET.fromstring(seen[1]['xml'])
        self.assertEqual(tree.find('desktops').get('number'), '2')
        self.assertEqual([rule.get('title') for rule in tree.findall('./windowRules/windowRule')],
                         ['HydraOS: first', 'HydraOS: second'])
        self.assertFalse(manager.paths('first').disk.exists())
    def test_no_running_vms_opens_recovery_without_start_or_stop(self):
        terminal = self.root / 'terminal'
        capture = self.root / 'recovery-argv.json'
@@ -1,48 +0,0 @@
 """Evidence gates used by the real private-display integration probe."""
 import tempfile
 import unittest
 from pathlib import Path
 from dev import smoke
 class SmokeEvidenceTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='hydra-smoke-test-')
        self.addCleanup(self.tmp.cleanup)
        root = Path(self.tmp.name)
        self.hidden = root / 'hidden.log'
        self.visible = root / 'visible.log'
        self.hidden.write_text('wl_keyboard#21.enter(1)\nwl_keyboard#21.leave(2)\n')
        self.visible.write_text('wl_keyboard#21.enter(3)\n')
    def test_focus_parser_ignores_pointer_events(self):
        self.hidden.write_text(self.hidden.read_text() + 'wl_pointer#22.enter(4)\n')
        self.assertEqual(smoke.focus_events(self.hidden), ['enter', 'leave'])
    def test_old_focus_events_cannot_prove_a_new_switch(self):
        self.assertFalse(smoke.fresh_focus_transition(['enter', 'leave'], ['enter'],
                                                     ['enter', 'leave'], ['enter']))
        self.assertTrue(smoke.fresh_focus_transition(['enter'], ['leave'],
                                                    ['enter', 'leave'], ['leave', 'enter']))
    def test_hidden_guest_must_advance_during_hidden_interval(self):
        with self.assertRaisesRegex(AssertionError, 'hidden guest stopped'):
            smoke.measure_hidden_interval(lambda: 10, self.hidden, self.visible, duration=.01)
        readings = iter((10, 20))
        result = smoke.measure_hidden_interval(lambda: next(readings), self.hidden, self.visible, duration=.01)
        self.assertEqual((result['before'], result['after']), (10, 20))
        self.assertGreaterEqual(result['elapsed_seconds'], .01)
    def test_focus_change_during_sampling_invalidates_hidden_evidence(self):
        def reading():
            self.hidden.write_text(self.hidden.read_text() + 'wl_keyboard#21.enter(4)\n')
            return 10
        with self.assertRaisesRegex(AssertionError, 'focus changed'):
            smoke.measure_hidden_interval(reading, self.hidden, self.visible, duration=.01)
    def test_compositor_classifier_does_not_hide_other_errors(self):
        occupied = '[ERROR] [xwayland/sockets.c:47] Failed to bind socket @/tmp/.X11-unix/X2: Address already in use'
        self.assertEqual(smoke.classify_compositor_errors(occupied), [occupied])
        with self.assertRaisesRegex(AssertionError, 'Unexpected compositor'):
            smoke.classify_compositor_errors(occupied + '\n[ERROR] [config.c:1] invalid configuration')
Author	SHA1	Message	Date
Codex	4a636924df	Capture redacted guest serial while waiting for shutdown	2026-10-13 05:47:48 +03:00
Codex	17ade497a4	Wait for installer exit and live shell before followup probes	2026-10-13 05:39:59 +03:00
Codex	32e9334feb	Fix ISO tool checks and fail promptly on installer refusal	2026-10-13 05:33:27 +03:00
Codex	c490991054	Distinguish closed login scopes during lifecycle acceptance	2026-10-13 04:55:37 +03:00
Codex	972b82c24b	Probe installed user VM and audio continuity across logout	2026-10-13 04:49:54 +03:00
Codex	f46fef764b	Make ISO acceptance probes fail on forbidden state	2026-10-13 04:35:05 +03:00
Codex	e93820bd52	Add disposable UEFI ISO install acceptance harness	2026-10-13 04:30:20 +03:00
		`@@ -1 +0,0 @@`
			`bdff621037ce9220767b7869499823c9b840b76c0b38268bc7f56a7b54390ce0 hydra-os.iso`
		`@@ -1 +0,0 @@`
			`DIST brightnessctl-0.5.1.tar.gz 8284 BLAKE2B 7d2226aae10fddc606c5ce71ed4a04eda583acf245ea1532d8730c73dc02ea7aeed47a0e25ca7f9d40e7b0170bb08d87617e9f4aa082495e78894982870ca7cc SHA512 41ac86357b3b5a3d54a043140a6bdb1fb68d5080a16e86692059ad58f4b9134a8532432a3f664c70c665a4ad73ea8a94cc9b88fa84abc28f0356b8b397aaebc7`
		`@@ -1 +0,0 @@`
			`HydraOS live medium: changes to the host root are not persistent.`