test(runtime-adapter): cover the preflight contract and execute taxonomy

Fake-engine/fake-inventory tests over a real UDS gRPC server plus a fast
direct-executor path: health/capabilities shape and version identity, the
Go-preflight port passing with a READY model and failing closed without
one, only-READY-counts semantics, digest stability/sensitivity, execute
happy path (manifest checksum matches the written WAV), deadline
enforcement, cancel race with idempotent dispositions, slot exhaustion,
duplicate attempts, URL/relative handle rejection, checksum mismatch, and
the input/model-load/inference/GPU/storage failure classification.

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_runtime_adapter_capabilities.py

@@ -0,0 +1,161 @@
+"""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,
+)
+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 len(ready.model_version) == 40
+    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_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,331 @@
+"""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}
+
+
+# ── 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