From fb9166ed9e8d51fdeaf60188e0c42827baa8d883 Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Tue, 29 Sep 2026 15:33:27 -0700 Subject: [PATCH] feat(settings): run local VocaGateway from Advanced (podman-first) (#774) * feat(settings): run local VocaGateway from Speech Engine (podman-first) Add an optional Local VocaGateway panel next to Remote Server so Linux users can start the pinned v0.1.0 compose project with podman (docker fallback), watch Live/Pairable/Ready status, pair a phone without loopback QR URLs, and preset remote_api to the gateway transcriptions endpoint. Stop preserves volumes; Flatpak fails closed; tray offers Stop only when this session manages the gateway. Default on-device engine stays whisper.cpp. * fix(gateway-embed): drop nonexistent compose cpu profile VocaGateway v0.1.0 exposes default service gateway plus native/cuda/vulkan profiles, not cpu. Start and stop the default service without --profile, keep the latest image pin rejection, and prefer save_config for the remote_api preset. * fix(gateway-embed): keep compose --profile cpu per embed contract v0.1.0 has no cpu profile entry, but Compose still starts the unprofiled gateway service when --profile cpu is set, and keeps cuda/vulkan/native off. That matches the frozen desktop embed contract and stays forward-compatible. * fix(gateway-embed): harden embed path for settings and compose Keep podman/docker probing off the GTK thread, reject link-local and loopback public URLs for phone QR, sanitize compose .env image overrides, and cap bootstrap token reads so a hostile file cannot blow memory. * fix(gateway-embed): keep LAN toggle and phone QR in sync When Allow LAN flips while compose is already managed, rewrite publish host and PUBLIC_URL and recreate the gateway before Pairable comes back. Also reject default docker/podman bridge addresses for phone QR. * test(gateway-embed): pin bridge and LAN-gate pairing cases Cover 172.17.0.2 / link-local not becoming Pairable, and the mid-run LAN toggle path that must wait for compose republish before advertising a phone URL. * style(gateway-embed): black format embed modules * fix(gateway-embed): reject short-form loopback and localhost.* for QR Align Pairable/QR host checks with vocagateway desktop-embed policy so 127.1 / localhost.* never become phone-reachable display URLs. * fix(gateway-embed): detach Settings status listener on destroy Gateway poll callbacks were left registered after Settings closed, so idle_add could touch destroyed GTK widgets while compose stayed up. * fix(gateway-embed): pass host_env to bridge ip probe _bridge_iface_ips shells out to host ip; without host_env the AST gate fails and AppImage bundles can hand the wrong GLib to ip. * fix(packaging): ship gateway_embed in the AppImage The explicit setuptools packages list omitted the new package, so the bundle booted from source trees but died with ModuleNotFoundError on every distro boot test. * style(packaging): black the gateway_embed package guard * fix(settings): pack local VocaGateway into the Advanced island The rebase onto main auto-merged textually onto the simple Speech Model page. Keep the gateway card on advanced_box, next to Remote Server. * fix(gateway-embed): treat missing Compose as unavailable When podman or docker is present but Compose is not, stop advertising Run. Return empty compose_args, show an install-Compose hint, and cover it with a unit test. Also add module loggers on the new gateway_embed modules and annotate the gateway Settings/tray callbacks Greptile flagged. just lock found no lockfile drift. * fix(gateway-embed): keep Allow LAN bind matched to the switch Snapshot the LAN flag for start/republish, record what compose actually got, then republish again if the switch moved mid-flight so 0.0.0.0 never lingers while Allow LAN reads off. * fix(gateway-embed): detect leftover compose after Quit Quit does not stop compose, so a LAN-published gateway can keep running. Settings and the tray treated that as Stopped because managed_by_us is only process memory. Probe compose after runtime detection, show Stop when it is actually up, and let Stop work even if this session did not start it. * fix(gateway-embed): restore LAN switch from leftover compose Quit leaves compose running, so the next launch now reads VOCAGATEWAY_PUBLISH_HOST from the embed .env, adopts the project as managed, and keeps Allow LAN honest so republish can close 0.0.0.0. * fix(gateway-embed): restore .env if LAN republish fails compose recreate can fail after write_env_file already pointed VOCAGATEWAY_PUBLISH_HOST at loopback. Restore the previous .env so orphan adopt still sees a live 0.0.0.0 bind, and snap Allow LAN back to the running publish mode. * style(gateway-embed): type helper args and black MagicMock * refactor(types): move Readable protocol to common_types AGENTS.md keeps Protocol interfaces in common_types. Pairing now imports the shared Readable surface instead of a local _Readable helper. * fix(gateway-embed): satisfy mypy on rebased embed code * fix(audio): keep duck restore point when a set reports failure --- README.md | 3 +- docs/GATEWAY_EMBED.md | 77 ++ docs/HTTP_REMOTE.md | 4 + packaging/appimage/boot-test.sh | 3 +- pyproject.toml | 1 + src/vocalinux/audio/playback_ducker.py | 26 +- src/vocalinux/common_types.py | 6 + src/vocalinux/gateway_embed/__init__.py | 28 + src/vocalinux/gateway_embed/health.py | 58 ++ src/vocalinux/gateway_embed/manager.py | 445 ++++++++++ src/vocalinux/gateway_embed/pairing.py | 146 +++ src/vocalinux/gateway_embed/paths_embed.py | 66 ++ src/vocalinux/gateway_embed/preset.py | 49 ++ src/vocalinux/gateway_embed/runner.py | 483 ++++++++++ src/vocalinux/gateway_embed/runtime.py | 142 +++ src/vocalinux/gateway_embed/sandbox.py | 57 ++ src/vocalinux/gateway_embed/status.py | 51 ++ src/vocalinux/gateway_embed/urls.py | 184 ++++ src/vocalinux/ui/config_manager.py | 4 + src/vocalinux/ui/settings_dialog.py | 307 ++++++- src/vocalinux/ui/tray_indicator.py | 40 +- tests/test_appimage_packaging.py | 24 + tests/test_gateway_embed.py | 978 +++++++++++++++++++++ tests/test_playback_ducker.py | 41 + tests/test_settings_dialog.py | 2 + 25 files changed, 3214 insertions(+), 11 deletions(-) create mode 100644 docs/GATEWAY_EMBED.md create mode 100644 src/vocalinux/gateway_embed/__init__.py create mode 100644 src/vocalinux/gateway_embed/health.py create mode 100644 src/vocalinux/gateway_embed/manager.py create mode 100644 src/vocalinux/gateway_embed/pairing.py create mode 100644 src/vocalinux/gateway_embed/paths_embed.py create mode 100644 src/vocalinux/gateway_embed/preset.py create mode 100644 src/vocalinux/gateway_embed/runner.py create mode 100644 src/vocalinux/gateway_embed/runtime.py create mode 100644 src/vocalinux/gateway_embed/sandbox.py create mode 100644 src/vocalinux/gateway_embed/status.py create mode 100644 src/vocalinux/gateway_embed/urls.py create mode 100644 tests/test_gateway_embed.py diff --git a/README.md b/README.md index f4c6cd7..18b7ebb 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ No Voca account is required. Models download once. After that, speech-to-text st - **On-device after model download**: Local engines; speech-to-text stays on your machine - **X11 and Wayland**: Text injection via xdotool, IBus, wtype, ydotool, or clipboard fallback -- **Several engines**: whisper.cpp (default), OpenAI Whisper, Faster Whisper, VOSK, Parakeet, plus optional remote HTTP API +- **Several engines**: whisper.cpp (default), OpenAI Whisper, Faster Whisper, VOSK, Parakeet, plus optional remote HTTP API or local VocaGateway from Settings (not on-device) - **GPU acceleration**: Vulkan for AMD, Intel, and NVIDIA with whisper.cpp - **Toggle or push-to-talk**: New installs default to hold Right Alt; existing configs keep their shortcut - **System tray + settings**: Searchable sidebar, Speech Model simple setup with Advanced as an island, status icons, audio feedback @@ -309,6 +309,7 @@ Neural VAD (Silero) is used when `onnxruntime` is available; install via `pip in | [Support](SUPPORT.md) | Where to get help | | [Distribution compatibility](docs/DISTRO_COMPATIBILITY.md) | Distro matrix and session notes | | [Remote HTTP API](docs/HTTP_REMOTE.md) | Offload transcription to a server | +| [Local VocaGateway (optional)](docs/GATEWAY_EMBED.md) | Start a local gateway from Settings; not on-device | | [Contributing](CONTRIBUTING.md) | Dev setup, style, PR process | | [Security](SECURITY.md) | Supported versions and vulnerability reporting | | [Docs index](docs/README.md) | Full documentation map | diff --git a/docs/GATEWAY_EMBED.md b/docs/GATEWAY_EMBED.md new file mode 100644 index 0000000..81c4af1 --- /dev/null +++ b/docs/GATEWAY_EMBED.md @@ -0,0 +1,77 @@ +# Local VocaGateway from Vocalinux (optional) + +Vocalinux can start a **local** [VocaGateway](https://github.com/VocaHQ/vocagateway) +container from **Settings → Speech Model → Advanced**, next to Remote Server. This is an +optional power-user path. It is **not** on-device processing: microphone audio +still leaves the desktop client and is transcribed by the gateway process on +your machine (or LAN). + +The default speech engine remains **whisper.cpp** on-device. Enabling a local +gateway never changes that default by itself. Use **Use this Gateway** only when +you want Vocalinux to talk to the gateway over the existing Remote API settings +(`remote_api_*`, endpoint `/v1/audio/transcriptions`). See [HTTP_REMOTE.md](HTTP_REMOTE.md). + +## Requirements (Linux MVP) + +- **podman** (preferred) or **docker**, with a working `compose` plugin (or + `podman-compose` / `docker-compose`). +- Vocalinux does **not** bundle a container engine inside the AppImage. +- **Flatpak**: the sandbox cannot reach the host container socket. Run the + native package or AppImage, or start VocaGateway on the host yourself and + point Remote Server at it. + +## What Vocalinux does + +1. Pins upstream **VocaGateway `v0.1.0`** (git tag) under your XDG cache and uses + that checkout's `compose.yaml` with Compose project name `vocagateway`. +2. Writes a private bootstrap token under `~/.config/vocalinux/gateway_embed/` + (mode `600`) and a matching Compose `.env`. +3. Runs `compose -f compose.yaml --profile cpu up -d` for the `gateway` service. + The v0.1.0 GitHub release has no published container image assets, so the first + start **builds** the local image `vocagateway:v0.1.0` from that tag (override with + `VOCAGATEWAY_IMAGE`; never `latest`). Later starts reuse the image/volume. +4. Polls `GET /health/live` and `GET /health/ready`, and (when live) authenticated + `GET /v1/admin/pairing` for phone pairing. +5. Stops with `compose down` **without** `-v`, so the named data volume (models, + config) is kept. + +## Status labels + +| Status | Meaning | +| --- | --- | +| Stopped | Not running. | +| Starting | Compose up in progress, or live probe not green yet. | +| Live | Process answers `/health/live`. | +| Pairable | Live, plus a **non-loopback** phone URL and bearer token (QR-safe). | +| Ready | `/health/ready` is 200 (model selected in the gateway WebUI). | +| Error | Start/stop or probe failure (see the subtitle). | + +Pairable can appear before Ready: you can pair a phone while a model is still +downloading. + +## LAN / phone + +By default the published host is `127.0.0.1` (desktop-only). Turn on **Allow LAN +access for Phone** to set `VOCAGATEWAY_PUBLISH_HOST=0.0.0.0` and to advertise a +LAN URL in pairing. Open port `8765` in your firewall only on trusted networks. +Never put `127.0.0.1` / `localhost` in the phone QR. If a LAN toggle fails, +compose keeps the previous bind, Vocalinux restores the embed `.env`, and the +switch snaps back with an error (Stop and Run again, or retry). + +## Tray and Quit + +The tray menu offers **Stop local Gateway** whenever compose is actually running +(Starting, Live, Pairable, Ready, or Error), including a leftover from a previous +Vocalinux session. Vocalinux does not auto-start the gateway on login in v1. + +**Quit does not stop compose.** The pinned compose file uses `restart: +unless-stopped`, so a LAN-published gateway on `0.0.0.0:8765` keeps running after +you quit Vocalinux. The next launch restores the Allow LAN switch from the running +compose publish host. Use **Stop local Gateway** in Settings or the tray when you +want it gone. Do not assume Quit tears it down. + +## Honesty + +VocaGateway is optional self-hosted infrastructure for the Voca family. There is +no Voca account and no hosted Voca cloud. Prefer a trusted LAN, Tailscale, or +HTTPS. Do not expose port 8765 to the public internet. diff --git a/docs/HTTP_REMOTE.md b/docs/HTTP_REMOTE.md index 85f9dc7..ab3d65e 100644 --- a/docs/HTTP_REMOTE.md +++ b/docs/HTTP_REMOTE.md @@ -11,6 +11,10 @@ Pick whichever your server exposes. The rest of this guide applies to both. > **Tip: share the server with your phone.** These HTTP formats are open enough that the same self-hosted server can also back mobile dictation apps. On Android, apps like *Dictate* and *Transcribro* speak OpenAI-compatible Whisper; on iOS, Shortcuts-based dictation clients can hit the same endpoint. Run one Whisper server, point your laptop and phone at it, and you get consistent dictation everywhere without uploading audio to a third party. +## Local VocaGateway (optional) + +On Linux you can start an optional local [VocaGateway](https://github.com/VocaHQ/vocagateway) from **Settings → Speech Model → Advanced** (podman-first, docker fallback). That path uses the same Remote API settings described below, with endpoint `/v1/audio/transcriptions`. It is **not** on-device recognition: audio goes to the gateway container on your machine. Details: [GATEWAY_EMBED.md](GATEWAY_EMBED.md). + ## How it works When the **Remote API** engine is active, Vocalinux: diff --git a/packaging/appimage/boot-test.sh b/packaging/appimage/boot-test.sh index e6dd47e..4b92601 100755 --- a/packaging/appimage/boot-test.sh +++ b/packaging/appimage/boot-test.sh @@ -91,8 +91,9 @@ if IBus.Engine.__gtype__.name in ("void", "invalid"): raise SystemExit("IBus typelib loaded but its library did not") from vocalinux.text_injection import text_injector # noqa: F401 +from vocalinux import gateway_embed # noqa: F401 -print(" Gtk 3.0, IBus and text_injection all import") +print(" Gtk 3.0, IBus, text_injection and gateway_embed all import") PY echo "== Host binaries still run when the app spawns them ==" diff --git a/pyproject.toml b/pyproject.toml index e74e101..38dc5a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -125,6 +125,7 @@ required-version = ">=0.12,<0.13" packages = [ "vocalinux", "vocalinux.audio", + "vocalinux.gateway_embed", "vocalinux.resources", "vocalinux.resources.icons", "vocalinux.resources.icons.scalable", diff --git a/src/vocalinux/audio/playback_ducker.py b/src/vocalinux/audio/playback_ducker.py index 4a07b3b..74e83bb 100644 --- a/src/vocalinux/audio/playback_ducker.py +++ b/src/vocalinux/audio/playback_ducker.py @@ -500,9 +500,24 @@ class PlaybackDucker: logger.warning("Could not lower sink %s", sink_id, exc_info=True) return if not applied: - # The sink was not changed, so the record would restore nothing - # useful and might later skip a real user volume. - self._forget() + # A reported failure is only the tool's word for it: the + # command can still have applied before the failure was + # reported (a dropped reply, a daemon disconnect). Re-read + # the sink and drop the record only when it provably stayed + # at the original level; otherwise restore() is the only way + # back and the record has to stay. + try: + verify = self._control.volume_of(sink_id) + except Exception: + verify = None + if verify is not None and channels_match(verify, original_channels): + self._forget() + return + logger.warning( + "Sink %s reported a failed volume set without confirming it was " + "unchanged; keeping the restore point", + sink_id, + ) return logger.info( "Lowered sink %s from %s to %s (%d%%) while dictating", @@ -568,10 +583,7 @@ class PlaybackDucker: self._forget() def _current_percent(self) -> int: - from ..ui.config_manager import ( - DEFAULT_PLAYBACK_DUCK_PERCENT, - clamp_playback_duck_percent, - ) + from ..ui.config_manager import DEFAULT_PLAYBACK_DUCK_PERCENT, clamp_playback_duck_percent try: return clamp_playback_duck_percent(self._percent_fn()) diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index 5a4cae9..ee62bf7 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -127,3 +127,9 @@ class CancelableTimer(Protocol): def cancel(self) -> None: """Drop the scheduled call if it has not started.""" ... + + +class Readable(Protocol): + """Minimal read() surface for urlopen-like responses.""" + + def read(self, n: int = ...) -> bytes: ... diff --git a/src/vocalinux/gateway_embed/__init__.py b/src/vocalinux/gateway_embed/__init__.py new file mode 100644 index 0000000..fe49e9f --- /dev/null +++ b/src/vocalinux/gateway_embed/__init__.py @@ -0,0 +1,28 @@ +"""Optional local VocaGateway lifecycle helpers for Vocalinux. + +This is an opt-in path for power users who want to run +https://github.com/VocaHQ/vocagateway on the same Linux box via podman +(or docker). It is not on-device processing: audio still leaves the +desktop client and goes to the local gateway container. +""" + +import logging + +from .manager import GatewayEmbedManager, get_gateway_embed_manager +from .preset import remote_api_preset_from_pairing +from .runtime import ContainerRuntime, detect_container_runtime +from .status import GatewayStatus +from .urls import is_loopback_url, reject_loopback_url + +logger = logging.getLogger(__name__) + +__all__ = [ + "ContainerRuntime", + "GatewayEmbedManager", + "GatewayStatus", + "detect_container_runtime", + "get_gateway_embed_manager", + "is_loopback_url", + "reject_loopback_url", + "remote_api_preset_from_pairing", +] diff --git a/src/vocalinux/gateway_embed/health.py b/src/vocalinux/gateway_embed/health.py new file mode 100644 index 0000000..d2d0f72 --- /dev/null +++ b/src/vocalinux/gateway_embed/health.py @@ -0,0 +1,58 @@ +"""HTTP health probes for a local VocaGateway.""" + +from __future__ import annotations + +import logging +import urllib.error +import urllib.request +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +DEFAULT_BASE_URL = "http://127.0.0.1:8765" +DEFAULT_TIMEOUT = 2.5 + + +@dataclass(frozen=True) +class HealthSnapshot: + live: bool + ready: bool + live_status: int | None = None + ready_status: int | None = None + error: str = "" + + +def _get_status(url: str, *, timeout: float) -> tuple[int | None, str]: + request = urllib.request.Request(url, method="GET") + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return int(response.status), "" + except urllib.error.HTTPError as exc: + return int(exc.code), "" + except Exception as exc: # noqa: BLE001 - surface as soft error for UI + return None, str(exc) + + +def probe_health( + base_url: str = DEFAULT_BASE_URL, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> HealthSnapshot: + """Probe ``/health/live`` and ``/health/ready``.""" + root = base_url.rstrip("/") + live_status, live_err = _get_status(f"{root}/health/live", timeout=timeout) + ready_status, ready_err = _get_status(f"{root}/health/ready", timeout=timeout) + live = live_status == 200 + ready = ready_status == 200 + error = "" + if not live and live_err: + error = live_err + elif not ready and ready_err and ready_status is None: + error = ready_err + return HealthSnapshot( + live=live, + ready=ready, + live_status=live_status, + ready_status=ready_status, + error=error, + ) diff --git a/src/vocalinux/gateway_embed/manager.py b/src/vocalinux/gateway_embed/manager.py new file mode 100644 index 0000000..154457f --- /dev/null +++ b/src/vocalinux/gateway_embed/manager.py @@ -0,0 +1,445 @@ +"""Session-scoped manager shared by Settings and the tray.""" + +from __future__ import annotations + +import logging +import os +import socket +import threading +from typing import Any, Callable, List, Optional + +from .health import probe_health +from .pairing import PairingInfo, fetch_pairing +from .preset import apply_remote_api_preset, remote_api_preset_from_pairing +from .runner import GatewayRunner, read_lan_publish_from_env +from .status import GatewayStatus +from .urls import reject_loopback_url + +logger = logging.getLogger(__name__) + +StatusListener = Callable[[GatewayStatus, str], None] + + +def _guess_lan_url(port: int = 8765) -> Optional[str]: + """Best-effort LAN IPv4 for PUBLIC_URL when user opts into LAN publish.""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect(("8.8.8.8", 80)) + ip = sock.getsockname()[0] + finally: + sock.close() + except OSError: + return None + if not ip or ip.startswith("127."): + return None + candidate = f"http://{ip}:{port}" + return reject_loopback_url(candidate) + + +class GatewayEmbedManager: + """Owns start/stop, health polling, and managed-by-us tray visibility.""" + + def __init__(self, runner: GatewayRunner | None = None): + self.runner = runner or GatewayRunner() + self._status = GatewayStatus.STOPPED + self._status_detail = "" + self._pairing: Optional[PairingInfo] = None + self._token: str = "" + self._listeners: List[StatusListener] = [] + self._lock = threading.RLock() + self._poll_stop = threading.Event() + self._poll_thread: Optional[threading.Thread] = None + self.lan_publish = False + # What compose was last started/republished with (pairing uses this). + self._compose_lan_publish: Optional[bool] = None + self._runtime_detect_started = False + self._republish_started = False + if not self.runner.runtime_ready: + self.begin_runtime_detection() + + # ------------------------------------------------------------------ + # listeners (GTK should register wrappers that idle_add into UI) + # ------------------------------------------------------------------ + def add_listener(self, callback: StatusListener) -> None: + with self._lock: + if callback not in self._listeners: + self._listeners.append(callback) + + def remove_listener(self, callback: StatusListener) -> None: + with self._lock: + if callback in self._listeners: + self._listeners.remove(callback) + + def _emit(self, status: GatewayStatus, detail: str = "") -> None: + self._status = status + self._status_detail = detail + listeners = list(self._listeners) + for callback in listeners: + try: + callback(status, detail) + except Exception: # noqa: BLE001 + logger.exception("gateway status listener failed") + + @property + def status(self) -> GatewayStatus: + return self._status + + @property + def status_detail(self) -> str: + return self._status_detail + + @property + def pairing(self) -> Optional[PairingInfo]: + return self._pairing + + @property + def managed_by_us(self) -> bool: + return bool(self.runner.managed_by_us) + + @property + def available(self) -> bool: + return self.runner.available + + @property + def unavailable_hint(self) -> str: + return self.runner.unavailable_hint + + @property + def runtime_ready(self) -> bool: + return self.runner.runtime_ready + + def begin_runtime_detection(self) -> None: + """Probe podman/docker off the GTK main thread, then adopt leftover compose.""" + with self._lock: + if self._runtime_detect_started: + return + self._runtime_detect_started = True + + def _worker() -> None: + try: + self.runner.ensure_runtime() + except Exception: # noqa: BLE001 + logger.exception("container runtime detection failed") + if not self.runner.available: + self._emit(self._status, self.runner.unavailable_hint) + return + try: + from .paths_embed import ensure_token_file, token_file_path + + if os.path.isfile(token_file_path()): + self._token = ensure_token_file() + except Exception: # noqa: BLE001 + logger.exception("could not load existing gateway token") + self._adopt_orphan_compose() + status = self.refresh_status() + if status is not GatewayStatus.STOPPED: + self._start_polling() + + # Runtime already resolved (tests, or a runner passed in): probe now. + # Production still uses the worker thread so GTK never waits on compose ps. + if self.runner.runtime_ready: + _worker() + return + threading.Thread( + target=_worker, + name="vocalinux-gateway-runtime", + daemon=True, + ).start() + + def _adopt_orphan_compose(self) -> None: + """Restore LAN truth and take ownership of leftover compose after Quit.""" + if self.runner.managed_by_us: + return + if not self.runner.is_compose_running(): + return + published = read_lan_publish_from_env() + self._compose_lan_publish = published + self.lan_publish = published + self.runner.managed_by_us = True + logger.info("adopted leftover vocagateway compose (lan_publish=%s)", published) + + def start_async(self, *, lan_publish: bool = False) -> None: + self.lan_publish = lan_publish + thread = threading.Thread( + target=self._start_worker, + name="vocalinux-gateway-start", + daemon=True, + ) + thread.start() + + def stop_async(self) -> None: + thread = threading.Thread( + target=self._stop_worker, + name="vocalinux-gateway-stop", + daemon=True, + ) + thread.start() + + def apply_lan_publish(self, lan_publish: bool) -> None: + """Persist desired LAN flag; republish compose when we manage it. + + Until republish finishes, pairing stays non-Pairable so QR never claims + LAN while still bound to 127.0.0.1 (or the previous publish mode). + """ + self.lan_publish = bool(lan_publish) + if not self.runner.managed_by_us: + # Next Run will pick up lan_publish; clear stale QR if any. + if ( + self._compose_lan_publish is not None + and self._compose_lan_publish != self.lan_publish + ): + self._pairing = None + return + if self._compose_lan_publish == self.lan_publish: + return + # Drop Pairable immediately; republish worker restores it when safe. + self._pairing = None + with self._lock: + if self._republish_started: + return + self._republish_started = True + thread = threading.Thread( + target=self._republish_worker, + name="vocalinux-gateway-republish", + daemon=True, + ) + thread.start() + + def _republish_worker(self) -> None: + succeeded = False + try: + with self._lock: + published = bool(self.lan_publish) + self._emit( + GatewayStatus.STARTING, + "Updating LAN publish so the pairing URL matches…", + ) + public_url = _guess_lan_url() if published else None + if published and not public_url: + logger.info("LAN republish enabled but LAN IP could not be guessed") + result = self.runner.republish(lan_publish=published, public_url=public_url) + if not result.ok: + # Recreate failed: live bind is still the previous publish mode. + if self._compose_lan_publish is not None: + self.lan_publish = bool(self._compose_lan_publish) + self._emit( + GatewayStatus.ERROR, + "Could not update LAN publish; the previous bind is still in use. " + "Stop and Run again, or retry.", + ) + return + self._compose_lan_publish = published + self._pairing = None + self.refresh_status() + succeeded = True + finally: + with self._lock: + self._republish_started = False + if succeeded: + self._reconcile_lan_after_compose() + + def _effective_lan_for_pairing(self) -> bool: + """True only when desired LAN matches what compose actually published.""" + return bool(self.lan_publish) and self._compose_lan_publish is True + + def _reconcile_lan_after_compose(self) -> None: + """If Allow LAN moved during start/republish, match compose to the switch.""" + if not self.runner.managed_by_us: + return + if self._compose_lan_publish != bool(self.lan_publish): + self.apply_lan_publish(self.lan_publish) + + def _start_worker(self) -> None: + with self._lock: + published = bool(self.lan_publish) + self._emit(GatewayStatus.STARTING, "Fetching VocaGateway v0.1.0 and starting compose…") + public_url = _guess_lan_url() if published else None + if published and not public_url: + # Still start with 0.0.0.0 publish; user can set PUBLIC_URL later. + logger.info("LAN publish enabled but LAN IP could not be guessed") + + try: + from .paths_embed import ensure_token_file + + self.runner.ensure_runtime() + self._token = ensure_token_file() + result = self.runner.start(lan_publish=published, public_url=public_url) + except Exception as exc: # noqa: BLE001 + self.runner.managed_by_us = False + self._emit(GatewayStatus.ERROR, str(exc)) + return + + if not result.ok: + self._emit(GatewayStatus.ERROR, result.message) + return + + self._compose_lan_publish = published + self._start_polling() + self.refresh_status() + self._reconcile_lan_after_compose() + + def _stop_worker(self) -> None: + self._stop_polling() + with self._lock: + self._emit(GatewayStatus.STARTING, "Stopping local VocaGateway…") + result = self.runner.stop(wipe_volumes=False) + self._pairing = None + self._compose_lan_publish = None + if not result.ok: + self._emit(GatewayStatus.ERROR, result.message) + return + self._emit(GatewayStatus.STOPPED, "") + + def _start_polling(self) -> None: + self._stop_polling() + self._poll_stop = threading.Event() + + def _loop() -> None: + while not self._poll_stop.wait(2.0): + try: + self.refresh_status() + except Exception: # noqa: BLE001 + logger.exception("gateway poll failed") + + self._poll_thread = threading.Thread( + target=_loop, name="vocalinux-gateway-poll", daemon=True + ) + self._poll_thread.start() + + def _stop_polling(self) -> None: + self._poll_stop.set() + self._poll_thread = None + + def refresh_status(self) -> GatewayStatus: + """Probe health (+ pairing when live) and emit status.""" + if self._status is GatewayStatus.STARTING and not self.runner.managed_by_us: + # Mid start/stop worker; leave label alone unless we have live. + pass + + health = probe_health(self.runner.local_base_url()) + running = self.runner.managed_by_us or self.runner.is_compose_running() + + pairable = False + if health.live and self._token: + try: + public = _guess_lan_url() if self._effective_lan_for_pairing() else None + # Avoid re-downloading QR SVG every poll (hot-path memory churn). + need_qr = True + cached = self._pairing + if cached is not None and cached.qr_svg and cached.display_url: + need_qr = False + info = fetch_pairing( + self.runner.local_base_url(), + self._token, + public_url=public, + fetch_qr=need_qr, + ) + if ( + not info.qr_svg + and cached is not None + and cached.qr_svg + and cached.display_url == info.display_url + ): + info = PairingInfo( + version=info.version, + url=info.url, + token=info.token, + display_url=info.display_url, + raw_payload=info.raw_payload, + qr_svg=cached.qr_svg, + ) + self._pairing = info + pairable = info.pairable + except Exception as exc: # noqa: BLE001 + # Live but pairing endpoint not ready / auth mismatch. + logger.info("pairing fetch failed: %s", type(exc).__name__) + # If LAN publish is off, loopback URL is expected: not Pairable. + pairable = False + + status = GatewayStatus.from_health( + live=health.live, + ready=health.ready, + pairable=pairable, + error=False, + starting=running and not health.live, + running=running, + ) + detail = "" + if status is GatewayStatus.LIVE: + if self.lan_publish and self._compose_lan_publish is not True: + detail = ( + "LAN is on in settings, but publish still needs a restart. " + "Stop and Run again (or wait for republish) before pairing a phone." + ) + elif not self.lan_publish and self._compose_lan_publish is True: + detail = ( + "LAN is off in settings, but publish is still open on the LAN. " + "Wait for republish to finish (or Stop and Run again)." + ) + else: + detail = "Gateway process is up. Download a model in the WebUI to become Ready." + elif status is GatewayStatus.PAIRABLE: + detail = "Phone can pair. Dictation Ready still needs a selected model." + elif status is GatewayStatus.READY: + detail = "Ready for dictation." + elif status is GatewayStatus.STARTING: + detail = self._status_detail or "Starting…" + elif status is GatewayStatus.STOPPED and health.error and running: + detail = health.error + status = GatewayStatus.ERROR + + self._emit(status, detail) + return status + + def use_this_gateway(self, config_manager: Any) -> dict[str, str]: + """Fill remote_api_* from the current pairing (Ready or Pairable).""" + info = self._pairing + if info is None or not info.token: + raise RuntimeError("No pairing info yet. Wait until status is Pairable or Ready.") + + # Prefer non-loopback; for desktop-only use, allow loopback URL in remote_api + # (phone QR still rejects loopback separately). + url = info.display_url or info.url + if self._effective_lan_for_pairing(): + guessed = _guess_lan_url() + if guessed: + url = guessed + # Desktop "Use this Gateway" may legitimately use 127.0.0.1 when LAN is off. + if not url: + raise RuntimeError("Gateway URL missing from pairing payload.") + + preset = remote_api_preset_from_pairing( + url=url, + token=info.token, + prefer_public=bool(reject_loopback_url(url)), + ) + # When prefer_public rejected loopback, still allow explicit loopback for desktop. + if not reject_loopback_url(url): + preset = remote_api_preset_from_pairing( + url=url, + token=info.token, + prefer_public=False, + ) + apply_remote_api_preset(config_manager, preset) + return dict(preset) + + +_MANAGER: Optional[GatewayEmbedManager] = None +_MANAGER_LOCK = threading.Lock() + + +def get_gateway_embed_manager() -> GatewayEmbedManager: + global _MANAGER + with _MANAGER_LOCK: + if _MANAGER is None: + _MANAGER = GatewayEmbedManager() + return _MANAGER + + +def reset_gateway_embed_manager_for_tests() -> None: + global _MANAGER + with _MANAGER_LOCK: + if _MANAGER is not None: + _MANAGER._stop_polling() + _MANAGER = None diff --git a/src/vocalinux/gateway_embed/pairing.py b/src/vocalinux/gateway_embed/pairing.py new file mode 100644 index 0000000..d529544 --- /dev/null +++ b/src/vocalinux/gateway_embed/pairing.py @@ -0,0 +1,146 @@ +"""Fetch pairing payload and optional QR SVG from a local VocaGateway.""" + +from __future__ import annotations + +import json +import logging +import urllib.error +import urllib.request +from dataclasses import dataclass +from typing import Any, Optional + +from ..common_types import Readable +from .urls import is_unusable_phone_url, reject_loopback_url + +logger = logging.getLogger(__name__) + +# Adversarial: never pull an unbounded SVG into the desktop process. +MAX_QR_SVG_BYTES = 512 * 1024 +MAX_PAIRING_JSON_BYTES = 64 * 1024 +DEFAULT_TIMEOUT = 3.0 + + +@dataclass(frozen=True) +class PairingInfo: + """Decoded vocaphone-pair-v1 style payload.""" + + version: int + url: str + token: str + display_url: Optional[str] + raw_payload: dict[str, Any] + qr_svg: Optional[bytes] = None + + @property + def pairable(self) -> bool: + """True when a non-loopback phone URL and token are present.""" + return bool(self.display_url and self.token) + + +def _read_capped(response: Readable, limit: int) -> bytes: + chunks: list[bytes] = [] + total = 0 + while True: + piece = response.read(min(65536, limit - total + 1)) + if not piece: + break + total += len(piece) + if total > limit: + raise ValueError(f"response exceeded {limit} byte cap") + chunks.append(piece) + return b"".join(chunks) + + +def _authorized_get( + url: str, + token: str, + *, + timeout: float, + limit: int, +) -> bytes: + headers = {} + if token: + headers["Authorization"] = f"Bearer {token}" + request = urllib.request.Request(url, headers=headers, method="GET") + with urllib.request.urlopen(request, timeout=timeout) as response: + return _read_capped(response, limit) + + +def decode_pairing_payload(data: dict[str, Any] | str) -> PairingInfo: + """Decode gateway pairing JSON. + + Accepts either the gateway envelope ``{"payload": {...}}`` / ``{"payload": + ""}`` or a bare ``{v,url,token}`` object. + """ + if isinstance(data, str): + data = json.loads(data) + if not isinstance(data, dict): + raise ValueError("pairing response must be an object") + + payload = data.get("payload", data) + if isinstance(payload, str): + payload = json.loads(payload) + if not isinstance(payload, dict): + raise ValueError("pairing payload must be an object") + + version = int(payload.get("v") or payload.get("version") or 1) + url = str(payload.get("url") or "").strip() + token = str(payload.get("token") or "").strip() + if not url or not token: + raise ValueError("pairing payload missing url or token") + + display = reject_loopback_url(url) + return PairingInfo( + version=version, + url=url, + token=token, + display_url=display, + raw_payload=dict(payload), + ) + + +def fetch_pairing( + base_url: str, + bearer_token: str, + *, + public_url: str | None = None, + fetch_qr: bool = True, + timeout: float = DEFAULT_TIMEOUT, +) -> PairingInfo: + """GET ``/v1/admin/pairing`` and optionally ``/qr.svg``.""" + root = base_url.rstrip("/") + query = "" + if public_url and not is_unusable_phone_url(public_url): + from urllib.parse import quote + + query = f"?url={quote(public_url, safe=':/')}" + + body = _authorized_get( + f"{root}/v1/admin/pairing{query}", + bearer_token, + timeout=timeout, + limit=MAX_PAIRING_JSON_BYTES, + ) + info = decode_pairing_payload(json.loads(body.decode("utf-8"))) + + qr_svg: Optional[bytes] = None + if fetch_qr and info.display_url: + try: + qr_svg = _authorized_get( + f"{root}/v1/admin/pairing/qr.svg{query}", + bearer_token, + timeout=timeout, + limit=MAX_QR_SVG_BYTES, + ) + except Exception as exc: # noqa: BLE001 + # Soft failure: UI falls back to URL+token text. + logger.info("QR SVG unavailable: %s", type(exc).__name__) + + return PairingInfo( + version=info.version, + url=info.url, + token=info.token, + display_url=info.display_url, + raw_payload=info.raw_payload, + qr_svg=qr_svg, + ) diff --git a/src/vocalinux/gateway_embed/paths_embed.py b/src/vocalinux/gateway_embed/paths_embed.py new file mode 100644 index 0000000..505df5e --- /dev/null +++ b/src/vocalinux/gateway_embed/paths_embed.py @@ -0,0 +1,66 @@ +"""XDG paths for the embedded gateway checkout and secrets.""" + +from __future__ import annotations + +import logging +import os +import re +import secrets + +from vocalinux.utils.paths import config_dir + +logger = logging.getLogger(__name__) + +# Keep pin explicit: never float on latest. +GATEWAY_RELEASE_TAG = "v0.1.0" +GATEWAY_GIT_URL = "https://github.com/VocaHQ/vocagateway.git" +COMPOSE_PROJECT = "vocagateway" +DEFAULT_PORT = 8765 + +# Bootstrap tokens are hex; refuse unbounded or hostile token files. +MAX_TOKEN_FILE_BYTES = 4096 +_TOKEN_RE = re.compile(r"^[0-9a-fA-F]{32,2048}$") + + +def xdg_cache_home() -> str: + return os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache") + + +def gateway_cache_dir() -> str: + return os.path.join(xdg_cache_home(), "vocalinux", f"vocagateway-{GATEWAY_RELEASE_TAG}") + + +def gateway_embed_config_dir() -> str: + return os.path.join(config_dir(), "gateway_embed") + + +def token_file_path() -> str: + return os.path.join(gateway_embed_config_dir(), "token") + + +def env_file_path() -> str: + return os.path.join(gateway_embed_config_dir(), ".env") + + +def ensure_token_file(path: str | None = None) -> str: + """Create a mode-600 bootstrap token if missing; return the token string.""" + token_path = path or token_file_path() + os.makedirs(os.path.dirname(token_path), mode=0o700, exist_ok=True) + if os.path.isfile(token_path): + with open(token_path, "rb") as handle: + raw = handle.read(MAX_TOKEN_FILE_BYTES + 1) + if len(raw) <= MAX_TOKEN_FILE_BYTES: + existing = raw.decode("utf-8", errors="replace").strip() + if _TOKEN_RE.fullmatch(existing): + return existing + token = secrets.token_hex(32) + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + fd = os.open(token_path, flags, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(token) + handle.write("\n") + try: + os.chmod(token_path, 0o600) + except OSError: + pass + return token diff --git a/src/vocalinux/gateway_embed/preset.py b/src/vocalinux/gateway_embed/preset.py new file mode 100644 index 0000000..929df3a --- /dev/null +++ b/src/vocalinux/gateway_embed/preset.py @@ -0,0 +1,49 @@ +"""Map a paired local gateway onto Vocalinux remote_api settings.""" + +from __future__ import annotations + +import logging +from typing import Any, Mapping + +from .urls import reject_loopback_url + +logger = logging.getLogger(__name__) + +GATEWAY_TRANSCRIPTIONS_ENDPOINT = "/v1/audio/transcriptions" + + +def remote_api_preset_from_pairing( + *, + url: str, + token: str, + prefer_public: bool = True, +) -> dict[str, str]: + """Return speech_recognition keys for "Use this Gateway". + + Matches docs/HTTP_REMOTE.md / Gateway docs/configuration.md: + engine ``remote_api``, OpenAI-compatible transcriptions path, bearer key. + """ + cleaned = (url or "").strip().rstrip("/") + if prefer_public: + display = reject_loopback_url(cleaned) + if display: + cleaned = display.rstrip("/") + if not cleaned: + raise ValueError("gateway URL is required") + if not (token or "").strip(): + raise ValueError("gateway token is required") + + return { + "engine": "remote_api", + "remote_api_url": cleaned, + "remote_api_key": token.strip(), + "remote_api_endpoint": GATEWAY_TRANSCRIPTIONS_ENDPOINT, + "remote_api_model": "whisper-1", + } + + +def apply_remote_api_preset(config_manager: Any, preset: Mapping[str, str]) -> None: + """Write *preset* into an existing ConfigManager and save.""" + for key, value in preset.items(): + config_manager.set("speech_recognition", key, value) + config_manager.save_config() diff --git a/src/vocalinux/gateway_embed/runner.py b/src/vocalinux/gateway_embed/runner.py new file mode 100644 index 0000000..c6c973c --- /dev/null +++ b/src/vocalinux/gateway_embed/runner.py @@ -0,0 +1,483 @@ +"""Compose lifecycle for project ``vocagateway`` pinned to release v0.1.0.""" + +from __future__ import annotations + +import logging +import os +import re +import shutil +import subprocess +import threading +from dataclasses import dataclass +from typing import Any, Callable, Mapping, Optional, Sequence + +from vocalinux.utils.host_process import host_env + +from .paths_embed import ( + COMPOSE_PROJECT, + DEFAULT_PORT, + GATEWAY_GIT_URL, + GATEWAY_RELEASE_TAG, + ensure_token_file, + env_file_path, + gateway_cache_dir, +) +from .runtime import ContainerRuntime, RuntimeInfo, resolve_runtime_info +from .sandbox import SandboxState, detect_sandbox +from .urls import reject_loopback_url + +DEFAULT_IMAGE = "vocagateway:v0.1.0" +# Compose --profile cpu: selects a future cpu profile if present; on v0.1.0 +# the unprofiled gateway service still starts, while cuda/vulkan/native stay off. + +# Image refs only: block newline / metachar injection into the compose .env. +_IMAGE_RE = re.compile( + r"^(?:[A-Za-z0-9]+(?:[._-][A-Za-z0-9]+)*)" + r"(?:/(?:[A-Za-z0-9]+(?:[._-][A-Za-z0-9]+)*))*" + r"(?::[A-Za-z0-9_][A-Za-z0-9._-]{0,127})?$" +) + + +def _default_run(args: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess: + """Run a host binary with AppImage library paths stripped.""" + env = kwargs.pop("env", None) + return subprocess.run(args, env=host_env(env), **kwargs) + + +logger = logging.getLogger(__name__) + + +@dataclass +class RunnerResult: + ok: bool + message: str = "" + returncode: int = 0 + + +def _env_line_safe(value: str) -> bool: + """True when *value* can sit on a single compose .env line.""" + return bool(value) and not any(ch in value for ch in "\n\r\0") + + +def pin_gateway_image(raw: str | None) -> str: + """Return a pinned image ref; never ``:latest`` or hostile overrides.""" + image = (raw or "").strip() or DEFAULT_IMAGE + if ( + not _env_line_safe(image) + or image.endswith(":latest") + or image == "latest" + or not _IMAGE_RE.fullmatch(image) + ): + return DEFAULT_IMAGE + return image + + +def write_env_file( + *, + token: str, + lan_publish: bool, + public_url: str | None = None, + path: str | None = None, +) -> str: + """Write compose ``.env`` (mode 600). Never log the token.""" + if not _env_line_safe(token): + raise ValueError("gateway token contains invalid characters for .env") + env_path = path or env_file_path() + os.makedirs(os.path.dirname(env_path), mode=0o700, exist_ok=True) + publish_host = "0.0.0.0" if lan_publish else "127.0.0.1" + image = pin_gateway_image(os.environ.get("VOCAGATEWAY_IMAGE")) + lines = [ + f"VOCAGATEWAY_TOKEN={token}", + f"VOCAGATEWAY_PUBLISH_HOST={publish_host}", + f"VOCAGATEWAY_PUBLISH_PORT={DEFAULT_PORT}", + f"VOCAGATEWAY_PORT={DEFAULT_PORT}", + # Pin image; v0.1.0 release has no published GHCR assets yet, so the + # first start builds locally into this tag (or VOCAGATEWAY_IMAGE). + f"VOCAGATEWAY_IMAGE={image}", + ] + # Only phone-reachable PUBLIC_URL belongs in compose discovery / QR. + safe_public = reject_loopback_url(public_url) if public_url else None + if safe_public and _env_line_safe(safe_public): + lines.append(f"VOCAGATEWAY_PUBLIC_URL={safe_public}") + lines.append(f"VOCAGATEWAY_PAIRING_URL={safe_public}") + content = "\n".join(lines) + "\n" + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + fd = os.open(env_path, flags, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + try: + os.chmod(env_path, 0o600) + except OSError: + pass + return env_path + + +def read_lan_publish_from_env(path: str | None = None) -> bool: + """True when embed ``.env`` publishes ``0.0.0.0`` (Allow LAN on).""" + env_path = path or env_file_path() + try: + with open(env_path, encoding="utf-8") as handle: + for line in handle: + if line.startswith("VOCAGATEWAY_PUBLISH_HOST="): + return line.split("=", 1)[1].strip() == "0.0.0.0" + except OSError: + return False + return False + + +def _read_env_bytes(path: str) -> bytes | None: + """Return existing compose ``.env`` bytes, or None when missing.""" + try: + with open(path, "rb") as handle: + return handle.read() + except OSError: + return None + + +def _write_env_bytes(path: str, content: bytes) -> None: + """Replace compose ``.env`` with *content* (mode 600). Never log it.""" + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + fd = os.open(path, flags, 0o600) + with os.fdopen(fd, "wb") as handle: + handle.write(content) + try: + os.chmod(path, 0o600) + except OSError: + pass + + +def _restore_env_snapshot(path: str, snapshot: bytes | None) -> None: + """Put a pre-republish ``.env`` back so file truth matches a live bind.""" + if snapshot is None: + return + try: + _write_env_bytes(path, snapshot) + except OSError: + logger.warning("could not restore gateway embed .env after failed republish") + + +def ensure_gateway_checkout( + *, + cache_dir: str | None = None, + run: Callable[..., subprocess.CompletedProcess] = _default_run, +) -> str: + """Clone or fetch vocagateway @ v0.1.0 into the XDG cache.""" + dest = cache_dir or gateway_cache_dir() + os.makedirs(os.path.dirname(dest), mode=0o755, exist_ok=True) + git = shutil.which("git") + if not git: + raise RuntimeError("git is required to fetch VocaGateway v0.1.0 compose sources") + + if os.path.isdir(os.path.join(dest, ".git")): + run( + [git, "-C", dest, "fetch", "--tags", "--force", "origin", GATEWAY_RELEASE_TAG], + check=False, + capture_output=True, + timeout=120, + ) + completed = run( + [git, "-C", dest, "checkout", "-f", GATEWAY_RELEASE_TAG], + check=False, + capture_output=True, + timeout=60, + ) + if completed.returncode != 0: + raise RuntimeError(f"failed to checkout {GATEWAY_RELEASE_TAG} in {dest}") + return dest + + if os.path.exists(dest): + shutil.rmtree(dest) + + completed = run( + [ + git, + "clone", + "--depth", + "1", + "--branch", + GATEWAY_RELEASE_TAG, + GATEWAY_GIT_URL, + dest, + ], + check=False, + capture_output=True, + timeout=180, + ) + if completed.returncode != 0: + stderr = (completed.stderr or b"").decode("utf-8", errors="replace")[:400] + raise RuntimeError(f"git clone of vocagateway {GATEWAY_RELEASE_TAG} failed: {stderr}") + return dest + + +class GatewayRunner: + """Start/stop the compose project without wiping volumes.""" + + def __init__( + self, + runtime: RuntimeInfo | None = None, + sandbox: SandboxState | None = None, + *, + run: Callable[..., subprocess.CompletedProcess] = _default_run, + lazy_runtime: bool = True, + ) -> None: + self.sandbox = sandbox if sandbox is not None else detect_sandbox() + self._run = run + self._lock = threading.RLock() + self._runtime_lock = threading.Lock() + self.managed_by_us = False + self.last_error = "" + self._checkout: Optional[str] = None + if runtime is not None: + self._runtime: RuntimeInfo = runtime + self._runtime_ready = True + elif lazy_runtime: + # Do not probe podman/docker on the GTK main thread. + self._runtime = RuntimeInfo( + kind=ContainerRuntime.NONE, + binary=None, + hint="Detecting container runtime…", + ) + self._runtime_ready = False + else: + self._runtime = resolve_runtime_info() + self._runtime_ready = True + + @property + def runtime(self) -> RuntimeInfo: + return self._runtime + + @runtime.setter + def runtime(self, value: RuntimeInfo) -> None: + self._runtime = value + + @property + def runtime_ready(self) -> bool: + return self._runtime_ready + + def ensure_runtime(self) -> RuntimeInfo: + """Resolve container runtime; safe to call from a worker thread.""" + if self._runtime_ready: + return self._runtime + with self._runtime_lock: + if not self._runtime_ready: + self._runtime = resolve_runtime_info() + self._runtime_ready = True + return self._runtime + + @property + def available(self) -> bool: + if not self._runtime_ready: + return False + return ( + not self.sandbox.blocked + and self._runtime.kind is not ContainerRuntime.NONE + and bool(self._runtime.compose_args) + ) + + @property + def unavailable_hint(self) -> str: + if self.sandbox.blocked: + return self.sandbox.hint + if not self._runtime_ready: + return "Detecting container runtime…" + if self._runtime.hint: + return self._runtime.hint + return "No container runtime available." + + def local_base_url(self) -> str: + return f"http://127.0.0.1:{DEFAULT_PORT}" + + def _compose( + self, + args: Sequence[str], + *, + cwd: str, + env_file: str, + extra_env: Mapping[str, str] | None = None, + timeout: float = 600, + ) -> subprocess.CompletedProcess: + self.ensure_runtime() + argv = list(self._runtime.compose_args) + ["--env-file", env_file, "-p", COMPOSE_PROJECT] + argv.extend(args) + env = host_env() + if extra_env: + env.update(extra_env) + logger.info( + "gateway compose: %s (cwd=%s)", + " ".join(argv[:6] + ["…"] if len(argv) > 6 else argv), + cwd, + ) + return self._run( + argv, + cwd=cwd, + env=env, + check=False, + capture_output=True, + timeout=timeout, + ) + + def prepare(self, *, lan_publish: bool, public_url: str | None = None) -> tuple[str, str, str]: + """Ensure checkout + token + env. Returns (checkout, token, env_path).""" + with self._lock: + self.ensure_runtime() + if self.sandbox.blocked: + raise RuntimeError(self.sandbox.hint) + if not self.available: + raise RuntimeError(self.unavailable_hint) + checkout = ensure_gateway_checkout(run=self._run) + self._checkout = checkout + token = ensure_token_file() + env_path = write_env_file( + token=token, + lan_publish=lan_publish, + public_url=public_url, + ) + compose_file = os.path.join(checkout, "compose.yaml") + if not os.path.isfile(compose_file): + raise RuntimeError(f"compose.yaml missing in {checkout}") + return checkout, token, env_path + + def start(self, *, lan_publish: bool = False, public_url: str | None = None) -> RunnerResult: + """``compose --profile cpu up -d`` for service gateway (builds if needed).""" + with self._lock: + try: + checkout, _token, env_path = self.prepare( + lan_publish=lan_publish, public_url=public_url + ) + except Exception as exc: # noqa: BLE001 + self.last_error = str(exc) + return RunnerResult(ok=False, message=str(exc), returncode=1) + + completed = self._compose( + [ + "-f", + "compose.yaml", + "--profile", + "cpu", + "up", + "-d", + "--build", + "gateway", + ], + cwd=checkout, + env_file=env_path, + timeout=900, + ) + if completed.returncode != 0: + stderr = (completed.stderr or b"").decode("utf-8", errors="replace") + # Never include .env contents; stderr from compose is OK to trim. + message = stderr.strip()[-800:] or "compose up failed" + self.last_error = message + self.managed_by_us = False + return RunnerResult(ok=False, message=message, returncode=completed.returncode) + + self.managed_by_us = True + self.last_error = "" + return RunnerResult(ok=True, message="started", returncode=0) + + def stop(self, *, wipe_volumes: bool = False) -> RunnerResult: + """``compose down`` without ``-v`` by default (preserve models/token volume).""" + with self._lock: + if wipe_volumes: + # Explicitly refused in v1 UI path; keep for tests only. + raise RuntimeError("volume wipe is not allowed from the desktop UI") + checkout = self._checkout or gateway_cache_dir() + env_path = env_file_path() + self.ensure_runtime() + if not os.path.isdir(checkout) or not self._runtime.compose_args: + self.managed_by_us = False + return RunnerResult(ok=True, message="nothing to stop") + if not os.path.isfile(env_path): + # Still try down with a throwaway env so project name matches. + token = ensure_token_file() + env_path = write_env_file(token=token, lan_publish=False) + + completed = self._compose( + ["-f", "compose.yaml", "--profile", "cpu", "down"], + cwd=checkout, + env_file=env_path, + timeout=180, + ) + self.managed_by_us = False + if completed.returncode != 0: + stderr = (completed.stderr or b"").decode("utf-8", errors="replace") + message = stderr.strip()[-800:] or "compose down failed" + self.last_error = message + return RunnerResult(ok=False, message=message, returncode=completed.returncode) + self.last_error = "" + return RunnerResult(ok=True, message="stopped", returncode=0) + + def republish(self, *, lan_publish: bool, public_url: str | None = None) -> RunnerResult: + """Rewrite publish host / PUBLIC_URL and recreate the gateway service. + + Used when LAN is toggled while compose is already managed by us so the + phone-facing bind and pairing URL match the new setting. + + ``prepare`` writes the new ``VOCAGATEWAY_PUBLISH_HOST`` before + ``compose up --force-recreate``. If recreate fails, the old container + is still bound to the previous host, so restore the snapshot ``.env``. + """ + with self._lock: + if not self.managed_by_us: + return RunnerResult( + ok=False, + message="gateway is not managed by this session", + returncode=1, + ) + env_path = env_file_path() + previous_env = _read_env_bytes(env_path) + try: + checkout, _token, env_path = self.prepare( + lan_publish=lan_publish, public_url=public_url + ) + except Exception as exc: # noqa: BLE001 + _restore_env_snapshot(env_file_path(), previous_env) + self.last_error = str(exc) + return RunnerResult(ok=False, message=str(exc), returncode=1) + + completed = self._compose( + [ + "-f", + "compose.yaml", + "--profile", + "cpu", + "up", + "-d", + "--force-recreate", + "--no-build", + "gateway", + ], + cwd=checkout, + env_file=env_path, + timeout=300, + ) + if completed.returncode != 0: + _restore_env_snapshot(env_path, previous_env) + stderr = (completed.stderr or b"").decode("utf-8", errors="replace") + message = stderr.strip()[-800:] or "compose republish failed" + self.last_error = message + return RunnerResult(ok=False, message=message, returncode=completed.returncode) + + self.managed_by_us = True + self.last_error = "" + return RunnerResult(ok=True, message="republished", returncode=0) + + def is_compose_running(self) -> bool: + """Best-effort ``compose ps -q gateway``.""" + checkout = self._checkout or gateway_cache_dir() + env_path = env_file_path() + if not os.path.isdir(checkout) or not os.path.isfile(env_path): + return False + if not self._runtime_ready or not self._runtime.compose_args: + return False + try: + completed = self._compose( + ["-f", "compose.yaml", "--profile", "cpu", "ps", "-q", "gateway"], + cwd=checkout, + env_file=env_path, + timeout=30, + ) + except Exception: # noqa: BLE001 + return False + out = (completed.stdout or b"").decode("utf-8", errors="replace").strip() + return completed.returncode == 0 and bool(out) diff --git a/src/vocalinux/gateway_embed/runtime.py b/src/vocalinux/gateway_embed/runtime.py new file mode 100644 index 0000000..b2ef582 --- /dev/null +++ b/src/vocalinux/gateway_embed/runtime.py @@ -0,0 +1,142 @@ +"""Detect a usable container runtime (podman first, then docker).""" + +from __future__ import annotations + +import logging +import os +import shutil +import subprocess +from dataclasses import dataclass +from enum import Enum +from typing import Callable, Optional, Sequence + +from vocalinux.utils.host_process import host_env + +logger = logging.getLogger(__name__) + + +class ContainerRuntime(str, Enum): + PODMAN = "podman" + DOCKER = "docker" + NONE = "none" + + +@dataclass(frozen=True) +class RuntimeInfo: + kind: ContainerRuntime + binary: Optional[str] = None + compose_args: tuple[str, ...] = () + hint: str = "" + + +def _which(name: str) -> Optional[str]: + return shutil.which(name) + + +def _run_ok(argv: Sequence[str], *, timeout: float = 3.0) -> bool: + try: + completed = subprocess.run( + list(argv), + check=False, + capture_output=True, + timeout=timeout, + env=host_env(), + ) + return completed.returncode == 0 + except (OSError, subprocess.SubprocessError): + return False + + +def _compose_argv( + binary: str, + kind: ContainerRuntime, + *, + path_lookup: Callable[[str], Optional[str]], + probe: Callable[..., bool], +) -> tuple[str, ...]: + if probe([binary, "compose", "version"]): + return (binary, "compose") + legacy = "podman-compose" if kind is ContainerRuntime.PODMAN else "docker-compose" + legacy_path = path_lookup(legacy) + if legacy_path and probe([legacy_path, "version"]): + return (legacy_path,) + return () + + +def detect_container_runtime( + *, + path_lookup: Callable[[str], Optional[str]] = _which, + probe: Callable[..., bool] = _run_ok, +) -> RuntimeInfo: + """Prefer podman, fall back to docker, else none with an install hint.""" + podman = path_lookup("podman") + if podman and (probe([podman, "info"]) or probe([podman, "version"])): + compose_args = _compose_argv( + podman, ContainerRuntime.PODMAN, path_lookup=path_lookup, probe=probe + ) + return RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary=podman, + compose_args=compose_args, + hint=( + "" + if compose_args + else ( + "podman is installed but Compose is missing. " + "Install the Compose plugin or podman-compose, then reopen Settings." + ) + ), + ) + + docker = path_lookup("docker") + if docker and (probe([docker, "info"]) or probe([docker, "version"])): + compose_args = _compose_argv( + docker, ContainerRuntime.DOCKER, path_lookup=path_lookup, probe=probe + ) + return RuntimeInfo( + kind=ContainerRuntime.DOCKER, + binary=docker, + compose_args=compose_args, + hint=( + "" + if compose_args + else ( + "docker is installed but Compose is missing. " + "Install the Compose plugin or docker-compose, then reopen Settings." + ) + ), + ) + + if podman: + return RuntimeInfo( + kind=ContainerRuntime.NONE, + binary=None, + hint=( + "podman is installed but not usable right now. " + "Start the podman service or check permissions, then reopen Settings." + ), + ) + if docker: + return RuntimeInfo( + kind=ContainerRuntime.NONE, + binary=None, + hint=( + "docker is installed but not usable right now. " + "Start the docker service or add your user to the docker group, " + "then reopen Settings." + ), + ) + + return RuntimeInfo( + kind=ContainerRuntime.NONE, + binary=None, + hint=( + "Install podman (preferred) or docker to run a local VocaGateway. " + "Vocalinux does not bundle a container engine in the AppImage." + ), + ) + + +def resolve_runtime_info() -> RuntimeInfo: + """Production entry point.""" + return detect_container_runtime() diff --git a/src/vocalinux/gateway_embed/sandbox.py b/src/vocalinux/gateway_embed/sandbox.py new file mode 100644 index 0000000..17d7b3a --- /dev/null +++ b/src/vocalinux/gateway_embed/sandbox.py @@ -0,0 +1,57 @@ +"""Sandbox / packaging detection for container socket access.""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SandboxState: + """Whether the desktop process can reach a host container engine.""" + + blocked: bool + kind: str # "", "flatpak", "appimage" + hint: str = "" + + +def detect_sandbox( + environ: dict[str, str] | None = None, +) -> SandboxState: + """Fail closed inside Flatpak; warn for AppImage with a docs hint. + + Flatpak cannot see the host podman/docker socket unless the user has + explicitly granted filesystem/socket permissions. We do not try to + smuggle a daemon into the sandbox. + """ + env = environ if environ is not None else os.environ + flatpak_id = (env.get("FLATPAK_ID") or "").strip() + if flatpak_id: + return SandboxState( + blocked=True, + kind="flatpak", + hint=( + "Vocalinux is running as a Flatpak and cannot reach the host " + "podman/docker socket. Run Vocalinux from the native package or " + "AppImage, or start VocaGateway on the host and point Remote " + "Server at it. See docs/GATEWAY_EMBED.md." + ), + ) + + appimage = (env.get("APPIMAGE") or "").strip() + if appimage: + # AppImage often can reach the host socket; still surface honesty. + return SandboxState( + blocked=False, + kind="appimage", + hint=( + "Running from an AppImage. Local VocaGateway uses the host " + "podman/docker socket when available; the AppImage does not " + "bundle a container engine." + ), + ) + + return SandboxState(blocked=False, kind="", hint="") diff --git a/src/vocalinux/gateway_embed/status.py b/src/vocalinux/gateway_embed/status.py new file mode 100644 index 0000000..1261594 --- /dev/null +++ b/src/vocalinux/gateway_embed/status.py @@ -0,0 +1,51 @@ +"""Status labels for the embedded local VocaGateway.""" + +from __future__ import annotations + +import logging +from enum import Enum + +logger = logging.getLogger(__name__) + + +class GatewayStatus(str, Enum): + """User-visible lifecycle states for the local gateway.""" + + STOPPED = "Stopped" + STARTING = "Starting" + LIVE = "Live" + PAIRABLE = "Pairable" + READY = "Ready" + ERROR = "Error" + + @classmethod + def from_health( + cls, + *, + live: bool, + ready: bool, + pairable: bool, + error: bool = False, + starting: bool = False, + running: bool = False, + ) -> "GatewayStatus": + """Map health probes into a single status label. + + Ready implies Live. Pairable is Live plus a phone-reachable + (non-loopback) URL and auth material, and may precede Ready. + """ + if error: + return cls.ERROR + if starting: + return cls.STARTING + if not running and not live: + return cls.STOPPED + if ready: + return cls.READY + if pairable: + return cls.PAIRABLE + if live: + return cls.LIVE + if running: + return cls.STARTING + return cls.STOPPED diff --git a/src/vocalinux/gateway_embed/urls.py b/src/vocalinux/gateway_embed/urls.py new file mode 100644 index 0000000..bed871d --- /dev/null +++ b/src/vocalinux/gateway_embed/urls.py @@ -0,0 +1,184 @@ +"""URL helpers for pairing / QR display (never advertise unreachable hosts). + +Phone QR and Pairable must only use hosts a phone can open on the LAN (or +an explicit public override). We reject: + +* loopback / unspecified: 127.0.0.0/8 (including short-form 127.1), ::1, + localhost / localhost.*, 0.0.0.0, :: +* link-local: 169.254.0.0/16, fe80::/10 +* default container bridges (not host-facing LAN): 172.17.0.0/16 (docker0), + 10.88.0.0/16 (podman default). Broader RFC1918 (10/8, 172.16/12, 192.168/16) + stays allowed when lan_publish is on so real LAN NICs still work. + +Optional: if docker0/podman0/cni-podman0 addresses are readable, those exact +interface IPs are also rejected even outside the default ranges. + +Aligned with vocagateway desktop-embed pairing host policy. +""" + +from __future__ import annotations + +import ipaddress +import logging +import socket +from functools import lru_cache +from urllib.parse import urlparse + +from vocalinux.utils.host_process import host_env + +logger = logging.getLogger(__name__) + +_LOOPBACK_HOSTS = frozenset( + { + "localhost", + "127.0.0.1", + "::1", + "[::1]", + "0.0.0.0", + "[::]", + } +) + +# Default engine bridges only — not the whole of 172.16.0.0/12 (private LAN). +_CONTAINER_BRIDGE_NETWORKS = ( + ipaddress.ip_network("172.17.0.0/16"), # docker0 default + ipaddress.ip_network("10.88.0.0/16"), # podman default +) + +_BRIDGE_IFACE_NAMES = frozenset( + { + "docker0", + "podman0", + "cni-podman0", + } +) + + +def _normalize_host(host: str | None) -> str | None: + if not host: + return None + cleaned = host.strip().lower().rstrip(".") + if cleaned.startswith("[") and cleaned.endswith("]"): + cleaned = cleaned[1:-1] + # Zone indices (fe80::1%eth0) are not phone-QR material. + if "%" in cleaned: + cleaned = cleaned.split("%", 1)[0] + return cleaned or None + + +def _parse_short_ipv4_host(host: str) -> ipaddress.IPv4Address | None: + """Parse abbreviated IPv4 (``127.1``) that ``ip_address`` rejects.""" + try: + return ipaddress.IPv4Address(socket.inet_aton(host)) + except OSError: + return None + + +def _parse_host_ip(host: str | None) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None: + cleaned = _normalize_host(host) + if not cleaned: + return None + try: + return ipaddress.ip_address(cleaned) + except ValueError: + return _parse_short_ipv4_host(cleaned) + + +def is_loopback_host(host: str | None) -> bool: + """Return True when *host* is loopback, unspecified, or localhost[.].""" + cleaned = _normalize_host(host) + if cleaned is None: + return True + if cleaned in _LOOPBACK_HOSTS or cleaned == "localhost" or cleaned.startswith("localhost."): + return True + addr = _parse_host_ip(cleaned) + if addr is None: + return False + return bool(addr.is_loopback or addr.is_unspecified) + + +def is_link_local_host(host: str | None) -> bool: + """Return True when *host* is link-local (not phone-reachable across LAN).""" + addr = _parse_host_ip(host) + if addr is None: + return False + return bool(addr.is_link_local) + + +@lru_cache(maxsize=1) +def _bridge_iface_ips() -> frozenset[str]: + """Best-effort IPs bound to docker0 / podman0 / cni-podman0.""" + import subprocess + + found: set[str] = set() + for iface in _BRIDGE_IFACE_NAMES: + try: + completed = subprocess.run( + ["ip", "-o", "addr", "show", "dev", iface], + check=False, + capture_output=True, + timeout=1.0, + text=True, + env=host_env(), + ) + except (OSError, subprocess.SubprocessError): + continue + if completed.returncode != 0 or not completed.stdout: + continue + for token in completed.stdout.replace("/", " ").split(): + ip = _parse_host_ip(token) + if ip is not None and not ip.is_loopback: + found.add(str(ip)) + return frozenset(found) + + +def is_container_bridge_host(host: str | None) -> bool: + """True for default docker0/podman bridge addresses (not phone-reachable LAN).""" + addr = _parse_host_ip(host) + if addr is None: + return False + if any(addr in net for net in _CONTAINER_BRIDGE_NETWORKS): + return True + try: + return str(addr) in _bridge_iface_ips() + except Exception: # noqa: BLE001 + logger.debug("bridge iface probe failed", exc_info=True) + return False + + +def is_unusable_phone_host(host: str | None) -> bool: + """True when a phone QR must not advertise this host.""" + return is_loopback_host(host) or is_link_local_host(host) or is_container_bridge_host(host) + + +def is_loopback_url(url: str | None) -> bool: + """Return True when *url* points at a loopback / unspecified host.""" + if not url or not str(url).strip(): + return True + try: + parsed = urlparse(str(url).strip()) + except Exception: + return True + return is_loopback_host(parsed.hostname) + + +def is_unusable_phone_url(url: str | None) -> bool: + """True when *url* is loopback, link-local, or a container-bridge address.""" + if not url or not str(url).strip(): + return True + try: + parsed = urlparse(str(url).strip()) + except Exception: + return True + return is_unusable_phone_host(parsed.hostname) + + +def reject_loopback_url(url: str | None) -> str | None: + """Return *url* when phone-reachable, else None. + + Rejects loopback, link-local, and default container-bridge hosts so callers + never put those into a QR or Pairable display. + """ + if is_unusable_phone_url(url): + return None + return str(url).strip() diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 4ed41bd..5c37821 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -212,6 +212,10 @@ DEFAULT_CONFIG = { # Tag last announced via desktop notification (avoids re-notifying every 6h). "last_notified_version": "", }, + # Optional local VocaGateway (podman/docker). Never flips the default engine. + "gateway_embed": { + "lan_publish": False, # VOCAGATEWAY_PUBLISH_HOST=0.0.0.0 when True (Phone on LAN) + }, } diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 6e50067..edfa276 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -33,6 +33,7 @@ gi.require_version("Gdk", "3.0") from gi.repository import Gdk, GLib, GObject, Gtk, Pango # noqa: E402 from ..common_types import RecognitionState, _EvdevCaptureDevice # noqa: E402 +from ..gateway_embed import GatewayStatus, get_gateway_embed_manager # noqa: E402 from ..speech_recognition.silero_vad import is_silero_available # noqa: E402 from ..utils import parakeet_model_info as parakeet # noqa: E402 from ..utils.faster_whisper_model_info import delete_model as delete_faster_whisper_model @@ -2582,6 +2583,7 @@ class SettingsDialog(Gtk.Dialog): self._build_simple_model_section() self._build_engine_section() self._build_remote_server_section() + self._build_gateway_embed_section() self._build_audio_section() self._build_auto_pause_section() self._build_model_keepalive_section() @@ -3669,7 +3671,7 @@ class SettingsDialog(Gtk.Dialog): _, ) = _make_expander_card( "Advanced", - "Engine, model, and remote server", + "Engine, model, remote server, and local VocaGateway", ) self.advanced_box.set_spacing(12) self.advanced_box.set_margin_top(8) @@ -5480,6 +5482,309 @@ class SettingsDialog(Gtk.Dialog): self.remote_server_group.hide() self.remote_status_label.hide() + def _build_gateway_embed_section(self) -> None: + """Optional local VocaGateway controls (podman-first). Lives on the Advanced island.""" + self._gateway_manager = get_gateway_embed_manager() + # Do not clobber LAN restored from leftover compose (orphan adopt). + if not self._gateway_manager.managed_by_us: + self._gateway_manager.lan_publish = bool( + self.config_manager.get("gateway_embed", "lan_publish", False) + ) + lan_publish = bool(self._gateway_manager.lan_publish) + + self.gateway_embed_group = PreferencesGroup( + title="Local VocaGateway", + description=( + "Optional self-hosted gateway via podman (or docker). " + "Audio goes to that container on this machine; it is not on-device " + "whisper.cpp. Pinned to VocaGateway v0.1.0." + ), + keywords=( + "vocagateway", + "gateway", + "podman", + "docker", + "pairing", + "phone", + "remote", + ), + ) + + self.gateway_run_btn = Gtk.Button(label="Run VocaGateway locally") + self.gateway_run_btn.set_size_request(_CONTROL_WIDTH, -1) + self.gateway_run_btn.connect("clicked", self._on_gateway_run_clicked) + self.gateway_embed_group.add_row( + PreferenceRow( + title="Local gateway", + subtitle="Start or stop the compose project vocagateway", + widget=self.gateway_run_btn, + ) + ) + + self.gateway_status_label = Gtk.Label(label="Stopped", xalign=0) + self.gateway_status_label.get_style_context().add_class("status-info") + self.gateway_embed_group.add_row( + PreferenceRow( + title="Status", + subtitle="Stopped, Starting, Live, Pairable, Ready, or Error", + widget=self.gateway_status_label, + ) + ) + + self.gateway_detail_label = Gtk.Label(label="", xalign=0, wrap=True) + self.gateway_detail_label.set_max_width_chars(55) + self.gateway_detail_label.get_style_context().add_class("preference-row-subtitle") + detail_row = Gtk.ListBoxRow() + detail_row.set_activatable(False) + detail_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) + detail_box.set_margin_start(16) + detail_box.set_margin_end(16) + detail_box.set_margin_top(4) + detail_box.set_margin_bottom(8) + detail_box.pack_start(self.gateway_detail_label, False, False, 0) + detail_row.add(detail_box) + self.gateway_embed_group.add_row(detail_row) + + self.gateway_lan_switch = Gtk.Switch() + self.gateway_lan_switch.set_active(lan_publish) + self.gateway_lan_switch.connect("notify::active", self._on_gateway_lan_toggled) + self.gateway_embed_group.add_row( + PreferenceRow( + title="Allow LAN access for Phone", + subtitle=( + "Publishes 0.0.0.0:8765 and prefers a LAN URL in pairing. " + "Open the port in your firewall on trusted networks only." + ), + widget=self.gateway_lan_switch, + ) + ) + + self.gateway_use_btn = Gtk.Button(label="Use this Gateway") + self.gateway_use_btn.set_sensitive(False) + self.gateway_use_btn.set_tooltip_text( + "Fill Remote Server with this gateway URL and bearer token " + "(engine remote_api, endpoint /v1/audio/transcriptions)" + ) + self.gateway_use_btn.connect("clicked", self._on_gateway_use_clicked) + self.gateway_embed_group.add_row( + PreferenceRow( + title="Remote API preset", + subtitle="Points Vocalinux at the local gateway", + widget=self.gateway_use_btn, + ) + ) + + self.gateway_qr_image = Gtk.Image() + self.gateway_qr_image.set_no_show_all(True) + self.gateway_pairing_label = Gtk.Label(label="", xalign=0, wrap=True, selectable=True) + self.gateway_pairing_label.set_max_width_chars(55) + self.gateway_pairing_label.get_style_context().add_class("preference-row-subtitle") + pairing_row = Gtk.ListBoxRow() + pairing_row.set_activatable(False) + pairing_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) + pairing_box.set_margin_start(16) + pairing_box.set_margin_end(16) + pairing_box.set_margin_top(8) + pairing_box.set_margin_bottom(12) + title = Gtk.Label(label="Phone pairing", xalign=0) + title.get_style_context().add_class("preference-row-title") + pairing_box.pack_start(title, False, False, 0) + pairing_box.pack_start(self.gateway_qr_image, False, False, 0) + pairing_box.pack_start(self.gateway_pairing_label, False, False, 0) + pairing_row.add(pairing_box) + self.gateway_embed_group.add_row(pairing_row) + + if self.advanced_box is not None: + self.advanced_box.pack_start(self.gateway_embed_group, False, False, 0) + + self._gateway_manager.add_listener(self._on_gateway_status_from_worker) + self.connect("destroy", self._on_gateway_embed_dialog_destroy) + # Runtime probe is async so Settings never blocks on podman/docker. + # When runtime is already ready, this probes leftover compose immediately. + self._gateway_manager.begin_runtime_detection() + if not self._gateway_manager.runtime_ready: + self.gateway_run_btn.set_sensitive(False) + self.gateway_detail_label.set_text("Detecting container runtime…") + elif not self._gateway_manager.available: + self.gateway_run_btn.set_sensitive(False) + self.gateway_detail_label.set_text(self._gateway_manager.unavailable_hint) + else: + hint = getattr(self._gateway_manager.runner.sandbox, "hint", "") or "" + if hint: + self.gateway_detail_label.set_text(hint) + + self._apply_gateway_status_ui( + self._gateway_manager.status, self._gateway_manager.status_detail + ) + + def _on_gateway_embed_dialog_destroy(self, *_args) -> None: + """Drop status listener so polls never touch a destroyed Settings dialog.""" + manager = getattr(self, "_gateway_manager", None) + if manager is not None: + manager.remove_listener(self._on_gateway_status_from_worker) + + def _on_gateway_status_from_worker(self, status: GatewayStatus, detail: str) -> None: + """Marshal status updates onto the GTK main loop.""" + GLib.idle_add(self._apply_gateway_status_ui, status, detail) + + def _apply_gateway_status_ui(self, status: GatewayStatus, detail: str) -> bool: + if not hasattr(self, "gateway_status_label"): + return False + if not self._dialog_is_alive(): + return False + label = status.value if isinstance(status, GatewayStatus) else str(status) + self.gateway_status_label.set_text(label) + if detail: + self.gateway_detail_label.set_text(detail) + + runningish = status in { + GatewayStatus.STARTING, + GatewayStatus.LIVE, + GatewayStatus.PAIRABLE, + GatewayStatus.READY, + GatewayStatus.ERROR, + } + if not self._gateway_manager.runtime_ready: + self.gateway_run_btn.set_sensitive(False) + if not detail: + self.gateway_detail_label.set_text("Detecting container runtime…") + elif self._gateway_manager.available: + if runningish or self._gateway_manager.managed_by_us: + self.gateway_run_btn.set_label("Stop local Gateway") + else: + self.gateway_run_btn.set_label("Run VocaGateway locally") + self.gateway_run_btn.set_sensitive(status is not GatewayStatus.STARTING) + else: + self.gateway_run_btn.set_sensitive(False) + if not detail: + self.gateway_detail_label.set_text(self._gateway_manager.unavailable_hint) + + can_use = status in {GatewayStatus.PAIRABLE, GatewayStatus.READY} + self.gateway_use_btn.set_sensitive(can_use) + self._sync_gateway_lan_switch() + self._update_gateway_pairing_widgets() + return False + + def _sync_gateway_lan_switch(self) -> None: + """Keep Allow LAN honest with compose; skip the toggle handler.""" + desired = bool(self._gateway_manager.lan_publish) + if bool(self.gateway_lan_switch.get_active()) != desired: + was_init = bool(getattr(self, "_initializing", False)) + self._initializing = True + try: + self.gateway_lan_switch.set_active(desired) + finally: + self._initializing = was_init + saved = bool(self.config_manager.get("gateway_embed", "lan_publish", False)) + if saved != desired: + self.config_manager.set("gateway_embed", "lan_publish", desired) + self.config_manager.save_config() + + def _update_gateway_pairing_widgets(self) -> None: + info = self._gateway_manager.pairing + if info is None: + self._gateway_qr_cache_key = None + self.gateway_qr_image.hide() + self.gateway_qr_image.clear() + self.gateway_pairing_label.set_text( + "Pairing QR appears when the gateway is Live with a phone-reachable URL " + "(enable LAN for Phone, or set VOCAGATEWAY_PUBLIC_URL)." + ) + return + if not info.display_url: + self._gateway_qr_cache_key = None + self.gateway_qr_image.hide() + self.gateway_qr_image.clear() + self.gateway_pairing_label.set_text( + "Gateway is live on loopback only. Enable LAN access for Phone " + "(or configure a non-loopback PUBLIC_URL) before pairing another device. " + "Token is held for Use this Gateway and is not shown here." + ) + return + + self.gateway_pairing_label.set_text( + f"URL: {info.display_url}\n" + "Token is available to Use this Gateway and the QR; it is not logged." + ) + if info.qr_svg: + cache_key = (info.display_url, len(info.qr_svg)) + if getattr(self, "_gateway_qr_cache_key", None) != cache_key: + pixbuf = self._pixbuf_from_svg_bytes(info.qr_svg) + if pixbuf is not None: + self.gateway_qr_image.set_from_pixbuf(pixbuf) + self._gateway_qr_cache_key = cache_key + if getattr(self, "_gateway_qr_cache_key", None) == cache_key: + self.gateway_qr_image.show() + else: + self._gateway_qr_cache_key = None + self.gateway_qr_image.hide() + self.gateway_qr_image.clear() + + def _pixbuf_from_svg_bytes(self, data: bytes) -> Any: + """Best-effort SVG to GdkPixbuf; returns None when loaders are missing.""" + try: + from gi.repository import GdkPixbuf + + loader = GdkPixbuf.PixbufLoader.new_with_type("svg") + loader.write(data[: 512 * 1024]) + loader.close() + return loader.get_pixbuf() + except Exception: + return None + + def _on_gateway_run_clicked(self, widget: Any) -> None: + if not self._gateway_manager.available: + return + status = self._gateway_manager.status + if ( + status + in { + GatewayStatus.STARTING, + GatewayStatus.LIVE, + GatewayStatus.PAIRABLE, + GatewayStatus.READY, + GatewayStatus.ERROR, + } + or self._gateway_manager.managed_by_us + ): + self._gateway_manager.stop_async() + else: + lan = bool(self.gateway_lan_switch.get_active()) + self._gateway_manager.lan_publish = lan + self._gateway_manager.start_async(lan_publish=lan) + + def _on_gateway_lan_toggled(self, widget: Any, _pspec: Any = None) -> None: + if getattr(self, "_initializing", False): + return + active = bool(self.gateway_lan_switch.get_active()) + self.config_manager.set("gateway_embed", "lan_publish", active) + self.config_manager.save_config() + # Republish when compose is already running so QR/bind match LAN. + self._gateway_manager.apply_lan_publish(active) + self._update_gateway_pairing_widgets() + + def _on_gateway_use_clicked(self, widget: Any) -> None: + try: + preset = self._gateway_manager.use_this_gateway(self.config_manager) + except Exception as exc: + self.gateway_detail_label.set_text(str(exc)) + return + self._applying_settings = True + try: + self.remote_api_url_entry.set_text(preset["remote_api_url"]) + self.remote_api_key_entry.set_text(preset["remote_api_key"]) + self.remote_api_endpoint_combo.set_active_id(preset["remote_api_endpoint"]) + self.remote_api_model_entry.set_text(preset.get("remote_api_model") or "whisper-1") + if hasattr(self, "engine_combo"): + self.engine_combo.set_active_id("remote_api") + finally: + self._applying_settings = False + self._auto_apply_settings() + self.gateway_detail_label.set_text( + "Remote Server now points at this gateway " f"({preset['remote_api_endpoint']})." + ) + self._update_engine_specific_ui() + def _on_power_user_toggled(self, widget, state): """Handle the power-user opt-in toggle.""" if self._initializing or self._applying_settings: diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index c0a783a..d2040c6 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -10,7 +10,7 @@ import os import signal import threading import time -from typing import Callable, Optional +from typing import Any, Callable, Optional import gi @@ -37,6 +37,7 @@ from gi.repository import GdkPixbuf, Gio, GLib, GObject, Gtk # Import local modules - Use protocols to avoid circular imports from ..auto_pause_monitor import DEFAULT_POLL_INTERVAL_SECONDS, AutoPauseMonitor from ..common_types import RecognitionState, SpeechRecognitionManagerProtocol, TextInjectorProtocol +from ..gateway_embed import GatewayStatus, get_gateway_embed_manager from ..model_keepalive import DEFAULT_IDLE_TIMEOUT_SECONDS, ModelKeepAlive from ..suspend_handler import SuspendHandler from ..utils.host_process import host_env @@ -343,6 +344,11 @@ class TrayIndicator: self._add_menu_separator() self._add_menu_item("Settings", self._on_settings_clicked) self._add_menu_item("View Logs", self._on_logs_clicked) + self._gateway_stop_menu_item = self._add_menu_item( + "Stop local Gateway", self._on_stop_local_gateway_clicked + ) + self._gateway_stop_menu_item.set_no_show_all(True) + self._gateway_stop_menu_item.hide() self._add_menu_separator() # Hidden until a background check finds a newer release. self._update_menu_item = self._add_menu_item( @@ -363,6 +369,12 @@ class TrayIndicator: self._update_menu_item.hide() else: self._show_update_menu_item(self._pending_update) + self._gateway_stop_menu_item.hide() + self._gateway_manager = get_gateway_embed_manager() + self._gateway_manager.add_listener(self._on_gateway_status_for_tray) + # Runtime detect + leftover compose probe (Quit does not stop compose). + self._gateway_manager.begin_runtime_detection() + self._sync_gateway_stop_menu(self._gateway_manager.status) # Update the UI based on the initial state self._update_ui(RecognitionState.IDLE) @@ -1154,6 +1166,32 @@ class TrayIndicator: self._input_monitor.cancel() self._input_monitor = None + def _on_gateway_status_for_tray(self, status: GatewayStatus, detail: str) -> None: + """Update tray Stop local Gateway visibility from a worker thread.""" + GLib.idle_add(self._sync_gateway_stop_menu, status) + + def _sync_gateway_stop_menu(self, status: GatewayStatus) -> bool: + item = getattr(self, "_gateway_stop_menu_item", None) + if item is None: + return False + # Show Stop for leftover compose too; managed_by_us is session memory only. + show = status in { + GatewayStatus.STARTING, + GatewayStatus.LIVE, + GatewayStatus.PAIRABLE, + GatewayStatus.READY, + GatewayStatus.ERROR, + } + if show: + item.show() + else: + item.hide() + return False + + def _on_stop_local_gateway_clicked(self, widget: Any) -> None: + """Stop local compose, including leftovers from a previous session.""" + get_gateway_embed_manager().stop_async() + def _on_quit_clicked(self, widget): """Handle click on the Quit menu item.""" logger.debug("Quit clicked") diff --git a/tests/test_appimage_packaging.py b/tests/test_appimage_packaging.py index d4257c4..98b6b5f 100644 --- a/tests/test_appimage_packaging.py +++ b/tests/test_appimage_packaging.py @@ -1,6 +1,7 @@ """Regression guards for AppImage packaging.""" import re +import tomllib from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] @@ -17,6 +18,7 @@ BOOT_FAMILIES = { } PINS = APPIMAGE / "tool_checksums.txt" REQUIREMENTS = REPO_ROOT / "requirements" +PYPROJECT = REPO_ROOT / "pyproject.toml" WORKFLOWS = REPO_ROOT / ".github" / "workflows" #: Workflows that build an AppImage, and so decide which distros it runs on. APPIMAGE_WORKFLOWS = ("unified-pipeline.yml", "release.yml", "nightly.yml") @@ -300,3 +302,25 @@ def test_boot_test_has_a_package_recipe_for_every_distro_it_runs_on(): if family is None or family not in case_block.group(0): missing.append(distro) assert not missing, f"boot-test.sh installs nothing for: {missing}" + + +def test_setuptools_packages_cover_every_python_subpackage(): + """AppImage and Flatpak install the wheel built from an explicit packages + list. A new subpackage left off that list boots fine from a source tree and + dies in the bundle with ModuleNotFoundError (#774). + """ + declared = set( + tomllib.loads(PYPROJECT.read_text(encoding="utf-8"))["tool"]["setuptools"]["packages"] + ) + src_root = REPO_ROOT / "src" / "vocalinux" + discovered = { + ( + "vocalinux" + if init.parent == src_root + else "vocalinux." + ".".join(init.parent.relative_to(src_root).parts) + ) + for init in src_root.rglob("__init__.py") + if "__pycache__" not in init.parts + } + missing = sorted(discovered - declared) + assert not missing, f"add these to tool.setuptools.packages in pyproject.toml: {missing}" diff --git a/tests/test_gateway_embed.py b/tests/test_gateway_embed.py new file mode 100644 index 0000000..c2ec224 --- /dev/null +++ b/tests/test_gateway_embed.py @@ -0,0 +1,978 @@ +"""Unit tests for local VocaGateway embed helpers (no real containers).""" + +from __future__ import annotations + +import io +import json +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from vocalinux.gateway_embed.pairing import ( + MAX_QR_SVG_BYTES, + decode_pairing_payload, +) +from vocalinux.gateway_embed.preset import ( + GATEWAY_TRANSCRIPTIONS_ENDPOINT, + remote_api_preset_from_pairing, +) +from vocalinux.gateway_embed.runtime import ContainerRuntime, detect_container_runtime +from vocalinux.gateway_embed.sandbox import detect_sandbox +from vocalinux.gateway_embed.status import GatewayStatus +from vocalinux.gateway_embed.urls import is_loopback_url, reject_loopback_url +from vocalinux.ui.config_manager import DEFAULT_CONFIG + + +class TestLoopbackRejection(unittest.TestCase): + def test_rejects_localhost_variants(self): + for url in ( + "http://127.0.0.1:8765", + "http://localhost:8765", + "http://[::1]:8765", + "https://127.0.0.1/", + "", + None, + ): + self.assertTrue(is_loopback_url(url), url) + self.assertIsNone(reject_loopback_url(url)) + + def test_rejects_short_form_loopback_and_localhost_star(self): + # Contract / gateway policy: 127.1 and localhost.* must never be Pairable. + for url in ( + "http://127.1:8765", + "http://127.0.1:8765", + "http://localhost.localdomain:8765", + "http://localhost.foo:8765", + ): + self.assertTrue(is_loopback_url(url), url) + self.assertIsNone(reject_loopback_url(url), url) + + def test_rejects_link_local_for_qr(self): + for url in ( + "http://169.254.10.20:8765", + "http://[fe80::1]:8765", + ): + self.assertIsNone(reject_loopback_url(url), url) + + def test_rejects_default_container_bridge(self): + for url in ( + "http://172.17.0.1:8765", + "http://172.17.0.2:8765", + "http://10.88.0.1:8765", + ): + self.assertIsNone(reject_loopback_url(url), url) + + def test_accepts_lan(self): + url = "http://192.168.1.20:8765" + self.assertFalse(is_loopback_url(url)) + self.assertEqual(reject_loopback_url(url), url) + + +class TestRuntimeDetector(unittest.TestCase): + def test_prefers_podman(self): + def lookup(name): + return { + "podman": "/usr/bin/podman", + "docker": "/usr/bin/docker", + }.get(name) + + def probe(argv, **_kwargs): + return argv[0] in {"/usr/bin/podman", "/usr/bin/docker"} and argv[1] in { + "info", + "version", + "compose", + } + + info = detect_container_runtime(path_lookup=lookup, probe=probe) + self.assertEqual(info.kind, ContainerRuntime.PODMAN) + self.assertEqual(info.binary, "/usr/bin/podman") + + def test_falls_back_to_docker(self): + def lookup(name): + return {"docker": "/usr/bin/docker"}.get(name) + + def probe(argv, **_kwargs): + return argv[0] == "/usr/bin/docker" + + info = detect_container_runtime(path_lookup=lookup, probe=probe) + self.assertEqual(info.kind, ContainerRuntime.DOCKER) + + def test_none_with_hint(self): + info = detect_container_runtime(path_lookup=lambda _n: None, probe=lambda *_a, **_k: False) + self.assertEqual(info.kind, ContainerRuntime.NONE) + self.assertIn("podman", info.hint.lower()) + + def test_engine_without_compose_is_unavailable(self): + def lookup(name): + return {"podman": "/usr/bin/podman"}.get(name) + + def probe(argv, **_kwargs): + # Engine probes succeed; the compose plugin probe (argv[1]=="compose") fails. + return ( + bool(argv) + and argv[0] == "/usr/bin/podman" + and len(argv) > 1 + and argv[1] in {"info", "version"} + ) + + info = detect_container_runtime(path_lookup=lookup, probe=probe) + self.assertEqual(info.kind, ContainerRuntime.PODMAN) + self.assertEqual(info.compose_args, ()) + self.assertIn("compose", info.hint.lower()) + + from vocalinux.gateway_embed.runner import GatewayRunner + + runner = GatewayRunner( + runtime=info, + sandbox=detect_sandbox({}), + run=MagicMock(), + ) + self.assertFalse(runner.available) + self.assertEqual(runner.unavailable_hint, info.hint) + self.assertNotIn("No container runtime", runner.unavailable_hint) + + +class TestStatusMachine(unittest.TestCase): + def test_ready_beats_pairable(self): + self.assertEqual( + GatewayStatus.from_health(live=True, ready=True, pairable=True, running=True), + GatewayStatus.READY, + ) + + def test_pairable_before_ready(self): + self.assertEqual( + GatewayStatus.from_health(live=True, ready=False, pairable=True, running=True), + GatewayStatus.PAIRABLE, + ) + + def test_live_only(self): + self.assertEqual( + GatewayStatus.from_health(live=True, ready=False, pairable=False, running=True), + GatewayStatus.LIVE, + ) + + def test_starting(self): + self.assertEqual( + GatewayStatus.from_health( + live=False, ready=False, pairable=False, running=True, starting=True + ), + GatewayStatus.STARTING, + ) + + def test_stopped(self): + self.assertEqual( + GatewayStatus.from_health(live=False, ready=False, pairable=False, running=False), + GatewayStatus.STOPPED, + ) + + def test_error(self): + self.assertEqual( + GatewayStatus.from_health( + live=False, ready=False, pairable=False, error=True, running=False + ), + GatewayStatus.ERROR, + ) + + +class TestRemoteApiPreset(unittest.TestCase): + def test_maps_openai_path(self): + preset = remote_api_preset_from_pairing( + url="http://192.168.1.20:8765", + token="a" * 32, + ) + self.assertEqual(preset["engine"], "remote_api") + self.assertEqual(preset["remote_api_url"], "http://192.168.1.20:8765") + self.assertEqual(preset["remote_api_key"], "a" * 32) + self.assertEqual(preset["remote_api_endpoint"], GATEWAY_TRANSCRIPTIONS_ENDPOINT) + + def test_rejects_empty(self): + with self.assertRaises(ValueError): + remote_api_preset_from_pairing(url="", token="x" * 32) + + +class TestPairingDecode(unittest.TestCase): + def test_payload_object(self): + info = decode_pairing_payload( + { + "payload": { + "v": 1, + "url": "http://10.0.0.5:8765", + "token": "tokentokentokentokentokentoken12", + } + } + ) + self.assertEqual(info.display_url, "http://10.0.0.5:8765") + self.assertTrue(info.pairable) + + def test_payload_string_and_loopback_hidden(self): + raw = json.dumps({"v": 1, "url": "http://127.0.0.1:8765", "token": "t" * 32}) + info = decode_pairing_payload({"payload": raw}) + self.assertIsNone(info.display_url) + self.assertFalse(info.pairable) + + def test_docker_bridge_not_pairable(self): + info = decode_pairing_payload( + { + "payload": { + "v": 1, + "url": "http://172.17.0.2:8765", + "token": "tokentokentokentokentokentoken12", + } + } + ) + self.assertIsNone(info.display_url) + self.assertFalse(info.pairable) + + def test_link_local_not_pairable(self): + info = decode_pairing_payload( + { + "payload": { + "v": 1, + "url": "http://169.254.1.2:8765", + "token": "tokentokentokentokentokentoken12", + } + } + ) + self.assertIsNone(info.display_url) + self.assertFalse(info.pairable) + + def test_short_form_loopback_not_pairable(self): + info = decode_pairing_payload( + { + "payload": { + "v": 1, + "url": "http://127.1:8765", + "token": "tokentokentokentokentokentoken12", + } + } + ) + self.assertIsNone(info.display_url) + self.assertFalse(info.pairable) + + def test_qr_download_capped(self): + from vocalinux.gateway_embed.pairing import _read_capped + + class FakeResp: + def __init__(self, payload: bytes): + self._buf = io.BytesIO(payload) + + def read(self, n=-1): + return self._buf.read(n if n is not None else -1) + + huge = b"x" * (MAX_QR_SVG_BYTES + 10) + with self.assertRaises(ValueError): + _read_capped(FakeResp(huge), MAX_QR_SVG_BYTES) + + +class TestSandbox(unittest.TestCase): + def test_flatpak_fails_closed(self): + state = detect_sandbox({"FLATPAK_ID": "com.vocahq.Vocalinux"}) + self.assertTrue(state.blocked) + self.assertEqual(state.kind, "flatpak") + self.assertIn("Flatpak", state.hint) + + def test_host_ok(self): + state = detect_sandbox({}) + self.assertFalse(state.blocked) + + +class TestDefaultEngineUnchanged(unittest.TestCase): + def test_default_still_whisper_cpp(self): + self.assertEqual(DEFAULT_CONFIG["speech_recognition"]["engine"], "whisper_cpp") + self.assertIn("gateway_embed", DEFAULT_CONFIG) + self.assertFalse(DEFAULT_CONFIG["gateway_embed"]["lan_publish"]) + + +class TestRunnerNoVolumeWipe(unittest.TestCase): + def test_stop_refuses_wipe_flag(self): + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + runner = GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=MagicMock(), + ) + with self.assertRaises(RuntimeError): + runner.stop(wipe_volumes=True) + + +class TestComposeCpuProfile(unittest.TestCase): + def test_start_uses_profile_cpu(self): + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + calls = [] + + def fake_run(argv, **kwargs): + calls.append(list(argv)) + completed = MagicMock() + completed.returncode = 0 + completed.stdout = b"" + completed.stderr = b"" + return completed + + runner = GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=fake_run, + ) + with patch.object(runner, "prepare", return_value=("/tmp/gw", "t" * 32, "/tmp/.env")): + result = runner.start(lan_publish=False) + self.assertTrue(result.ok) + compose_calls = [c for c in calls if c[:2] == ["/usr/bin/podman", "compose"]] + self.assertTrue(compose_calls) + argv = compose_calls[0] + self.assertIn("--profile", argv) + self.assertEqual(argv[argv.index("--profile") + 1], "cpu") + self.assertIn("up", argv) + self.assertIn("-d", argv) + self.assertIn("gateway", argv) + + +class TestImagePin(unittest.TestCase): + def test_rejects_latest_override(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + with patch.dict("os.environ", {"VOCAGATEWAY_IMAGE": "vocagateway:latest"}): + write_env_file(token="a" * 32, lan_publish=False, path=env_path) + body = Path(env_path).read_text(encoding="utf-8") + self.assertIn("VOCAGATEWAY_IMAGE=vocagateway:v0.1.0", body) + self.assertNotIn(":latest", body) + + def test_rejects_newline_image_injection(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + evil = "vocagateway:v0.1.0\nEVIL=1" + with patch.dict("os.environ", {"VOCAGATEWAY_IMAGE": evil}): + write_env_file(token="a" * 32, lan_publish=False, path=env_path) + body = Path(env_path).read_text(encoding="utf-8") + self.assertIn("VOCAGATEWAY_IMAGE=vocagateway:v0.1.0\n", body) + self.assertNotIn("EVIL=", body) + + def test_omits_loopback_public_url(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file( + token="a" * 32, + lan_publish=True, + public_url="http://127.0.0.1:8765", + path=env_path, + ) + body = Path(env_path).read_text(encoding="utf-8") + self.assertNotIn("VOCAGATEWAY_PUBLIC_URL=", body) + + def test_read_lan_publish_from_env_matches_write(self): + import tempfile + + from vocalinux.gateway_embed.runner import ( + read_lan_publish_from_env, + write_env_file, + ) + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file(token="a" * 32, lan_publish=True, path=env_path) + self.assertTrue(read_lan_publish_from_env(env_path)) + write_env_file(token="a" * 32, lan_publish=False, path=env_path) + self.assertFalse(read_lan_publish_from_env(env_path)) + self.assertFalse(read_lan_publish_from_env(f"{tmp}/missing.env")) + + +class TestTokenFileCap(unittest.TestCase): + def test_oversized_token_file_regenerated(self): + import tempfile + + from vocalinux.gateway_embed.paths_embed import ensure_token_file + + with tempfile.TemporaryDirectory() as tmp: + path = f"{tmp}/token" + with open(path, "wb") as handle: + handle.write(b"a" * 9000) + token = ensure_token_file(path) + self.assertEqual(len(token), 64) + self.assertLess(Path(path).stat().st_size, 200) + + +class TestLanPublishGate(unittest.TestCase): + def _manager(self): + from vocalinux.gateway_embed.manager import GatewayEmbedManager + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + runner = GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=MagicMock(), + ) + return GatewayEmbedManager(runner=runner), runner + + def test_lan_toggle_without_republish_not_pairable(self): + """Desired LAN alone must not advertise Pairable on a loopback bind.""" + from vocalinux.gateway_embed.pairing import PairingInfo + + manager, runner = self._manager() + runner.managed_by_us = True + manager.lan_publish = True + manager._compose_lan_publish = False + manager._token = "t" * 32 + self.assertFalse(manager._effective_lan_for_pairing()) + + captured = {} + + def fake_fetch(base_url, token, *, public_url=None, fetch_qr=True, timeout=3.0): + captured["public_url"] = public_url + return PairingInfo( + version=1, + url="http://127.0.0.1:8765", + token=token, + display_url=None, + raw_payload={"v": 1, "url": "http://127.0.0.1:8765", "token": token}, + ) + + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch("vocalinux.gateway_embed.manager.fetch_pairing", side_effect=fake_fetch): + status = manager.refresh_status() + self.assertIsNone(captured.get("public_url")) + self.assertEqual(status, GatewayStatus.LIVE) + self.assertIsNotNone(manager.pairing) + self.assertFalse(manager.pairing.pairable) + + def test_apply_lan_publish_republishes_when_managed(self): + from vocalinux.gateway_embed.pairing import PairingInfo + from vocalinux.gateway_embed.runner import RunnerResult + + manager, runner = self._manager() + runner.managed_by_us = True + manager.lan_publish = False + manager._compose_lan_publish = False + manager._pairing = PairingInfo( + version=1, + url="http://127.0.0.1:8765", + token="t" * 32, + display_url=None, + raw_payload={}, + ) + calls = {"n": 0} + + def fake_republish(*, lan_publish, public_url=None): + calls["n"] += 1 + self.assertTrue(lan_publish) + return RunnerResult(ok=True, message="republished") + + with patch("threading.Thread") as fake_thread: + fake_thread.return_value = MagicMock() + with patch.object(runner, "republish", side_effect=fake_republish): + with patch.object(manager, "refresh_status", return_value=GatewayStatus.LIVE): + manager.apply_lan_publish(True) + self.assertIsNone(manager.pairing) + self.assertTrue(manager._republish_started) + # Run worker synchronously (Thread.start was mocked). + manager._republish_worker() + self.assertEqual(calls["n"], 1) + self.assertTrue(manager._compose_lan_publish) + self.assertTrue(manager.lan_publish) + self.assertFalse(manager._republish_started) + + def test_start_worker_snapshots_lan_if_switch_flips_mid_start(self): + from vocalinux.gateway_embed.runner import RunnerResult + + manager, runner = self._manager() + manager.lan_publish = True + manager._compose_lan_publish = None + + def fake_start(*, lan_publish, public_url=None): + self.assertTrue(lan_publish) + manager.lan_publish = False + runner.managed_by_us = True + return RunnerResult(ok=True, message="started") + + with patch("threading.Thread") as fake_thread: + fake_thread.return_value = MagicMock() + with patch( + "vocalinux.gateway_embed.paths_embed.ensure_token_file", + return_value="t" * 32, + ): + with patch.object(runner, "ensure_runtime"): + with patch.object(runner, "start", side_effect=fake_start): + with patch.object(manager, "_start_polling"): + with patch.object(manager, "refresh_status"): + manager._start_worker() + + self.assertIs(manager._compose_lan_publish, True) + self.assertFalse(manager.lan_publish) + self.assertTrue(manager._republish_started) + + def test_republish_worker_snapshots_lan_if_switch_flips_mid_republish(self): + from vocalinux.gateway_embed.runner import RunnerResult + + manager, runner = self._manager() + runner.managed_by_us = True + manager.lan_publish = True + manager._compose_lan_publish = False + manager._republish_started = True + + def fake_republish(*, lan_publish, public_url=None): + self.assertTrue(lan_publish) + manager.lan_publish = False + return RunnerResult(ok=True, message="republished") + + with patch("threading.Thread") as fake_thread: + fake_thread.return_value = MagicMock() + with patch.object(runner, "republish", side_effect=fake_republish): + with patch.object(manager, "refresh_status"): + manager._republish_worker() + + self.assertIs(manager._compose_lan_publish, True) + self.assertFalse(manager.lan_publish) + self.assertTrue(manager._republish_started) + + def test_live_detail_warns_when_lan_off_but_compose_still_open(self): + from vocalinux.gateway_embed.pairing import PairingInfo + + manager, runner = self._manager() + runner.managed_by_us = True + manager.lan_publish = False + manager._compose_lan_publish = True + manager._token = "t" * 32 + + def fake_fetch(base_url, token, *, public_url=None, fetch_qr=True, timeout=3.0): + return PairingInfo( + version=1, + url="http://127.0.0.1:8765", + token=token, + display_url=None, + raw_payload={"v": 1, "url": "http://127.0.0.1:8765", "token": token}, + ) + + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch("vocalinux.gateway_embed.manager.fetch_pairing", side_effect=fake_fetch): + status = manager.refresh_status() + self.assertEqual(status, GatewayStatus.LIVE) + self.assertIn("still open on the LAN", manager.status_detail) + self.assertFalse(manager._effective_lan_for_pairing()) + + def test_omits_bridge_public_url_from_env(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file( + token="a" * 32, + lan_publish=True, + public_url="http://172.17.0.2:8765", + path=env_path, + ) + body = Path(env_path).read_text(encoding="utf-8") + self.assertNotIn("VOCAGATEWAY_PUBLIC_URL=", body) + + +class TestOrphanComposeDetect(unittest.TestCase): + """Leftover compose after Quit must not look Stopped.""" + + def _manager(self): + from vocalinux.gateway_embed.manager import GatewayEmbedManager + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + runner = GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=MagicMock(), + ) + runner.managed_by_us = False + return GatewayEmbedManager(runner=runner), runner + + def test_refresh_status_does_not_stay_stopped_for_orphan(self): + manager, runner = self._manager() + self.assertFalse(manager.managed_by_us) + self.assertEqual(manager.status, GatewayStatus.STOPPED) + + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + status = manager.refresh_status() + + self.assertEqual(status, GatewayStatus.LIVE) + self.assertNotEqual(status, GatewayStatus.STOPPED) + self.assertFalse(manager.managed_by_us) + stoppable = { + GatewayStatus.STARTING, + GatewayStatus.LIVE, + GatewayStatus.PAIRABLE, + GatewayStatus.READY, + GatewayStatus.ERROR, + } + self.assertIn(status, stoppable) + + def test_refresh_status_starting_when_compose_up_but_not_live(self): + manager, runner = self._manager() + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=False, ready=False, error="") + status = manager.refresh_status() + self.assertEqual(status, GatewayStatus.STARTING) + self.assertNotEqual(status, GatewayStatus.STOPPED) + + def test_runtime_ready_probe_adopts_orphan_and_starts_polling(self): + manager, runner = self._manager() + self.assertTrue(manager.runtime_ready) + self.assertFalse(manager._runtime_detect_started) + + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch.object(manager, "_start_polling") as poll: + manager.begin_runtime_detection() + + self.assertEqual(manager.status, GatewayStatus.LIVE) + poll.assert_called_once() + + def test_orphan_probe_loads_existing_token(self): + import tempfile + + manager, runner = self._manager() + with tempfile.TemporaryDirectory() as tmp: + token_path = f"{tmp}/token" + existing = "ab" * 32 + with open(token_path, "w", encoding="utf-8") as handle: + handle.write(existing + "\n") + with patch( + "vocalinux.gateway_embed.paths_embed.token_file_path", + return_value=token_path, + ): + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=False, ready=False, error="") + with patch.object(manager, "_start_polling"): + manager.begin_runtime_detection() + self.assertEqual(manager._token, existing) + + def test_stop_worker_stops_orphan_without_managed_by_us(self): + from vocalinux.gateway_embed.runner import RunnerResult + + manager, runner = self._manager() + self.assertFalse(manager.managed_by_us) + + with patch.object( + runner, "stop", return_value=RunnerResult(ok=True, message="stopped") + ) as stop: + manager._stop_worker() + + stop.assert_called_once_with(wipe_volumes=False) + self.assertEqual(manager.status, GatewayStatus.STOPPED) + self.assertFalse(manager.managed_by_us) + + def test_stop_async_starts_worker_for_orphan(self): + manager, runner = self._manager() + self.assertFalse(manager.managed_by_us) + with patch("threading.Thread") as fake_thread: + started = MagicMock() + fake_thread.return_value = started + manager.stop_async() + fake_thread.assert_called_once() + kwargs = fake_thread.call_args.kwargs + self.assertEqual(kwargs.get("name"), "vocalinux-gateway-stop") + self.assertEqual(kwargs.get("target").__func__, manager._stop_worker.__func__) + started.start.assert_called_once() + + def test_orphan_adopt_restores_lan_on_from_env(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + manager, runner = self._manager() + self.assertFalse(runner.managed_by_us) + manager.lan_publish = False + manager._compose_lan_publish = None + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file(token="a" * 32, lan_publish=True, path=env_path) + with patch( + "vocalinux.gateway_embed.runner.env_file_path", + return_value=env_path, + ): + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch.object(manager, "_start_polling"): + manager.begin_runtime_detection() + + self.assertIs(manager._compose_lan_publish, True) + self.assertTrue(manager.lan_publish) + self.assertTrue(runner.managed_by_us) + self.assertTrue(manager.managed_by_us) + + def test_orphan_adopt_restores_lan_off_from_loopback_env(self): + import tempfile + + from vocalinux.gateway_embed.runner import write_env_file + + manager, runner = self._manager() + manager.lan_publish = True + manager._compose_lan_publish = None + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file(token="a" * 32, lan_publish=False, path=env_path) + with patch( + "vocalinux.gateway_embed.runner.env_file_path", + return_value=env_path, + ): + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch.object(manager, "_start_polling"): + manager.begin_runtime_detection() + + self.assertIs(manager._compose_lan_publish, False) + self.assertFalse(manager.lan_publish) + self.assertTrue(runner.managed_by_us) + + def test_apply_lan_publish_republishes_after_orphan_lan_adopt(self): + import tempfile + + from vocalinux.gateway_embed.runner import RunnerResult, write_env_file + + manager, runner = self._manager() + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + write_env_file(token="a" * 32, lan_publish=True, path=env_path) + with patch( + "vocalinux.gateway_embed.runner.env_file_path", + return_value=env_path, + ): + with patch.object(runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch.object(manager, "_start_polling"): + manager.begin_runtime_detection() + + self.assertTrue(runner.managed_by_us) + self.assertIs(manager._compose_lan_publish, True) + + calls = {"n": 0} + + def fake_republish(*, lan_publish, public_url=None): + calls["n"] += 1 + self.assertFalse(lan_publish) + return RunnerResult(ok=True, message="republished") + + with patch("threading.Thread") as fake_thread: + fake_thread.return_value = MagicMock() + with patch.object(runner, "republish", side_effect=fake_republish): + with patch.object(manager, "refresh_status", return_value=GatewayStatus.LIVE): + manager.apply_lan_publish(False) + self.assertTrue(manager._republish_started) + manager._republish_worker() + + self.assertEqual(calls["n"], 1) + self.assertIs(manager._compose_lan_publish, False) + self.assertFalse(manager.lan_publish) + + +class TestFailedRepublishEnvRestore(unittest.TestCase): + """Failed force-recreate must not leave .env lying about the live bind.""" + + def _podman_runner(self, run): + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + return GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=run, + ) + + def test_failed_republish_restores_previous_env_publish_host(self): + import os + import tempfile + + from vocalinux.gateway_embed.runner import ( + read_lan_publish_from_env, + write_env_file, + ) + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + checkout = f"{tmp}/checkout" + os.makedirs(checkout) + Path(checkout, "compose.yaml").write_text("services: {}\n", encoding="utf-8") + write_env_file(token="a" * 32, lan_publish=True, path=env_path) + self.assertTrue(read_lan_publish_from_env(env_path)) + + def fake_run(argv, **_kwargs): + completed = MagicMock() + completed.returncode = 1 + completed.stdout = b"" + completed.stderr = b"force-recreate failed" + return completed + + runner = self._podman_runner(fake_run) + runner.managed_by_us = True + with patch("vocalinux.gateway_embed.runner.env_file_path", return_value=env_path): + with patch( + "vocalinux.gateway_embed.runner.ensure_gateway_checkout", + return_value=checkout, + ): + with patch( + "vocalinux.gateway_embed.runner.ensure_token_file", + return_value="a" * 32, + ): + result = runner.republish(lan_publish=False) + + self.assertFalse(result.ok) + self.assertTrue(read_lan_publish_from_env(env_path)) + body = Path(env_path).read_text(encoding="utf-8") + self.assertIn("VOCAGATEWAY_PUBLISH_HOST=0.0.0.0", body) + self.assertNotIn("VOCAGATEWAY_PUBLISH_HOST=127.0.0.1", body) + + def test_failed_republish_reverts_manager_lan_publish_to_compose(self): + from vocalinux.gateway_embed.manager import GatewayEmbedManager + from vocalinux.gateway_embed.runner import RunnerResult + + runner = self._podman_runner(MagicMock()) + runner.managed_by_us = True + manager = GatewayEmbedManager(runner=runner) + manager.lan_publish = False + manager._compose_lan_publish = True + manager._republish_started = True + + def fake_republish(*, lan_publish, public_url=None): + self.assertFalse(lan_publish) + return RunnerResult(ok=False, message="force-recreate failed") + + with patch("threading.Thread") as fake_thread: + fake_thread.return_value = MagicMock() + with patch.object(runner, "republish", side_effect=fake_republish): + manager._republish_worker() + + self.assertIs(manager._compose_lan_publish, True) + self.assertTrue(manager.lan_publish) + self.assertEqual(manager.status, GatewayStatus.ERROR) + self.assertIn("previous bind is still in use", manager.status_detail) + self.assertIn("Stop and Run", manager.status_detail) + self.assertFalse(manager._republish_started) + + def test_orphan_adopt_after_failed_republish_still_sees_lan_from_env(self): + import os + import tempfile + + from vocalinux.gateway_embed.manager import GatewayEmbedManager + from vocalinux.gateway_embed.runner import ( + read_lan_publish_from_env, + write_env_file, + ) + + with tempfile.TemporaryDirectory() as tmp: + env_path = f"{tmp}/.env" + checkout = f"{tmp}/checkout" + os.makedirs(checkout) + Path(checkout, "compose.yaml").write_text("services: {}\n", encoding="utf-8") + write_env_file(token="a" * 32, lan_publish=True, path=env_path) + + def fake_run(argv, **_kwargs): + completed = MagicMock() + completed.returncode = 1 + completed.stdout = b"" + completed.stderr = b"force-recreate failed" + return completed + + runner = self._podman_runner(fake_run) + runner.managed_by_us = True + with patch("vocalinux.gateway_embed.runner.env_file_path", return_value=env_path): + with patch( + "vocalinux.gateway_embed.runner.ensure_gateway_checkout", + return_value=checkout, + ): + with patch( + "vocalinux.gateway_embed.runner.ensure_token_file", + return_value="a" * 32, + ): + result = runner.republish(lan_publish=False) + self.assertFalse(result.ok) + self.assertTrue(read_lan_publish_from_env(env_path)) + + next_runner = self._podman_runner(MagicMock()) + manager = GatewayEmbedManager(runner=next_runner) + manager.lan_publish = False + manager._compose_lan_publish = None + with patch.object(next_runner, "is_compose_running", return_value=True): + with patch("vocalinux.gateway_embed.manager.probe_health") as health: + health.return_value = MagicMock(live=True, ready=False, error="") + with patch.object(manager, "_start_polling"): + manager.begin_runtime_detection() + + self.assertTrue(manager.lan_publish) + self.assertIs(manager._compose_lan_publish, True) + self.assertTrue(next_runner.managed_by_us) + + +class TestManagerListenerCleanup(unittest.TestCase): + def test_remove_listener_stops_callbacks(self): + """Settings destroy must be able to detach without further emits.""" + from vocalinux.gateway_embed.manager import GatewayEmbedManager + from vocalinux.gateway_embed.runner import GatewayRunner + from vocalinux.gateway_embed.runtime import RuntimeInfo + + runner = GatewayRunner( + runtime=RuntimeInfo( + kind=ContainerRuntime.PODMAN, + binary="/usr/bin/podman", + compose_args=("/usr/bin/podman", "compose"), + ), + sandbox=detect_sandbox({}), + run=MagicMock(), + ) + manager = GatewayEmbedManager(runner=runner) + seen: list[str] = [] + + def listener(status, detail): + seen.append(detail) + + manager.add_listener(listener) + manager._emit(manager.status, "one") + manager.remove_listener(listener) + manager._emit(manager.status, "two") + self.assertEqual(seen, ["one"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_playback_ducker.py b/tests/test_playback_ducker.py index b1061b4..6b37096 100644 --- a/tests/test_playback_ducker.py +++ b/tests/test_playback_ducker.py @@ -264,6 +264,47 @@ def test_unwritable_restore_point_does_not_lower_the_sink( assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() +def test_failed_set_that_applied_keeps_the_restore_point( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A set that reports failure can still have applied; keep the way back.""" + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + + def apply_but_report_false(sink_id: str, linear: tuple[float, ...]) -> bool: + recorded: Union[float, tuple[float, ...]] = linear[0] if len(linear) == 1 else linear + sink.sets.append((sink_id, recorded)) + sink._channels = tuple(linear) + return False + + monkeypatch.setattr(sink, "set_volume", apply_but_report_false) + ducker.duck() + + assert sink.volume == pytest.approx(0.1) + assert (tmp_path / duck.PENDING_RECORD_NAME).exists() + + ducker.restore() + assert sink.volume == pytest.approx(0.5) + + +def test_failed_set_confirmed_unchanged_drops_the_record( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + sink = FakeSink(volume=0.5) + ducker = _ducker(tmp_path, sink, percent=20) + + def reject(sink_id: str, linear: tuple[float, ...]) -> bool: + recorded: Union[float, tuple[float, ...]] = linear[0] if len(linear) == 1 else linear + sink.sets.append((sink_id, recorded)) + return False + + monkeypatch.setattr(sink, "set_volume", reject) + ducker.duck() + + assert sink.volume == pytest.approx(0.5) + assert not (tmp_path / duck.PENDING_RECORD_NAME).exists() + + def test_percent_100_does_not_change_the_sink(tmp_path: Path) -> None: sink = FakeSink(volume=0.5) ducker = _ducker(tmp_path, sink, percent=100) diff --git a/tests/test_settings_dialog.py b/tests/test_settings_dialog.py index 41b28b7..db66da7 100644 --- a/tests/test_settings_dialog.py +++ b/tests/test_settings_dialog.py @@ -534,6 +534,8 @@ class TestSettingsDialogInstantApply(unittest.TestCase): # separate Advanced page, which the advanced_tab assertions below still guard. self.assertIn("self.advanced_box.pack_start(self.remote_server_group", source_code) self.assertIn("self.advanced_box.pack_start(self.remote_status_label", source_code) + self.assertIn("self.advanced_box.pack_start(self.gateway_embed_group", source_code) + self.assertNotIn("self.content_box.pack_start(self.gateway_embed_group", source_code) self.assertIn("self.remote_api_model_entry", source_code) self.assertIn("OpenAI/FunASR", source_code) self.assertNotIn("advanced_tab.pack_start(self.remote_server_group", source_code)