feat: add fullscreen HydraOS session

dev/tools.nix

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

hydra/cli.py

@@ -3,11 +3,13 @@
 import argparse
 import json
 import os
+import subprocess
 import sys
 from pathlib import Path
 
 from .profiles import Profile
 from .runtime import Runtime, RuntimeError
+from .session import launch_session, session_helper, view
 
 
 def parser() -> argparse.ArgumentParser:
@@ -33,6 +35,14 @@ def parser() -> argparse.ArgumentParser:
         cmd.add_argument('name')
         if command == 'stop':
             cmd.add_argument('--force', action='store_true', required=True)
+    viewer = commands.add_parser('view', help='Connect a viewer without changing VM power')
+    viewer.add_argument('name')
+    viewer.add_argument('--windowed', action='store_true')
+    session = commands.add_parser('session', help='Start the minimal HydraOS labwc session')
+    session.add_argument('--nested', action='store_true')
+    helper = commands.add_parser('_session-helper', help=argparse.SUPPRESS)
+    helper.add_argument('operation')
+    helper.add_argument('manifest', type=Path)
     return cli
 
 
@@ -40,7 +50,13 @@ def main(argv: list[str] | None = None) -> int:
     args = parser().parse_args(argv)
     runtime = Runtime(args.config, args.state, args.runtime)
     try:
-        if args.command == 'add':
+        if args.command == 'view':
+            return view(runtime, args.name, windowed=args.windowed)
+        elif args.command == 'session':
+            return launch_session(runtime, nested=args.nested)
+        elif args.command == '_session-helper':
+            return session_helper(args.operation, args.manifest)
+        elif args.command == 'add':
             profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size,
                               args.graphics, args.network, args.bridge)
             runtime.add(profile)
@@ -55,6 +71,6 @@ def main(argv: list[str] | None = None) -> int:
             result = getattr(runtime, args.command)(args.name)
         print(json.dumps(result, indent=2))
         return 0
-    except (RuntimeError, OSError, ValueError) as exc:
+    except (RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc:
         print(f'hydra: {exc}', file=sys.stderr)
         return 1

hydra/session.py

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

patches/virt-viewer-hydra-session.patch

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

session/hydra-session

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

tests/test_session.py

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