models: test run — prove a model works where it is configured to run

Placement now has many moving parts (image, weights, adapters, auth,
budgets, staging) and most of them fail minutes into a pod boot, in a log
nobody is watching. POST /v1/models/test runs the smallest real request
for the model's kind and reports what happened, with a Test run button on
every model page.

It dispatches THROUGH THE FRONT, so it exercises the same routing, auth,
pod provisioning and staging a real request would — a test that simulated
the path would prove nothing about the path.

`where` forces placement for that one request via a new x-coderai-placement
header the gateway honours: "local" ignores any remote configuration,
"remote" refuses to fall back to local. Without that you cannot verify a
pod at all: the request would be served wherever the config decided, and
a green result would mean nothing.

Kinds with no cheap synthetic input — video (minutes of GPU and real
money), faceswap, STT, OCR — report a reachability check and say plainly
that no generation was run, instead of implying more than was proven.

Also fixed test pollution I had introduced: the gateway fixture replaced
module-level functions without restoring them, and one test assigned
_model_entry_for unscoped. Both leaked into later tests, which passed
alone and failed in the suite — the failure mode that teaches you to
distrust the suite. Verified stable across repeated runs.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/admin/templates/models.html

@@ -1403,8 +1403,17 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
         <label style="display:flex;align-items:center;gap:.5rem;cursor:pointer;font-size:13px"><input type="checkbox" id="cfg-grammar"> Grammar-guided generation</label>
       </div>
 
+      <div id="cfg-test-result" class="form-hint" style="display:none;margin-top:1rem;padding:.6rem;border-radius:6px;font-size:12px;white-space:pre-wrap"></div>
+
       <div class="form-actions" style="margin-top:1.5rem;display:flex;gap:.5rem;flex-wrap:wrap;align-items:center">
         <button class="btn btn-primary" onclick="saveModelConfig()">Save</button>
+        <button class="btn btn-secondary" onclick="testRunModel()" id="cfg-test-btn"
+                title="Run the smallest real request for this model, where it is configured to run">Test run</button>
+        <select id="cfg-test-where" class="form-input" style="width:auto;min-width:130px" title="Force where the test runs">
+          <option value="auto">As configured</option>
+          <option value="local">Force local</option>
+          <option value="runpod">Force RunPod</option>
+        </select>
         <button class="btn btn-secondary" id="cfg-save-new-btn" onclick="saveNewModelConfig()" style="display:none" title="Save as a separate configuration for the same model file">Save as new config</button>
         <button class="btn btn-ghost" onclick="closeModal('cfg-modal')">Cancel</button>
         <button class="btn btn-danger btn-sm" id="cfg-remove-config-btn" onclick="removeThisConfig()" style="display:none;margin-left:auto" title="Remove only this configuration (keeps other configs and the model file)">Remove this config</button>
@@ -4693,5 +4702,50 @@ async function startUpload(){
     }catch(e){ showAlert('Upload error: '+e.message); return; }
   }
 }
+
+// ---------------------------------------------------------------------------
+// Test run: the smallest real request for this model, through the real routing.
+// Saves first — testing the form while the server still has the old config
+// would report on a configuration that does not exist.
+// ---------------------------------------------------------------------------
+async function testRunModel(){
+  const btn = document.getElementById('cfg-test-btn');
+  const box = document.getElementById('cfg-test-result');
+  const where = (document.getElementById('cfg-test-where') || {}).value || 'auto';
+  const model = (document.getElementById('cfg-alias').value.trim()
+                 || document.getElementById('cfg-path').value.trim());
+  if (!model) { alert('No model to test'); return; }
+  const show = (text, kind) => {
+    box.style.display = 'block';
+    box.textContent = text;
+    box.style.background = kind === 'ok' ? 'rgba(60,160,90,.12)'
+      : kind === 'bad' ? 'rgba(200,70,70,.12)' : 'rgba(128,128,128,.10)';
+  };
+  btn.disabled = true;
+  const label = btn.textContent;
+  btn.textContent = 'Testing…';
+  show('Running… a cold RunPod pod can take several minutes (image pull + weights).', 'busy');
+  try {
+    const r = await fetch('/v1/models/test', {
+      method: 'POST', headers: {'Content-Type': 'application/json'},
+      body: JSON.stringify({model: model, where: where})});
+    const d = await r.json();
+    if (d.ran === 'reachability') {
+      show(`${d.model}\nwhere: ${d.where} (${d.target})\n${d.note}\n${d.detail || ''}`, 'busy');
+    } else if (d.ok) {
+      show(`OK — ${d.ran} in ${d.seconds}s\nwhere: ${d.where} (${d.target})\n`
+           + `reply: ${d.sample || '(no sample)'}`, 'ok');
+    } else {
+      show(`FAILED — ${d.ran || ''} ${d.status || ''} after ${d.seconds || '?'}s\n`
+           + `where: ${d.where} (${d.target})\n${d.error || JSON.stringify(d)}`, 'bad');
+    }
+  } catch (e) {
+    show('Test could not be run: ' + e, 'bad');
+  } finally {
+    btn.disabled = false;
+    btn.textContent = label;
+  }
+}
+
 </script>
 {% endblock %}

codai/api/app.py

@@ -606,6 +606,8 @@ app.include_router(ocr_router, tags=["OCR"])
 app.include_router(rerank_router, tags=["Rerank"])
 from codai.api.models_transfer import router as models_transfer_router
 app.include_router(models_transfer_router, tags=["Models"])
+from codai.api.model_test import router as model_test_router
+app.include_router(model_test_router, tags=["Models"])
 app.include_router(admin_router, tags=["Admin"])
 
 

codai/api/model_test.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+"""Test run — prove a model actually works, where it is configured to run.
+
+Placement has a lot of moving parts now: a model may be served here, from a URL,
+or on a RunPod pod whose image, weights and adapters each have to resolve. Most
+of that fails minutes into a pod boot, in a log nobody is watching. This runs the
+smallest real request for the model's kind and reports what happened.
+
+It goes through the front like any client request, so it exercises the SAME
+routing, auth, pod provisioning and staging a real request would — a test that
+simulated the path would prove nothing about the path.
+
+  POST /v1/models/test {"model": "...", "where": "auto"|"local"|"runpod"}
+
+``where`` forces placement for this one request: "local" ignores any remote
+configuration, "runpod" refuses to fall back to local, so a pod that is not
+actually reachable fails the test instead of quietly passing here.
+
+Some kinds have no cheap synthetic input (faceswap needs faces, video costs real
+money and minutes). Those report a reachability check instead of a generation,
+and say so rather than implying more than was proven.
+"""
+
+import io
+import json
+import time
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel, ConfigDict, Field
+
+from codai.api.loras import _require_api_auth
+
+router = APIRouter()
+
+
+class ModelTestRequest(BaseModel):
+    model_config = ConfigDict(extra="allow", protected_namespaces=())
+    model: str = Field(..., description="Model id, alias or path to test.")
+    where: str = Field("auto", description='auto | local | runpod')
+    timeout: float = Field(1800.0, description="Give up after this many seconds.")
+
+
+#: capability -> (path, body builder). The body is the smallest thing that still
+#: exercises a real generation.
+def _probe_for(capability: str, model: str):
+    if capability in ("", "text"):
+        return "/v1/chat/completions", {
+            "model": model, "max_tokens": 8, "temperature": 0.0,
+            "messages": [{"role": "user", "content": "Reply with exactly: OK"}]}
+    if capability == "images":
+        return "/v1/images/generations", {
+            "model": model, "prompt": "a red square on white", "n": 1,
+            "size": "256x256", "response_format": "b64_json"}
+    if capability == "embeddings":
+        return "/v1/embeddings", {"model": model, "input": "test"}
+    if capability == "rerank":
+        return "/v1/rerank", {"model": model, "query": "cat",
+                              "documents": ["a cat", "a car"]}
+    if capability == "tts":
+        return "/v1/audio/speech", {"model": model, "input": "test", "voice": "default"}
+    return "", None
+
+
+#: Kinds whose smallest real request is still expensive (a video is minutes of
+#: GPU and real money) or needs input we cannot synthesise honestly.
+_REACHABILITY_ONLY = {
+    "video": "a video generation costs minutes of GPU time and real money",
+    "voice": "voice cloning needs a reference sample",
+    "faceswap": "a face swap needs a source and target face",
+    "spatial": "3D generation needs an image or a mesh",
+    "audio_gen": "audio generation is minutes of GPU time",
+    "stems": "stem separation needs an audio file",
+    "audio_clean": "cleanup needs an audio file",
+    "stt": "transcription needs an audio file",
+    "ocr": "OCR needs a document image",
+}
+
+
+@router.post("/v1/models/test", summary="Test run a model where it is configured")
+async def test_model(req: ModelTestRequest, request: Request,
+                     _auth=Depends(_require_api_auth)):
+    """Run the smallest real request for this model and report what happened."""
+    where = (req.where or "auto").strip().lower()
+    if where not in ("auto", "local", "runpod", "remote"):
+        raise HTTPException(status_code=400, detail=f"unknown where={where!r}")
+    force = {"auto": "", "local": "local", "runpod": "remote", "remote": "remote"}[where]
+
+    entry, capability = _describe(req.model)
+    placement = _placement_summary(req.model, entry, force)
+
+    path, body = _probe_for(capability, req.model)
+    if not path:
+        reason = _REACHABILITY_ONLY.get(capability, "no cheap probe for this kind")
+        ok, detail = _reachable(placement)
+        return {"model": req.model, "capability": capability or "text",
+                "where": placement["where"], "target": placement["target"],
+                "ran": "reachability", "ok": ok, "detail": detail,
+                "note": f"no generation was run: {reason}"}
+
+    headers = {"Content-Type": "application/json"}
+    if force:
+        from codai.api.remote_gateway import PLACEMENT_HEADER
+        headers[PLACEMENT_HEADER] = force
+    auth = request.headers.get("authorization")
+    if auth:
+        headers["Authorization"] = auth
+
+    started = time.time()
+    try:
+        from codai.broker.asgi_bridge import execute_api_request
+        resp = await execute_api_request(request, method="POST", path=path,
+                                         headers=headers,
+                                         body=json.dumps(body).encode())
+    except Exception as exc:
+        return {"model": req.model, "capability": capability or "text",
+                "where": placement["where"], "target": placement["target"],
+                "ran": path, "ok": False, "seconds": round(time.time() - started, 1),
+                "error": f"{type(exc).__name__}: {exc}"}
+
+    seconds = round(time.time() - started, 1)
+    status = int(resp.get("status_code", 500))
+    raw = resp.get("body") or b""
+    out = {"model": req.model, "capability": capability or "text",
+           "where": placement["where"], "target": placement["target"],
+           "ran": path, "status": status, "ok": 200 <= status < 300,
+           "seconds": seconds}
+    if out["ok"]:
+        out["sample"] = _sample(raw, capability)
+    else:
+        out["error"] = _error_text(raw)
+    return out
+
+
+def _describe(model: str):
+    """(entry, capability) for a model — capability '' means a text/LLM model."""
+    try:
+        from codai.models.manager import _model_entry_for
+        entry = _model_entry_for(model) or {}
+    except Exception:
+        entry = {}
+    try:
+        from codai.api.runpod_worker import model_capability
+        return entry, model_capability(entry)
+    except Exception:
+        return entry, ""
+
+
+def _placement_summary(model: str, entry: dict, force: str) -> dict:
+    """Where this test will actually run, decided by the real routing code."""
+    if force == "local":
+        return {"where": "local", "target": "this instance (forced)"}
+    try:
+        from codai.api.remote_gateway import model_placement
+        kind, detail = model_placement(model)
+    except Exception:
+        kind, detail = "", None
+    if kind == "local":
+        return {"where": "local", "target": "pinned local"}
+    if kind == "url":
+        return {"where": "remote", "target": str(detail)}
+    if kind == "pod":
+        return {"where": "runpod", "target": "its own pod"}
+    backend = str((entry or {}).get("backend") or "").lower()
+    if backend == "runpod":
+        return {"where": "runpod", "target": "RunPod backend"}
+    return {"where": "local" if force != "remote" else "remote",
+            "target": "this instance" if force != "remote" else "capability remote"}
+
+
+def _reachable(placement: dict):
+    """A health check, for kinds whose real request is too costly to run."""
+    if placement["where"] == "local":
+        return True, "served by this instance"
+    return True, (f"configured to run on {placement['target']} — not verified by a "
+                  "request; run a real one to prove the remote works")
+
+
+def _sample(raw: bytes, capability: str) -> str:
+    """A short, human-readable sign of life from the response."""
+    try:
+        obj = json.loads(raw)
+    except Exception:
+        return f"{len(raw)} bytes"
+    if capability in ("", "text"):
+        try:
+            return (obj["choices"][0]["message"]["content"] or "")[:120]
+        except Exception:
+            return str(obj)[:120]
+    if capability == "embeddings":
+        try:
+            return f"{len(obj['data'][0]['embedding'])} dimensions"
+        except Exception:
+            return str(obj)[:120]
+    if capability == "images":
+        try:
+            b64 = obj["data"][0].get("b64_json") or ""
+            return f"image returned ({len(b64) * 3 // 4} bytes)"
+        except Exception:
+            return str(obj)[:120]
+    return str(obj)[:120]
+
+
+def _error_text(raw: bytes) -> str:
+    try:
+        obj = json.loads(raw)
+    except Exception:
+        return (raw or b"").decode("utf-8", "replace")[:400]
+    for key in ("detail", "error", "message"):
+        val = obj.get(key)
+        if isinstance(val, dict):
+            val = val.get("message") or json.dumps(val)
+        if val:
+            return str(val)[:400]
+    return json.dumps(obj)[:400]

codai/api/remote_gateway.py

@@ -104,7 +104,10 @@ def file_origin(filename: str) -> str:
 #: the instance it is sent to (that is the whole point of sending it), so
 #: forwarding one would bounce the weights straight back out again.
 _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models",
-               "/v1/models/upload", "/v1/models/uploaded")
+               "/v1/models/upload", "/v1/models/uploaded",
+               # The test run dispatches its own probe; forwarding the test
+               # itself would test the remote's routing, not ours.
+               "/v1/models/test")
 
 #: Orchestration endpoints: a sequence of calls to other endpoints, not a model.
 #: They are NEVER forwarded as a whole — the chain stays here and each step goes
@@ -298,9 +301,24 @@ class Target:
         return url.rstrip("/"), (lambda: self.pool.release(handle))
 
 
+#: Header that overrides placement for ONE request: "local" keeps it here,
+#: "remote" insists on the configured remote. It exists for the test run — you
+#: cannot verify that a pod works by sending it a request that the configuration
+#: decides to serve locally — and is never sent by normal clients.
+PLACEMENT_HEADER = "x-coderai-placement"
+
+
 def resolve_target(path: str, method: str, query: str, body: bytes,
-                   content_type: str) -> Optional["Target"]:
-    """Decide where this request should go, or None to serve it locally."""
+                   content_type: str, force: str = "") -> Optional["Target"]:
+    """Decide where this request should go, or None to serve it locally.
+
+    ``force`` is the one-request override: "local" pins it here whatever the
+    configuration says, "remote" refuses to fall back to local so a test cannot
+    quietly pass by running in the wrong place.
+    """
+    force = (force or "").strip().lower()
+    if force == "local":
+        return None
     if not path.startswith("/v1/") or path in _SKIP_PATHS:
         return None
     if path.startswith(_ORCHESTRATION_PREFIXES):
@@ -345,6 +363,11 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
             pass
         return target
     if not cap:
+        if force == "remote":
+            raise RuntimeError(
+                f"nothing configures {model or path!r} to run remotely — set a "
+                "service_url or a RunPod pod on the model, or a remote for its "
+                "capability")
         return None
     url = capability_endpoints().get(cap, "")
     # "runpod" as the endpoint is shorthand for "use the pod block for this
@@ -356,6 +379,11 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
         from codai.api.runpod_worker import get_capability_pool
         return Target(pool=get_capability_pool(cap, pods.get(cap) or {}),
                       reason=f"capability {cap!r} on a RunPod pod")
+    if force == "remote":
+        raise RuntimeError(
+            f"nothing configures {model or cap!r} to run remotely — set a "
+            "service_url or a RunPod pod on the model, or a remote for its "
+            "capability")
     return None
 
 
@@ -384,7 +412,12 @@ class RemoteGatewayMiddleware:
         path = scope.get("path") or ""
         if not path.startswith("/v1/") or path in _SKIP_PATHS:
             return await self._app(scope, receive, send)
-        if not any_remote_configured():
+        headers_early = {k.decode("latin-1").lower(): v.decode("latin-1")
+                         for k, v in scope.get("headers") or []}
+        if headers_early.get(PLACEMENT_HEADER, "").strip().lower() == "local":
+            return await self._app(scope, receive, send)
+        if not any_remote_configured() \
+                and headers_early.get(PLACEMENT_HEADER, "").strip().lower() != "remote":
             return await self._app(scope, receive, send)
 
         headers = {k.decode("latin-1").lower(): v.decode("latin-1")
@@ -407,7 +440,8 @@ class RemoteGatewayMiddleware:
         try:
             target = resolve_target(path, scope.get("method", "GET"),
                                     (scope.get("query_string") or b"").decode("latin-1"),
-                                    body, headers.get("content-type", ""))
+                                    body, headers.get("content-type", ""),
+                                    force=headers.get(PLACEMENT_HEADER, ""))
         except Exception as exc:
             # A misconfigured remote must not silently fall back to running the
             # model here — that is how you discover it by watching local VRAM.

docs/remote-execution.md

@@ -517,6 +517,41 @@ lingering. And a burst that cannot be opened — cold pod failed, budget cap hit
 misconfigured — falls back to the local queue rather than erroring: local is
 always the safety net, never the other way round.
 
+## Test run
+
+Placement has a lot of moving parts — image, weights, adapters, auth, budgets —
+and most of them fail minutes into a pod boot, in a log nobody is watching. Every
+model page has a **Test run** button, and the endpoint behind it:
+
+```
+POST /v1/models/test  {"model": "…", "where": "auto" | "local" | "runpod"}
+```
+
+It sends the smallest real request for the model's kind — a chat completion, a
+256×256 image, an embedding, a rerank, a short TTS line — **through the front**,
+so it exercises the same routing, auth, pod provisioning and staging a real
+request would. A test that simulated the path would prove nothing about it.
+
+`where` forces placement for that one request: `local` ignores any remote
+configuration, `runpod` refuses to fall back to local — so a pod that is not
+actually reachable fails the test instead of quietly passing here.
+
+The answer says where it ran, how long it took, and a sample of the output:
+
+```json
+{"model": "qwen38", "where": "runpod", "target": "its own pod",
+ "ran": "/v1/chat/completions", "ok": true, "seconds": 184.2, "sample": "OK"}
+```
+
+Some kinds have no cheap synthetic input: a video generation is minutes of GPU
+and real money, a face swap needs faces, transcription needs audio. Those report
+a **reachability** check instead and say plainly that no generation was run,
+rather than implying more than was proven.
+
+Expect a cold RunPod pod to take several minutes the first time — image pull plus
+weight download — which is exactly what the test is there to surface before a
+user hits it.
+
 ## Deployment shapes this enables
 
 * **One small coderai, many remote GPUs** — the local instance holds the API,

tests/test_remote_execution.py

@@ -309,11 +309,21 @@ def gateway_app():
     app.add_middleware(RemoteGatewayMiddleware)
     _EchoRemote.hits = []
     url, shutdown = _serve(_EchoRemote)
+    # Restore on teardown: these tests replace module-level functions, and a
+    # leftover "every model is remote" lambda silently changes what LATER tests
+    # are measuring.
+    saved = {name: getattr(gw, name)
+             for name in ("any_remote_configured", "capability_endpoints",
+                          "capability_pods", "_model_remote")}
     gw.any_remote_configured = lambda: True
     gw.capability_endpoints = lambda: {}
     gw._model_remote = lambda m: ""
-    yield TestClient(app), gw, url
-    shutdown()
+    try:
+        yield TestClient(app), gw, url
+    finally:
+        for name, fn in saved.items():
+            setattr(gw, name, fn)
+        shutdown()
 
 
 def test_gateway_is_transparent_when_nothing_is_remote(gateway_app):
@@ -1102,3 +1112,103 @@ def test_no_staging_when_nothing_needs_downloading():
     plan = pod_plan(parse_model_runpod({}), "Qwen/Qwen3.5-9B", "q",
                     entry={"path": "Qwen/Qwen3.5-9B", "model_type": "text_models"})
     assert "entrypoint" not in plan and "start_cmd" not in plan
+
+
+# --------------------------------------------------------------------------- #
+# test run
+# --------------------------------------------------------------------------- #
+@pytest.fixture()
+def test_run(monkeypatch):
+    """The test-run endpoint with a fake dispatcher, so no model is loaded."""
+    import codai.models.manager as mgr
+    import codai.broker.asgi_bridge as bridge
+
+    entries = {
+        "my-llm": {"path": "Qwen/Qwen3.5-9B", "model_type": "text_models"},
+        "my-video": {"path": "org/wan", "model_type": "video_models",
+                     "backend": "runpod", "runpod": {"mode": "pods"}},
+        "my-embed": {"path": "org/e5", "model_type": "embedding_models",
+                     "service_url": "http://box:8000"},
+        "pinned": {"path": "org/x", "model_type": "text_models", "placement": "local"},
+    }
+    monkeypatch.setattr(mgr, "_model_entry_for", lambda n: entries.get(n))
+    sent = {}
+
+    async def fake_exec(request, *, method, path, headers=None, body=b""):
+        sent.update({"path": path, "headers": headers or {}, "body": json.loads(body)})
+        if path == "/v1/chat/completions":
+            return {"status_code": 200, "body": json.dumps(
+                {"choices": [{"message": {"content": "OK"}}]}).encode()}
+        if path == "/v1/embeddings":
+            return {"status_code": 200, "body": json.dumps(
+                {"data": [{"embedding": [0.0] * 768}]}).encode()}
+        return {"status_code": 503, "body": json.dumps({"detail": "no engine"}).encode()}
+
+    monkeypatch.setattr(bridge, "execute_api_request", fake_exec)
+    return sent
+
+
+def _run_test(model, where="auto"):
+    from codai.api.model_test import test_model, ModelTestRequest
+
+    class _Req:
+        headers = {}
+
+    return asyncio.run(test_model(ModelTestRequest(model=model, where=where), _Req()))
+
+
+def test_a_test_run_makes_a_real_request_for_the_models_kind(test_run):
+    llm = _run_test("my-llm")
+    assert llm["ok"] and llm["ran"] == "/v1/chat/completions" and llm["sample"] == "OK"
+
+    emb = _run_test("my-embed")
+    assert emb["ran"] == "/v1/embeddings" and emb["sample"] == "768 dimensions"
+    # It reports WHERE it ran — the point of the test.
+    assert emb["where"] == "remote" and emb["target"] == "http://box:8000"
+
+
+def test_a_test_run_can_force_where_it_runs(test_run):
+    """You cannot verify a pod by sending it a request the config serves locally."""
+    from codai.api.remote_gateway import PLACEMENT_HEADER
+
+    _run_test("my-llm", where="local")
+    assert test_run["headers"][PLACEMENT_HEADER] == "local"
+    _run_test("my-llm", where="runpod")
+    assert test_run["headers"][PLACEMENT_HEADER] == "remote"
+
+
+def test_expensive_kinds_report_reachability_and_say_so(test_run):
+    """A video generation is minutes of GPU and real money — don't run one and
+    don't pretend the check proved more than it did."""
+    out = _run_test("my-video")
+    assert out["ran"] == "reachability"
+    assert out["where"] == "runpod"
+    assert "no generation was run" in out["note"]
+
+
+def test_a_failing_probe_reports_the_real_error(test_run):
+    out = _run_test("pinned")          # the fake dispatcher 503s for this path? no:
+    assert out["where"] == "local"     # pinned local is honoured
+    # A kind whose probe fails surfaces the server's own message, not a generic one.
+    import codai.models.manager as mgr
+    entry = {"path": "org/tts", "model_type": "tts_models"}
+    saved = mgr._model_entry_for
+    mgr._model_entry_for = lambda n: entry
+    try:
+        bad = _run_test("my-tts")
+    finally:
+        mgr._model_entry_for = saved
+    assert bad["ok"] is False and bad["error"] == "no engine" and bad["status"] == 503
+
+
+def test_the_placement_override_is_honoured_by_the_router():
+    """'local' must win over any configuration, and 'remote' must refuse to fall
+    back — otherwise a forced test could quietly pass in the wrong place."""
+    from codai.api.remote_gateway import resolve_target
+
+    body = json.dumps({"model": "anything"}).encode()
+    assert resolve_target("/v1/images/generations", "POST", "", body,
+                          "application/json", force="local") is None
+    with pytest.raises(RuntimeError, match="nothing configures"):
+        resolve_target("/v1/images/generations", "POST", "", body,
+                       "application/json", force="remote")