Merge remote-tracking branch 'origin/velixio' into velixio

This reverts commit 2926ce615a.

CHANGELOG.md

@@ -241,7 +241,7 @@ the frozen-backend fallback mirror it for their toolchains.
 - The stdio wire protocol every engine sidecar speaks is now tested once across all nine of them, instead of against a single engine — a bug in any one sidecar's copy gets caught — thanks @paoloantinori! (#1408)
 - Windows smoke tests stopped silently passing a broken ffmpeg install, and every smoke leg is now budgeted for a cold dependency install. (#1290)
 - Test suites no longer leak config paths or model-manager shutdown state into one another, which had been failing unrelated pull requests. (#1269)
-- The nightly preview build stopped refusing to publish its own healthy updater manifest when the macOS legs finished a few minutes ahead of the slowest one — Preview-channel users were silently left without new builds. 
+- The nightly preview build stopped refusing to publish its own healthy updater manifest when the macOS legs finished a few minutes ahead of the slowest one — Preview-channel users were silently left without new builds.
 
 ## [0.4.2] — 2026-07-28
 

backend/api/routers/generation.py

@@ -1146,6 +1146,9 @@ async def generate_speech(
     # classic flow, so streaming is purely a delivery channel — engine-agnostic
     # (text-level chunking, no per-engine token streaming).
     stream: bool = Form(False),
+    # Explicit opt-in. The absence of this field preserves the local-first
+    # /generate contract even when an administrator configured hosted values.
+    hosted: bool = Form(False),
 ):
     # #502: NFC-normalize the input text so decomposed (NFD) diacritics — common
     # in pasted Vietnamese and other Latin-with-marks text — are composed to the
@@ -1156,6 +1159,36 @@ async def generate_speech(
     import unicodedata
     text = unicodedata.normalize("NFC", text)
 
+    if hosted:
+        # Hosted execution accepts only a previously, explicitly synchronized
+        # consent-verified profile. Never silently sync a local recording from
+        # a synthesis request: that would make normal offline use an upload.
+        if not profile_id:
+            raise HTTPException(status_code=422, detail="Hosted synthesis requires a synchronized voice profile.")
+        from services.hosted_voice_api import HostedSettings, HostedVoiceClient, HostedVoiceError
+        try:
+            settings = HostedSettings.from_environment()
+        except HostedVoiceError as exc:
+            raise HTTPException(status_code=503, detail=str(exc)) from exc
+        if settings is None:
+            raise HTTPException(status_code=409, detail="Hosted synthesis is not configured on this device.")
+        with db_conn() as conn:
+            profile = conn.execute("SELECT hosted_voice_id, language FROM voice_profiles WHERE id=?", (profile_id,)).fetchone()
+        if not profile:
+            raise HTTPException(status_code=404, detail="Voice profile not found")
+        if not profile["hosted_voice_id"]:
+            raise HTTPException(status_code=422, detail="Sync this consent-verified profile to hosted before hosted synthesis.")
+        client = HostedVoiceClient(settings)
+        try:
+            audio = await client.synthesize(
+                text=text, profile_voice_id=profile["hosted_voice_id"], language=language or profile["language"],
+            )
+        except HostedVoiceError as exc:
+            raise HTTPException(status_code=502, detail=str(exc)) from exc
+        finally:
+            await client.aclose()
+        return StreamingResponse(io.BytesIO(audio), media_type="audio/wav", headers={"X-OmniVoice-Execution": "hosted"})
+
     # ── Engine resolution (issue #312) ──────────────────────────────────────
     # The request runs on the engine selected in Settings (POST /engines/select,
     # env var OMNIVOICE_TTS_BACKEND wins), or an explicit per-request `engine`

backend/api/routers/profiles.py

@@ -14,6 +14,7 @@ from core import event_bus
 from core.personalities import get_personalities
 from omnivoice.utils.voice_design import heal_design_instruct, sanitize_instruct
 from core.path_security import UnsafePath, resolve_within
+from services.hosted_voice_api import HostedSettings, HostedVoiceClient, HostedVoiceError
 
 router = APIRouter()
 
@@ -184,6 +185,50 @@ def get_profile(profile_id: str):
     return dict(row)
 
 
+@router.post("/profiles/{profile_id}/hosted-sync")
+async def sync_profile_to_hosted(profile_id: str):
+    """Explicitly copy a consent-verified local clone to the hosted library.
+
+    This is deliberately not part of local profile creation: merely creating a
+    profile must never upload biometric source audio. The hosted service records
+    the existing spoken-consent evidence as its versioned attestation; it does
+    not receive the consent recording itself.
+    """
+    try:
+        settings = HostedSettings.from_environment()
+    except HostedVoiceError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    if settings is None:
+        raise HTTPException(status_code=409, detail="Hosted voice sync is not configured on this device.")
+    with db_conn() as conn:
+        row = conn.execute(
+            "SELECT id, name, description, ref_text, ref_audio_path, verified_own_voice, consent_text, hosted_voice_id "
+            "FROM voice_profiles WHERE id=?", (profile_id,)
+        ).fetchone()
+    if not row:
+        raise HTTPException(status_code=404, detail="Profile not found")
+    if row["hosted_voice_id"]:
+        return {"profile_id": profile_id, "hosted_voice_id": row["hosted_voice_id"], "state": "already_synced"}
+    if not row["verified_own_voice"] or not row["consent_text"].strip():
+        raise HTTPException(status_code=422, detail="Record the voice-ownership consent statement before hosted sync.")
+    reference_path = _voices_path(row["ref_audio_path"] or "")
+    if not reference_path or not os.path.isfile(reference_path):
+        raise HTTPException(status_code=422, detail="This profile has no local reference recording to sync.")
+    client = HostedVoiceClient(settings)
+    try:
+        hosted_voice_id = await client.create_voice(
+            name=row["name"], description=row["description"] or row["ref_text"] or "", reference_path=reference_path,
+        )
+    except HostedVoiceError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    finally:
+        await client.aclose()
+    with db_conn() as conn:
+        conn.execute("UPDATE voice_profiles SET hosted_voice_id=? WHERE id=? AND hosted_voice_id=''", (hosted_voice_id, profile_id))
+        persisted = conn.execute("SELECT hosted_voice_id FROM voice_profiles WHERE id=?", (profile_id,)).fetchone()["hosted_voice_id"]
+    return {"profile_id": profile_id, "hosted_voice_id": persisted, "state": "synced"}
+
+
 @router.put("/profiles/{profile_id}")
 def update_profile(profile_id: str, patch: ProfileUpdate):
     """Partial update — only fields set on the payload are changed."""

backend/core/db.py

@@ -57,6 +57,9 @@ _BASE_SCHEMA = """
         consent_recorded_at REAL DEFAULT NULL,
         kind TEXT DEFAULT 'clone',
         vd_states TEXT DEFAULT NULL,
+        -- Hosted Voice ID is opt-in synchronization metadata. Local synthesis
+        -- never depends on it, so existing offline profiles remain useful.
+        hosted_voice_id TEXT DEFAULT '',
         created_at REAL
     );
     CREATE TABLE IF NOT EXISTS generation_history (

backend/migrations/versions/0011_hosted_voice_sync.py

@@ -0,0 +1,30 @@
+"""Opt-in hosted Voice ID on local profiles.
+
+Revision ID: 0011_hosted_voice_sync
+Revises: 0010_remote_worker_schema
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "0011_hosted_voice_sync"
+down_revision: Union[str, None] = "0010_remote_worker_schema"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _has_column(table: str, column: str) -> bool:
+    rows = op.get_bind().execute(sa.text(f"PRAGMA table_info({table})")).fetchall()
+    return any(row[1] == column for row in rows)
+
+
+def upgrade() -> None:
+    if not _has_column("voice_profiles", "hosted_voice_id"):
+        op.add_column("voice_profiles", sa.Column("hosted_voice_id", sa.Text(), nullable=True, server_default=""))
+
+
+def downgrade() -> None:
+    if _has_column("voice_profiles", "hosted_voice_id"):
+        op.drop_column("voice_profiles", "hosted_voice_id")

backend/migrations/versions/0012_mark_archetype_profiles_design.py

@@ -0,0 +1,54 @@
+"""Mark materialized gallery archetypes as voice-design profiles.
+
+Revision ID: 0012_mark_archetype_profiles_design
+Revises: 0011_hosted_voice_sync
+Create Date: 2026-08-15 00:00:00.000000
+
+``POST /archetypes/{id}/use`` stores the archetype id in ``personality`` and
+also stores a locally rendered identity WAV.  That WAV must not make the
+profile a clone: the archetype's instruct recipe is authoritative.  Older
+rows relied on the ``kind='clone'`` default and therefore selected the clone
+generation path.  This data-only migration fixes every row whose personality
+is a current archetype id, leaving unrelated persona and marketplace imports
+untouched.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision: str = "0012_mark_archetype_profiles_design"
+down_revision: Union[str, None] = "0011_hosted_voice_sync"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = inspect(bind)
+    if "voice_profiles" not in inspector.get_table_names():
+        return
+    columns = {column["name"] for column in inspector.get_columns("voice_profiles")}
+    if not {"kind", "personality"}.issubset(columns):
+        return
+
+    # The catalog is intentionally a value object, so checking an id against
+    # its current generated list is the precise provenance test.  The
+    # parameterized update avoids treating any other personality string as an
+    # archetype.
+    from core import archetypes
+
+    archetype_ids = [item["id"] for item in archetypes.list_archetypes()]
+    for archetype_id in archetype_ids:
+        bind.exec_driver_sql(
+            "UPDATE voice_profiles SET kind = 'design' "
+            "WHERE personality = ? AND (kind IS NULL OR kind = '' OR kind = 'clone')",
+            (archetype_id,),
+        )
+
+
+def downgrade() -> None:
+    # Do not silently convert voice-design profiles back to clones: that would
+    # reintroduce the generation mismatch for existing user data.
+    pass

backend/runtime_adapter/README.md

@@ -0,0 +1,95 @@
+# VoiceStudio runtime adapter
+
+A local gRPC server implementing the vssaas GPU-node runtime contract
+`voicestudio.runtime.v1.RuntimeAdapterService`, so a vssaas GPU Gateway can
+drive this VoiceStudio backend as its inference runtime.
+
+## Boundary (deliberate non-capabilities)
+
+- Binds **only** a Unix-domain socket (default `/run/voicestudio/runtime.sock`,
+  override with `VOICE_STUDIO_RUNTIME_SOCKET`). No HTTP listener, no TCP.
+- Never reaches PostgreSQL, customer credentials, or arbitrary network URLs.
+  `Execute` accepts **local file handles only** — absolute paths generated by
+  the Gateway; any URL-shaped or relative handle is rejected as invalid input.
+- The Gateway owns leases, artifact transfer, retries, and billing. This
+  adapter owns approved model loading and inference only.
+
+## Running
+
+```sh
+# serve (production socket):
+VOICE_STUDIO_RUNTIME_SOCKET=/run/voicestudio/runtime.sock \
+  python -m backend.runtime_adapter
+
+# self-check: starts the server on a private temp socket and validates the
+# same expectations the Go preflight (cmd/runtime-adapter-preflight) enforces:
+python -m backend.runtime_adapter --selfcheck
+```
+
+Environment:
+
+| Variable | Default | Meaning |
+| --- | --- | --- |
+| `VOICE_STUDIO_RUNTIME_SOCKET` | `/run/voicestudio/runtime.sock` | Unix socket path (must be absolute; parent dir must exist and not be world-writable). |
+| `VOICE_STUDIO_RUNTIME_SLOTS` | `1` | Concurrent execution slots per device. |
+
+## Wire contract and generated stubs
+
+`runtime_adapter.proto` is a **byte-identical vendored copy** of the vssaas
+contract `api/proto/voicestudio/runtime/v1/runtime_adapter.proto`. Do not edit
+it here; re-vendor from vssaas when the contract changes, then regenerate.
+
+The `gen/` stubs are committed (same policy as `backend/worker/protocol/gen/`).
+Regenerate with:
+
+```sh
+uv run python scripts/gen_runtime_adapter_protocol.py
+```
+
+`tests/test_runtime_adapter_gen.py` fails if the committed stubs drift from
+the proto.
+
+## Preflight expectations honoured
+
+The Go preflight (`internal/gateway/preflight.go`) fails closed unless:
+
+- the socket path is absolute, a real Unix socket (not a symlink), and its
+  parent directory is not world-writable — `server.prepare_socket` enforces
+  the same rules at bind time;
+- `Health` returns `SERVING_STATE_READY` with nonempty runtime + adapter
+  versions, and `GetCapabilities` returns **identical** versions — both
+  handlers read the same constants, so they cannot disagree;
+- at least one device with nonempty id/hardware class, nonzero VRAM and
+  slots, `free_slots <= total_slots`, unique ids;
+- at least one model **explicitly READY** with `catalog_model_id`,
+  `model_version`, `model_digest`, and ≥1 precision. A loading, installed,
+  or failed model is reported with its true state and never as READY.
+
+## Model identity
+
+- `catalog_model_id` — the VoiceStudio TTS engine id (`omnivoice`,
+  `voxcpm2`, …) from `services.tts_backend`'s registry.
+- `model_version` — an immutable catalog version comprising the installed
+  Hugging Face revision (40-char commit SHA) and the first 16 hex characters
+  of the attested snapshot digest. This creates a new catalog identity when
+  snapshot bytes change; it never rewrites an identity retained by a Job.
+- `model_digest` — `sha256:<hex>` computed over the installed snapshot files
+  (sorted relative path + per-file SHA-256), cached next to the repo cache
+  keyed by (revision, file list, sizes, mtimes) so multi-GB weights are
+  hashed once. See `digest.py`.
+
+## Failure taxonomy
+
+Stable codes (prefix `RTA_`) map onto the proto's `RuntimeFailureClass`:
+invalid input (`RTA_INPUT_*`), model load (`RTA_MODEL_LOAD_FAILED`),
+inference (`RTA_INFERENCE_*`), GPU resource (`RTA_GPU_*`), local storage
+(`RTA_STORAGE_*`), cancellation (terminal `ExecutionCanceled`), and adapter
+crash (`RTA_RUNTIME_CRASH`). See `codes.py`.
+
+## Tests
+
+```sh
+uv run pytest backend/tests/test_runtime_adapter_capabilities.py \
+              backend/tests/test_runtime_adapter_execute.py \
+              tests/test_runtime_adapter_gen.py
+```

backend/runtime_adapter/__init__.py

@@ -0,0 +1,18 @@
+"""VoiceStudio runtime adapter — the vssaas GPU-node runtime boundary.
+
+Implements ``voicestudio.runtime.v1.RuntimeAdapterService`` over a private
+Unix-domain socket so a vssaas GPU Gateway can drive VoiceStudio's TTS
+engines as its inference runtime. No HTTP listener, no database access, no
+outbound network: the adapter reads and writes only the local file handles
+each ``Execute`` request carries. See ``README.md`` in this directory.
+"""
+from __future__ import annotations
+
+#: Version of this adapter layer (the gRPC boundary), independent of the app
+#: version, which is reported as ``runtime_version``. Bump on any behavioral
+#: change to the adapter itself.
+ADAPTER_VERSION = "0.1.0"
+
+DEFAULT_SOCKET_PATH = "/run/voicestudio/runtime.sock"
+SOCKET_ENV = "VOICE_STUDIO_RUNTIME_SOCKET"
+SLOTS_ENV = "VOICE_STUDIO_RUNTIME_SLOTS"

backend/runtime_adapter/__main__.py

@@ -0,0 +1,65 @@
+"""Entry point: ``python -m backend.runtime_adapter``.
+
+Serves the runtime adapter on a private Unix-domain socket (default
+``/run/voicestudio/runtime.sock``, override ``VOICE_STUDIO_RUNTIME_SOCKET``
+or ``--socket``). ``--selfcheck`` instead starts the server on a temp socket
+and validates the GPU Gateway preflight expectations against it.
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+
+from ._paths import ensure_backend_on_path
+
+
+def main(argv: list[str] | None = None) -> int:
+    ensure_backend_on_path()
+    parser = argparse.ArgumentParser(
+        prog="backend.runtime_adapter",
+        description="VoiceStudio runtime adapter (vssaas GPU-node gRPC server)",
+    )
+    parser.add_argument(
+        "--socket",
+        default=None,
+        help="absolute Unix socket path (default: $VOICE_STUDIO_RUNTIME_SOCKET "
+        "or /run/voicestudio/runtime.sock)",
+    )
+    parser.add_argument(
+        "--selfcheck",
+        action="store_true",
+        help="start on a temp socket and validate the preflight expectations",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=10.0,
+        help="selfcheck RPC timeout in seconds (default: 10)",
+    )
+    parser.add_argument(
+        "--no-prewarm",
+        action="store_true",
+        help="serve immediately without loading models first (the first "
+        "execution then pays weight loading and compilation)",
+    )
+    args = parser.parse_args(argv)
+
+    if args.selfcheck:
+        from .selfcheck import selfcheck  # noqa: PLC0415
+
+        return selfcheck(timeout_s=args.timeout)
+
+    from .production import build_runtime_context, prewarm_engines  # noqa: PLC0415
+    from .server import resolve_socket_path, serve  # noqa: PLC0415
+
+    context = build_runtime_context()
+    if not args.no_prewarm:
+        # Deliberately before the socket exists: the Gateway's preflight and
+        # first offer should both find a runtime that can start inference at
+        # once, rather than one that spends an attempt lease compiling.
+        prewarm_engines(context)
+    return serve(context, resolve_socket_path(args.socket))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/runtime_adapter/_paths.py

@@ -0,0 +1,20 @@
+"""Import-path bootstrap for running outside the FastAPI app.
+
+The backend is laid out to run with ``--app-dir backend`` (imports like
+``services.tts_backend`` resolve against the ``backend/`` directory). When
+the adapter is launched as ``python -m backend.runtime_adapter`` from the
+repo root, ``backend/`` is a namespace package but not on ``sys.path`` — so
+call :func:`ensure_backend_on_path` before any ``services.*`` / ``core.*``
+import. Idempotent; mirrors ``backend/tests/conftest.py``.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+
+def ensure_backend_on_path() -> str:
+    backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if backend_dir not in sys.path:
+        sys.path.insert(0, backend_dir)
+    return backend_dir

backend/runtime_adapter/codes.py

@@ -0,0 +1,163 @@
+"""Stable failure codes and exception classification for Execute.
+
+The vssaas API Gateway keys retry and customer-charge policy off these codes,
+so they are a wire contract: never rename an existing code, only add. Every
+code maps to exactly one proto ``RuntimeFailureClass``.
+"""
+from __future__ import annotations
+
+import re
+
+from .gen import runtime_adapter_pb2 as pb2
+
+# ── invalid approved input ────────────────────────────────────────────────
+INPUT_ATTEMPT_IDENTITY = "RTA_INPUT_ATTEMPT_IDENTITY"
+INPUT_ATTEMPT_DUPLICATE = "RTA_INPUT_ATTEMPT_DUPLICATE"
+INPUT_MODEL_UNKNOWN = "RTA_INPUT_MODEL_UNKNOWN"
+INPUT_MODEL_NOT_READY = "RTA_INPUT_MODEL_NOT_READY"
+INPUT_MODEL_DIGEST_MISMATCH = "RTA_INPUT_MODEL_DIGEST_MISMATCH"
+INPUT_MODEL_PRECISION = "RTA_INPUT_MODEL_PRECISION_UNSUPPORTED"
+INPUT_DEVICE_UNKNOWN = "RTA_INPUT_DEVICE_UNKNOWN"
+INPUT_HANDLE_INVALID = "RTA_INPUT_HANDLE_INVALID"
+INPUT_ARTIFACTS_INVALID = "RTA_INPUT_ARTIFACTS_INVALID"
+INPUT_CHECKSUM_MISMATCH = "RTA_INPUT_CHECKSUM_MISMATCH"
+INPUT_TEXT_EMPTY = "RTA_INPUT_TEXT_EMPTY"
+INPUT_TEXT_TOO_LARGE = "RTA_INPUT_TEXT_TOO_LARGE"
+INPUT_TEXT_ENCODING = "RTA_INPUT_TEXT_ENCODING"
+INPUT_PARAMETER_UNKNOWN = "RTA_INPUT_PARAMETER_UNKNOWN"
+INPUT_PARAMETER_TYPE = "RTA_INPUT_PARAMETER_TYPE"
+INPUT_PARAMETER_RANGE = "RTA_INPUT_PARAMETER_RANGE"
+INPUT_DEADLINE_INVALID = "RTA_INPUT_DEADLINE_INVALID"
+INPUT_REJECTED = "RTA_INPUT_REJECTED"  # engine-level TTSInputError
+
+# ── model load / inference ────────────────────────────────────────────────
+MODEL_LOAD_FAILED = "RTA_MODEL_LOAD_FAILED"
+MODEL_LOAD_DEADLINE = "RTA_MODEL_LOAD_DEADLINE_EXCEEDED"
+INFERENCE_FAILED = "RTA_INFERENCE_FAILED"
+INFERENCE_BAD_OUTPUT = "RTA_INFERENCE_BAD_OUTPUT"
+INFERENCE_DEADLINE = "RTA_INFERENCE_DEADLINE_EXCEEDED"
+
+# ── GPU resource ──────────────────────────────────────────────────────────
+GPU_OUT_OF_MEMORY = "RTA_GPU_OUT_OF_MEMORY"
+GPU_SLOTS_EXHAUSTED = "RTA_GPU_SLOTS_EXHAUSTED"
+
+# ── local storage ─────────────────────────────────────────────────────────
+STORAGE_READ_FAILED = "RTA_STORAGE_READ_FAILED"
+STORAGE_WRITE_FAILED = "RTA_STORAGE_WRITE_FAILED"
+
+# ── adapter crash ─────────────────────────────────────────────────────────
+RUNTIME_CRASH = "RTA_RUNTIME_CRASH"
+
+_INPUT = pb2.RUNTIME_FAILURE_CLASS_INPUT
+_MODEL_LOAD = pb2.RUNTIME_FAILURE_CLASS_MODEL_LOAD
+_INFERENCE = pb2.RUNTIME_FAILURE_CLASS_INFERENCE
+_GPU = pb2.RUNTIME_FAILURE_CLASS_GPU_RESOURCE
+_STORAGE = pb2.RUNTIME_FAILURE_CLASS_LOCAL_STORAGE
+_RUNTIME = pb2.RUNTIME_FAILURE_CLASS_RUNTIME
+
+CODE_CLASS: dict[str, int] = {
+    INPUT_ATTEMPT_IDENTITY: _INPUT,
+    INPUT_ATTEMPT_DUPLICATE: _INPUT,
+    INPUT_MODEL_UNKNOWN: _INPUT,
+    INPUT_MODEL_NOT_READY: _INPUT,
+    INPUT_MODEL_DIGEST_MISMATCH: _INPUT,
+    INPUT_MODEL_PRECISION: _INPUT,
+    INPUT_DEVICE_UNKNOWN: _INPUT,
+    INPUT_HANDLE_INVALID: _INPUT,
+    INPUT_ARTIFACTS_INVALID: _INPUT,
+    INPUT_CHECKSUM_MISMATCH: _INPUT,
+    INPUT_TEXT_EMPTY: _INPUT,
+    INPUT_TEXT_TOO_LARGE: _INPUT,
+    INPUT_TEXT_ENCODING: _INPUT,
+    INPUT_PARAMETER_UNKNOWN: _INPUT,
+    INPUT_PARAMETER_TYPE: _INPUT,
+    INPUT_PARAMETER_RANGE: _INPUT,
+    INPUT_DEADLINE_INVALID: _INPUT,
+    INPUT_REJECTED: _INPUT,
+    MODEL_LOAD_FAILED: _MODEL_LOAD,
+    MODEL_LOAD_DEADLINE: _MODEL_LOAD,
+    INFERENCE_FAILED: _INFERENCE,
+    INFERENCE_BAD_OUTPUT: _INFERENCE,
+    INFERENCE_DEADLINE: _INFERENCE,
+    GPU_OUT_OF_MEMORY: _GPU,
+    GPU_SLOTS_EXHAUSTED: _GPU,
+    STORAGE_READ_FAILED: _STORAGE,
+    STORAGE_WRITE_FAILED: _STORAGE,
+    RUNTIME_CRASH: _RUNTIME,
+}
+
+
+class ExecutionFailure(Exception):
+    """A classified, wire-safe execution failure."""
+
+    def __init__(self, stable_code: str, safe_detail: str = ""):
+        if stable_code not in CODE_CLASS:  # programming error, not a wire case
+            raise ValueError(f"unknown stable code {stable_code!r}")
+        super().__init__(stable_code)
+        self.stable_code = stable_code
+        self.failure_class = CODE_CLASS[stable_code]
+        self.safe_detail = scrub_detail(safe_detail)
+
+
+_PATHISH = re.compile(r"(?:[A-Za-z]:)?[/\\][^\s'\"]+")
+_MAX_DETAIL = 240
+
+
+def scrub_detail(detail: str) -> str:
+    """Bound and de-path a detail string before it crosses the wire.
+
+    Local handles are server-generated, but engine exceptions routinely embed
+    checkpoint paths, cache dirs, and home directories. None of that belongs
+    in an event the Gateway relays upstream.
+    """
+    scrubbed = _PATHISH.sub("<path>", detail or "").strip()
+    return scrubbed[:_MAX_DETAIL]
+
+
+_OOM_MARKERS = (
+    "out of memory",
+    "cuda error: out of memory",
+    "mps backend out of memory",
+    "hip out of memory",
+    "cublas_status_alloc_failed",
+)
+
+
+def _is_oom(exc: BaseException) -> bool:
+    if type(exc).__name__ == "OutOfMemoryError":  # torch.cuda.OutOfMemoryError
+        return True
+    message = str(exc).lower()
+    return any(marker in message for marker in _OOM_MARKERS)
+
+
+def _is_engine_input_error(exc: BaseException) -> bool:
+    try:
+        from services.tts_backend import TTSInputError  # noqa: PLC0415
+    except Exception:
+        return False
+    return isinstance(exc, TTSInputError)
+
+
+def classify_engine_error(exc: BaseException, phase: str) -> ExecutionFailure:
+    """Map an engine exception to a stable failure code.
+
+    ``phase`` is ``"model_load"`` or ``"synthesis"`` — the phase the engine
+    thread was in when it raised.
+    """
+    if isinstance(exc, ExecutionFailure):
+        return exc
+    detail = f"{type(exc).__name__}: {exc}"
+    if _is_oom(exc):
+        return ExecutionFailure(GPU_OUT_OF_MEMORY, detail)
+    if _is_engine_input_error(exc):
+        return ExecutionFailure(INPUT_REJECTED, detail)
+    if isinstance(exc, OSError):
+        return ExecutionFailure(STORAGE_READ_FAILED, detail)
+    if phase == "model_load":
+        return ExecutionFailure(MODEL_LOAD_FAILED, detail)
+    return ExecutionFailure(INFERENCE_FAILED, detail)
+
+
+def deadline_failure(phase: str) -> ExecutionFailure:
+    code = MODEL_LOAD_DEADLINE if phase == "model_load" else INFERENCE_DEADLINE
+    return ExecutionFailure(code, "attempt deadline exceeded")

backend/runtime_adapter/digest.py

@@ -0,0 +1,112 @@
+"""Stable digests for locally installed model snapshots.
+
+``model_digest`` in the wire contract pins the exact bytes a READY model will
+execute with. Hugging Face snapshots are symlink farms into ``blobs/``, so the
+digest is computed over the *resolved* file contents: SHA-256 of the sorted
+sequence ``<posix relpath>\\n<file sha256>\\n``. That is stable across hosts,
+cache locations, and symlink layout, and changes whenever any weight byte or
+the file set changes.
+
+Hashing multi-GB weights on every ``GetCapabilities`` call would be absurd, so
+the result is cached in a JSON sidecar keyed by a cheap fingerprint of the
+file list (relpath, size, mtime_ns). Any file change invalidates the cache and
+forces a full re-hash.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+
+DIGEST_PREFIX = "sha256:"
+_CHUNK = 1024 * 1024
+
+
+def file_sha256(path: str | os.PathLike[str]) -> str:
+    hasher = hashlib.sha256()
+    with open(path, "rb") as fh:
+        while True:
+            chunk = fh.read(_CHUNK)
+            if not chunk:
+                break
+            hasher.update(chunk)
+    return hasher.hexdigest()
+
+
+def _manifest(root: Path) -> list[tuple[str, int, int]]:
+    """Sorted (relpath, size, mtime_ns) for every regular file under root.
+
+    Follows symlinks (HF snapshot layout); a dangling symlink raises
+    ``FileNotFoundError`` — callers treat that as an incomplete install.
+    """
+    entries: list[tuple[str, int, int]] = []
+    for current, dirs, files in os.walk(root, followlinks=True):
+        dirs.sort()
+        for name in sorted(files):
+            path = Path(current) / name
+            stat = path.stat()  # resolves symlinks; raises if dangling
+            rel = path.relative_to(root).as_posix()
+            entries.append((rel, stat.st_size, stat.st_mtime_ns))
+    entries.sort()
+    return entries
+
+
+def _fingerprint(entries: list[tuple[str, int, int]]) -> str:
+    return hashlib.sha256(
+        json.dumps(entries, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+
+
+def snapshot_digest(root: str | os.PathLike[str], cache_path: str | os.PathLike[str] | None = None) -> str:
+    """``sha256:<hex>`` digest of the snapshot at ``root``.
+
+    Raises ``FileNotFoundError`` for a missing/empty snapshot or dangling
+    symlink and ``OSError`` for unreadable files — callers classify those as
+    not-READY rather than fabricating a digest.
+    """
+    root = Path(root)
+    entries = _manifest(root)
+    if not entries:
+        raise FileNotFoundError(f"empty model snapshot: {root}")
+    fingerprint = _fingerprint(entries)
+
+    if cache_path is not None:
+        cached = _read_cache(cache_path)
+        if cached is not None and cached.get("fingerprint") == fingerprint:
+            digest = cached.get("digest", "")
+            if isinstance(digest, str) and digest.startswith(DIGEST_PREFIX):
+                return digest
+
+    hasher = hashlib.sha256()
+    for rel, _size, _mtime in entries:
+        hasher.update(rel.encode("utf-8"))
+        hasher.update(b"\n")
+        hasher.update(file_sha256(root / rel).encode("ascii"))
+        hasher.update(b"\n")
+    digest = DIGEST_PREFIX + hasher.hexdigest()
+
+    if cache_path is not None:
+        _write_cache(cache_path, fingerprint, digest)
+    return digest
+
+
+def _read_cache(cache_path: str | os.PathLike[str]) -> dict | None:
+    try:
+        with open(cache_path, encoding="utf-8") as fh:
+            data = json.load(fh)
+        return data if isinstance(data, dict) else None
+    except (OSError, ValueError):
+        return None
+
+
+def _write_cache(cache_path: str | os.PathLike[str], fingerprint: str, digest: str) -> None:
+    cache_path = Path(cache_path)
+    payload = json.dumps({"fingerprint": fingerprint, "digest": digest})
+    try:
+        cache_path.parent.mkdir(parents=True, exist_ok=True)
+        temporary = cache_path.with_suffix(f".tmp-{os.getpid()}")
+        temporary.write_text(payload, encoding="utf-8")
+        os.replace(temporary, cache_path)
+    except OSError:
+        pass  # cache is an optimization; the digest itself is already computed

backend/runtime_adapter/executor.py

@@ -0,0 +1,664 @@
+"""Execute/Cancel: attempt registry, validation, and the event stream.
+
+One ``Execute`` call is one *attempt*. The generator emits::
+
+    started → progress* → exactly one of completed | failed | canceled
+
+The engine call itself (``ensure_ready`` + ``generate``) runs on a daemon
+worker thread; the streaming generator polls it, emitting bounded heartbeat
+progress and enforcing the request deadline and cancellation. A blocking
+engine cannot be interrupted mid-kernel, so on cancel/deadline the thread is
+abandoned and its result discarded — the terminal event is what the Gateway
+acts on, and slot accounting is released only when the thread actually exits.
+
+The adapter never turns a customer string into a filesystem path: it touches
+exactly the local handles the request carries, after validation.
+"""
+from __future__ import annotations
+
+import os
+import threading
+import time
+from collections import OrderedDict
+from dataclasses import dataclass, field
+
+from . import codes
+from ._paths import ensure_backend_on_path
+from .digest import file_sha256
+from .gen import runtime_adapter_pb2 as pb2
+from .inventory import STATE_READY
+
+_MAX_TEXT_BYTES = 512_000
+_MAX_REF_AUDIO_BYTES = 100 * 1024 * 1024
+_MAX_DEADLINE_S = 24 * 3600.0
+_MAX_PROGRESS_EVENTS = 512
+
+#: Typed, bounded Execute parameters → the engine ``generate()`` kwarg of the
+#: same name. Kinds: ("string", max_len) / ("integer", lo, hi) /
+#: ("number", lo, hi) / ("boolean",).
+PARAMETER_SPECS: dict[str, tuple] = {
+    "language": ("string", 32),
+    "ref_text": ("string", 4096),
+    "instruct": ("string", 2048),
+    "description": ("string", 2048),
+    "speed": ("number", 0.25, 4.0),
+    "guidance_scale": ("number", 0.0, 16.0),
+    "num_step": ("integer", 1, 128),
+    # Gallery reference voices persist their OSS design seed.  Accept it at
+    # the hosted runtime boundary so a selected voice produces the same take.
+    "seed": ("integer", 0, 4_294_967_295),
+}
+
+
+# ── attempt registry ──────────────────────────────────────────────────────
+
+
+@dataclass
+class AttemptRecord:
+    job_id: str
+    attempt_id: str
+    cancel: threading.Event = field(default_factory=threading.Event)
+    terminal: str | None = None  # "completed" | "failed" | "canceled"
+
+
+class AttemptRegistry:
+    """Attempt bookkeeping: admission, idempotent cancel, bounded history."""
+
+    def __init__(self, max_terminal: int = 4096):
+        self._lock = threading.Lock()
+        self._active: dict[str, AttemptRecord] = {}
+        self._terminal: OrderedDict[str, AttemptRecord] = OrderedDict()
+        self._max_terminal = max_terminal
+
+    def begin(self, job_id: str, attempt_id: str, slot_limit: int) -> AttemptRecord:
+        with self._lock:
+            if attempt_id in self._active or attempt_id in self._terminal:
+                raise codes.ExecutionFailure(
+                    codes.INPUT_ATTEMPT_DUPLICATE, "attempt id already used"
+                )
+            if len(self._active) >= max(1, slot_limit):
+                raise codes.ExecutionFailure(
+                    codes.GPU_SLOTS_EXHAUSTED, "no free execution slot"
+                )
+            record = AttemptRecord(job_id=job_id, attempt_id=attempt_id)
+            self._active[attempt_id] = record
+            return record
+
+    def finish(self, attempt_id: str, terminal: str) -> None:
+        with self._lock:
+            record = self._active.pop(attempt_id, None)
+            if record is None:
+                return
+            record.terminal = terminal
+            self._terminal[attempt_id] = record
+            while len(self._terminal) > self._max_terminal:
+                self._terminal.popitem(last=False)
+
+    def active_count(self) -> int:
+        with self._lock:
+            return len(self._active)
+
+    def cancel(self, job_id: str, attempt_id: str) -> int:
+        """Idempotent by attempt id; returns a proto CancelDisposition."""
+        with self._lock:
+            record = self._active.get(attempt_id)
+            if record is not None:
+                if job_id and record.job_id and job_id != record.job_id:
+                    return pb2.CANCEL_DISPOSITION_NOT_FOUND
+                record.cancel.set()
+                return pb2.CANCEL_DISPOSITION_ACCEPTED
+            record = self._terminal.get(attempt_id)
+            if record is not None:
+                if job_id and record.job_id and job_id != record.job_id:
+                    return pb2.CANCEL_DISPOSITION_NOT_FOUND
+                return pb2.CANCEL_DISPOSITION_ALREADY_TERMINAL
+        return pb2.CANCEL_DISPOSITION_NOT_FOUND
+
+
+# ── request validation ────────────────────────────────────────────────────
+
+
+@dataclass
+class ValidatedRequest:
+    text: str
+    output_handle: str
+    output_media_type: str
+    output_size_bound: int
+    engine_kwargs: dict
+    deadline_monotonic: float
+    catalog_model_id: str
+
+
+def _validate_handle(handle: str, code: str = codes.INPUT_HANDLE_INVALID) -> str:
+    cleaned = (handle or "").strip()
+    if (
+        not cleaned
+        or "\x00" in cleaned
+        or "://" in cleaned
+        or not os.path.isabs(cleaned)
+        or os.path.normpath(cleaned) != cleaned
+    ):
+        raise codes.ExecutionFailure(code, "local handle must be an absolute path")
+    return cleaned
+
+
+def _read_input_file(artifact, max_bytes: int) -> bytes:
+    path = _validate_handle(artifact.local_handle)
+    try:
+        stat = os.lstat(path)
+    except OSError as exc:
+        raise codes.ExecutionFailure(
+            codes.STORAGE_READ_FAILED, f"input handle unreadable: {type(exc).__name__}"
+        )
+    import stat as stat_module  # noqa: PLC0415
+
+    if not stat_module.S_ISREG(stat.st_mode):
+        raise codes.ExecutionFailure(
+            codes.INPUT_HANDLE_INVALID, "input handle must be a regular file"
+        )
+    bound = max_bytes
+    if 0 < artifact.expected_size_bytes <= max_bytes:
+        bound = artifact.expected_size_bytes
+    if stat.st_size > bound:
+        raise codes.ExecutionFailure(
+            codes.INPUT_TEXT_TOO_LARGE, "input exceeds its size bound"
+        )
+    try:
+        with open(path, "rb") as fh:
+            data = fh.read(bound + 1)
+    except OSError as exc:
+        raise codes.ExecutionFailure(
+            codes.STORAGE_READ_FAILED, f"input read failed: {type(exc).__name__}"
+        )
+    if len(data) > bound:
+        raise codes.ExecutionFailure(
+            codes.INPUT_TEXT_TOO_LARGE, "input exceeds its size bound"
+        )
+    expected = (artifact.expected_sha256 or "").strip().lower().removeprefix("sha256:")
+    if expected:
+        import hashlib  # noqa: PLC0415
+
+        if hashlib.sha256(data).hexdigest() != expected:
+            raise codes.ExecutionFailure(
+                codes.INPUT_CHECKSUM_MISMATCH, "input checksum mismatch"
+            )
+    return data
+
+
+def _typed_parameter(name: str, value) -> object:
+    spec = PARAMETER_SPECS.get(name)
+    if spec is None:
+        raise codes.ExecutionFailure(
+            codes.INPUT_PARAMETER_UNKNOWN, f"unknown parameter {name!r}"
+        )
+    kind = spec[0]
+    which = value.WhichOneof("value")
+    if kind == "string":
+        if which != "string_value":
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_TYPE, f"parameter {name!r} must be a string"
+            )
+        text = value.string_value
+        if len(text) > spec[1]:
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_RANGE, f"parameter {name!r} too long"
+            )
+        return text
+    if kind == "integer":
+        if which != "integer_value":
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_TYPE, f"parameter {name!r} must be an integer"
+            )
+        number = value.integer_value
+        if not spec[1] <= number <= spec[2]:
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_RANGE, f"parameter {name!r} out of range"
+            )
+        return int(number)
+    if kind == "number":
+        if which == "number_value":
+            number = value.number_value
+        elif which == "integer_value":
+            number = float(value.integer_value)
+        else:
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_TYPE, f"parameter {name!r} must be a number"
+            )
+        if not spec[1] <= number <= spec[2]:
+            raise codes.ExecutionFailure(
+                codes.INPUT_PARAMETER_RANGE, f"parameter {name!r} out of range"
+            )
+        return float(number)
+    if which != "boolean_value":
+        raise codes.ExecutionFailure(
+            codes.INPUT_PARAMETER_TYPE, f"parameter {name!r} must be a boolean"
+        )
+    return bool(value.boolean_value)
+
+
+# ── the executor ──────────────────────────────────────────────────────────
+
+
+class Executor:
+    """Validates and runs attempts against an inventory + engine provider."""
+
+    def __init__(
+        self,
+        inventory,
+        engine_provider,
+        registry: AttemptRegistry,
+        *,
+        slot_limit: int = 1,
+        progress_interval: float = 0.5,
+        poll_interval: float = 0.02,
+        clock=time.monotonic,
+    ):
+        self._inventory = inventory
+        self._engine_provider = engine_provider
+        self._registry = registry
+        self._slot_limit = max(1, slot_limit)
+        self._progress_interval = progress_interval
+        self._poll_interval = poll_interval
+        self._clock = clock
+
+    # -- validation ----------------------------------------------------
+
+    def _validate(self, request) -> ValidatedRequest:
+        now_ms = int(time.time() * 1000)
+        if request.deadline_unix_ms <= now_ms:
+            raise codes.ExecutionFailure(
+                codes.INPUT_DEADLINE_INVALID, "deadline is not in the future"
+            )
+        budget_s = min((request.deadline_unix_ms - now_ms) / 1000.0, _MAX_DEADLINE_S)
+
+        model = self._validate_model(request.model)
+        self._validate_device(request.device_id)
+
+        text_artifact, ref_artifact = self._split_inputs(request.inputs)
+        output = self._single_output(request.outputs)
+        output_handle = _validate_handle(output.local_handle)
+        parent = os.path.dirname(output_handle)
+        if not os.path.isdir(parent):
+            raise codes.ExecutionFailure(
+                codes.INPUT_HANDLE_INVALID, "output handle directory does not exist"
+            )
+
+        raw = _read_input_file(text_artifact, _MAX_TEXT_BYTES)
+        try:
+            text = raw.decode("utf-8").strip()
+        except UnicodeDecodeError:
+            raise codes.ExecutionFailure(
+                codes.INPUT_TEXT_ENCODING, "input text is not valid UTF-8"
+            )
+        if not text:
+            raise codes.ExecutionFailure(codes.INPUT_TEXT_EMPTY, "input text is empty")
+
+        engine_kwargs: dict = {}
+        for name in sorted(request.parameters):
+            engine_kwargs[name] = _typed_parameter(name, request.parameters[name])
+        if ref_artifact is not None:
+            _read_input_file(ref_artifact, _MAX_REF_AUDIO_BYTES)  # existence/bounds/checksum
+            engine_kwargs["ref_audio"] = _validate_handle(ref_artifact.local_handle)
+
+        return ValidatedRequest(
+            text=text,
+            output_handle=output_handle,
+            output_media_type=output.media_type or "audio/wav",
+            output_size_bound=int(output.expected_size_bytes),
+            engine_kwargs=engine_kwargs,
+            deadline_monotonic=self._clock() + budget_s,
+            catalog_model_id=request.model.catalog_model_id,
+        )
+
+    def _validate_model(self, spec):
+        wanted = (spec.catalog_model_id or "").strip()
+        if not wanted:
+            raise codes.ExecutionFailure(
+                codes.INPUT_MODEL_UNKNOWN, "catalog model id is required"
+            )
+        matches = [
+            model
+            for model in self._inventory.models()
+            if model.catalog_model_id == wanted
+        ]
+        if not matches:
+            raise codes.ExecutionFailure(codes.INPUT_MODEL_UNKNOWN, "model not present")
+        model = matches[0]
+        if model.state != STATE_READY:
+            raise codes.ExecutionFailure(
+                codes.INPUT_MODEL_NOT_READY, "model is not READY"
+            )
+        if spec.model_version and spec.model_version != model.model_version:
+            raise codes.ExecutionFailure(
+                codes.INPUT_MODEL_UNKNOWN, "model version mismatch"
+            )
+        if not spec.model_digest or spec.model_digest != model.model_digest:
+            raise codes.ExecutionFailure(
+                codes.INPUT_MODEL_DIGEST_MISMATCH, "approved model digest mismatch"
+            )
+        if spec.precision and spec.precision not in model.precisions:
+            raise codes.ExecutionFailure(
+                codes.INPUT_MODEL_PRECISION, "precision not offered by this model"
+            )
+        return model
+
+    def _validate_device(self, device_id: str) -> None:
+        wanted = (device_id or "").strip()
+        if not wanted:
+            raise codes.ExecutionFailure(
+                codes.INPUT_DEVICE_UNKNOWN, "device id is required"
+            )
+        known = {device.device_id for device in self._inventory.devices()}
+        if wanted not in known:
+            raise codes.ExecutionFailure(
+                codes.INPUT_DEVICE_UNKNOWN, "device id not in inventory"
+            )
+
+    @staticmethod
+    def _split_inputs(inputs):
+        text_artifacts, audio_artifacts = [], []
+        for artifact in inputs:
+            if artifact.operation != pb2.LOCAL_ARTIFACT_OPERATION_READ:
+                raise codes.ExecutionFailure(
+                    codes.INPUT_ARTIFACTS_INVALID, "inputs must be READ artifacts"
+                )
+            media = artifact.media_type or ""
+            if media.startswith("audio/"):
+                audio_artifacts.append(artifact)
+            elif media == "" or media.startswith("text/"):
+                text_artifacts.append(artifact)
+            else:
+                raise codes.ExecutionFailure(
+                    codes.INPUT_ARTIFACTS_INVALID, f"unsupported input media {media!r}"
+                )
+        if len(text_artifacts) != 1 or len(audio_artifacts) > 1:
+            raise codes.ExecutionFailure(
+                codes.INPUT_ARTIFACTS_INVALID,
+                "tts needs exactly one text input and at most one reference audio",
+            )
+        return text_artifacts[0], (audio_artifacts[0] if audio_artifacts else None)
+
+    @staticmethod
+    def _single_output(outputs):
+        if len(outputs) != 1:
+            raise codes.ExecutionFailure(
+                codes.INPUT_ARTIFACTS_INVALID, "tts needs exactly one output artifact"
+            )
+        output = outputs[0]
+        if output.operation != pb2.LOCAL_ARTIFACT_OPERATION_WRITE:
+            raise codes.ExecutionFailure(
+                codes.INPUT_ARTIFACTS_INVALID, "output must be a WRITE artifact"
+            )
+        media = output.media_type or ""
+        if media and not media.startswith("audio/"):
+            raise codes.ExecutionFailure(
+                codes.INPUT_ARTIFACTS_INVALID, f"unsupported output media {media!r}"
+            )
+        return output
+
+    # -- execution -----------------------------------------------------
+
+    def execute(self, request, grpc_context=None):
+        """Generator of ``pb2.ExecuteResponse``. Never raises for a
+        classified failure — failures become terminal events."""
+        session = _Session(self, request)
+        return session.run(grpc_context)
+
+
+class _Session:
+    def __init__(self, executor: Executor, request):
+        self._x = executor
+        self.request = request
+        self.job_id = request.job_id
+        self.attempt_id = request.attempt_id
+        self.sequence = 0
+        self.phase = "model_load"
+        self.terminal_sent = False
+        self.chars = 0
+        self.gpu_ms = 0
+        self.cpu_ms = 0
+        self.output_audio_ms = 0
+
+    # event builders ---------------------------------------------------
+
+    def _event(self, **payload):
+        self.sequence += 1
+        return pb2.ExecuteResponse(
+            event=pb2.ExecutionEvent(
+                job_id=self.job_id,
+                attempt_id=self.attempt_id,
+                sequence=self.sequence,
+                observed_at_unix_ms=int(time.time() * 1000),
+                **payload,
+            )
+        )
+
+    def _measurements(self):
+        return pb2.RuntimeMeasurements(
+            normalized_input_characters=self.chars,
+            output_audio_ms=self.output_audio_ms,
+            gpu_execution_ms=self.gpu_ms,
+            cpu_execution_ms=self.cpu_ms,
+        )
+
+    def _failed(self, failure: codes.ExecutionFailure):
+        self.terminal_sent = True
+        return self._event(
+            failed=pb2.ExecutionFailed(
+                failure_class=failure.failure_class,
+                stable_code=failure.stable_code,
+                safe_detail=failure.safe_detail,
+                measurements=self._measurements(),
+            )
+        )
+
+    def _canceled(self):
+        self.terminal_sent = True
+        return self._event(
+            canceled=pb2.ExecutionCanceled(measurements=self._measurements())
+        )
+
+    # main flow --------------------------------------------------------
+
+    def run(self, grpc_context):
+        if not self.attempt_id.strip() or not self.job_id.strip():
+            yield self._failed(
+                codes.ExecutionFailure(
+                    codes.INPUT_ATTEMPT_IDENTITY, "job and attempt ids are required"
+                )
+            )
+            return
+        registry = self._x._registry
+        try:
+            record = registry.begin(self.job_id, self.attempt_id, self._x._slot_limit)
+        except codes.ExecutionFailure as failure:
+            yield self._failed(failure)
+            return
+        try:
+            yield from self._run_admitted(record, grpc_context)
+        finally:
+            terminal = "canceled"
+            if self.terminal_sent:
+                terminal = self._terminal_kind or "failed"
+            registry.finish(self.attempt_id, terminal)
+
+    _terminal_kind: str | None = None
+
+    def _run_admitted(self, record, grpc_context):
+        try:
+            validated = self._x._validate(self.request)
+        except codes.ExecutionFailure as failure:
+            self._terminal_kind = "failed"
+            yield self._failed(failure)
+            return
+        except Exception as exc:  # adapter bug — still a classified event
+            self._terminal_kind = "failed"
+            yield self._failed(
+                codes.ExecutionFailure(codes.RUNTIME_CRASH, f"{type(exc).__name__}")
+            )
+            return
+
+        self.chars = len(validated.text)
+        yield self._event(started=pb2.ExecutionStarted())
+
+        worker = _EngineWorker(self._x._engine_provider, validated, self)
+        worker.start()
+
+        clock = self._x._clock
+        next_progress = clock() + self._x._progress_interval
+        progress_events = 0
+        while not worker.done.wait(self._x._poll_interval):
+            if record.cancel.is_set() or (
+                grpc_context is not None and not grpc_context.is_active()
+            ):
+                self._terminal_kind = "canceled"
+                yield self._canceled()
+                return
+            now = clock()
+            if now >= validated.deadline_monotonic:
+                self._terminal_kind = "failed"
+                yield self._failed(codes.deadline_failure(self.phase))
+                return
+            if now >= next_progress and progress_events < _MAX_PROGRESS_EVENTS:
+                progress_events += 1
+                next_progress = now + self._x._progress_interval
+                permille = 100 if self.phase == "model_load" else 550
+                yield self._event(
+                    progress=pb2.ExecutionProgress(
+                        progress_permille=permille, stage_code=self.phase
+                    )
+                )
+
+        if record.cancel.is_set():
+            self._terminal_kind = "canceled"
+            yield self._canceled()
+            return
+        if worker.error is not None:
+            self._terminal_kind = "failed"
+            yield self._failed(codes.classify_engine_error(worker.error, worker.phase))
+            return
+
+        try:
+            manifest = self._write_output(worker, validated)
+        except codes.ExecutionFailure as failure:
+            self._terminal_kind = "failed"
+            yield self._failed(failure)
+            return
+        self._terminal_kind = "completed"
+        self.terminal_sent = True
+        yield self._event(
+            completed=pb2.ExecutionCompleted(
+                outputs=[manifest], measurements=self._measurements()
+            )
+        )
+
+    def _write_output(self, worker, validated: ValidatedRequest):
+        ensure_backend_on_path()
+        tensor = worker.result
+        sample_rate = worker.sample_rate
+        if tensor is None or not hasattr(tensor, "numel") or tensor.numel() == 0:
+            raise codes.ExecutionFailure(
+                codes.INFERENCE_BAD_OUTPUT, "engine returned no audio"
+            )
+        if not isinstance(sample_rate, int) or sample_rate <= 0:
+            raise codes.ExecutionFailure(
+                codes.INFERENCE_BAD_OUTPUT, "engine reported no sample rate"
+            )
+        try:
+            from services.audio_io import atomic_save_wav  # noqa: PLC0415
+
+            atomic_save_wav(validated.output_handle, tensor.detach().cpu(), sample_rate)
+        except codes.ExecutionFailure:
+            raise
+        except Exception as exc:
+            raise codes.ExecutionFailure(
+                codes.STORAGE_WRITE_FAILED, f"{type(exc).__name__}: {exc}"
+            )
+        try:
+            size = os.stat(validated.output_handle).st_size
+            sha = file_sha256(validated.output_handle)
+        except OSError as exc:
+            raise codes.ExecutionFailure(
+                codes.STORAGE_WRITE_FAILED, f"{type(exc).__name__}"
+            )
+        if 0 < validated.output_size_bound < size:
+            raise codes.ExecutionFailure(
+                codes.STORAGE_WRITE_FAILED, "output exceeds its size bound"
+            )
+        samples = tensor.numel() if tensor.dim() == 1 else tensor.shape[-1]
+        self.output_audio_ms = int(samples * 1000 / sample_rate)
+        return pb2.LocalArtifactManifest(
+            artifact_id=self.request.outputs[0].artifact_id,
+            local_handle=validated.output_handle,
+            size_bytes=size,
+            sha256=sha,
+            media_type=validated.output_media_type,
+            duration_ms=self.output_audio_ms,
+        )
+
+
+class _EngineWorker:
+    """Runs the engine on a daemon thread, recording phase and timings."""
+
+    def __init__(self, engine_provider, validated: ValidatedRequest, session: _Session):
+        self._engine_provider = engine_provider
+        self._validated = validated
+        self._session = session
+        self.done = threading.Event()
+        self.error: BaseException | None = None
+        self.result = None
+        self.sample_rate: int | None = None
+        self.phase = "model_load"
+
+    def start(self) -> None:
+        thread = threading.Thread(
+            target=self._run,
+            name=f"runtime-adapter-attempt-{self._session.attempt_id}",
+            daemon=True,
+        )
+        thread.start()
+
+    @staticmethod
+    def _synthesize(engine, text: str, params: dict):
+        """Use the same seeded native path as OSS Gallery and ovnode workers."""
+        from services import tts_backend  # noqa: PLC0415
+
+        if isinstance(engine, tts_backend.OmniVoiceBackend):
+            from api.routers.generation import _run_inference  # noqa: PLC0415
+
+            with tts_backend.engine_in_use(engine):
+                return _run_inference(
+                    engine._model, text, params.get("language"),
+                    params.get("ref_audio"), params.get("ref_text"),
+                    params.get("instruct"), params.get("duration"),
+                    params.get("num_step", 16), params.get("guidance_scale", 2.0),
+                    params.get("speed", 1.0), params.get("t_shift"),
+                    params.get("denoise", True), params.get("postprocess_output", True),
+                    params.get("layer_penalty_factor"),
+                    params.get("position_temperature"),
+                    params.get("class_temperature"), params.get("seed"),
+                )
+        return engine.generate(text, **params)
+
+    def _run(self) -> None:
+        wall_start = time.monotonic()
+        cpu_start = time.process_time()
+        try:
+            engine = self._engine_provider(self._validated.catalog_model_id)
+            ensure_ready = getattr(engine, "ensure_ready", None)
+            if callable(ensure_ready):
+                ensure_ready()
+            self.phase = "synthesis"
+            self._session.phase = "synthesis"
+            synth_start = time.monotonic()
+            self.result = self._synthesize(engine, self._validated.text, self._validated.engine_kwargs)
+            rate = getattr(engine, "sample_rate", None)
+            self.sample_rate = int(rate) if isinstance(rate, (int, float)) and rate else None
+            self._session.gpu_ms = int((time.monotonic() - synth_start) * 1000)
+        except BaseException as exc:  # classified later, never lost
+            self.error = exc
+        finally:
+            self._session.cpu_ms = int((time.process_time() - cpu_start) * 1000)
+            if self._session.gpu_ms == 0 and self.error is None:
+                self._session.gpu_ms = int((time.monotonic() - wall_start) * 1000)
+            self.done.set()

backend/runtime_adapter/gen/__init__.py

@@ -0,0 +1,5 @@
+"""Generated protocol stubs — DO NOT EDIT.
+
+Regenerate with ``uv run python scripts/gen_runtime_adapter_protocol.py``
+after any change to ``../runtime_adapter.proto``.
+"""

backend/runtime_adapter/gen/runtime_adapter_pb2.pyi

@@ -0,0 +1,330 @@
+from google.protobuf.internal import containers as _containers
+from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from collections.abc import Iterable as _Iterable, Mapping as _Mapping
+from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class ServingState(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    SERVING_STATE_UNSPECIFIED: _ClassVar[ServingState]
+    SERVING_STATE_READY: _ClassVar[ServingState]
+    SERVING_STATE_DEGRADED: _ClassVar[ServingState]
+    SERVING_STATE_UNHEALTHY: _ClassVar[ServingState]
+
+class RuntimeModelState(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    RUNTIME_MODEL_STATE_UNSPECIFIED: _ClassVar[RuntimeModelState]
+    RUNTIME_MODEL_STATE_INSTALLED: _ClassVar[RuntimeModelState]
+    RUNTIME_MODEL_STATE_LOADING: _ClassVar[RuntimeModelState]
+    RUNTIME_MODEL_STATE_READY: _ClassVar[RuntimeModelState]
+    RUNTIME_MODEL_STATE_FAILED: _ClassVar[RuntimeModelState]
+
+class LocalArtifactOperation(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    LOCAL_ARTIFACT_OPERATION_UNSPECIFIED: _ClassVar[LocalArtifactOperation]
+    LOCAL_ARTIFACT_OPERATION_READ: _ClassVar[LocalArtifactOperation]
+    LOCAL_ARTIFACT_OPERATION_WRITE: _ClassVar[LocalArtifactOperation]
+
+class RuntimeFailureClass(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    RUNTIME_FAILURE_CLASS_UNSPECIFIED: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_INPUT: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_MODEL_LOAD: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_INFERENCE: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_GPU_RESOURCE: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_LOCAL_STORAGE: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_RUNTIME: _ClassVar[RuntimeFailureClass]
+    RUNTIME_FAILURE_CLASS_CANCELED: _ClassVar[RuntimeFailureClass]
+
+class CancelDisposition(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    CANCEL_DISPOSITION_UNSPECIFIED: _ClassVar[CancelDisposition]
+    CANCEL_DISPOSITION_ACCEPTED: _ClassVar[CancelDisposition]
+    CANCEL_DISPOSITION_ALREADY_TERMINAL: _ClassVar[CancelDisposition]
+    CANCEL_DISPOSITION_NOT_FOUND: _ClassVar[CancelDisposition]
+SERVING_STATE_UNSPECIFIED: ServingState
+SERVING_STATE_READY: ServingState
+SERVING_STATE_DEGRADED: ServingState
+SERVING_STATE_UNHEALTHY: ServingState
+RUNTIME_MODEL_STATE_UNSPECIFIED: RuntimeModelState
+RUNTIME_MODEL_STATE_INSTALLED: RuntimeModelState
+RUNTIME_MODEL_STATE_LOADING: RuntimeModelState
+RUNTIME_MODEL_STATE_READY: RuntimeModelState
+RUNTIME_MODEL_STATE_FAILED: RuntimeModelState
+LOCAL_ARTIFACT_OPERATION_UNSPECIFIED: LocalArtifactOperation
+LOCAL_ARTIFACT_OPERATION_READ: LocalArtifactOperation
+LOCAL_ARTIFACT_OPERATION_WRITE: LocalArtifactOperation
+RUNTIME_FAILURE_CLASS_UNSPECIFIED: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_INPUT: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_MODEL_LOAD: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_INFERENCE: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_GPU_RESOURCE: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_LOCAL_STORAGE: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_RUNTIME: RuntimeFailureClass
+RUNTIME_FAILURE_CLASS_CANCELED: RuntimeFailureClass
+CANCEL_DISPOSITION_UNSPECIFIED: CancelDisposition
+CANCEL_DISPOSITION_ACCEPTED: CancelDisposition
+CANCEL_DISPOSITION_ALREADY_TERMINAL: CancelDisposition
+CANCEL_DISPOSITION_NOT_FOUND: CancelDisposition
+
+class ExecuteResponse(_message.Message):
+    __slots__ = ("event",)
+    EVENT_FIELD_NUMBER: _ClassVar[int]
+    event: ExecutionEvent
+    def __init__(self, event: _Optional[_Union[ExecutionEvent, _Mapping]] = ...) -> None: ...
+
+class HealthRequest(_message.Message):
+    __slots__ = ()
+    def __init__(self) -> None: ...
+
+class HealthResponse(_message.Message):
+    __slots__ = ("state", "runtime_version", "adapter_version", "health_flags")
+    STATE_FIELD_NUMBER: _ClassVar[int]
+    RUNTIME_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ADAPTER_VERSION_FIELD_NUMBER: _ClassVar[int]
+    HEALTH_FLAGS_FIELD_NUMBER: _ClassVar[int]
+    state: ServingState
+    runtime_version: str
+    adapter_version: str
+    health_flags: _containers.RepeatedScalarFieldContainer[str]
+    def __init__(self, state: _Optional[_Union[ServingState, str]] = ..., runtime_version: _Optional[str] = ..., adapter_version: _Optional[str] = ..., health_flags: _Optional[_Iterable[str]] = ...) -> None: ...
+
+class GetCapabilitiesRequest(_message.Message):
+    __slots__ = ()
+    def __init__(self) -> None: ...
+
+class GetCapabilitiesResponse(_message.Message):
+    __slots__ = ("runtime_version", "adapter_version", "devices", "models")
+    RUNTIME_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ADAPTER_VERSION_FIELD_NUMBER: _ClassVar[int]
+    DEVICES_FIELD_NUMBER: _ClassVar[int]
+    MODELS_FIELD_NUMBER: _ClassVar[int]
+    runtime_version: str
+    adapter_version: str
+    devices: _containers.RepeatedCompositeFieldContainer[RuntimeDevice]
+    models: _containers.RepeatedCompositeFieldContainer[RuntimeModel]
+    def __init__(self, runtime_version: _Optional[str] = ..., adapter_version: _Optional[str] = ..., devices: _Optional[_Iterable[_Union[RuntimeDevice, _Mapping]]] = ..., models: _Optional[_Iterable[_Union[RuntimeModel, _Mapping]]] = ...) -> None: ...
+
+class RuntimeDevice(_message.Message):
+    __slots__ = ("device_id", "hardware_class", "total_vram_bytes", "total_slots", "free_slots")
+    DEVICE_ID_FIELD_NUMBER: _ClassVar[int]
+    HARDWARE_CLASS_FIELD_NUMBER: _ClassVar[int]
+    TOTAL_VRAM_BYTES_FIELD_NUMBER: _ClassVar[int]
+    TOTAL_SLOTS_FIELD_NUMBER: _ClassVar[int]
+    FREE_SLOTS_FIELD_NUMBER: _ClassVar[int]
+    device_id: str
+    hardware_class: str
+    total_vram_bytes: int
+    total_slots: int
+    free_slots: int
+    def __init__(self, device_id: _Optional[str] = ..., hardware_class: _Optional[str] = ..., total_vram_bytes: _Optional[int] = ..., total_slots: _Optional[int] = ..., free_slots: _Optional[int] = ...) -> None: ...
+
+class RuntimeModel(_message.Message):
+    __slots__ = ("catalog_model_id", "model_version", "model_digest", "precisions", "features", "state")
+    CATALOG_MODEL_ID_FIELD_NUMBER: _ClassVar[int]
+    MODEL_VERSION_FIELD_NUMBER: _ClassVar[int]
+    MODEL_DIGEST_FIELD_NUMBER: _ClassVar[int]
+    PRECISIONS_FIELD_NUMBER: _ClassVar[int]
+    FEATURES_FIELD_NUMBER: _ClassVar[int]
+    STATE_FIELD_NUMBER: _ClassVar[int]
+    catalog_model_id: str
+    model_version: str
+    model_digest: str
+    precisions: _containers.RepeatedScalarFieldContainer[str]
+    features: _containers.RepeatedScalarFieldContainer[str]
+    state: RuntimeModelState
+    def __init__(self, catalog_model_id: _Optional[str] = ..., model_version: _Optional[str] = ..., model_digest: _Optional[str] = ..., precisions: _Optional[_Iterable[str]] = ..., features: _Optional[_Iterable[str]] = ..., state: _Optional[_Union[RuntimeModelState, str]] = ...) -> None: ...
+
+class ExecuteRequest(_message.Message):
+    __slots__ = ("job_id", "attempt_id", "device_id", "slot_id", "model", "parameters", "inputs", "outputs", "deadline_unix_ms", "maximum_preview_bytes")
+    class ParametersEntry(_message.Message):
+        __slots__ = ("key", "value")
+        KEY_FIELD_NUMBER: _ClassVar[int]
+        VALUE_FIELD_NUMBER: _ClassVar[int]
+        key: str
+        value: ParameterValue
+        def __init__(self, key: _Optional[str] = ..., value: _Optional[_Union[ParameterValue, _Mapping]] = ...) -> None: ...
+    JOB_ID_FIELD_NUMBER: _ClassVar[int]
+    ATTEMPT_ID_FIELD_NUMBER: _ClassVar[int]
+    DEVICE_ID_FIELD_NUMBER: _ClassVar[int]
+    SLOT_ID_FIELD_NUMBER: _ClassVar[int]
+    MODEL_FIELD_NUMBER: _ClassVar[int]
+    PARAMETERS_FIELD_NUMBER: _ClassVar[int]
+    INPUTS_FIELD_NUMBER: _ClassVar[int]
+    OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    DEADLINE_UNIX_MS_FIELD_NUMBER: _ClassVar[int]
+    MAXIMUM_PREVIEW_BYTES_FIELD_NUMBER: _ClassVar[int]
+    job_id: str
+    attempt_id: str
+    device_id: str
+    slot_id: str
+    model: ModelSpec
+    parameters: _containers.MessageMap[str, ParameterValue]
+    inputs: _containers.RepeatedCompositeFieldContainer[LocalArtifact]
+    outputs: _containers.RepeatedCompositeFieldContainer[LocalArtifact]
+    deadline_unix_ms: int
+    maximum_preview_bytes: int
+    def __init__(self, job_id: _Optional[str] = ..., attempt_id: _Optional[str] = ..., device_id: _Optional[str] = ..., slot_id: _Optional[str] = ..., model: _Optional[_Union[ModelSpec, _Mapping]] = ..., parameters: _Optional[_Mapping[str, ParameterValue]] = ..., inputs: _Optional[_Iterable[_Union[LocalArtifact, _Mapping]]] = ..., outputs: _Optional[_Iterable[_Union[LocalArtifact, _Mapping]]] = ..., deadline_unix_ms: _Optional[int] = ..., maximum_preview_bytes: _Optional[int] = ...) -> None: ...
+
+class ModelSpec(_message.Message):
+    __slots__ = ("catalog_model_id", "model_version", "model_digest", "precision")
+    CATALOG_MODEL_ID_FIELD_NUMBER: _ClassVar[int]
+    MODEL_VERSION_FIELD_NUMBER: _ClassVar[int]
+    MODEL_DIGEST_FIELD_NUMBER: _ClassVar[int]
+    PRECISION_FIELD_NUMBER: _ClassVar[int]
+    catalog_model_id: str
+    model_version: str
+    model_digest: str
+    precision: str
+    def __init__(self, catalog_model_id: _Optional[str] = ..., model_version: _Optional[str] = ..., model_digest: _Optional[str] = ..., precision: _Optional[str] = ...) -> None: ...
+
+class ParameterValue(_message.Message):
+    __slots__ = ("string_value", "integer_value", "number_value", "boolean_value")
+    STRING_VALUE_FIELD_NUMBER: _ClassVar[int]
+    INTEGER_VALUE_FIELD_NUMBER: _ClassVar[int]
+    NUMBER_VALUE_FIELD_NUMBER: _ClassVar[int]
+    BOOLEAN_VALUE_FIELD_NUMBER: _ClassVar[int]
+    string_value: str
+    integer_value: int
+    number_value: float
+    boolean_value: bool
+    def __init__(self, string_value: _Optional[str] = ..., integer_value: _Optional[int] = ..., number_value: _Optional[float] = ..., boolean_value: _Optional[bool] = ...) -> None: ...
+
+class LocalArtifact(_message.Message):
+    __slots__ = ("artifact_id", "local_handle", "operation", "expected_size_bytes", "expected_sha256", "media_type")
+    ARTIFACT_ID_FIELD_NUMBER: _ClassVar[int]
+    LOCAL_HANDLE_FIELD_NUMBER: _ClassVar[int]
+    OPERATION_FIELD_NUMBER: _ClassVar[int]
+    EXPECTED_SIZE_BYTES_FIELD_NUMBER: _ClassVar[int]
+    EXPECTED_SHA256_FIELD_NUMBER: _ClassVar[int]
+    MEDIA_TYPE_FIELD_NUMBER: _ClassVar[int]
+    artifact_id: str
+    local_handle: str
+    operation: LocalArtifactOperation
+    expected_size_bytes: int
+    expected_sha256: str
+    media_type: str
+    def __init__(self, artifact_id: _Optional[str] = ..., local_handle: _Optional[str] = ..., operation: _Optional[_Union[LocalArtifactOperation, str]] = ..., expected_size_bytes: _Optional[int] = ..., expected_sha256: _Optional[str] = ..., media_type: _Optional[str] = ...) -> None: ...
+
+class ExecutionEvent(_message.Message):
+    __slots__ = ("job_id", "attempt_id", "sequence", "observed_at_unix_ms", "started", "progress", "preview", "completed", "failed", "canceled")
+    JOB_ID_FIELD_NUMBER: _ClassVar[int]
+    ATTEMPT_ID_FIELD_NUMBER: _ClassVar[int]
+    SEQUENCE_FIELD_NUMBER: _ClassVar[int]
+    OBSERVED_AT_UNIX_MS_FIELD_NUMBER: _ClassVar[int]
+    STARTED_FIELD_NUMBER: _ClassVar[int]
+    PROGRESS_FIELD_NUMBER: _ClassVar[int]
+    PREVIEW_FIELD_NUMBER: _ClassVar[int]
+    COMPLETED_FIELD_NUMBER: _ClassVar[int]
+    FAILED_FIELD_NUMBER: _ClassVar[int]
+    CANCELED_FIELD_NUMBER: _ClassVar[int]
+    job_id: str
+    attempt_id: str
+    sequence: int
+    observed_at_unix_ms: int
+    started: ExecutionStarted
+    progress: ExecutionProgress
+    preview: PreviewChunk
+    completed: ExecutionCompleted
+    failed: ExecutionFailed
+    canceled: ExecutionCanceled
+    def __init__(self, job_id: _Optional[str] = ..., attempt_id: _Optional[str] = ..., sequence: _Optional[int] = ..., observed_at_unix_ms: _Optional[int] = ..., started: _Optional[_Union[ExecutionStarted, _Mapping]] = ..., progress: _Optional[_Union[ExecutionProgress, _Mapping]] = ..., preview: _Optional[_Union[PreviewChunk, _Mapping]] = ..., completed: _Optional[_Union[ExecutionCompleted, _Mapping]] = ..., failed: _Optional[_Union[ExecutionFailed, _Mapping]] = ..., canceled: _Optional[_Union[ExecutionCanceled, _Mapping]] = ...) -> None: ...
+
+class ExecutionStarted(_message.Message):
+    __slots__ = ()
+    def __init__(self) -> None: ...
+
+class ExecutionProgress(_message.Message):
+    __slots__ = ("progress_permille", "stage_code")
+    PROGRESS_PERMILLE_FIELD_NUMBER: _ClassVar[int]
+    STAGE_CODE_FIELD_NUMBER: _ClassVar[int]
+    progress_permille: int
+    stage_code: str
+    def __init__(self, progress_permille: _Optional[int] = ..., stage_code: _Optional[str] = ...) -> None: ...
+
+class PreviewChunk(_message.Message):
+    __slots__ = ("sequence", "media_type", "data")
+    SEQUENCE_FIELD_NUMBER: _ClassVar[int]
+    MEDIA_TYPE_FIELD_NUMBER: _ClassVar[int]
+    DATA_FIELD_NUMBER: _ClassVar[int]
+    sequence: int
+    media_type: str
+    data: bytes
+    def __init__(self, sequence: _Optional[int] = ..., media_type: _Optional[str] = ..., data: _Optional[bytes] = ...) -> None: ...
+
+class ExecutionCompleted(_message.Message):
+    __slots__ = ("outputs", "measurements")
+    OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    MEASUREMENTS_FIELD_NUMBER: _ClassVar[int]
+    outputs: _containers.RepeatedCompositeFieldContainer[LocalArtifactManifest]
+    measurements: RuntimeMeasurements
+    def __init__(self, outputs: _Optional[_Iterable[_Union[LocalArtifactManifest, _Mapping]]] = ..., measurements: _Optional[_Union[RuntimeMeasurements, _Mapping]] = ...) -> None: ...
+
+class LocalArtifactManifest(_message.Message):
+    __slots__ = ("artifact_id", "local_handle", "size_bytes", "sha256", "media_type", "duration_ms")
+    ARTIFACT_ID_FIELD_NUMBER: _ClassVar[int]
+    LOCAL_HANDLE_FIELD_NUMBER: _ClassVar[int]
+    SIZE_BYTES_FIELD_NUMBER: _ClassVar[int]
+    SHA256_FIELD_NUMBER: _ClassVar[int]
+    MEDIA_TYPE_FIELD_NUMBER: _ClassVar[int]
+    DURATION_MS_FIELD_NUMBER: _ClassVar[int]
+    artifact_id: str
+    local_handle: str
+    size_bytes: int
+    sha256: str
+    media_type: str
+    duration_ms: int
+    def __init__(self, artifact_id: _Optional[str] = ..., local_handle: _Optional[str] = ..., size_bytes: _Optional[int] = ..., sha256: _Optional[str] = ..., media_type: _Optional[str] = ..., duration_ms: _Optional[int] = ...) -> None: ...
+
+class ExecutionFailed(_message.Message):
+    __slots__ = ("failure_class", "stable_code", "safe_detail", "measurements")
+    FAILURE_CLASS_FIELD_NUMBER: _ClassVar[int]
+    STABLE_CODE_FIELD_NUMBER: _ClassVar[int]
+    SAFE_DETAIL_FIELD_NUMBER: _ClassVar[int]
+    MEASUREMENTS_FIELD_NUMBER: _ClassVar[int]
+    failure_class: RuntimeFailureClass
+    stable_code: str
+    safe_detail: str
+    measurements: RuntimeMeasurements
+    def __init__(self, failure_class: _Optional[_Union[RuntimeFailureClass, str]] = ..., stable_code: _Optional[str] = ..., safe_detail: _Optional[str] = ..., measurements: _Optional[_Union[RuntimeMeasurements, _Mapping]] = ...) -> None: ...
+
+class ExecutionCanceled(_message.Message):
+    __slots__ = ("measurements",)
+    MEASUREMENTS_FIELD_NUMBER: _ClassVar[int]
+    measurements: RuntimeMeasurements
+    def __init__(self, measurements: _Optional[_Union[RuntimeMeasurements, _Mapping]] = ...) -> None: ...
+
+class RuntimeMeasurements(_message.Message):
+    __slots__ = ("normalized_input_characters", "input_audio_ms", "output_audio_ms", "gpu_execution_ms", "cpu_execution_ms")
+    NORMALIZED_INPUT_CHARACTERS_FIELD_NUMBER: _ClassVar[int]
+    INPUT_AUDIO_MS_FIELD_NUMBER: _ClassVar[int]
+    OUTPUT_AUDIO_MS_FIELD_NUMBER: _ClassVar[int]
+    GPU_EXECUTION_MS_FIELD_NUMBER: _ClassVar[int]
+    CPU_EXECUTION_MS_FIELD_NUMBER: _ClassVar[int]
+    normalized_input_characters: int
+    input_audio_ms: int
+    output_audio_ms: int
+    gpu_execution_ms: int
+    cpu_execution_ms: int
+    def __init__(self, normalized_input_characters: _Optional[int] = ..., input_audio_ms: _Optional[int] = ..., output_audio_ms: _Optional[int] = ..., gpu_execution_ms: _Optional[int] = ..., cpu_execution_ms: _Optional[int] = ...) -> None: ...
+
+class CancelRequest(_message.Message):
+    __slots__ = ("job_id", "attempt_id", "reason_code", "deadline_unix_ms")
+    JOB_ID_FIELD_NUMBER: _ClassVar[int]
+    ATTEMPT_ID_FIELD_NUMBER: _ClassVar[int]
+    REASON_CODE_FIELD_NUMBER: _ClassVar[int]
+    DEADLINE_UNIX_MS_FIELD_NUMBER: _ClassVar[int]
+    job_id: str
+    attempt_id: str
+    reason_code: str
+    deadline_unix_ms: int
+    def __init__(self, job_id: _Optional[str] = ..., attempt_id: _Optional[str] = ..., reason_code: _Optional[str] = ..., deadline_unix_ms: _Optional[int] = ...) -> None: ...
+
+class CancelResponse(_message.Message):
+    __slots__ = ("disposition",)
+    DISPOSITION_FIELD_NUMBER: _ClassVar[int]
+    disposition: CancelDisposition
+    def __init__(self, disposition: _Optional[_Union[CancelDisposition, str]] = ...) -> None: ...

backend/runtime_adapter/gen/runtime_adapter_pb2_grpc.py

@@ -0,0 +1,229 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+import warnings
+
+from . import runtime_adapter_pb2 as runtime__adapter__pb2
+
+GRPC_GENERATED_VERSION = '1.81.1'
+GRPC_VERSION = grpc.__version__
+_version_not_supported = False
+
+try:
+    from grpc._utilities import first_version_is_lower
+    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
+except ImportError:
+    _version_not_supported = True
+
+if _version_not_supported:
+    raise RuntimeError(
+        f'The grpc package installed is at version {GRPC_VERSION},'
+        + ' but the generated code in runtime_adapter_pb2_grpc.py depends on'
+        + f' grpcio>={GRPC_GENERATED_VERSION}.'
+        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
+    )
+
+
+class RuntimeAdapterServiceStub:
+    """RuntimeAdapterService is local to a GPU Node and is never publicly exposed.
+    """
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.Health = channel.unary_unary(
+                '/voicestudio.runtime.v1.RuntimeAdapterService/Health',
+                request_serializer=runtime__adapter__pb2.HealthRequest.SerializeToString,
+                response_deserializer=runtime__adapter__pb2.HealthResponse.FromString,
+                _registered_method=True)
+        self.GetCapabilities = channel.unary_unary(
+                '/voicestudio.runtime.v1.RuntimeAdapterService/GetCapabilities',
+                request_serializer=runtime__adapter__pb2.GetCapabilitiesRequest.SerializeToString,
+                response_deserializer=runtime__adapter__pb2.GetCapabilitiesResponse.FromString,
+                _registered_method=True)
+        self.Execute = channel.unary_stream(
+                '/voicestudio.runtime.v1.RuntimeAdapterService/Execute',
+                request_serializer=runtime__adapter__pb2.ExecuteRequest.SerializeToString,
+                response_deserializer=runtime__adapter__pb2.ExecuteResponse.FromString,
+                _registered_method=True)
+        self.Cancel = channel.unary_unary(
+                '/voicestudio.runtime.v1.RuntimeAdapterService/Cancel',
+                request_serializer=runtime__adapter__pb2.CancelRequest.SerializeToString,
+                response_deserializer=runtime__adapter__pb2.CancelResponse.FromString,
+                _registered_method=True)
+
+
+class RuntimeAdapterServiceServicer:
+    """RuntimeAdapterService is local to a GPU Node and is never publicly exposed.
+    """
+
+    def Health(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def GetCapabilities(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Execute(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Cancel(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_RuntimeAdapterServiceServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'Health': grpc.unary_unary_rpc_method_handler(
+                    servicer.Health,
+                    request_deserializer=runtime__adapter__pb2.HealthRequest.FromString,
+                    response_serializer=runtime__adapter__pb2.HealthResponse.SerializeToString,
+            ),
+            'GetCapabilities': grpc.unary_unary_rpc_method_handler(
+                    servicer.GetCapabilities,
+                    request_deserializer=runtime__adapter__pb2.GetCapabilitiesRequest.FromString,
+                    response_serializer=runtime__adapter__pb2.GetCapabilitiesResponse.SerializeToString,
+            ),
+            'Execute': grpc.unary_stream_rpc_method_handler(
+                    servicer.Execute,
+                    request_deserializer=runtime__adapter__pb2.ExecuteRequest.FromString,
+                    response_serializer=runtime__adapter__pb2.ExecuteResponse.SerializeToString,
+            ),
+            'Cancel': grpc.unary_unary_rpc_method_handler(
+                    servicer.Cancel,
+                    request_deserializer=runtime__adapter__pb2.CancelRequest.FromString,
+                    response_serializer=runtime__adapter__pb2.CancelResponse.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'voicestudio.runtime.v1.RuntimeAdapterService', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+    server.add_registered_method_handlers('voicestudio.runtime.v1.RuntimeAdapterService', rpc_method_handlers)
+
+
+ # This class is part of an EXPERIMENTAL API.
+class RuntimeAdapterService:
+    """RuntimeAdapterService is local to a GPU Node and is never publicly exposed.
+    """
+
+    @staticmethod
+    def Health(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/voicestudio.runtime.v1.RuntimeAdapterService/Health',
+            runtime__adapter__pb2.HealthRequest.SerializeToString,
+            runtime__adapter__pb2.HealthResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def GetCapabilities(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/voicestudio.runtime.v1.RuntimeAdapterService/GetCapabilities',
+            runtime__adapter__pb2.GetCapabilitiesRequest.SerializeToString,
+            runtime__adapter__pb2.GetCapabilitiesResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Execute(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_stream(
+            request,
+            target,
+            '/voicestudio.runtime.v1.RuntimeAdapterService/Execute',
+            runtime__adapter__pb2.ExecuteRequest.SerializeToString,
+            runtime__adapter__pb2.ExecuteResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Cancel(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/voicestudio.runtime.v1.RuntimeAdapterService/Cancel',
+            runtime__adapter__pb2.CancelRequest.SerializeToString,
+            runtime__adapter__pb2.CancelResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)

backend/runtime_adapter/inventory.py

@@ -0,0 +1,318 @@
+"""Device and model inventory reported through Health/GetCapabilities.
+
+The server is written against the small protocol at the top of this module so
+tests can substitute fakes; :class:`ProductionInventory` is the real thing,
+wired to ``services.tts_backend``'s engine registry, ``services.hf_revisions``
+pinned revisions, and :mod:`runtime_adapter.digest`.
+
+State rules (mirrors the Go preflight's expectations):
+
+- READY is **explicit**: engine registered, availability probe passed, the
+  pinned snapshot fully present on disk, and a digest computed. Anything
+  less is INSTALLED / LOADING / FAILED — never READY.
+- A loading or failed model is still listed (with its true state) so the
+  Gateway can observe it; only READY models are schedulable.
+"""
+from __future__ import annotations
+
+import os
+import threading
+import time
+from dataclasses import dataclass, field
+
+from . import SLOTS_ENV
+from ._paths import ensure_backend_on_path
+from .digest import snapshot_digest
+
+STATE_INSTALLED = "installed"
+STATE_LOADING = "loading"
+STATE_READY = "ready"
+STATE_FAILED = "failed"
+
+
+@dataclass(frozen=True)
+class DeviceInfo:
+    device_id: str
+    hardware_class: str
+    total_vram_bytes: int
+    total_slots: int
+    free_slots: int
+
+
+@dataclass(frozen=True)
+class ModelInfo:
+    catalog_model_id: str
+    model_version: str
+    model_digest: str
+    precisions: tuple[str, ...] = ()
+    features: tuple[str, ...] = ()
+    state: str = STATE_INSTALLED
+
+
+#: Engines this adapter can attest as digest-pinned models: TTS engine id →
+#: curated Hugging Face repo (must be pinned in ``services.hf_revisions``).
+#: Engines without a single pinned weights repo (external API servers,
+#: multi-model muxes) are deliberately absent — they cannot be digest-pinned.
+ENGINE_MODEL_REPOS: dict[str, str] = {
+    "omnivoice": "k2-fsa/OmniVoice",
+    "voxcpm2": "openbmb/VoxCPM2",
+    "moss-tts-nano": "OpenMOSS-Team/MOSS-TTS-Nano-100M",
+    "kittentts": "KittenML/kitten-tts-mini-0.8",
+    "cosyvoice": "FunAudioLLM/Fun-CosyVoice3-0.5B-2512",
+    "moss-tts-v15": "OpenMOSS-Team/MOSS-TTS-v1.5",
+}
+
+
+def catalog_model_version(revision: str, model_digest: str) -> str:
+    """Return the immutable catalog version for an attested model snapshot.
+
+    A Hugging Face revision names source history, not necessarily the exact
+    snapshot bytes installed on a node.  The catalog version therefore carries
+    a short, deterministic digest suffix.  A changed snapshot becomes a new
+    catalog identity instead of mutating an identity retained by Jobs.
+    """
+    digest = model_digest.removeprefix("sha256:")
+    if len(revision) != 40 or len(digest) != 64:
+        raise ValueError("model identity requires a SHA revision and SHA-256 digest")
+    return f"{revision}+sha256-{digest[:16]}"
+
+
+def slots_per_device(default: int = 1) -> int:
+    raw = os.environ.get(SLOTS_ENV, "").strip()
+    try:
+        value = int(raw) if raw else default
+    except ValueError:
+        return default
+    return max(1, min(value, 64))
+
+
+@dataclass
+class ProductionInventory:
+    """Real host inventory. All heavy imports happen inside methods.
+
+    ``models()`` is memoized for ``model_ttl_s`` under a lock: the first call
+    hashes every installed snapshot (minutes for multi-GB weights, then cached
+    in the on-disk digest sidecar), and Health + GetCapabilities arrive
+    back-to-back. Call :meth:`warm` before serving so the first RPC never
+    pays the hashing cost inside its deadline.
+    """
+
+    slots: int = field(default_factory=slots_per_device)
+    model_ttl_s: float = 15.0
+
+    def __post_init__(self):
+        self._model_lock = threading.Lock()
+        self._model_cache: list[ModelInfo] | None = None
+        self._model_cache_at = 0.0
+
+    def warm(self) -> None:
+        self.models()
+
+    def devices(self, busy_slots: int = 0) -> list[DeviceInfo]:
+        ensure_backend_on_path()
+        devices = self._accelerators() or [self._cpu_device()]
+        return [self._with_slots(device, busy_slots) for device in devices]
+
+    def _with_slots(self, device: DeviceInfo, busy_slots: int) -> DeviceInfo:
+        free = max(0, min(device.total_slots - busy_slots, device.total_slots))
+        return DeviceInfo(
+            device_id=device.device_id,
+            hardware_class=device.hardware_class,
+            total_vram_bytes=device.total_vram_bytes,
+            total_slots=device.total_slots,
+            free_slots=free,
+        )
+
+    def _accelerators(self) -> list[DeviceInfo]:
+        try:
+            import torch  # noqa: PLC0415
+        except Exception:
+            return []
+        found: list[DeviceInfo] = []
+        try:
+            if torch.cuda.is_available():
+                for index in range(torch.cuda.device_count()):
+                    props = torch.cuda.get_device_properties(index)
+                    found.append(
+                        DeviceInfo(
+                            device_id=f"cuda:{index}",
+                            hardware_class=torch.cuda.get_device_name(index),
+                            total_vram_bytes=int(props.total_memory),
+                            total_slots=self.slots,
+                            free_slots=self.slots,
+                        )
+                    )
+                return found
+        except Exception:
+            pass
+        try:
+            if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
+                vram = 0
+                recommended = getattr(torch.mps, "recommended_max_memory", None)
+                if callable(recommended):
+                    try:
+                        vram = int(recommended())
+                    except Exception:
+                        vram = 0
+                if vram <= 0:
+                    vram = _system_memory_bytes()
+                return [
+                    DeviceInfo(
+                        device_id="mps:0",
+                        hardware_class="apple-silicon-mps",
+                        total_vram_bytes=vram,
+                        total_slots=self.slots,
+                        free_slots=self.slots,
+                    )
+                ]
+        except Exception:
+            pass
+        return []
+
+    def _cpu_device(self) -> DeviceInfo:
+        # A CPU-only node is a valid (slow) execution device. total_vram_bytes
+        # carries system memory so the Gateway's ">0" validity check reflects
+        # real capacity rather than a made-up constant.
+        import platform  # noqa: PLC0415
+
+        return DeviceInfo(
+            device_id="cpu:0",
+            hardware_class=platform.processor() or platform.machine() or "cpu",
+            total_vram_bytes=_system_memory_bytes(),
+            total_slots=self.slots,
+            free_slots=self.slots,
+        )
+
+    def models(self) -> list[ModelInfo]:
+        with self._model_lock:
+            now = time.monotonic()
+            if (
+                self._model_cache is not None
+                and now - self._model_cache_at < self.model_ttl_s
+            ):
+                return list(self._model_cache)
+            self._model_cache = self._scan_models()
+            self._model_cache_at = time.monotonic()
+            return list(self._model_cache)
+
+    def _scan_models(self) -> list[ModelInfo]:
+        ensure_backend_on_path()
+        from services.hf_cache_repair import repo_cache_dir  # noqa: PLC0415
+        from services.hf_revisions import installed_revision  # noqa: PLC0415
+        from services.tts_backend import get_backend_class  # noqa: PLC0415
+
+        models: list[ModelInfo] = []
+        for engine_id, repo_id in sorted(ENGINE_MODEL_REPOS.items()):
+            try:
+                backend_cls = get_backend_class(engine_id)
+            except Exception:
+                continue  # engine not registered in this build
+            repo_dir = repo_cache_dir(repo_id)
+            try:
+                revision = installed_revision(repo_id, os.path.dirname(repo_dir))
+            except ValueError:
+                continue  # repo not in the curated catalog — cannot attest
+            snapshot = os.path.join(repo_dir, "snapshots", revision)
+            if not os.path.isdir(snapshot):
+                continue  # weights not installed at the pinned revision
+            models.append(
+                self._model_state(engine_id, backend_cls, repo_dir, revision, snapshot)
+            )
+        return models
+
+    def _model_state(
+        self, engine_id: str, backend_cls, repo_dir: str, revision: str, snapshot: str
+    ) -> ModelInfo:
+        base = ModelInfo(
+            catalog_model_id=engine_id,
+            model_version=revision,
+            model_digest="",
+            precisions=self._precisions(backend_cls),
+            features=self._features(backend_cls),
+        )
+        try:
+            ok, _message = backend_cls.is_available()
+        except Exception:
+            return _replace_state(base, STATE_FAILED)
+        if not ok:
+            return _replace_state(base, STATE_INSTALLED)
+        if _snapshot_incomplete(repo_dir, snapshot):
+            return _replace_state(base, STATE_LOADING)
+        try:
+            model_digest = snapshot_digest(
+                snapshot,
+                cache_path=os.path.join(repo_dir, f"voicestudio-digest-{revision}.json"),
+            )
+        except OSError:
+            return _replace_state(base, STATE_LOADING)
+        return ModelInfo(
+            catalog_model_id=base.catalog_model_id,
+            model_version=catalog_model_version(base.model_version, model_digest),
+            model_digest=model_digest,
+            precisions=base.precisions,
+            features=base.features,
+            state=STATE_READY,
+        )
+
+    def _precisions(self, backend_cls) -> tuple[str, ...]:
+        # Advisory execution precisions. fp32 always works; fp16 is offered
+        # when the engine targets an accelerator this host actually has.
+        compat = tuple(getattr(backend_cls, "gpu_compat", ("cpu",)))
+        try:
+            from core.device_caps import detect_host_caps  # noqa: PLC0415
+
+            family = detect_host_caps().family
+        except Exception:
+            family = "cpu"
+        if family != "cpu" and family in compat:
+            return ("fp16", "fp32")
+        return ("fp32",)
+
+    def _features(self, backend_cls) -> tuple[str, ...]:
+        features = ["tts"]
+        if getattr(backend_cls, "supports_cloning", False) is True:
+            features.append("voice_clone")
+        if getattr(backend_cls, "supports_voice_design", False):
+            features.append("voice_design")
+        if getattr(backend_cls, "supports_emotion", False):
+            features.append("emotion")
+        return tuple(features)
+
+
+def _replace_state(model: ModelInfo, state: str) -> ModelInfo:
+    return ModelInfo(
+        catalog_model_id=model.catalog_model_id,
+        model_version=model.model_version,
+        model_digest=model.model_digest,
+        precisions=model.precisions,
+        features=model.features,
+        state=state,
+    )
+
+
+def _snapshot_incomplete(repo_dir: str, snapshot: str) -> bool:
+    """A download in flight leaves ``*.incomplete`` blobs or dangling links."""
+    blobs = os.path.join(repo_dir, "blobs")
+    try:
+        if any(name.endswith(".incomplete") for name in os.listdir(blobs)):
+            return True
+    except OSError:
+        pass
+    for current, _dirs, files in os.walk(snapshot):
+        for name in files:
+            path = os.path.join(current, name)
+            if not os.path.exists(path):  # dangling symlink
+                return True
+    return False
+
+
+def _system_memory_bytes() -> int:
+    try:
+        import psutil  # noqa: PLC0415
+
+        return int(psutil.virtual_memory().total)
+    except Exception:
+        try:
+            return os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES")
+        except (ValueError, OSError, AttributeError):
+            return 1  # still nonzero: the preflight requires > 0

backend/runtime_adapter/production.py

@@ -0,0 +1,63 @@
+"""Wires the adapter to the real VoiceStudio backend.
+
+Kept separate from ``server.py`` so tests can build a
+:class:`~runtime_adapter.server.RuntimeContext` from fakes without importing
+torch or the engine registry.
+"""
+from __future__ import annotations
+
+import sys
+
+from . import ADAPTER_VERSION
+from ._paths import ensure_backend_on_path
+from .inventory import ProductionInventory, slots_per_device
+from .server import RuntimeContext
+
+
+def production_engine_provider(catalog_model_id: str):
+    """Resolve a READY catalog model id to its cached engine instance."""
+    ensure_backend_on_path()
+    from services.tts_backend import get_engine_instance_for  # noqa: PLC0415
+
+    return get_engine_instance_for(catalog_model_id)
+
+
+def build_runtime_context() -> RuntimeContext:
+    ensure_backend_on_path()
+    from core.version import APP_VERSION  # noqa: PLC0415
+
+    slots = slots_per_device()
+    return RuntimeContext(
+        runtime_version=APP_VERSION,
+        adapter_version=ADAPTER_VERSION,
+        inventory=ProductionInventory(slots=slots),
+        engine_provider=production_engine_provider,
+        slot_limit=slots,
+    )
+
+def prewarm_engines(context: RuntimeContext) -> None:
+    """Load and compile every READY model before the socket accepts work.
+
+    The GPU Gateway leases an attempt for a bounded window and renews it from
+    execution evidence. A cold engine produces no evidence: weight loading and
+    torch compilation can run for minutes emitting nothing, so the lease
+    expires mid-load, the attempt is fenced, the Job requeues, and the next
+    attempt pays the same cost — a loop that never yields audio.
+
+    Paying that cost once at startup, before the adapter is reachable, means
+    the first real Execute begins inference immediately. Preflight already
+    refuses a runtime with no READY model, so a failure here is reported and
+    the model is dropped from the advertised set rather than being offered as
+    schedulable capacity the node cannot actually serve promptly.
+    """
+    ensure_backend_on_path()
+    for model in context.inventory.models():
+        if model.state != "ready":
+            continue
+        try:
+            context.engine_provider(model.catalog_model_id)
+        except Exception as error:  # noqa: BLE001 - reported, never fatal
+            print(
+                f"runtime adapter: prewarm of {model.catalog_model_id} failed: {error}",
+                file=sys.stderr,
+            )

backend/runtime_adapter/runtime_adapter.proto

@@ -0,0 +1,199 @@
+syntax = "proto3";
+
+package voicestudio.runtime.v1;
+
+option go_package = "github.com/velixio/vssaas/api/gen/runtime/v1;runtimev1";
+
+// RuntimeAdapterService is local to a GPU Node and is never publicly exposed.
+service RuntimeAdapterService {
+  rpc Health(HealthRequest) returns (HealthResponse);
+  rpc GetCapabilities(GetCapabilitiesRequest) returns (GetCapabilitiesResponse);
+  rpc Execute(ExecuteRequest) returns (stream ExecuteResponse);
+  rpc Cancel(CancelRequest) returns (CancelResponse);
+}
+
+message ExecuteResponse { ExecutionEvent event = 1; }
+
+message HealthRequest {}
+
+message HealthResponse {
+  ServingState state = 1;
+  string runtime_version = 2;
+  string adapter_version = 3;
+  repeated string health_flags = 4;
+}
+
+enum ServingState {
+  SERVING_STATE_UNSPECIFIED = 0;
+  SERVING_STATE_READY = 1;
+  SERVING_STATE_DEGRADED = 2;
+  SERVING_STATE_UNHEALTHY = 3;
+}
+
+message GetCapabilitiesRequest {}
+
+message GetCapabilitiesResponse {
+  string runtime_version = 1;
+  string adapter_version = 2;
+  repeated RuntimeDevice devices = 3;
+  repeated RuntimeModel models = 4;
+}
+
+message RuntimeDevice {
+  string device_id = 1;
+  string hardware_class = 2;
+  uint64 total_vram_bytes = 3;
+  uint32 total_slots = 4;
+  uint32 free_slots = 5;
+}
+
+message RuntimeModel {
+  string catalog_model_id = 1;
+  string model_version = 2;
+  string model_digest = 3;
+  repeated string precisions = 4;
+  repeated string features = 5;
+  RuntimeModelState state = 6;
+}
+
+enum RuntimeModelState {
+  RUNTIME_MODEL_STATE_UNSPECIFIED = 0;
+  RUNTIME_MODEL_STATE_INSTALLED = 1;
+  RUNTIME_MODEL_STATE_LOADING = 2;
+  RUNTIME_MODEL_STATE_READY = 3;
+  RUNTIME_MODEL_STATE_FAILED = 4;
+}
+
+message ExecuteRequest {
+  string job_id = 1;
+  string attempt_id = 2;
+  string device_id = 3;
+  string slot_id = 4;
+  ModelSpec model = 5;
+  map<string, ParameterValue> parameters = 6;
+  repeated LocalArtifact inputs = 7;
+  repeated LocalArtifact outputs = 8;
+  int64 deadline_unix_ms = 9;
+  uint32 maximum_preview_bytes = 10;
+}
+
+message ModelSpec {
+  string catalog_model_id = 1;
+  string model_version = 2;
+  string model_digest = 3;
+  string precision = 4;
+}
+
+message ParameterValue {
+  oneof value {
+    string string_value = 1;
+    int64 integer_value = 2;
+    double number_value = 3;
+    bool boolean_value = 4;
+  }
+}
+
+message LocalArtifact {
+  string artifact_id = 1;
+  string local_handle = 2;
+  LocalArtifactOperation operation = 3;
+  uint64 expected_size_bytes = 4;
+  string expected_sha256 = 5;
+  string media_type = 6;
+}
+
+enum LocalArtifactOperation {
+  LOCAL_ARTIFACT_OPERATION_UNSPECIFIED = 0;
+  LOCAL_ARTIFACT_OPERATION_READ = 1;
+  LOCAL_ARTIFACT_OPERATION_WRITE = 2;
+}
+
+message ExecutionEvent {
+  string job_id = 1;
+  string attempt_id = 2;
+  uint64 sequence = 3;
+  int64 observed_at_unix_ms = 4;
+  oneof payload {
+    ExecutionStarted started = 10;
+    ExecutionProgress progress = 11;
+    PreviewChunk preview = 12;
+    ExecutionCompleted completed = 13;
+    ExecutionFailed failed = 14;
+    ExecutionCanceled canceled = 15;
+  }
+}
+
+message ExecutionStarted {}
+
+message ExecutionProgress {
+  uint32 progress_permille = 1;
+  string stage_code = 2;
+}
+
+message PreviewChunk {
+  uint64 sequence = 1;
+  string media_type = 2;
+  bytes data = 3;
+}
+
+message ExecutionCompleted {
+  repeated LocalArtifactManifest outputs = 1;
+  RuntimeMeasurements measurements = 2;
+}
+
+message LocalArtifactManifest {
+  string artifact_id = 1;
+  string local_handle = 2;
+  uint64 size_bytes = 3;
+  string sha256 = 4;
+  string media_type = 5;
+  uint64 duration_ms = 6;
+}
+
+message ExecutionFailed {
+  RuntimeFailureClass failure_class = 1;
+  string stable_code = 2;
+  string safe_detail = 3;
+  RuntimeMeasurements measurements = 4;
+}
+
+message ExecutionCanceled {
+  RuntimeMeasurements measurements = 1;
+}
+
+enum RuntimeFailureClass {
+  RUNTIME_FAILURE_CLASS_UNSPECIFIED = 0;
+  RUNTIME_FAILURE_CLASS_INPUT = 1;
+  RUNTIME_FAILURE_CLASS_MODEL_LOAD = 2;
+  RUNTIME_FAILURE_CLASS_INFERENCE = 3;
+  RUNTIME_FAILURE_CLASS_GPU_RESOURCE = 4;
+  RUNTIME_FAILURE_CLASS_LOCAL_STORAGE = 5;
+  RUNTIME_FAILURE_CLASS_RUNTIME = 6;
+  RUNTIME_FAILURE_CLASS_CANCELED = 7;
+}
+
+message RuntimeMeasurements {
+  uint64 normalized_input_characters = 1;
+  uint64 input_audio_ms = 2;
+  uint64 output_audio_ms = 3;
+  uint64 gpu_execution_ms = 4;
+  uint64 cpu_execution_ms = 5;
+}
+
+message CancelRequest {
+  string job_id = 1;
+  string attempt_id = 2;
+  string reason_code = 3;
+  int64 deadline_unix_ms = 4;
+}
+
+message CancelResponse {
+  CancelDisposition disposition = 1;
+}
+
+enum CancelDisposition {
+  CANCEL_DISPOSITION_UNSPECIFIED = 0;
+  CANCEL_DISPOSITION_ACCEPTED = 1;
+  CANCEL_DISPOSITION_ALREADY_TERMINAL = 2;
+  CANCEL_DISPOSITION_NOT_FOUND = 3;
+}

backend/runtime_adapter/selfcheck.py

@@ -0,0 +1,168 @@
+"""``--selfcheck``: validate the Go preflight's expectations against ourselves.
+
+Starts the server on a private temp socket, then runs a Python port of
+``internal/gateway/preflight.go``'s checks over the wire: socket-path safety,
+READY health with version evidence, identical versions across Health and
+GetCapabilities, valid unique devices, and at least one explicitly READY,
+digest-pinned model with a version and precisions. Prints only a bounded
+readiness summary (never handles, paths, or credentials) and exits nonzero on
+any failed expectation — the same fail-closed behavior a node deployment gets
+from ``cmd/runtime-adapter-preflight``.
+"""
+from __future__ import annotations
+
+import os
+import stat as stat_module
+import tempfile
+from dataclasses import dataclass
+
+import grpc
+
+from .gen import runtime_adapter_pb2 as pb2
+from .gen import runtime_adapter_pb2_grpc as pb2_grpc
+
+_MAX_UINT32 = 2**32 - 1
+
+
+class PreflightError(Exception):
+    """One failed preflight expectation, with a bounded message."""
+
+
+@dataclass(frozen=True)
+class PreflightSummary:
+    socket_path: str
+    runtime_version: str
+    adapter_version: str
+    device_count: int
+    ready_model_count: int
+    total_slots: int
+    free_slots: int
+
+    def render(self) -> str:
+        return (
+            f"runtime={self.runtime_version} adapter={self.adapter_version} "
+            f"devices={self.device_count} ready_models={self.ready_model_count} "
+            f"slots={self.free_slots}/{self.total_slots}"
+        )
+
+
+def validate_socket_file(socket_path: str) -> None:
+    if not socket_path or not os.path.isabs(socket_path):
+        raise PreflightError("socket path must be absolute")
+    info = os.lstat(socket_path)
+    if stat_module.S_ISLNK(info.st_mode) or not stat_module.S_ISSOCK(info.st_mode):
+        raise PreflightError("endpoint must be a local Unix socket")
+    parent = os.stat(os.path.dirname(socket_path))
+    if not stat_module.S_ISDIR(parent.st_mode) or parent.st_mode & 0o002:
+        raise PreflightError("socket directory is unsafe")
+
+
+def run_preflight(socket_path: str, timeout_s: float = 10.0) -> PreflightSummary:
+    """Port of ``PreflightRuntime`` + ``validateRuntimeCapabilities``."""
+    validate_socket_file(socket_path)
+    with grpc.insecure_channel(f"unix:{socket_path}") as channel:
+        stub = pb2_grpc.RuntimeAdapterServiceStub(channel)
+        try:
+            health = stub.Health(pb2.HealthRequest(), timeout=timeout_s)
+        except grpc.RpcError as exc:
+            raise PreflightError(f"health call failed: {exc.code().name}")
+        if (
+            health.state != pb2.SERVING_STATE_READY
+            or not health.runtime_version.strip()
+            or not health.adapter_version.strip()
+        ):
+            raise PreflightError("runtime is not ready with versioned adapter evidence")
+        try:
+            caps = stub.GetCapabilities(pb2.GetCapabilitiesRequest(), timeout=timeout_s)
+        except grpc.RpcError as exc:
+            raise PreflightError(f"capabilities call failed: {exc.code().name}")
+    return _validate_capabilities(socket_path, health, caps)
+
+
+def _validate_capabilities(socket_path, health, caps) -> PreflightSummary:
+    if not caps.runtime_version.strip() or not caps.adapter_version.strip():
+        raise PreflightError("capabilities lack version evidence")
+    if (
+        caps.runtime_version != health.runtime_version
+        or caps.adapter_version != health.adapter_version
+    ):
+        raise PreflightError("health and capabilities versions disagree")
+    if not caps.devices:
+        raise PreflightError("no execution devices reported")
+    total_slots = free_slots = 0
+    seen_devices: set[str] = set()
+    for device in caps.devices:
+        if (
+            not device.device_id.strip()
+            or not device.hardware_class.strip()
+            or device.total_vram_bytes == 0
+            or device.total_slots == 0
+            or device.free_slots > device.total_slots
+        ):
+            raise PreflightError("invalid execution device reported")
+        if device.device_id in seen_devices:
+            raise PreflightError("duplicate execution device reported")
+        seen_devices.add(device.device_id)
+        if (
+            total_slots + device.total_slots > _MAX_UINT32
+            or free_slots + device.free_slots > _MAX_UINT32
+        ):
+            raise PreflightError("slot total overflows protocol limit")
+        total_slots += device.total_slots
+        free_slots += device.free_slots
+    ready = 0
+    seen_models: set[tuple[str, str, str]] = set()
+    for model in caps.models:
+        if model.state != pb2.RUNTIME_MODEL_STATE_READY:
+            continue
+        if (
+            not model.catalog_model_id.strip()
+            or not model.model_version.strip()
+            or not model.model_digest.strip()
+            or not model.precisions
+        ):
+            raise PreflightError("invalid ready model reported")
+        identity = (model.catalog_model_id, model.model_version, model.model_digest)
+        if identity in seen_models:
+            raise PreflightError("duplicate ready model reported")
+        seen_models.add(identity)
+        ready += 1
+    if ready == 0:
+        raise PreflightError("no ready model reported")
+    return PreflightSummary(
+        socket_path=socket_path,
+        runtime_version=health.runtime_version,
+        adapter_version=health.adapter_version,
+        device_count=len(caps.devices),
+        ready_model_count=ready,
+        total_slots=total_slots,
+        free_slots=free_slots,
+    )
+
+
+def selfcheck(timeout_s: float = 10.0) -> int:
+    """Start the production server on a temp socket and preflight it."""
+    from .production import build_runtime_context  # noqa: PLC0415
+    from .server import create_server  # noqa: PLC0415
+
+    context = build_runtime_context()
+    warm = getattr(context.inventory, "warm", None)
+    if callable(warm):
+        print("selfcheck: warming model inventory (first run hashes weights)…")
+        warm()
+    # Short prefix: macOS caps Unix-socket paths at 103 characters and the
+    # default macOS tempdir is already ~60 characters deep.
+    with tempfile.TemporaryDirectory(prefix="vs-rta-") as tmp:
+        os.chmod(tmp, 0o700)
+        socket_path = os.path.join(tmp, "runtime.sock")
+        server = create_server(context, socket_path)
+        server.start()
+        try:
+            summary = run_preflight(socket_path, timeout_s=timeout_s)
+        except PreflightError as failure:
+            print(f"selfcheck: FAIL: {failure}")
+            return 1
+        finally:
+            server.stop(grace=2).wait()
+    print(f"selfcheck: OK: {summary.render()}")
+    return 0

backend/runtime_adapter/server.py

@@ -0,0 +1,208 @@
+"""The gRPC server: Unix-domain socket only, no HTTP, no TCP.
+
+``Health`` and ``GetCapabilities`` read the same version constants from one
+:class:`RuntimeContext`, so the "identical versions" preflight expectation
+holds by construction. Socket-path safety mirrors the Go preflight's checks
+(absolute path, no symlink, parent directory not world-writable) at bind time
+so an unsafe deployment fails closed on our side too.
+"""
+from __future__ import annotations
+
+import os
+import stat as stat_module
+import threading
+from concurrent import futures
+from dataclasses import dataclass, field
+
+import grpc
+
+from . import ADAPTER_VERSION, DEFAULT_SOCKET_PATH, SOCKET_ENV
+from .executor import AttemptRegistry, Executor
+from .gen import runtime_adapter_pb2 as pb2
+from .gen import runtime_adapter_pb2_grpc as pb2_grpc
+from .inventory import (
+    STATE_FAILED,
+    STATE_INSTALLED,
+    STATE_LOADING,
+    STATE_READY,
+)
+
+_MODEL_STATE_TO_PB = {
+    STATE_INSTALLED: pb2.RUNTIME_MODEL_STATE_INSTALLED,
+    STATE_LOADING: pb2.RUNTIME_MODEL_STATE_LOADING,
+    STATE_READY: pb2.RUNTIME_MODEL_STATE_READY,
+    STATE_FAILED: pb2.RUNTIME_MODEL_STATE_FAILED,
+}
+
+
+@dataclass
+class RuntimeContext:
+    """Everything the servicer needs; tests build it from fakes."""
+
+    runtime_version: str
+    inventory: object
+    engine_provider: object
+    adapter_version: str = ADAPTER_VERSION
+    slot_limit: int = 1
+    progress_interval: float = 0.5
+    poll_interval: float = 0.02
+    registry: AttemptRegistry = field(default_factory=AttemptRegistry)
+
+    def executor(self) -> Executor:
+        return Executor(
+            self.inventory,
+            self.engine_provider,
+            self.registry,
+            slot_limit=self.slot_limit,
+            progress_interval=self.progress_interval,
+            poll_interval=self.poll_interval,
+        )
+
+
+class RuntimeAdapterServicer(pb2_grpc.RuntimeAdapterServiceServicer):
+    def __init__(self, context: RuntimeContext):
+        self._context = context
+        self._executor = context.executor()
+
+    def Health(self, request, grpc_context):
+        flags: list[str] = []
+        state = pb2.SERVING_STATE_READY
+        try:
+            devices = self._context.inventory.devices(
+                busy_slots=self._context.registry.active_count()
+            )
+            models = self._context.inventory.models()
+        except Exception:
+            return pb2.HealthResponse(
+                state=pb2.SERVING_STATE_UNHEALTHY,
+                runtime_version=self._context.runtime_version,
+                adapter_version=self._context.adapter_version,
+                health_flags=["inventory-error"],
+            )
+        if not devices:
+            state = pb2.SERVING_STATE_UNHEALTHY
+            flags.append("no-device")
+        if not any(model.state == STATE_READY for model in models):
+            state = max(state, pb2.SERVING_STATE_DEGRADED)
+            flags.append("no-ready-model")
+        return pb2.HealthResponse(
+            state=state,
+            runtime_version=self._context.runtime_version,
+            adapter_version=self._context.adapter_version,
+            health_flags=flags,
+        )
+
+    def GetCapabilities(self, request, grpc_context):
+        busy = self._context.registry.active_count()
+        response = pb2.GetCapabilitiesResponse(
+            runtime_version=self._context.runtime_version,
+            adapter_version=self._context.adapter_version,
+        )
+        for device in self._context.inventory.devices(busy_slots=busy):
+            response.devices.append(
+                pb2.RuntimeDevice(
+                    device_id=device.device_id,
+                    hardware_class=device.hardware_class,
+                    total_vram_bytes=device.total_vram_bytes,
+                    total_slots=device.total_slots,
+                    free_slots=device.free_slots,
+                )
+            )
+        for model in self._context.inventory.models():
+            response.models.append(
+                pb2.RuntimeModel(
+                    catalog_model_id=model.catalog_model_id,
+                    model_version=model.model_version,
+                    model_digest=model.model_digest,
+                    precisions=list(model.precisions),
+                    features=list(model.features),
+                    state=_MODEL_STATE_TO_PB.get(
+                        model.state, pb2.RUNTIME_MODEL_STATE_UNSPECIFIED
+                    ),
+                )
+            )
+        return response
+
+    def Execute(self, request, grpc_context):
+        yield from self._executor.execute(request, grpc_context)
+
+    def Cancel(self, request, grpc_context):
+        disposition = self._context.registry.cancel(request.job_id, request.attempt_id)
+        return pb2.CancelResponse(disposition=disposition)
+
+
+def resolve_socket_path(explicit: str | None = None) -> str:
+    return (
+        (explicit or "").strip()
+        or os.environ.get(SOCKET_ENV, "").strip()
+        or DEFAULT_SOCKET_PATH
+    )
+
+
+def prepare_socket(socket_path: str) -> str:
+    """Fail closed on any unsafe socket placement; remove only a stale socket."""
+    if not socket_path or not os.path.isabs(socket_path):
+        raise ValueError("runtime socket path must be absolute")
+    parent = os.path.dirname(socket_path)
+    try:
+        parent_stat = os.stat(parent)
+    except OSError as exc:
+        raise ValueError(f"runtime socket directory is missing: {exc}") from exc
+    if not stat_module.S_ISDIR(parent_stat.st_mode) or parent_stat.st_mode & 0o002:
+        raise ValueError("runtime socket directory is unsafe (world-writable?)")
+    try:
+        existing = os.lstat(socket_path)
+    except FileNotFoundError:
+        return socket_path
+    if stat_module.S_ISSOCK(existing.st_mode):
+        os.unlink(socket_path)  # stale socket from a previous run
+        return socket_path
+    raise ValueError("runtime socket path exists and is not a socket")
+
+
+def create_server(
+    context: RuntimeContext, socket_path: str, *, max_workers: int | None = None
+) -> grpc.Server:
+    prepare_socket(socket_path)
+    workers = max_workers or max(8, context.slot_limit * 2 + 4)
+    server = grpc.server(
+        futures.ThreadPoolExecutor(
+            max_workers=workers, thread_name_prefix="runtime-adapter"
+        )
+    )
+    pb2_grpc.add_RuntimeAdapterServiceServicer_to_server(
+        RuntimeAdapterServicer(context), server
+    )
+    bound = server.add_insecure_port(f"unix:{socket_path}")
+    if bound == 0:
+        raise RuntimeError("failed to bind the runtime adapter socket")
+    return server
+
+
+def serve(context: RuntimeContext, socket_path: str) -> int:
+    """Run until SIGINT/SIGTERM. Returns a process exit code."""
+    import signal  # noqa: PLC0415
+
+    warm = getattr(context.inventory, "warm", None)
+    if callable(warm):
+        warm()  # hash installed snapshots before the socket exists
+    server = create_server(context, socket_path)
+    server.start()
+    try:
+        os.chmod(socket_path, 0o660)  # gateway runs under the same service identity
+    except OSError:
+        pass
+    stop = threading.Event()
+
+    def _stop(_signum, _frame):
+        stop.set()
+
+    signal.signal(signal.SIGTERM, _stop)
+    signal.signal(signal.SIGINT, _stop)
+    stop.wait()
+    server.stop(grace=10).wait()
+    try:
+        os.unlink(socket_path)
+    except OSError:
+        pass
+    return 0

backend/services/hosted_voice_api.py

@@ -0,0 +1,145 @@
+"""Opt-in adapter from OSS profiles/generation to the hosted v1 contract.
+
+Local VoiceStudio never calls this module unless the caller explicitly requests
+``hosted`` execution *and* all VSS_HOSTED_* settings are present.  It stages
+text/reference bytes as hosted Artifacts, creates a consent-backed Voice, and
+uses durable Jobs; no local path, source recording URL, or plaintext text is
+sent in a Job snapshot.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import os
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+
+class HostedVoiceError(RuntimeError):
+    """A safe, user-actionable hosted adapter failure."""
+
+
+@dataclass(frozen=True)
+class HostedSettings:
+    base_url: str
+    token: str
+    project_id: str
+    model_id: str
+    model_version: str
+    base_voice_id: str
+    consent_text_version: str
+
+    @classmethod
+    def from_environment(cls) -> "HostedSettings | None":
+        values = {
+            name: os.environ.get(name, "").strip()
+            for name in (
+                "VSS_HOSTED_API_BASE", "VSS_HOSTED_API_TOKEN",
+                "VSS_HOSTED_PROJECT_ID", "VSS_HOSTED_MODEL_ID",
+                "VSS_HOSTED_MODEL_VERSION", "VSS_HOSTED_BASE_VOICE_ID",
+            )
+        }
+        if not any(values.values()):
+            return None
+        missing = [name for name, value in values.items() if not value]
+        if missing:
+            raise HostedVoiceError("Hosted execution is incomplete; configure " + ", ".join(missing) + ".")
+        base_url = values["VSS_HOSTED_API_BASE"].rstrip("/")
+        if not base_url.startswith(("http://", "https://")):
+            raise HostedVoiceError("VSS_HOSTED_API_BASE must be an http(s) URL.")
+        return cls(
+            base_url=base_url, token=values["VSS_HOSTED_API_TOKEN"],
+            project_id=values["VSS_HOSTED_PROJECT_ID"], model_id=values["VSS_HOSTED_MODEL_ID"],
+            model_version=values["VSS_HOSTED_MODEL_VERSION"], base_voice_id=values["VSS_HOSTED_BASE_VOICE_ID"],
+            consent_text_version=os.environ.get("VSS_HOSTED_CONSENT_TEXT_VERSION", "oss-spoken-consent-v1").strip() or "oss-spoken-consent-v1",
+        )
+
+
+class HostedVoiceClient:
+    def __init__(self, settings: HostedSettings, client: httpx.AsyncClient | None = None):
+        self.settings = settings
+        self.client = client or httpx.AsyncClient(base_url=settings.base_url, timeout=60)
+        self._owns_client = client is None
+
+    async def aclose(self) -> None:
+        if self._owns_client:
+            await self.client.aclose()
+
+    def _headers(self, *, idempotency: bool = False) -> dict[str, str]:
+        headers = {"Authorization": f"Bearer {self.settings.token}"}
+        if idempotency:
+            headers["Idempotency-Key"] = str(uuid.uuid4())
+        return headers
+
+    async def _request(self, method: str, path: str, *, json: dict | None = None, headers: dict | None = None) -> httpx.Response:
+        response = await self.client.request(method, path, json=json, headers=headers)
+        if response.is_error:
+            detail = "hosted service rejected the request"
+            try:
+                body = response.json()
+                detail = body.get("error", {}).get("message") or body.get("detail") or detail
+            except ValueError:
+                pass
+            raise HostedVoiceError(f"Hosted request failed ({response.status_code}): {detail}")
+        return response
+
+    async def upload_artifact(self, *, purpose: str, media_type: str, payload: bytes) -> str:
+        digest = hashlib.sha256(payload).hexdigest()
+        grant = (await self._request("POST", "/v1/artifacts/upload-authorizations", json={
+            "project_id": self.settings.project_id, "purpose": purpose, "media_type": media_type,
+            "size_bytes": len(payload), "sha256": digest,
+        }, headers=self._headers())).json()
+        put_headers = {k: v for k, v in (grant.get("required_headers") or {}).items() if k.lower() not in {"host", "content-length"}}
+        put_headers.setdefault("Content-Type", media_type)
+        response = await self.client.request(grant.get("method", "PUT"), grant["url"], content=payload, headers=put_headers)
+        if response.is_error:
+            raise HostedVoiceError(f"Hosted Artifact upload failed ({response.status_code}).")
+        await self._request("POST", f"/v1/artifacts/{grant['artifact_id']}/complete", json={"size_bytes": len(payload), "sha256": digest}, headers=self._headers())
+        return grant["artifact_id"]
+
+    async def create_voice(self, *, name: str, description: str, reference_path: str) -> str:
+        payload = Path(reference_path).read_bytes()
+        if not payload:
+            raise HostedVoiceError("The reference recording is empty.")
+        suffix = Path(reference_path).suffix.lower()
+        media_type = {".wav": "audio/wav", ".mp3": "audio/mpeg", ".flac": "audio/flac"}.get(suffix, "audio/wav")
+        reference_id = await self.upload_artifact(purpose="reference_audio", media_type=media_type, payload=payload)
+        voice = await self._request("POST", "/v1/voices", json={
+            "project_id": self.settings.project_id, "display_name": name, "description": description[:1024],
+            "reference_audio_artifact_id": reference_id,
+            "consent": {"attestation_text_version": self.settings.consent_text_version},
+        }, headers=self._headers(idempotency=True))
+        return voice.json()["id"]
+
+    async def synthesize(self, *, text: str, profile_voice_id: str, language: str | None = None) -> bytes:
+        text_artifact = await self.upload_artifact(purpose="input", media_type="text/plain", payload=text.encode("utf-8"))
+        configuration = {"voice_id": self.settings.base_voice_id, "voice_reference_id": profile_voice_id, "output_format": "wav"}
+        if language and language != "Auto":
+            configuration["language"] = language
+        job = await self._request("POST", "/v1/jobs", json={
+            "project_id": self.settings.project_id, "workflow": "tts",
+            "model": {"id": self.settings.model_id, "version": self.settings.model_version},
+            "input": {"text_artifact_id": text_artifact}, "configuration": configuration,
+        }, headers=self._headers(idempotency=True))
+        job_id = job.json()["job_id"]
+        deadline = time.monotonic() + 15 * 60
+        while time.monotonic() < deadline:
+            view = (await self._request("GET", f"/v1/jobs/{job_id}", headers=self._headers())).json()
+            if view.get("state") == "succeeded":
+                outputs = view.get("output_artifact_ids") or []
+                if not outputs:
+                    raise HostedVoiceError("Hosted synthesis completed without audio output.")
+                grant = (await self._request("POST", f"/v1/artifacts/{outputs[0]}/download-authorization", headers=self._headers())).json()
+                audio = await self.client.request(grant.get("method", "GET"), grant["url"])
+                if audio.is_error:
+                    raise HostedVoiceError("Hosted synthesis output could not be downloaded.")
+                return audio.content
+            if view.get("state") in {"failed", "canceled"}:
+                raise HostedVoiceError("Hosted synthesis did not complete successfully.")
+            await asyncio.sleep(0.5)
+        raise HostedVoiceError("Hosted synthesis timed out waiting for its durable Job.")

backend/tests/_runtime_adapter_helpers.py

@@ -0,0 +1,205 @@
+"""Shared fakes and harness for the runtime-adapter tests.
+
+Not a test module (no ``test_`` prefix): imported by
+``test_runtime_adapter_capabilities.py`` and
+``test_runtime_adapter_execute.py``.
+"""
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import os
+import shutil
+import tempfile
+import threading
+import time
+
+import grpc
+
+from runtime_adapter.gen import runtime_adapter_pb2 as pb2
+from runtime_adapter.gen import runtime_adapter_pb2_grpc as pb2_grpc
+from runtime_adapter.inventory import (
+    STATE_READY,
+    DeviceInfo,
+    ModelInfo,
+)
+from runtime_adapter.server import RuntimeContext, create_server
+
+READY_MODEL = ModelInfo(
+    catalog_model_id="fake-tts",
+    model_version="a" * 40,
+    model_digest="sha256:" + "b" * 64,
+    precisions=("fp32",),
+    features=("tts",),
+    state=STATE_READY,
+)
+
+DEVICE = DeviceInfo(
+    device_id="cpu:0",
+    hardware_class="test-cpu",
+    total_vram_bytes=8 * 1024**3,
+    total_slots=1,
+    free_slots=1,
+)
+
+
+class FakeInventory:
+    def __init__(self, models=None, devices=None):
+        self._models = list(models) if models is not None else [READY_MODEL]
+        self._devices = list(devices) if devices is not None else [DEVICE]
+
+    def devices(self, busy_slots: int = 0):
+        return [
+            DeviceInfo(
+                device_id=d.device_id,
+                hardware_class=d.hardware_class,
+                total_vram_bytes=d.total_vram_bytes,
+                total_slots=d.total_slots,
+                free_slots=max(0, d.total_slots - busy_slots),
+            )
+            for d in self._devices
+        ]
+
+    def models(self):
+        return list(self._models)
+
+
+class FakeEngine:
+    """Half a second of silence at 24 kHz, instantly."""
+
+    sample_rate = 24000
+
+    def __init__(self):
+        self.generate_calls = []
+
+    def ensure_ready(self):
+        pass
+
+    def generate(self, text, **kw):
+        import torch
+
+        self.generate_calls.append((text, kw))
+        return torch.zeros(1, 12000)
+
+
+class SlowEngine(FakeEngine):
+    """Sleeps through generate in small slices so tests stay responsive."""
+
+    def __init__(self, seconds: float = 10.0):
+        super().__init__()
+        self.seconds = seconds
+        self.started = threading.Event()
+
+    def generate(self, text, **kw):
+        self.started.set()
+        deadline = time.monotonic() + self.seconds
+        while time.monotonic() < deadline:
+            time.sleep(0.01)
+        return super().generate(text, **kw)
+
+
+class FailingEngine(FakeEngine):
+    def __init__(self, exc: BaseException, phase: str = "synthesis"):
+        super().__init__()
+        self._exc = exc
+        self._phase = phase
+
+    def ensure_ready(self):
+        if self._phase == "model_load":
+            raise self._exc
+
+    def generate(self, text, **kw):
+        raise self._exc
+
+
+def make_context(engine=None, inventory=None, **kw) -> RuntimeContext:
+    engine = engine if engine is not None else FakeEngine()
+    engines = {READY_MODEL.catalog_model_id: engine}
+    kw.setdefault("progress_interval", 0.05)
+    kw.setdefault("poll_interval", 0.005)
+    return RuntimeContext(
+        runtime_version="1.2.3-test",
+        inventory=inventory if inventory is not None else FakeInventory(),
+        engine_provider=lambda model_id: engines[model_id],
+        **kw,
+    )
+
+
+@contextlib.contextmanager
+def serve_over_socket(context: RuntimeContext, tmp_path=None):
+    # A pytest tmp_path routinely exceeds the 103-character Unix-socket path
+    # limit on macOS, so the socket gets its own short private tempdir.
+    socket_dir = tempfile.mkdtemp(prefix="vs-rta-")
+    socket_path = os.path.join(socket_dir, "runtime.sock")
+    server = create_server(context, socket_path)
+    server.start()
+    channel = grpc.insecure_channel(f"unix:{socket_path}")
+    try:
+        yield pb2_grpc.RuntimeAdapterServiceStub(channel), socket_path
+    finally:
+        channel.close()
+        server.stop(grace=0).wait()
+        shutil.rmtree(socket_dir, ignore_errors=True)
+
+
+def make_execute_request(
+    tmp_path,
+    text: str = "hello runtime",
+    *,
+    attempt_id: str = "attempt-1",
+    job_id: str = "job-1",
+    model: ModelInfo = READY_MODEL,
+    device_id: str = "cpu:0",
+    deadline_in_s: float = 30.0,
+    parameters: dict | None = None,
+    input_sha256: str | None = None,
+    input_handle: str | None = None,
+    output_handle: str | None = None,
+) -> pb2.ExecuteRequest:
+    if input_handle is None:
+        input_path = tmp_path / "input.txt"
+        input_path.write_text(text, encoding="utf-8")
+        input_handle = str(input_path)
+    if input_sha256 is None and text is not None:
+        input_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
+    if output_handle is None:
+        output_handle = str(tmp_path / "output.wav")
+    return pb2.ExecuteRequest(
+        job_id=job_id,
+        attempt_id=attempt_id,
+        device_id=device_id,
+        slot_id="slot-0",
+        model=pb2.ModelSpec(
+            catalog_model_id=model.catalog_model_id,
+            model_version=model.model_version,
+            model_digest=model.model_digest,
+            precision="fp32",
+        ),
+        parameters=parameters or {},
+        inputs=[
+            pb2.LocalArtifact(
+                artifact_id="in-1",
+                local_handle=input_handle,
+                operation=pb2.LOCAL_ARTIFACT_OPERATION_READ,
+                expected_sha256=input_sha256 or "",
+                media_type="text/plain",
+            )
+        ],
+        outputs=[
+            pb2.LocalArtifact(
+                artifact_id="out-1",
+                local_handle=output_handle,
+                operation=pb2.LOCAL_ARTIFACT_OPERATION_WRITE,
+                media_type="audio/wav",
+            )
+        ],
+        deadline_unix_ms=int((time.time() + deadline_in_s) * 1000),
+        maximum_preview_bytes=0,
+    )
+
+
+def terminal_of(events):
+    last = events[-1].event
+    kind = last.WhichOneof("payload")
+    assert kind in ("completed", "failed", "canceled"), kind
+    return kind, last

backend/tests/test_hosted_voice_api.py

@@ -0,0 +1,71 @@
+import asyncio
+
+import httpx
+import pytest
+
+from services.hosted_voice_api import HostedSettings, HostedVoiceClient, HostedVoiceError
+
+
+_NAMES = (
+    "VSS_HOSTED_API_BASE", "VSS_HOSTED_API_TOKEN", "VSS_HOSTED_PROJECT_ID",
+    "VSS_HOSTED_MODEL_ID", "VSS_HOSTED_MODEL_VERSION", "VSS_HOSTED_BASE_VOICE_ID",
+)
+
+
+def test_hosted_adapter_is_disabled_without_configuration(monkeypatch):
+    for name in _NAMES:
+        monkeypatch.delenv(name, raising=False)
+    assert HostedSettings.from_environment() is None
+
+
+def test_hosted_adapter_refuses_partial_configuration(monkeypatch):
+    for name in _NAMES:
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("VSS_HOSTED_API_BASE", "http://127.0.0.1:8080")
+    with pytest.raises(HostedVoiceError, match="VSS_HOSTED_API_TOKEN"):
+        HostedSettings.from_environment()
+
+
+def test_hosted_adapter_requires_http_endpoint(monkeypatch):
+    values = {
+        "VSS_HOSTED_API_BASE": "not-a-url",
+        "VSS_HOSTED_API_TOKEN": "token",
+        "VSS_HOSTED_PROJECT_ID": "project",
+        "VSS_HOSTED_MODEL_ID": "model",
+        "VSS_HOSTED_MODEL_VERSION": "v1",
+        "VSS_HOSTED_BASE_VOICE_ID": "base",
+    }
+    for name, value in values.items():
+        monkeypatch.setenv(name, value)
+    with pytest.raises(HostedVoiceError, match="http"):
+        HostedSettings.from_environment()
+
+
+def test_create_voice_uses_artifact_grants_then_canonical_voice_resource(tmp_path):
+    reference = tmp_path / "reference.wav"
+    reference.write_bytes(b"reference-audio")
+    settings = HostedSettings("https://api.test", "token", "project", "model", "v1", "base", "oss-spoken-consent-v1")
+    requests = []
+
+    def handler(request):
+        requests.append(request)
+        if request.url.path == "/v1/artifacts/upload-authorizations":
+            return httpx.Response(200, json={"artifact_id": "artifact-ref", "method": "PUT", "url": "https://objects.test/ref", "required_headers": {}})
+        if request.url.host == "objects.test":
+            return httpx.Response(200)
+        if request.url.path == "/v1/artifacts/artifact-ref/complete":
+            return httpx.Response(200, json={})
+        if request.url.path == "/v1/voices":
+            return httpx.Response(201, json={"id": "hosted-voice"})
+        return httpx.Response(404)
+
+    async def create():
+        client = HostedVoiceClient(settings, httpx.AsyncClient(base_url=settings.base_url, transport=httpx.MockTransport(handler)))
+        return await client.create_voice(name="Local profile", description="description", reference_path=str(reference))
+
+    assert asyncio.run(create()) == "hosted-voice"
+    voice_request = next(request for request in requests if request.url.path == "/v1/voices")
+    body = __import__("json").loads(voice_request.content)
+    assert body["project_id"] == "project"
+    assert body["reference_audio_artifact_id"] == "artifact-ref"
+    assert body["consent"]["attestation_text_version"] == "oss-spoken-consent-v1"

backend/tests/test_runtime_adapter_capabilities.py

@@ -0,0 +1,171 @@
+"""Health/GetCapabilities shape, preflight parity, digest stability.
+
+Mirrors what ``internal/gateway/preflight.go`` in vssaas enforces: READY
+health with version evidence, identical versions across both calls, valid
+unique devices, and only explicitly-READY models counting as schedulable.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from _runtime_adapter_helpers import (  # noqa: E402
+    DEVICE,
+    READY_MODEL,
+    FakeInventory,
+    make_context,
+    serve_over_socket,
+)
+from runtime_adapter.digest import file_sha256, snapshot_digest
+from runtime_adapter.gen import runtime_adapter_pb2 as pb2
+from runtime_adapter.inventory import (
+    STATE_FAILED,
+    STATE_INSTALLED,
+    STATE_LOADING,
+    ModelInfo,
+    catalog_model_version,
+)
+from runtime_adapter.selfcheck import PreflightError, run_preflight
+from runtime_adapter.server import prepare_socket
+
+
+def _model(state, model_id="other-model", digest="sha256:" + "c" * 64):
+    return ModelInfo(
+        catalog_model_id=model_id,
+        model_version="d" * 40,
+        model_digest=digest,
+        precisions=("fp32",),
+        features=("tts",),
+        state=state,
+    )
+
+
+def test_health_and_capabilities_versions_are_identical_and_ready(tmp_path):
+    with serve_over_socket(make_context(), tmp_path) as (stub, _):
+        health = stub.Health(pb2.HealthRequest(), timeout=5)
+        caps = stub.GetCapabilities(pb2.GetCapabilitiesRequest(), timeout=5)
+
+    assert health.state == pb2.SERVING_STATE_READY
+    assert health.runtime_version == "1.2.3-test"
+    assert health.adapter_version.strip()
+    assert caps.runtime_version == health.runtime_version
+    assert caps.adapter_version == health.adapter_version
+
+
+def test_capabilities_report_device_and_ready_model_evidence(tmp_path):
+    inventory = FakeInventory(
+        models=[
+            READY_MODEL,
+            _model(STATE_LOADING, "loading-model"),
+            _model(STATE_FAILED, "failed-model"),
+            _model(STATE_INSTALLED, "installed-model"),
+        ]
+    )
+    with serve_over_socket(make_context(inventory=inventory), tmp_path) as (stub, _):
+        caps = stub.GetCapabilities(pb2.GetCapabilitiesRequest(), timeout=5)
+
+    [device] = caps.devices
+    assert device.device_id == DEVICE.device_id
+    assert device.hardware_class == DEVICE.hardware_class
+    assert device.total_vram_bytes > 0
+    assert 0 < device.free_slots <= device.total_slots
+
+    by_id = {model.catalog_model_id: model for model in caps.models}
+    ready = by_id[READY_MODEL.catalog_model_id]
+    assert ready.state == pb2.RUNTIME_MODEL_STATE_READY
+    assert ready.model_version.startswith("d" * 40 + "+sha256-")
+    assert ready.model_digest.startswith("sha256:")
+    assert list(ready.precisions)
+    # A loading/failed/installed model is reported truthfully, never READY.
+    assert by_id["loading-model"].state == pb2.RUNTIME_MODEL_STATE_LOADING
+    assert by_id["failed-model"].state == pb2.RUNTIME_MODEL_STATE_FAILED
+    assert by_id["installed-model"].state == pb2.RUNTIME_MODEL_STATE_INSTALLED
+
+
+def test_preflight_port_passes_against_a_ready_server(tmp_path):
+    inventory = FakeInventory(models=[READY_MODEL, _model(STATE_LOADING)])
+    with serve_over_socket(make_context(inventory=inventory), tmp_path) as (
+        stub,
+        socket_path,
+    ):
+        summary = run_preflight(socket_path, timeout_s=5)
+    assert summary.ready_model_count == 1  # the loading model must not count
+    assert summary.device_count == 1
+    assert summary.runtime_version == "1.2.3-test"
+    assert summary.total_slots == 1
+
+
+def test_preflight_fails_closed_without_a_ready_model(tmp_path):
+    inventory = FakeInventory(models=[_model(STATE_LOADING)])
+    with serve_over_socket(make_context(inventory=inventory), tmp_path) as (
+        stub,
+        socket_path,
+    ):
+        health = stub.Health(pb2.HealthRequest(), timeout=5)
+        assert health.state == pb2.SERVING_STATE_DEGRADED
+        assert "no-ready-model" in health.health_flags
+        with pytest.raises(PreflightError):
+            run_preflight(socket_path, timeout_s=5)
+
+
+def test_prepare_socket_rejects_unsafe_paths(tmp_path):
+    with pytest.raises(ValueError):
+        prepare_socket("relative/socket.sock")
+    regular = tmp_path / "not-a-socket"
+    regular.write_text("x")
+    with pytest.raises(ValueError):
+        prepare_socket(str(regular))
+    missing_parent = tmp_path / "nope" / "runtime.sock"
+    with pytest.raises(ValueError):
+        prepare_socket(str(missing_parent))
+
+
+def test_snapshot_digest_is_stable_and_content_sensitive(tmp_path):
+    snapshot = tmp_path / "snapshots" / "rev"
+    snapshot.mkdir(parents=True)
+    (snapshot / "weights.bin").write_bytes(b"\x01\x02\x03")
+    (snapshot / "config.json").write_text("{}")
+    cache = tmp_path / "digest-cache.json"
+
+    first = snapshot_digest(snapshot, cache_path=cache)
+    second = snapshot_digest(snapshot, cache_path=cache)  # served from cache
+    assert first == second
+    assert first.startswith("sha256:")
+    assert cache.exists()
+
+    # Any byte change must change the digest (cache invalidated by mtime/size).
+    (snapshot / "weights.bin").write_bytes(b"\x01\x02\x04")
+    assert snapshot_digest(snapshot, cache_path=cache) != first
+
+    with pytest.raises(FileNotFoundError):
+        snapshot_digest(tmp_path / "empty-none")
+
+
+def test_catalog_model_version_changes_when_attested_snapshot_changes():
+    revision = "d" * 40
+    first = catalog_model_version(revision, "sha256:" + "a" * 64)
+    second = catalog_model_version(revision, "sha256:" + "b" * 64)
+
+    assert first.startswith(revision + "+sha256-")
+    assert first != second
+
+
+def test_file_sha256_matches_hashlib(tmp_path):
+    import hashlib
+
+    payload = b"runtime adapter"
+    path = tmp_path / "f.bin"
+    path.write_bytes(payload)
+    assert file_sha256(path) == hashlib.sha256(payload).hexdigest()
+
+
+def test_socket_file_is_private_to_the_node(tmp_path):
+    import stat
+
+    with serve_over_socket(make_context(), tmp_path) as (_stub, socket_path):
+        mode = os.lstat(socket_path).st_mode
+        assert stat.S_ISSOCK(mode)

backend/tests/test_runtime_adapter_execute.py

@@ -0,0 +1,347 @@
+"""Execute/Cancel: happy path, deadline, cancel race, failure taxonomy."""
+from __future__ import annotations
+
+import hashlib
+import os
+import sys
+import threading
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from _runtime_adapter_helpers import (  # noqa: E402
+    READY_MODEL,
+    FailingEngine,
+    FakeInventory,
+    SlowEngine,
+    make_context,
+    make_execute_request,
+    serve_over_socket,
+    terminal_of,
+)
+from runtime_adapter import codes
+from runtime_adapter.gen import runtime_adapter_pb2 as pb2
+from runtime_adapter.inventory import STATE_INSTALLED, ModelInfo
+
+
+def _run_direct(context, request):
+    """Drive the executor without a live gRPC server (fast path for taxonomy)."""
+    return list(context.executor().execute(request, None))
+
+
+def _failure(events):
+    kind, last = terminal_of(events)
+    assert kind == "failed", f"expected failed terminal, got {kind}"
+    return last.failed
+
+
+# ── happy path ────────────────────────────────────────────────────────────
+
+
+def test_execute_happy_path_streams_and_writes_the_manifest(tmp_path):
+    context = make_context()
+    request = make_execute_request(tmp_path, text="hello runtime")
+    with serve_over_socket(context, tmp_path) as (stub, _):
+        events = list(stub.Execute(request, timeout=30))
+
+    payloads = [event.event.WhichOneof("payload") for event in events]
+    assert payloads[0] == "started"
+    assert payloads[-1] == "completed"
+    assert all(kind == "progress" for kind in payloads[1:-1])
+    sequences = [event.event.sequence for event in events]
+    assert sequences == sorted(sequences)
+    assert all(event.event.attempt_id == "attempt-1" for event in events)
+
+    completed = events[-1].event.completed
+    [manifest] = completed.outputs
+    output_path = tmp_path / "output.wav"
+    assert manifest.local_handle == str(output_path)
+    assert output_path.stat().st_size == manifest.size_bytes > 0
+    assert manifest.sha256 == hashlib.sha256(output_path.read_bytes()).hexdigest()
+    assert manifest.media_type == "audio/wav"
+    assert manifest.duration_ms == 500  # 12000 samples at 24 kHz
+
+    measurements = completed.measurements
+    assert measurements.normalized_input_characters == len("hello runtime")
+    assert measurements.output_audio_ms == 500
+
+
+def test_execute_passes_typed_parameters_to_the_engine(tmp_path):
+    from _runtime_adapter_helpers import FakeEngine
+
+    engine = FakeEngine()
+    context = make_context(engine=engine)
+    request = make_execute_request(
+        tmp_path,
+        parameters={
+            "speed": pb2.ParameterValue(number_value=1.5),
+            "language": pb2.ParameterValue(string_value="en"),
+            "num_step": pb2.ParameterValue(integer_value=8),
+        },
+    )
+    events = _run_direct(context, request)
+    assert terminal_of(events)[0] == "completed"
+    [(text, kwargs)] = engine.generate_calls
+    assert text == "hello runtime"
+    assert kwargs == {"speed": 1.5, "language": "en", "num_step": 8}
+
+
+def test_execute_passes_seed_to_the_engine(tmp_path):
+    """Hosted Gallery defaults must retain the OSS deterministic seed."""
+    from _runtime_adapter_helpers import FakeEngine
+
+    engine = FakeEngine()
+    context = make_context(engine=engine)
+    request = make_execute_request(
+        tmp_path,
+        parameters={"seed": pb2.ParameterValue(integer_value=42)},
+    )
+    events = _run_direct(context, request)
+    assert terminal_of(events)[0] == "completed"
+    [(_, kwargs)] = engine.generate_calls
+    assert kwargs == {"seed": 42}
+
+
+# ── deadline ──────────────────────────────────────────────────────────────
+
+
+def test_deadline_is_enforced_with_a_stable_code(tmp_path):
+    context = make_context(engine=SlowEngine(seconds=30))
+    request = make_execute_request(tmp_path, deadline_in_s=0.4)
+    start = time.monotonic()
+    events = _run_direct(context, request)
+    elapsed = time.monotonic() - start
+
+    failed = _failure(events)
+    assert failed.stable_code in (codes.INFERENCE_DEADLINE, codes.MODEL_LOAD_DEADLINE)
+    assert failed.failure_class in (
+        pb2.RUNTIME_FAILURE_CLASS_INFERENCE,
+        pb2.RUNTIME_FAILURE_CLASS_MODEL_LOAD,
+    )
+    assert elapsed < 5, "terminal event must arrive promptly after the deadline"
+
+
+def test_deadline_in_the_past_is_invalid_input(tmp_path):
+    context = make_context()
+    request = make_execute_request(tmp_path)
+    request.deadline_unix_ms = int(time.time() * 1000) - 1000
+    failed = _failure(_run_direct(context, request))
+    assert failed.stable_code == codes.INPUT_DEADLINE_INVALID
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_INPUT
+
+
+# ── cancel ────────────────────────────────────────────────────────────────
+
+
+def test_cancel_race_yields_canceled_terminal_and_idempotent_dispositions(tmp_path):
+    engine = SlowEngine(seconds=30)
+    context = make_context(engine=engine)
+    request = make_execute_request(tmp_path)
+    with serve_over_socket(context, tmp_path) as (stub, _):
+        stream = stub.Execute(request, timeout=30)
+        first = next(stream)
+        assert first.event.WhichOneof("payload") == "started"
+        assert engine.started.wait(5), "engine must be mid-generate for the race"
+
+        cancel = pb2.CancelRequest(job_id="job-1", attempt_id="attempt-1")
+        assert stub.Cancel(cancel, timeout=5).disposition == (
+            pb2.CANCEL_DISPOSITION_ACCEPTED
+        )
+        # Idempotent while still running.
+        assert stub.Cancel(cancel, timeout=5).disposition == (
+            pb2.CANCEL_DISPOSITION_ACCEPTED
+        )
+
+        events = [first, *stream]
+        kind, last = terminal_of(events)
+        assert kind == "canceled"
+        assert last.canceled.HasField("measurements")
+
+        # After the terminal event the same cancel is ALREADY_TERMINAL …
+        assert stub.Cancel(cancel, timeout=5).disposition == (
+            pb2.CANCEL_DISPOSITION_ALREADY_TERMINAL
+        )
+        # … and an unknown attempt is NOT_FOUND.
+        unknown = pb2.CancelRequest(job_id="job-1", attempt_id="nope")
+        assert stub.Cancel(unknown, timeout=5).disposition == (
+            pb2.CANCEL_DISPOSITION_NOT_FOUND
+        )
+
+
+def test_cancel_before_any_execute_is_not_found(tmp_path):
+    with serve_over_socket(make_context(), tmp_path) as (stub, _):
+        response = stub.Cancel(
+            pb2.CancelRequest(job_id="j", attempt_id="never-ran"), timeout=5
+        )
+    assert response.disposition == pb2.CANCEL_DISPOSITION_NOT_FOUND
+
+
+# ── failure classification ────────────────────────────────────────────────
+
+
+def test_model_load_failure_is_classified(tmp_path):
+    engine = FailingEngine(RuntimeError("weights corrupted"), phase="model_load")
+    failed = _failure(
+        _run_direct(make_context(engine=engine), make_execute_request(tmp_path))
+    )
+    assert failed.stable_code == codes.MODEL_LOAD_FAILED
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_MODEL_LOAD
+
+
+def test_inference_failure_is_classified(tmp_path):
+    engine = FailingEngine(ValueError("synthesis exploded"))
+    failed = _failure(
+        _run_direct(make_context(engine=engine), make_execute_request(tmp_path))
+    )
+    assert failed.stable_code == codes.INFERENCE_FAILED
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_INFERENCE
+
+
+def test_gpu_oom_is_classified_as_gpu_resource(tmp_path):
+    engine = FailingEngine(RuntimeError("CUDA out of memory. Tried to allocate…"))
+    failed = _failure(
+        _run_direct(make_context(engine=engine), make_execute_request(tmp_path))
+    )
+    assert failed.stable_code == codes.GPU_OUT_OF_MEMORY
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_GPU_RESOURCE
+
+
+def test_engine_input_rejection_is_invalid_input(tmp_path):
+    from services.tts_backend import TTSInputError
+
+    engine = FailingEngine(TTSInputError("text too long for this engine"))
+    failed = _failure(
+        _run_direct(make_context(engine=engine), make_execute_request(tmp_path))
+    )
+    assert failed.stable_code == codes.INPUT_REJECTED
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_INPUT
+
+
+def test_url_handles_are_rejected_never_fetched(tmp_path):
+    request = make_execute_request(
+        tmp_path, input_handle="https://evil.example/input.txt", input_sha256=""
+    )
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_HANDLE_INVALID
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_INPUT
+
+
+def test_relative_output_handle_is_rejected(tmp_path):
+    request = make_execute_request(tmp_path, output_handle="relative/out.wav")
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_HANDLE_INVALID
+
+
+def test_model_digest_mismatch_is_rejected(tmp_path):
+    request = make_execute_request(tmp_path)
+    request.model.model_digest = "sha256:" + "f" * 64
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_MODEL_DIGEST_MISMATCH
+
+
+def test_non_ready_model_is_rejected(tmp_path):
+    installed = ModelInfo(
+        catalog_model_id=READY_MODEL.catalog_model_id,
+        model_version=READY_MODEL.model_version,
+        model_digest=READY_MODEL.model_digest,
+        precisions=READY_MODEL.precisions,
+        features=READY_MODEL.features,
+        state=STATE_INSTALLED,
+    )
+    context = make_context(inventory=FakeInventory(models=[installed]))
+    failed = _failure(_run_direct(context, make_execute_request(tmp_path)))
+    assert failed.stable_code == codes.INPUT_MODEL_NOT_READY
+
+
+def test_unknown_model_is_rejected(tmp_path):
+    request = make_execute_request(tmp_path)
+    request.model.catalog_model_id = "who-dis"
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_MODEL_UNKNOWN
+
+
+def test_unknown_and_out_of_range_parameters_are_rejected(tmp_path):
+    unknown = make_execute_request(
+        tmp_path,
+        parameters={"exfiltrate": pb2.ParameterValue(string_value="x")},
+    )
+    assert _failure(_run_direct(make_context(), unknown)).stable_code == (
+        codes.INPUT_PARAMETER_UNKNOWN
+    )
+    out_of_range = make_execute_request(
+        tmp_path,
+        attempt_id="attempt-2",
+        parameters={"speed": pb2.ParameterValue(number_value=99.0)},
+    )
+    assert _failure(_run_direct(make_context(), out_of_range)).stable_code == (
+        codes.INPUT_PARAMETER_RANGE
+    )
+
+
+def test_input_checksum_mismatch_is_rejected(tmp_path):
+    request = make_execute_request(tmp_path, input_sha256="0" * 64)
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_CHECKSUM_MISMATCH
+
+
+def test_empty_text_is_rejected(tmp_path):
+    request = make_execute_request(tmp_path, text="   ")
+    failed = _failure(_run_direct(make_context(), request))
+    assert failed.stable_code == codes.INPUT_TEXT_EMPTY
+
+
+def test_unwritable_output_directory_is_local_storage(tmp_path):
+    locked = tmp_path / "locked"
+    locked.mkdir()
+    request = make_execute_request(tmp_path, output_handle=str(locked / "out.wav"))
+    locked.chmod(0o500)
+    try:
+        failed = _failure(_run_direct(make_context(), request))
+    finally:
+        locked.chmod(0o700)
+    assert failed.stable_code == codes.STORAGE_WRITE_FAILED
+    assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_LOCAL_STORAGE
+
+
+def test_duplicate_attempt_id_is_rejected(tmp_path):
+    context = make_context()
+    executor = context.executor()
+    first = make_execute_request(tmp_path)
+    assert terminal_of(list(executor.execute(first, None)))[0] == "completed"
+    duplicate = make_execute_request(tmp_path)
+    events = list(executor.execute(duplicate, None))
+    failed = _failure(events)
+    assert failed.stable_code == codes.INPUT_ATTEMPT_DUPLICATE
+
+
+def test_slot_exhaustion_is_gpu_resource(tmp_path):
+    engine = SlowEngine(seconds=30)
+    context = make_context(engine=engine, slot_limit=1)
+    executor = context.executor()
+    hog = make_execute_request(tmp_path, attempt_id="hog")
+    hog_events = []
+    hog_thread = threading.Thread(
+        target=lambda: hog_events.extend(executor.execute(hog, None)), daemon=True
+    )
+    hog_thread.start()
+    assert engine.started.wait(5)
+    try:
+        crowded = make_execute_request(tmp_path, attempt_id="crowded")
+        failed = _failure(list(executor.execute(crowded, None)))
+        assert failed.stable_code == codes.GPU_SLOTS_EXHAUSTED
+        assert failed.failure_class == pb2.RUNTIME_FAILURE_CLASS_GPU_RESOURCE
+    finally:
+        context.registry.cancel("job-1", "hog")
+        hog_thread.join(timeout=10)
+    assert terminal_of(hog_events)[0] == "canceled"
+
+
+def test_safe_detail_never_carries_local_paths(tmp_path):
+    engine = FailingEngine(RuntimeError(f"failed loading {tmp_path}/weights.bin"))
+    failed = _failure(
+        _run_direct(make_context(engine=engine), make_execute_request(tmp_path))
+    )
+    assert str(tmp_path) not in failed.safe_detail
+    assert "<path>" in failed.safe_detail

docs/hosted-voice.md

@@ -0,0 +1,29 @@
+# Hosted Voice integration
+
+VoiceStudio remains local-first. `/profiles` and `/generate` keep their local
+SQLite and on-device synthesis behaviour unless a caller explicitly asks for a
+hosted operation. No profile or generation is uploaded merely because hosted
+configuration exists.
+
+To enable the optional adapter, configure the backend environment:
+
+```text
+VSS_HOSTED_API_BASE=http://127.0.0.1:8080
+VSS_HOSTED_API_TOKEN=<scoped API credential>
+VSS_HOSTED_PROJECT_ID=<hosted project id>
+VSS_HOSTED_MODEL_ID=<approved TTS model id>
+VSS_HOSTED_MODEL_VERSION=<approved model version>
+VSS_HOSTED_BASE_VOICE_ID=<model-approved base voice>
+VSS_HOSTED_CONSENT_TEXT_VERSION=oss-spoken-consent-v1
+```
+
+First record ownership consent in the local profile UI, then explicitly call
+`POST /profiles/{profile_id}/hosted-sync`. The adapter uploads the reference
+recording through hosted Artifact grants and creates a consent-backed
+`/v1/voices` record; it never sends a local path or a consent recording. The
+returned hosted ID is stored only as local synchronization metadata.
+
+Call `POST /generate` with `hosted=true` and that synchronized `profile_id` to
+use the hosted durable `/v1/jobs` path. The adapter stages text as an Artifact,
+polls the durable Job, and downloads the result only through a temporary grant.
+Without `hosted=true`, `/generate` stays entirely on-device.

docs/superpowers/specs/2026-05-31-voice-gallery-design.md

@@ -131,7 +131,9 @@ Gallery tab has two zones (top toggle):
   - `GET /archetypes/{id}/preview` — serve pre-rendered WAV if present; else render
     via the voice-design engine and cache to disk keyed by instruct hash.
   - `POST /archetypes/{id}/use` — render a sample → create a `voice_profile`
-    (rendered WAV as `ref_audio`, archetype `instruct`/`language`) → return profile id.
+    with `kind='design'` (the rendered WAV is an identity sample; the
+    archetype `instruct`/`language` and deterministic render seed remain
+    authoritative) → return profile id.
   - Register in `backend/main.py` alongside the other routers.
 - **Preview cache** — `OUTPUTS_DIR/archetype_previews/<hash>.wav`, served via a new
   static mount or `FileResponse`. Pre-rendered featured WAVs live under

frontend/src/App.jsx

@@ -109,6 +109,7 @@ import { clearDubHistory as apiClearDubHistory } from './api/dub';
 import { isTauri, doubleClickMaximize, fileToMediaUrl, playBlobAudio } from './utils/media';
 import { browserDownload } from './utils/download';
 import { downloadMedia } from './utils/mediaDownload';
+import { installDesktopInteractionGuards } from './utils/desktopInteractions';
 import { checkForUpdate, fetchAppVersion } from './utils/updater';
 import { syncChannel } from './utils/channelControl';
 import i18n from './i18n';
@@ -411,6 +412,8 @@ function App() {
     insertTag,
     applyPreset,
     handleGenerate,
+    cancelGeneration,
+    cancelAllPendingJobs,
   } = useTTS({ selectedProfile, setSelectedProfile, loadHistory, profiles });
 
   const handleSaveProfile = () => _handleSaveProfile(refAudio, refText, instruct, language);
@@ -748,58 +751,19 @@ function App() {
   // ── DESKTOP NATIVE INTEGRATION ──
   useEffect(() => {
     if (typeof window === 'undefined') return;
-
-    // 1. Prevent default right-click to hide web nature
-    const handleContextMenu = (e) => {
-      // allow on inputs/textareas for copy/paste
-      if (['INPUT', 'TEXTAREA'].includes(e.target.tagName)) return;
-      e.preventDefault();
-    };
-
-    // 2. Prevent keyboard quicks (reload, zoom, print)
-    const handleKeyDown = (e) => {
-      if (!e.metaKey && !e.ctrlKey) return;
-      if (['r', 'p', '=', '-', '+'].includes(e.key.toLowerCase())) {
-        e.preventDefault();
-      }
-    };
-
-    // 3. Prevent pinch-to-zoom
-    const handleWheel = (e) => {
-      if (e.ctrlKey) e.preventDefault();
-    };
-
-    // 4. Global Drag and drop for seamless native feeling
-    const handleDrop = (e) => {
-      e.preventDefault();
-      const file = e.dataTransfer?.files[0];
-      if (!file) return;
-
-      const isVideo = file.name.match(/\.(mp4|mov|mkv|webm|avi)$/i);
-      const isAudio = file.name.match(/\.(mp3|wav|flac|m4a|ogg)$/i);
-      if (isVideo || isAudio) {
-        setMode('dub');
-        setDubVideoFile(file);
-        fileToMediaUrl(file, null).then((urls) => setDubLocalBlobUrl(urls));
-        setDubFilename(file.name);
-        setDubStep('idle');
-      }
-    };
-    const handleDragOver = (e) => e.preventDefault();
-
-    window.addEventListener('contextmenu', handleContextMenu);
-    window.addEventListener('keydown', handleKeyDown);
-    window.addEventListener('wheel', handleWheel, { passive: false });
-    window.addEventListener('drop', handleDrop);
-    window.addEventListener('dragover', handleDragOver);
-
-    return () => {
-      window.removeEventListener('contextmenu', handleContextMenu);
-      window.removeEventListener('keydown', handleKeyDown);
-      window.removeEventListener('wheel', handleWheel);
-      window.removeEventListener('drop', handleDrop);
-      window.removeEventListener('dragover', handleDragOver);
-    };
+    return installDesktopInteractionGuards({
+      onDrop: (file) => {
+        const isVideo = file.name.match(/\.(mp4|mov|mkv|webm|avi)$/i);
+        const isAudio = file.name.match(/\.(mp3|wav|flac|m4a|ogg)$/i);
+        if (isVideo || isAudio) {
+          setMode('dub');
+          setDubVideoFile(file);
+          fileToMediaUrl(file, null).then((urls) => setDubLocalBlobUrl(urls));
+          setDubFilename(file.name);
+          setDubStep('idle');
+        }
+      },
+    });
   }, []);
 
   // ── KEYBOARD SHORTCUTS ──
@@ -1755,6 +1719,8 @@ function App() {
                     handleSaveProfile={handleSaveProfile}
                     handleSaveDesignProfile={handleSaveDesignProfile}
                     handleGenerate={handleGenerate}
+                    cancelGeneration={cancelGeneration}
+                    cancelAllPendingJobs={cancelAllPendingJobs}
                     startRecording={startRecording}
                     stopRecording={stopRecording}
                     ingestRefAudio={ingestRefAudio}

frontend/src/api/client.apibase.test.ts

@@ -33,6 +33,15 @@ describe('_resolveApiBase', () => {
     expect(_resolveApiBase({ VITE_API_PORT: '4000' }, win)).toBe('http://127.0.0.1:4000');
   });
 
+  it('does not send a development UI back to itself when VITE_API_PORT matches its port', () => {
+    const win = {
+      location: { origin: 'http://127.0.0.1:3000', hostname: '127.0.0.1', port: '3000' },
+    };
+    expect(_resolveApiBase({ DEV: true, VITE_API_PORT: '3000' }, win)).toBe(
+      'http://127.0.0.1:3900',
+    );
+  });
+
   it('runtime window.__OMNIVOICE_API_BASE__ wins over everything (Docker prebuilt-image override)', () => {
     const win = {
       __TAURI__: {},

frontend/src/api/client.ts

@@ -46,7 +46,16 @@ export const LS_API_KEY = LEGACY_API_KEY_STORAGE_KEY;
 // Pure + exported for unit testing — takes env + window so tests don't need to
 // re-import the module or stub import.meta.env.
 export function _resolveApiBase(env: any, win: any): string {
-  const port = env?.VITE_API_PORT || '3900';
+  const defaultPort = '3900';
+  // A port override is useful for a deliberately moved backend, but pointing
+  // it at Vite itself can only return the SPA's 404 page. This commonly
+  // happens when a developer moves the UI to :3000 and copies that value into
+  // both variables. Preserve explicit API URLs (which may name a real proxy),
+  // while making the port-only configuration recover to the local backend.
+  const requestedPort = String(env?.VITE_API_PORT || defaultPort);
+  const port = env?.DEV && requestedPort === String(win?.location?.port || '')
+    ? defaultPort
+    : requestedPort;
   // Explicit override, in precedence order:
   //   1. localStorage ov_backend_url — the user's explicit "Remote backend"
   //      setting (Wave 2.3). Beats everything: it's the one override a

frontend/src/api/generate.ts

@@ -55,6 +55,19 @@ export async function generateSpeech(
   }
 }
 
+// Hosted builds replace this module and provide tenant-scoped durable Job
+// cancellation. Local VoiceStudio has no durable hosted Job queue, so its
+// equivalent is intentionally a no-op rather than a cloud dependency.
+export async function cancelPendingHostedJobs(): Promise<number> {
+  return 0;
+}
+
+// The local backend has no durable hosted Job to cancel. Returning false lets
+// the caller stop its local request directly.
+export async function cancelActiveHostedJob(_signal: AbortSignal): Promise<boolean> {
+  return false;
+}
+
 export async function listHistory(): Promise<unknown> {
   return apiJson('/history');
 }

frontend/src/api/types.ts

@@ -182,6 +182,8 @@ export interface Profile {
   ref_audio?: string;
   ref_text?: string;
   instruct?: string;
+  /** Deterministic identity seed for a designed profile. */
+  seed?: number | null;
   vd_states?: string | null;
   description?: string;
   created_at?: string;

frontend/src/components/clone/ActionBar.jsx

@@ -48,6 +48,8 @@ export default function ActionBar({
   outputPlaying,
   isGenerating,
   handleGenerate,
+  cancelGeneration,
+  cancelAllPendingJobs,
   generationTime,
   wasGeneratingRef,
 }) {
@@ -265,13 +267,12 @@ export default function ActionBar({
         <Button
           variant="primary"
           block
-          loading={isGenerating}
-          onClick={handleGenerate}
-          leading={!isGenerating && <Play size={14} />}
+          onClick={isGenerating ? cancelGeneration : handleGenerate}
+          leading={isGenerating ? <Square size={14} /> : <Play size={14} />}
           className="mt-[6px]"
         >
           {isGenerating
-            ? t('clone.synthesizing', { seconds: generationTime })
+            ? 'Cancel job'
             : t('clone.synthesize')}
         </Button>
       )}
@@ -283,6 +284,14 @@ export default function ActionBar({
           className="mt-[6px]"
         />
       )}
+      <Button
+        variant="ghost"
+        block
+        onClick={cancelAllPendingJobs}
+        className="mt-[4px]"
+      >
+        Cancel all pending jobs
+      </Button>
       {/* 10x P4 a11y (spec §3): persistent polite live region — screen
             readers hear generation start AND finish in-workspace, without
             relying on the FloatingPill. sr-only keeps it out of the

frontend/src/components/gallery/ArchetypeCard.jsx

@@ -3,7 +3,6 @@ import { BookOpen, Ellipsis, Headphones, Loader, Play, Star, UserPlus, Wand2 } f
 import { Menu } from '../../ui';
 import {
   ArchetypeAvatar,
-  AccentFlag,
   NowPlaying,
   USE_CASE_COLOR,
 } from '../../utils/archetypeIcons';
@@ -91,8 +90,7 @@ export default function ArchetypeCard({
       {hasChips && (
         <div className="flex flex-wrap items-center gap-[5px]">
           {accentLabel && (
-            <span className="inline-flex items-center gap-[5px] pl-[5px] pr-[8px] py-[2px] rounded-[7px] bg-[var(--color-bg-elev-2)] text-[var(--color-fg-muted)] text-[0.64rem] leading-[1.6]">
-              <AccentFlag accent={a.facets.accent} lang={a.language} size={14} />
+            <span className="inline-flex items-center px-[8px] py-[2px] rounded-[7px] bg-[var(--color-bg-elev-2)] text-[var(--color-fg-muted)] text-[0.64rem] leading-[1.6]">
               {accentLabel}
             </span>
           )}

frontend/src/components/gallery/ArchetypesZone.test.jsx

@@ -133,4 +133,35 @@ describe('ArchetypeCard accessibility', () => {
     fireEvent.click(await screen.findByRole('menuitem', { name: 'Set as Audiobook default' }));
     expect(onUseAsAudiobookDefault).toHaveBeenCalledWith(archetype);
   });
+
+  it('labels an accent without representing it with a country flag', () => {
+    const { container } = render(
+      <ArchetypeCard
+        a={{
+          id: 'librarian',
+          name: 'The Librarian',
+          language: 'English',
+          use_case: 'narration',
+          facets: {
+            gender: 'female',
+            age: 'middle aged',
+            pitch: 'low pitch',
+            accent: 'british accent',
+          },
+          attrs: {},
+        }}
+        t={t}
+        isFavorite={false}
+        isPlaying={false}
+        isLoadingPreview={false}
+        onPreview={vi.fn()}
+        onUse={vi.fn()}
+        onDesign={vi.fn()}
+        onToggleFavorite={vi.fn()}
+      />,
+    );
+
+    expect(screen.getByText('British')).toBeInTheDocument();
+    expect(container.querySelector('.accent-flag')).toBeNull();
+  });
 });

frontend/src/hooks/useProfiles.js

@@ -41,6 +41,8 @@ export default function useProfiles({ loadHistory, loadProfiles }) {
   const setLanguage = useAppStore((s) => s.setLanguage);
   const setVdStates = useAppStore((s) => s.setVdStates);
   const setDefineMethod = useAppStore((s) => s.setDefineMethod);
+  const setDesignSeed = useAppStore((s) => s.setDesignSeed);
+  const setKeepSeed = useAppStore((s) => s.setKeepSeed);
   const language = useAppStore((s) => s.language);
   const mode = useAppStore((s) => s.mode);
   const steps = useAppStore((s) => s.steps);
@@ -102,6 +104,14 @@ export default function useProfiles({ loadHistory, loadProfiles }) {
       // The profile's kind picks the "Define voice" method implicitly: design
       // profiles open the design controls, everything else the audio path.
       setDefineMethod(profile.kind === 'design' ? 'design' : 'audio');
+      // Gallery archetypes render their identity sample with the profile's
+      // stored seed. Reuse it when that profile is selected: otherwise the
+      // Design workspace sends a fresh random seed and the same archetype
+      // visibly drifts away from its gallery voice on every generation.
+      if (profile.kind === 'design' && Number.isInteger(profile.seed)) {
+        setDesignSeed(profile.seed);
+        setKeepSeed(true);
+      }
       // Design profiles (0005) carry their category picks — restore the sliders
       // so selecting one makes it re-editable, not just re-usable.
       if (profile.kind === 'design') {
@@ -132,7 +142,15 @@ export default function useProfiles({ loadHistory, loadProfiles }) {
         setInstruct(profile.instruct || '');
       }
     },
-    [setRefText, setInstruct, setLanguage, setVdStates, setDefineMethod],
+    [
+      setRefText,
+      setInstruct,
+      setLanguage,
+      setVdStates,
+      setDefineMethod,
+      setDesignSeed,
+      setKeepSeed,
+    ],
   );
 
   /** Save the current design (vd_states + instruct) as a reusable profile.

frontend/src/hooks/useTTS.js

@@ -1,6 +1,6 @@
 import { useState, useRef, useCallback } from 'react';
 import { useAppStore } from '../store';
-import { generateSpeech } from '../api/generate';
+import { cancelActiveHostedJob, cancelPendingHostedJobs, generateSpeech } from '../api/generate';
 import { pickDesignSeed } from '../utils/seed';
 import { playBlobAudio, playPing } from '../utils/media';
 import {
@@ -67,6 +67,8 @@ export default function useTTS({ selectedProfile, setSelectedProfile, loadHistor
   const [generationTime, setGenerationTime] = useState(0);
   const timerRef = useRef(null);
   const textAreaRef = useRef(null);
+  const generationAbortRef = useRef(null);
+  const canceledByUserRef = useRef(false);
 
   const ingestRefAudio = useCallback(
     async (file) => {
@@ -109,11 +111,43 @@ export default function useTTS({ selectedProfile, setSelectedProfile, loadHistor
     [text, insertTag],
   );
 
+  const cancelGeneration = useCallback(async () => {
+    // Hosted synthesis observes this signal after admission and sends the
+    // durable cancel command for the queued Job. Before admission it simply
+    // stops staging, so no unsubmitted work is left behind.
+    const controller = generationAbortRef.current;
+    if (!controller) return;
+    try {
+      // Hosted cancellation is durable. Do not change the UI until the server
+      // accepted it; a conflict leaves the active Job visible for retry.
+      await cancelActiveHostedJob(controller.signal);
+      canceledByUserRef.current = true;
+      controller.abort();
+    } catch (error) {
+      toastErrorWithReport(`Could not cancel the job: ${error.message}`, error);
+    }
+  }, []);
+
+  const cancelAllPendingJobs = useCallback(async () => {
+    try {
+      const canceled = await cancelPendingHostedJobs();
+      if (canceled > 0) {
+        toast.success(`Canceled ${canceled} pending ${canceled === 1 ? 'job' : 'jobs'}.`);
+      } else {
+        toast('No pending hosted jobs to cancel.');
+      }
+      await loadHistory();
+    } catch (error) {
+      toastErrorWithReport(`Could not cancel pending jobs: ${error.message}`, error);
+    }
+  }, [loadHistory]);
+
   const handleGenerate = useCallback(async () => {
     if (!text.trim()) return toast.error(t('tts_errors.enter_text'));
     if (defineMethod === 'audio' && !refAudio && !selectedProfile)
       return toast.error(t('tts_errors.upload_or_select'));
     addBreadcrumb(`generate:start (${defineMethod})`);
+    canceledByUserRef.current = false;
     setIsGenerating(true);
     setGenerationTime(0);
     const st = Date.now();
@@ -213,6 +247,7 @@ export default function useTTS({ selectedProfile, setSelectedProfile, loadHistor
       // is unreachable. The ceiling sits just above the backend's load timeout
       // so the backend's descriptive error wins in the normal case.
       const ac = new AbortController();
+      generationAbortRef.current = ac;
       abortTimer = setTimeout(() => ac.abort(), 21 * 60 * 1000);
 
       // #1330 — one voice for both delivery paths. A dropped chunk is not an
@@ -369,13 +404,14 @@ export default function useTTS({ selectedProfile, setSelectedProfile, loadHistor
       // Timeouts are user-recoverable (retry / shorter input) — plain toast.
       // Real generation failures get the "Report this bug" action.
       if (err?.name === 'AbortError') {
-        toast.error(t('tts_errors.timeout'));
+        toast.error(canceledByUserRef.current ? 'Job canceled.' : t('tts_errors.timeout'));
       } else if (modelNotDownloadedPayload(err)) {
         toastModelNotDownloaded(modelNotDownloadedPayload(err));
       } else {
         toastErrorWithReport(t('tts_errors.error_prefix', { message: err.message }), err);
       }
     } finally {
+      generationAbortRef.current = null;
       if (abortTimer) clearTimeout(abortTimer);
       clearInterval(timerRef.current);
       setIsGenerating(false);
@@ -418,5 +454,7 @@ export default function useTTS({ selectedProfile, setSelectedProfile, loadHistor
     insertTag,
     applyPreset,
     handleGenerate,
+    cancelGeneration,
+    cancelAllPendingJobs,
   };
 }

frontend/src/index.css

@@ -876,7 +876,7 @@ html[data-ui-scale-engine='native'] .app-container {
   padding: 3px 8px 3px 16px;
   background: var(--chrome-bg);
   border-bottom: 1px solid var(--chrome-border);
-  user-select: none;
+  user-select: text;
   position: relative;
   z-index: 100;
   grid-column: 1 / -1;
@@ -3478,7 +3478,7 @@ html[data-ui-scale-engine='native'] .app-bootstrap-scale {
   font-family: var(--font-sans);
   font-size: 12px;
   color: var(--chrome-fg);
-  user-select: none;
+  user-select: text;
 }
 
 /* Inside the shell the footer is a GRID ITEM (row 3), not a fixed overlay —

frontend/src/pages/CloneDesignTab.jsx

@@ -74,6 +74,8 @@ export default function CloneDesignTab(props) {
     handleSaveProfile,
     handleSaveDesignProfile,
     handleGenerate,
+    cancelGeneration,
+    cancelAllPendingJobs,
     startRecording,
     stopRecording,
     ingestRefAudio,
@@ -444,6 +446,8 @@ export default function CloneDesignTab(props) {
         outputPlaying={outputPlaying}
         isGenerating={isGenerating}
         handleGenerate={handleGenerate}
+        cancelGeneration={cancelGeneration}
+        cancelAllPendingJobs={cancelAllPendingJobs}
         generationTime={generationTime}
         wasGeneratingRef={wasGeneratingRef}
       />

frontend/src/test/useProfilesDesignSeed.test.jsx

@@ -0,0 +1,39 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { act, renderHook } from '@testing-library/react';
+import useProfiles from '../hooks/useProfiles';
+import { useAppStore } from '../store';
+
+describe('useProfiles designed-voice selection', () => {
+  beforeEach(() => {
+    useAppStore.setState({
+      defineMethod: 'audio',
+      designSeed: null,
+      keepSeed: false,
+      refText: '',
+      instruct: '',
+    });
+  });
+
+  it('pins a gallery design profile to its stored identity seed', () => {
+    const { result } = renderHook(() =>
+      useProfiles({ loadHistory: vi.fn(), loadProfiles: vi.fn() }),
+    );
+
+    act(() => {
+      result.current.handleSelectProfile({
+        id: 'whisper-gallery',
+        kind: 'design',
+        seed: 42,
+        ref_text: 'A gallery sample',
+        instruct: 'female, whisper',
+        language: 'English',
+      });
+    });
+
+    expect(useAppStore.getState()).toMatchObject({
+      defineMethod: 'design',
+      designSeed: 42,
+      keepSeed: true,
+    });
+  });
+});

frontend/src/test/useTTSAutoplayPref.test.jsx

@@ -8,7 +8,7 @@ import {
   streamGenerateSpeech,
   supportsStreamingPreview,
 } from '../utils/streamingTts';
-import { generateSpeech } from '../api/generate';
+import { cancelActiveHostedJob, generateSpeech } from '../api/generate';
 import toast from 'react-hot-toast';
 
 // #1032: Settings → Appearance "Auto-play preview" ("play the output as soon
@@ -31,6 +31,8 @@ vi.mock('../api/generate', async (importOriginal) => {
   const actual = await importOriginal();
   return {
     ...actual,
+    cancelActiveHostedJob: vi.fn().mockResolvedValue(false),
+    cancelPendingHostedJobs: vi.fn().mockResolvedValue(0),
     generateSpeech: vi.fn().mockImplementation(async () => {
       let served = false;
       return {
@@ -113,6 +115,73 @@ describe('useTTS auto-play pref (#1032)', () => {
   });
 });
 
+describe('useTTS hosted cancellation', () => {
+  it('aborts an in-flight hosted synthesis when the user cancels it', async () => {
+    let rejectGeneration;
+    vi.mocked(generateSpeech).mockImplementationOnce((_formData, { signal }) =>
+      new Promise((_resolve, reject) => {
+        rejectGeneration = reject;
+        signal.addEventListener('abort', () => {
+          reject(new DOMException('Aborted', 'AbortError'));
+        });
+      }),
+    );
+    const { result } = renderHook(() => useTTS(hookProps()));
+    let generation;
+    await act(async () => {
+      generation = result.current.handleGenerate();
+      await Promise.resolve();
+    });
+
+    expect(result.current.isGenerating).toBe(true);
+    await act(async () => {
+      await result.current.cancelGeneration();
+      await generation;
+    });
+
+    expect(rejectGeneration).toBeTypeOf('function');
+    expect(result.current.isGenerating).toBe(false);
+  });
+
+  it('keeps the job active when the cancellation API rejects it', async () => {
+    let resolveGeneration;
+    vi.mocked(generateSpeech).mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolveGeneration = resolve;
+        }),
+    );
+    vi.mocked(cancelActiveHostedJob).mockRejectedValueOnce(
+      new Error('The Job cancellation could not be accepted.'),
+    );
+
+    const { result } = renderHook(() => useTTS(hookProps()));
+    let generation;
+    await act(async () => {
+      generation = result.current.handleGenerate();
+      await Promise.resolve();
+    });
+
+    await act(async () => {
+      await result.current.cancelGeneration();
+    });
+
+    expect(result.current.isGenerating).toBe(true);
+
+    await act(async () => {
+      resolveGeneration({
+        body: {
+          getReader: () => ({
+            read: async () => ({ done: true, value: undefined }),
+          }),
+        },
+        headers: { get: () => null },
+      });
+      await generation;
+    });
+  });
+});
+
 describe('useTTS delivery path vs the chosen GPU', () => {
   beforeEach(() => {
     useAppStore.setState({ autoPlayPreview: true });

frontend/src/utils/desktopInteractions.js

@@ -0,0 +1,32 @@
+/**
+ * Keep the desktop shell from accidentally navigating away while preserving
+ * browser-native affordances such as selecting text, copying, and context
+ * menus. The latter are especially important in the web build.
+ */
+export function installDesktopInteractionGuards({ onDrop }) {
+  const handleKeyDown = (event) => {
+    if (!event.metaKey && !event.ctrlKey) return;
+    if (['r', 'p', '=', '-', '+'].includes(event.key.toLowerCase())) event.preventDefault();
+  };
+  const handleWheel = (event) => {
+    if (event.ctrlKey) event.preventDefault();
+  };
+  const handleDrop = (event) => {
+    event.preventDefault();
+    const file = event.dataTransfer?.files[0];
+    if (file) onDrop(file);
+  };
+  const handleDragOver = (event) => event.preventDefault();
+
+  window.addEventListener('keydown', handleKeyDown);
+  window.addEventListener('wheel', handleWheel, { passive: false });
+  window.addEventListener('drop', handleDrop);
+  window.addEventListener('dragover', handleDragOver);
+
+  return () => {
+    window.removeEventListener('keydown', handleKeyDown);
+    window.removeEventListener('wheel', handleWheel);
+    window.removeEventListener('drop', handleDrop);
+    window.removeEventListener('dragover', handleDragOver);
+  };
+}

frontend/src/utils/desktopInteractions.test.js

@@ -0,0 +1,33 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { installDesktopInteractionGuards } from './desktopInteractions';
+
+describe('installDesktopInteractionGuards', () => {
+  it('leaves the native context menu available for copying and browser commands', () => {
+    const dispose = installDesktopInteractionGuards({
+      onDrop: vi.fn(),
+    });
+    const event = new MouseEvent('contextmenu', { bubbles: true, cancelable: true });
+
+    window.dispatchEvent(event);
+
+    expect(event.defaultPrevented).toBe(false);
+    dispose();
+  });
+
+  it('still blocks browser reload, print, and zoom shortcuts', () => {
+    const dispose = installDesktopInteractionGuards({
+      onDrop: vi.fn(),
+    });
+
+    for (const key of ['r', 'p', '=', '-', '+']) {
+      const event = new KeyboardEvent('keydown', { key, ctrlKey: true, cancelable: true });
+      window.dispatchEvent(event);
+      expect(event.defaultPrevented).toBe(true);
+    }
+    const inspect = new KeyboardEvent('keydown', { key: 'i', ctrlKey: true, cancelable: true });
+    window.dispatchEvent(inspect);
+    expect(inspect.defaultPrevented).toBe(false);
+    dispose();
+  });
+});

scripts/gen_runtime_adapter_protocol.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Regenerate the runtime-adapter stubs from ``runtime_adapter.proto``.
+
+    uv run python scripts/gen_runtime_adapter_protocol.py
+
+The ``.proto`` is a byte-identical vendored copy of the vssaas contract
+``api/proto/voicestudio/runtime/v1/runtime_adapter.proto`` (the wire contract
+between the vssaas GPU Gateway and this runtime). The generated files are
+committed so that neither the installer, the frozen build, nor Docker needs
+``protoc`` — only developers changing the ``.proto`` do.
+``tests/test_runtime_adapter_gen.py`` regenerates into a temporary directory
+and fails if the committed output has drifted, so a forgotten regeneration is
+a red test rather than a runtime import error.
+
+The one post-processing step is the import fixup: ``protoc`` emits
+``import runtime_adapter_pb2`` in the gRPC stub, which only resolves if the
+output directory happens to be on ``sys.path``. Rewriting it to a relative
+import lets the package be imported normally.
+"""
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parent.parent
+_PROTO_DIR = _REPO / "backend" / "runtime_adapter"
+_OUT_DIR = _PROTO_DIR / "gen"
+_PROTO = _PROTO_DIR / "runtime_adapter.proto"
+
+_INIT = '''"""Generated protocol stubs — DO NOT EDIT.
+
+Regenerate with ``uv run python scripts/gen_runtime_adapter_protocol.py``
+after any change to ``../runtime_adapter.proto``.
+"""
+'''
+
+
+def generate(out_dir: Path) -> int:
+    """Run protoc into ``out_dir``. Returns protoc's exit code."""
+    from grpc_tools import protoc  # noqa: PLC0415 — dev-only dependency
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    code = protoc.main(
+        [
+            "protoc",
+            f"-I{_PROTO_DIR}",
+            f"--python_out={out_dir}",
+            f"--pyi_out={out_dir}",
+            f"--grpc_python_out={out_dir}",
+            str(_PROTO),
+        ]
+    )
+    if code != 0:
+        return code
+    _fix_imports(out_dir)
+    (out_dir / "__init__.py").write_text(_INIT, encoding="utf-8")
+    return 0
+
+
+def _fix_imports(out_dir: Path) -> None:
+    """Make protoc's flat sibling import work inside a package."""
+    stub = out_dir / "runtime_adapter_pb2_grpc.py"
+    if not stub.exists():
+        return
+    text = stub.read_text(encoding="utf-8")
+    text = re.sub(
+        r"^import (\w+_pb2) as (\w+)$",
+        r"from . import \1 as \2",
+        text,
+        flags=re.M,
+    )
+    stub.write_text(text, encoding="utf-8")
+
+
+def main() -> int:
+    code = generate(_OUT_DIR)
+    if code == 0:
+        print(f"Generated {_OUT_DIR.relative_to(_REPO)}")
+    return code
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/render_gallery.py

@@ -32,6 +32,12 @@ Usage (from the repo root, with the model cached):
 
     python3 scripts/render_gallery.py --out dist/gallery
     python3 scripts/render_gallery.py --out dist/gallery --featured-only
+
+Only the render step is on the GPU; marking, encoding, decoding and detection
+are all CPU, so clips are built ``--jobs`` at a time (4 by default) and the card
+does not sit idle through four stages per clip. ``--resume`` picks up whatever
+is already in the output directory, including MP3s from a run that was
+interrupted before it could write a manifest.
 """
 from __future__ import annotations
 
@@ -110,21 +116,30 @@ async def _build_one(archetype: dict, key: str, work: Path, out_previews: Path)
     mp3_path = out_previews / f"{key}.mp3"
 
     await _render_archetype_wav(archetype, raw_wav)
-    wav, sr = _load(raw_wav)
+
+    # Everything below that is CPU-bound goes through asyncio.to_thread. The
+    # AudioSeal embed and detection are each a real neural forward pass, and run
+    # inline they hold the event loop for the whole clip — so --jobs above 1
+    # would queue work behind a busy loop and buy nothing. torch releases the
+    # GIL inside those passes, which is what makes threads (rather than
+    # processes) the right tool: no second model copy, no IPC for the tensors.
+    wav, sr = await asyncio.to_thread(_load, raw_wav)
 
     # force=True: the published clip carries the mark regardless of whether the
     # machine doing the publishing has invisible watermarking switched on. Same
     # contract as persona_bundle's preview embed.
-    marked = mark_synthetic(wav, sr, force=True, context="gallery.publish")
+    marked = await asyncio.to_thread(
+        mark_synthetic, wav, sr, force=True, context="gallery.publish"
+    )
     from api.routers.generation import _safe_torchaudio_save
 
-    _safe_torchaudio_save(str(marked_wav), marked, sr)
+    await asyncio.to_thread(_safe_torchaudio_save, str(marked_wav), marked, sr)
     await _encode_mp3(marked_wav, mp3_path)
 
     check_wav = work / f"{key}.check.wav"
     await _decode_wav(mp3_path, check_wav)
-    decoded, decoded_sr = _load(check_wav)
-    verdict = detect_watermark(decoded, decoded_sr)
+    decoded, decoded_sr = await asyncio.to_thread(_load, check_wav)
+    verdict = await asyncio.to_thread(detect_watermark, decoded, decoded_sr)
     if not verdict.get("is_watermarked"):
         mp3_path.unlink(missing_ok=True)
         raise AssertionError(
@@ -146,6 +161,90 @@ async def _build_one(archetype: dict, key: str, work: Path, out_previews: Path)
     }
 
 
+async def _preflight_watermark() -> None:
+    """Prove the watermark works before rendering a thousand clips.
+
+    Every clip is verified individually, so a broken embed was always caught —
+    but only after the first full render, and the failure named the bitrate
+    ("raise the bitrate or fix the embed") when the real cause can be nothing to
+    do with audio at all. On a machine missing ``python3-dev``, AudioSeal's
+    forward pass dies inside Inductor (``Python.h: No such file``),
+    ``embed_watermark`` catches it, and the clip is returned *unmarked*. Five
+    seconds here beats discovering that at clip 1 of 1126.
+
+    Doubles as a single-threaded warm-up: the generator and detector are lazy
+    module globals, so touching them once before --jobs fans out avoids several
+    threads racing to load the same model.
+    """
+    import torch
+    from services.watermark import detect_watermark, mark_synthetic
+
+    sample_rate = 24000
+    tone = torch.sin(
+        2 * 3.14159 * 220 * torch.arange(sample_rate * 2) / sample_rate
+    ).unsqueeze(0) * 0.3
+    marked = await asyncio.to_thread(
+        mark_synthetic, tone, sample_rate, force=True, context="gallery.preflight"
+    )
+    verdict = await asyncio.to_thread(detect_watermark, marked, sample_rate)
+    if not verdict.get("is_watermarked"):
+        raise SystemExit(
+            "watermark preflight failed: mark_synthetic returned audio the "
+            f"detector does not recognise (confidence {verdict.get('confidence')}). "
+            "Publishing would ship unmarked audio, so this build stops here.\n"
+            "Most common cause: torch.compile/Inductor cannot build its helper "
+            "(missing Python headers — install python3-dev), which makes the "
+            "embed raise and silently pass the audio through unchanged. "
+            "TORCHDYNAMO_DISABLE=1 is the quick workaround."
+        )
+
+
+def _resume_from_disk(out_previews: Path, by_key: dict) -> dict:
+    """Rebuild manifest entries for previews already rendered.
+
+    The manifest is written once, at the end, so a run interrupted at clip 900
+    leaves 900 perfectly good MP3s that ``--resume`` cannot see — it keys off
+    the manifest, so it would render every one of them again. Everything an
+    entry needs is recoverable from the file itself, so recover it.
+    """
+    recovered: dict = {}
+    for mp3_path in sorted(out_previews.glob("*.mp3")):
+        key = mp3_path.stem
+        archetype = by_key.get(key)
+        if archetype is None:
+            continue  # a key from some older catalog — leave it out of the index
+        data = mp3_path.read_bytes()
+        if not data:
+            mp3_path.unlink(missing_ok=True)
+            continue
+        recovered[key] = {
+            "filename": mp3_path.name,
+            "sha256": _sha256(data),
+            "bytes": len(data),
+            "duration": _probe_duration(mp3_path),
+            "featured": bool(archetype.get("is_featured")),
+        }
+    return recovered
+
+
+def _probe_duration(path: Path) -> float:
+    """Duration in seconds, straight from the encoded file."""
+    import subprocess
+
+    from services.ffmpeg_utils import find_ffmpeg
+
+    ffprobe = str(Path(find_ffmpeg()).with_name("ffprobe"))
+    try:
+        out = subprocess.run(
+            [ffprobe, "-v", "error", "-show_entries", "format=duration",
+             "-of", "default=nw=1:nk=1", str(path)],
+            capture_output=True, text=True, timeout=30,
+        ).stdout.strip()
+        return round(float(out), 3)
+    except (OSError, ValueError, subprocess.SubprocessError):
+        return 0.0
+
+
 def _write_featured_tarball(out: Path, previews: dict) -> dict:
     """Bundle the featured previews so a first run costs one request, not 51."""
     featured = sorted(k for k, e in previews.items() if e["featured"])
@@ -197,28 +296,67 @@ async def _main(args: argparse.Namespace) -> int:
 
     previews: dict[str, dict] = {}
     manifest_path = out / "manifest.json"
-    if args.resume and manifest_path.is_file():
-        previous = json.loads(manifest_path.read_text(encoding="utf-8"))
-        previews = {
-            k: e for k, e in (previous.get("previews") or {}).items()
-            if (out_previews / f"{k}.mp3").is_file()
-        }
+    if args.resume:
+        if manifest_path.is_file():
+            previous = json.loads(manifest_path.read_text(encoding="utf-8"))
+            previews = {
+                k: e for k, e in (previous.get("previews") or {}).items()
+                if (out_previews / f"{k}.mp3").is_file()
+            }
+        # Also adopt clips on disk the manifest never got to describe — an
+        # interrupted run has no manifest at all, and re-rendering audio that
+        # is already correct is the most expensive way to do nothing.
+        for key, entry in _resume_from_disk(out_previews, by_key).items():
+            previews.setdefault(key, entry)
+        if previews:
+            print(f"resuming: {len(previews)} preview(s) already rendered", flush=True)
 
+    await _preflight_watermark()
+
+    pending = [k for k in keys if k not in previews]
     failures: list[str] = []
     with tempfile.TemporaryDirectory(prefix="gallery-render-") as tmp:
         work = Path(tmp)
-        for index, key in enumerate(keys, 1):
-            if key in previews:
-                continue
+        # Bounded fan-out. Each clip is render → embed → encode → decode →
+        # detect, and only the first of those is on the GPU: with one clip in
+        # flight the card idles through four CPU stages. The cap keeps that
+        # overlap from turning into unbounded memory (every concurrent clip
+        # holds decoded audio) and matches how the app itself bounds GPU work.
+        limit = asyncio.Semaphore(max(1, args.jobs))
+        completed = 0
+        state = asyncio.Lock()
+
+        async def build(key: str) -> None:
+            nonlocal completed
             archetype = by_key[key]
-            print(f"[{index}/{len(keys)}] {key} {archetype['name']}", flush=True)
-            try:
-                previews[key] = await _build_one(archetype, key, work, out_previews)
-            except AssertionError:
-                raise  # a lost watermark is a build failure, not a bad voice
-            except Exception as exc:
-                failures.append(f"{key} ({archetype['id']}): {type(exc).__name__}: {exc}")
-                print(f"    FAILED: {exc}", file=sys.stderr, flush=True)
+            async with limit:
+                entry = await _build_one(archetype, key, work, out_previews)
+            async with state:
+                previews[key] = entry
+                completed += 1
+                print(f"[{completed}/{len(pending)}] {key} {archetype['name']}", flush=True)
+
+        tasks = [asyncio.create_task(build(key), name=key) for key in pending]
+        try:
+            for task in asyncio.as_completed(tasks):
+                try:
+                    await task
+                except AssertionError:
+                    # A lost watermark is a build failure, not a bad voice —
+                    # stop the whole run rather than let the remaining jobs
+                    # keep writing clips nobody has verified.
+                    for other in tasks:
+                        other.cancel()
+                    raise
+                except asyncio.CancelledError:
+                    pass
+                except Exception as exc:
+                    failures.append(f"{type(exc).__name__}: {exc}")
+                    print(f"    FAILED: {exc}", file=sys.stderr, flush=True)
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
 
     if not previews:
         print("nothing rendered", file=sys.stderr)
@@ -258,8 +396,13 @@ def main() -> int:
     parser.add_argument("--featured-only", action="store_true",
                         help="render only the 51 featured archetypes")
     parser.add_argument("--limit", type=int, default=0, help="stop after N keys")
+    parser.add_argument(
+        "--jobs", type=int, default=4,
+        help="clips built concurrently (default 4); 1 restores serial rendering",
+    )
     parser.add_argument("--resume", action="store_true",
-                        help="keep previews already described by <out>/manifest.json")
+                        help="keep previews already in <out> (manifest entries "
+                             "and any MP3s an interrupted run left behind)")
     return asyncio.run(_main(parser.parse_args()))
 
 

tests/test_migration_0012_archetype_profiles.py

@@ -0,0 +1,62 @@
+"""Existing gallery voices keep voice-design conditioning after upgrade."""
+import os
+import sqlite3
+
+
+_BASE_PROFILES = """
+    CREATE TABLE voice_profiles (
+        id TEXT PRIMARY KEY,
+        name TEXT NOT NULL,
+        ref_audio_path TEXT,
+        ref_text TEXT DEFAULT '',
+        instruct TEXT DEFAULT '',
+        language TEXT DEFAULT 'Auto',
+        locked_audio_path TEXT DEFAULT '',
+        seed INTEGER DEFAULT NULL,
+        is_locked INTEGER DEFAULT 0,
+        personality TEXT DEFAULT '',
+        description TEXT DEFAULT '',
+        is_demo INTEGER DEFAULT 0,
+        created_at REAL
+    );
+"""
+
+
+def _repo_root() -> str:
+    root = os.path.abspath(os.path.dirname(__file__))
+    while root and root != "/" and not os.path.isfile(os.path.join(root, "alembic.ini")):
+        root = os.path.dirname(root)
+    assert os.path.isfile(os.path.join(root, "alembic.ini")), "alembic.ini not found"
+    return root
+
+
+def _upgrade(db_path: str, target: str = "head") -> None:
+    from alembic import command
+    from alembic.config import Config
+
+    config = Config(os.path.join(_repo_root(), "alembic.ini"))
+    config.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
+    command.upgrade(config, target)
+
+
+def test_migration_marks_only_materialized_archetypes_as_design(tmp_path):
+    database = tmp_path / "gallery-voices.db"
+    with sqlite3.connect(str(database)) as conn:
+        conn.executescript(_BASE_PROFILES)
+
+    _upgrade(str(database), target="0011_hosted_voice_sync")
+    with sqlite3.connect(str(database)) as conn:
+        conn.executemany(
+            "INSERT INTO voice_profiles (id, name, personality, kind) VALUES (?, ?, ?, 'clone')",
+            [
+                ("gallery", "The Librarian", "feat_00_the_librarian"),
+                ("import", "User import", "custom-import"),
+            ],
+        )
+
+    _upgrade(str(database))
+    with sqlite3.connect(str(database)) as conn:
+        kinds = dict(conn.execute("SELECT id, kind FROM voice_profiles").fetchall())
+
+    assert kinds["gallery"] == "design"
+    assert kinds["import"] == "clone"

tests/test_runtime_adapter_gen.py

@@ -0,0 +1,73 @@
+"""The committed runtime-adapter stubs must match the .proto they came from.
+
+Same contract as ``test_worker_protocol_gen.py``: the stubs are committed so
+that neither the installer, the frozen build, nor Docker needs ``protoc``.
+Regenerating into a temporary directory and diffing turns forgotten
+regeneration into a red test with an obvious fix.
+
+The vendored ``backend/runtime_adapter/runtime_adapter.proto`` must also stay
+byte-identical to the upstream vssaas contract
+(``api/proto/voicestudio/runtime/v1/runtime_adapter.proto``); provenance is
+documented in ``backend/runtime_adapter/README.md``.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+import pytest
+
+_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+_GEN_DIR = os.path.join(_REPO, "backend", "runtime_adapter", "gen")
+_GENERATED_FILES = (
+    "runtime_adapter_pb2.py",
+    "runtime_adapter_pb2_grpc.py",
+    "runtime_adapter_pb2.pyi",
+)
+
+pytest.importorskip(
+    "grpc_tools",
+    reason="grpcio-tools is a dev dependency; the committed stubs are what ship.",
+)
+
+sys.path.insert(0, os.path.join(_REPO, "scripts"))
+
+
+def _normalise(text: str) -> list[str]:
+    """Ignore trailing whitespace and blank-line churn between protoc builds."""
+    return [line.rstrip() for line in text.splitlines() if line.strip()]
+
+
+@pytest.mark.parametrize("filename", _GENERATED_FILES)
+def test_committed_stubs_match_the_proto(tmp_path, filename):
+    import gen_runtime_adapter_protocol
+
+    assert gen_runtime_adapter_protocol.generate(tmp_path) == 0, "protoc failed"
+
+    fresh = (tmp_path / filename).read_text(encoding="utf-8")
+    with open(os.path.join(_GEN_DIR, filename), encoding="utf-8") as fh:
+        committed = fh.read()
+
+    assert _normalise(committed) == _normalise(fresh), (
+        f"{filename} is out of date with runtime_adapter.proto. "
+        "Run: uv run python scripts/gen_runtime_adapter_protocol.py"
+    )
+
+
+def test_generated_package_is_importable():
+    """protoc emits a flat sibling import that only resolves if the output
+    directory happens to be on sys.path; the generator rewrites it."""
+    from runtime_adapter.gen import runtime_adapter_pb2 as pb
+    from runtime_adapter.gen import runtime_adapter_pb2_grpc as pb_grpc
+
+    assert hasattr(pb_grpc, "RuntimeAdapterServiceStub")
+    assert pb.ExecuteRequest(attempt_id="a").attempt_id == "a"
+
+
+def test_stub_import_is_relative():
+    with open(
+        os.path.join(_GEN_DIR, "runtime_adapter_pb2_grpc.py"), encoding="utf-8"
+    ) as fh:
+        source = fh.read()
+    assert "from . import runtime_adapter_pb2" in source
+    assert "\nimport runtime_adapter_pb2" not in source

tests/test_runtime_adapter_tts_parity.py

@@ -0,0 +1,28 @@
+"""The hosted runtime adapter must use the OSS OmniVoice render path."""
+from __future__ import annotations
+
+from contextlib import nullcontext
+
+
+def test_runtime_adapter_preserves_seeded_gallery_render_contract(monkeypatch):
+    from services import tts_backend
+    from runtime_adapter.executor import _EngineWorker
+    import api.routers.generation as generation
+
+    model = object()
+    backend = tts_backend.OmniVoiceBackend(model=model)
+    captured = {}
+    params = {
+        "ref_audio": "/runtime-inputs/gallery.wav", "ref_text": "sample",
+        "instruct": "female, whispering", "language": "English",
+        "num_step": 32, "guidance_scale": 2.0, "speed": 1.0,
+        "denoise": True, "postprocess_output": True, "seed": 42,
+    }
+    monkeypatch.setattr(tts_backend, "engine_in_use", lambda _backend: nullcontext())
+    monkeypatch.setattr(generation, "_run_inference", lambda *args: captured.update(args=args) or "audio")
+
+    assert _EngineWorker._synthesize(backend, "Test line.", params) == "audio"
+    assert captured["args"][0] is model
+    assert captured["args"][3] == "/runtime-inputs/gallery.wav"
+    assert captured["args"][5] == "female, whispering"
+    assert captured["args"][13:17] == (None, None, None, 42)