"""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)))) if names: labels = ET.SubElement(desktops, 'names') for name in names: ET.SubElement(labels, 'name').text = name if initial is not None: ET.SubElement(desktops, 'initial').text = str(names.index(initial) + 1) ET.SubElement(desktops, 'popupTime').text = '0' 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') _private_dir(runtime.runtime) while True: profiles = runtime.profiles() initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None) with tempfile.TemporaryDirectory(prefix='session-', dir=runtime.runtime) as tmp: directory = Path(tmp) config = directory / 'labwc' config.mkdir(mode=0o700) manifest = directory / 'session.json' manifest.write_text(json.dumps({'config': str(runtime.config), 'state': str(runtime.state), 'runtime': str(runtime.runtime), 'names': [p.name for p in profiles], 'terminal': terminal, 'directory': str(directory)}), encoding='utf-8') (config / 'rc.xml').write_text(build_labwc_config([p.name for p in profiles], _recovery_argv(manifest, terminal), brightness, initial=initial), encoding='utf-8') (config / 'autostart').write_text('#!/bin/sh\n' + shlex.join(_module_argv('clients', manifest)) + ' &\n', encoding='utf-8') (config / 'environment').touch() env['PYTHONUNBUFFERED'] = '1' compositor = subprocess.Popen([labwc, '-C', str(config)], env=env) refresh = False try: while True: if (directory / 'refresh.request').exists(): refresh = True break if compositor.poll() is not None: # A requesting recovery terminal can exit before polling. refresh = (directory / 'refresh.request').exists() break time.sleep(.1) finally: if compositor.poll() is None: compositor.terminate() compositor.wait(timeout=10) if not refresh: return compositor.returncode # Only our compositor was stopped. Its viewers disconnect; detached # QEMU instances retain their identity and running workloads. 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', 'refresh') and ( not command.startswith('reconnect:') or command.split(':', 1)[1] not in data['names']): raise SessionError('Invalid session helper operation') runtime = Runtime(Path(data['config']), Path(data['state']), Path(data['runtime'])) if command == 'refresh': fd = os.open(manifest.parent / 'refresh.request', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) os.close(fd) print('Refreshing workspaces and viewers; VMs keep running.') return 0 if command == 'recover': print('HydraOS local recovery. VMs keep running independently.\n' 'Ctrl+Alt+Shift+Left/Right switches workspaces; Ctrl+Alt+Shift+R releases input; Ctrl+Alt+Shift+U opens USB.\n' 'Enter a VM name to reconnect, :shell for a host shell, :refresh to rebuild workspaces/viewers, or :quit.') if not data['names']: print('No VM profiles yet. Use :shell to run hydra add NAME --iso PATH and hydra start NAME.\n' 'After adding or removing profiles, leave the shell and enter :refresh; running VMs are preserved.') while True: try: print(json.dumps([runtime.status(name) for name in data['names']], indent=2)) choice = input('hydra recovery> ').strip() except (EOFError, KeyboardInterrupt): 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.') 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