test-run: place ONE model on a pod, and prove the engine took the config

Testing an image model by setting remotes.endpoints["images"] moves every
request of that kind. An embeddings remote left on from an earlier test did
exactly that on a live box: production embedding traffic went to a rented pod
while the local GPU sat idle, and the operator noticed because the card was
quiet. tools/runpod_model_test.py writes a runpod block on the single model
entry instead — per-model placement already beats the capability map, which is
what lets two models of the same kind sit in different places.

Writing the config is also not enough. The front pushes a reload and the engine
takes it only when idle, so a harness that writes and immediately tests is
testing the PREVIOUS configuration. That produced a whole pass of false "nothing
configures this to run remotely" failures, and could just as easily have
reported a remote pass for a request served locally. GET /v1/models/test/state
answers for the process that serves the request; the harness polls it until the
placement it wrote is visible and refuses to test if it never is.

The new endpoint is in _SKIP_PATHS: forwarded, it would report a pod's view of a
config the pod does not have.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/api/model_test.py

@@ -50,6 +50,53 @@ class ModelTestRequest(BaseModel):
     timeout: float = Field(1800.0, description="Give up after this many seconds.")
 
 
+@router.get("/v1/models/test/state", summary="What this engine currently believes")
+async def test_state(model: str = "", _auth=Depends(_require_api_auth)):
+    """The engine's LIVE view of a model's placement and the remote config.
+
+    Editing config.json or models.json is not enough to change where a request
+    goes: the engine re-reads them only when the front pushes a reload, and that
+    push is deferred while the engine is busy. So a caller that writes a config
+    and immediately tests is testing the OLD one — which is exactly what
+    happened: a whole test pass reported "nothing configures this to run
+    remotely" while the file on disk said otherwise, and an earlier pass sent
+    production embeddings to a pod because the engine still held a stale remote.
+
+    Poll this until it reflects what you wrote, then test. It answers for the
+    process that actually serves the request, so "the engine has it" is a fact
+    here rather than an assumption.
+    """
+    try:
+        from codai.admin.routes import config_manager
+        remotes = getattr(getattr(config_manager, "config", None), "remotes", None)
+    except Exception:
+        remotes = None
+    out = {
+        "pid": os.getpid(),
+        "remotes_enabled": bool(getattr(remotes, "enabled", False)),
+        "capability_endpoints": sorted((getattr(remotes, "endpoints", None) or {}).keys()),
+        "capability_pods": sorted((getattr(remotes, "pods", None) or {}).keys()),
+    }
+    if model:
+        entry, capability = _describe(model)
+        where, _ = ("", None)
+        try:
+            from codai.api.remote_gateway import model_placement
+            where, _ = model_placement(model)
+        except Exception:
+            pass
+        out.update({
+            "model": model,
+            "known": bool(entry),
+            "capability": capability or "text",
+            "placement": where or "(none — falls through to the capability map)",
+            "backend": str((entry or {}).get("backend") or ""),
+            "has_runpod_block": isinstance((entry or {}).get("runpod"), dict)
+                                and bool(entry["runpod"]),
+        })
+    return out
+
+
 #: capability -> (path, body builder). The body is the smallest thing that still
 #: exercises a real generation.
 def _probe_for(capability: str, model: str):

codai/api/remote_gateway.py

@@ -107,7 +107,12 @@ _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models",
                "/v1/models/upload", "/v1/models/uploaded", "/v1/models/register",
                # The test run dispatches its own probe; forwarding the test
                # itself would test the remote's routing, not ours.
-               "/v1/models/test")
+               "/v1/models/test",
+               # …and this one answers "what does THIS process believe?", which
+               # is only meaningful locally. Forwarded, it would report the pod's
+               # view of a config the pod does not have — the exact confusion it
+               # was added to end.
+               "/v1/models/test/state")
 
 #: 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

tests/test_remote_execution.py

@@ -1491,3 +1491,73 @@ def test_an_upload_probe_falls_back_rather_than_failing_the_model(monkeypatch):
 
     monkeypatch.setattr(mt, "_spoken_wav", _boom)
     assert mt._upload_probe("stt", "whisper0") == (None, "", b"")
+
+
+# --- the test harness must not move a whole capability --------------------- #
+
+def test_the_state_endpoint_reports_the_serving_process_view(monkeypatch):
+    """Writing models.json does not change routing: the engine takes the reload
+    only when idle. A whole test pass reported false 'not configured' failures
+    because it trusted the file instead of asking the process."""
+    import asyncio
+    import codai.api.model_test as mt
+
+    entry = {"path": "org/m", "model_type": "image_models", "backend": "runpod",
+             "runpod": {"mode": "pods"}}
+    monkeypatch.setattr(mt, "_describe", lambda m: (entry, "images"))
+    monkeypatch.setattr("codai.api.remote_gateway.model_placement",
+                        lambda m: ("pod", (entry, entry["runpod"])))
+
+    state = asyncio.run(mt.test_state(model="org/m"))
+    assert state["placement"] == "pod"
+    assert state["backend"] == "runpod" and state["has_runpod_block"] is True
+    assert "pid" in state          # which process answered, not which file exists
+
+
+def test_the_harness_pins_one_model_and_never_a_capability(tmp_path, monkeypatch):
+    """remotes.endpoints[capability] moves EVERY request of that kind — that is
+    how production embeddings ended up on a rented pod. The harness must edit
+    the single model entry instead."""
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(
+        "runpod_model_test", "tools/runpod_model_test.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+
+    models = tmp_path / "models.json"
+    models.write_text(json.dumps({
+        "image_models": [{"path": "org/a", "alias": "a"},
+                         {"path": "org/b", "alias": "b"}],
+        "text_models": [{"path": "org/t"}],
+    }))
+    monkeypatch.setattr(mod, "MODELS", str(models))
+
+    assert mod._pin_to_runpod("a", "coderai", 24, 1.0) is True
+    after = json.loads(models.read_text())
+
+    pinned = after["image_models"][0]
+    assert pinned["backend"] == "runpod" and pinned["runpod"]["engine"] == "coderai"
+    # Its sibling of the SAME capability is untouched: that is the whole point.
+    assert after["image_models"][1] == {"path": "org/b", "alias": "b"}
+    assert after["text_models"] == [{"path": "org/t"}]
+    # And nothing anywhere set a capability-wide endpoint.
+    assert "remotes" not in after
+
+
+def test_the_harness_refuses_to_test_a_config_the_engine_has_not_taken(monkeypatch):
+    """Testing against a stale config is worse than not testing: it once
+    reported a remote pass for a request that ran somewhere else."""
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(
+        "runpod_model_test", "tools/runpod_model_test.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+
+    monkeypatch.setattr(mod, "CONFIG_TIMEOUT_S", 0.1)
+    monkeypatch.setattr(mod, "_curl",
+                        lambda *a, **k: {"placement": "(none)", "backend": "",
+                                         "has_runpod_block": False})
+    problem = mod._wait_for_engine("org/a")
+    assert "never picked up the config" in problem

tools/runpod_model_test.py

+#!/usr/bin/env python3
+"""Test one model on RunPod without moving anything else.
+
+Two rules, both learned the hard way on a live box.
+
+ONE MODEL, NOT A CAPABILITY. The obvious way to send an image model to a pod is
+``remotes.endpoints["images"] = "runpod"`` — and that is a production routing
+switch: it moves EVERY request of that kind, not the one under test. An
+embeddings remote left on from an earlier test quietly sent real traffic to a
+rented pod while the local GPU sat idle. So this writes a ``runpod`` block on the
+single model entry instead; per-model placement beats the capability map, which
+is what lets two models of the same kind sit in different places.
+
+CONFIRM THE ENGINE HAS IT. Writing models.json does not change where a request
+goes: the front pushes a reload and the engine takes it only when idle. A test
+that writes and immediately runs is testing the previous configuration — which
+produced both a full pass of false "not configured" failures and, worse, a pass
+that would have reported success for a request served in the wrong place. So it
+polls /v1/models/test/state until the serving process reports the placement it
+just wrote, and refuses to test if that never happens.
+
+Usage:
+    python3 tools/runpod_model_test.py <model> [<model> ...]
+    python3 tools/runpod_model_test.py --engine vllm Qwen/Qwen2.5-0.5B-Instruct
+
+models.json is restored on the way out, including on Ctrl-C or a crash.
+"""
+
+import argparse
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+
+CONFIG_DIR = os.environ.get("CODERAI_CONFIG_DIR",
+                            os.path.expanduser("~/.coderai"))
+MODELS = os.path.join(CONFIG_DIR, "models.json")
+BASE_URL = os.environ.get("CODERAI_URL", "http://127.0.0.1:8000")
+
+#: How long to wait for the engine to take a config push. Deferred while the
+#: engine is busy, so this is generous — but finite: silently testing a stale
+#: config is the failure this exists to prevent.
+CONFIG_TIMEOUT_S = 300
+
+
+def _token() -> str:
+    tok = os.environ.get("CODERAI_TOKEN", "").strip()
+    if tok:
+        return tok
+    with open(os.path.join(CONFIG_DIR, "auth.json")) as fh:
+        tokens = json.load(fh).get("tokens") or []
+    for t in tokens:
+        val = t.get("token") if isinstance(t, dict) else t
+        if val:
+            return val
+    raise SystemExit("no API token found — set CODERAI_TOKEN")
+
+
+def _curl(path: str, payload=None, timeout=2400):
+    cmd = ["curl", "-s", "-m", str(timeout),
+           "-H", f"Authorization: Bearer {_token()}"]
+    if payload is not None:
+        cmd += ["-H", "Content-Type: application/json", "-d", json.dumps(payload)]
+    cmd.append(BASE_URL.rstrip("/") + path)
+    out = subprocess.run(cmd, capture_output=True, text=True).stdout
+    try:
+        return json.loads(out)
+    except Exception:
+        return {"error": f"unparseable response: {out[:300]}"}
+
+
+def _load_models() -> dict:
+    with open(MODELS) as fh:
+        return json.load(fh)
+
+
+def _save_models(data: dict) -> None:
+    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(MODELS), suffix=".tmp")
+    with os.fdopen(fd, "w") as fh:
+        json.dump(data, fh, indent=1)
+    os.replace(tmp, MODELS)          # atomic: never leave a half-written catalogue
+
+
+def _find(data: dict, model: str):
+    """(section, index, entry) for a model named by path, alias or basename."""
+    name = model.strip().lower()
+    for section, entries in data.items():
+        if not isinstance(entries, list):
+            continue
+        for i, entry in enumerate(entries):
+            if not isinstance(entry, dict):
+                continue
+            path = str(entry.get("path") or "")
+            cands = {path.lower(), os.path.basename(path.rstrip("/")).lower(),
+                     str(entry.get("alias") or "").lower()}
+            if name in cands:
+                return section, i, entry
+    return None, None, None
+
+
+def _pin_to_runpod(model: str, engine: str, vram: float, usd: float) -> bool:
+    """Give this ONE model a runpod block. True when the catalogue changed."""
+    data = _load_models()
+    section, index, entry = _find(data, model)
+    if entry is None:
+        print(f"   {model}: not in models.json — nothing to place", flush=True)
+        return False
+    entry = dict(entry)
+    entry["backend"] = "runpod"
+    entry["runpod"] = {"mode": "pods", "engine": engine, "min_vram_gb": vram,
+                       "max_hourly_usd": usd, "max_pods": 1, "idle_timeout_s": 120}
+    data[section][index] = entry
+    _save_models(data)
+    os.utime(MODELS, None)           # the front watches this mtime to push a reload
+    return True
+
+
+def _wait_for_engine(model: str) -> str:
+    """Block until the SERVING process reports this model pinned to a pod.
+
+    Returns '' on success, or why it never happened.
+    """
+    deadline = time.time() + CONFIG_TIMEOUT_S
+    last = {}
+    while time.time() < deadline:
+        state = _curl(f"/v1/models/test/state?model={model}", timeout=30)
+        last = state
+        if state.get("placement") == "pod":
+            return ""
+        time.sleep(5)
+    return (f"the engine never picked up the config for {model!r} within "
+            f"{CONFIG_TIMEOUT_S}s (it is pushed only when the engine is idle). "
+            f"Last seen: placement={last.get('placement')!r} "
+            f"backend={last.get('backend')!r} "
+            f"has_runpod_block={last.get('has_runpod_block')!r}")
+
+
+def _reap() -> None:
+    """Terminate every pod this account has. A capability pod idles for 20
+    minutes before the reaper takes it, which is real money between tests."""
+    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    try:
+        from codai.config import ConfigManager
+        from codai.api.runpod_client import RunpodClient
+        import codai.api.runpod_worker as rw
+        cm = ConfigManager(CONFIG_DIR)
+        cm.load()
+        client = RunpodClient(cm.config.runpod)
+        for pod in client.list_pods() or []:
+            client.terminate_pod(pod["id"])
+            rw.unregister_pod(pod["id"])
+            print(f"   reaped {pod['id']} {pod.get('name')}", flush=True)
+    except Exception as exc:
+        print(f"   could not reap pods: {exc}", flush=True)
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("models", nargs="+")
+    ap.add_argument("--engine", default="auto",
+                    help="auto | vllm | llamacpp | coderai (default: auto)")
+    ap.add_argument("--vram", type=float, default=16.0, help="min VRAM GB")
+    ap.add_argument("--usd", type=float, default=1.0, help="max $/hr per pod")
+    ap.add_argument("--keep-pods", action="store_true",
+                    help="do not terminate pods after each test")
+    args = ap.parse_args()
+
+    backup = MODELS + ".testrun-backup"
+    shutil.copy(MODELS, backup)
+    restored = False
+
+    def restore(*_):
+        nonlocal restored
+        if not restored:
+            shutil.copy(backup, MODELS)
+            os.utime(MODELS, None)
+            os.remove(backup)
+            restored = True
+            print("== models.json restored", flush=True)
+
+    signal.signal(signal.SIGTERM, lambda *a: (restore(), sys.exit(1)))
+
+    results = []
+    try:
+        for model in args.models:
+            print(f"================ {model} ================", flush=True)
+            if not _pin_to_runpod(model, args.engine, args.vram, args.usd):
+                results.append((model, {"error": "not in models.json"}))
+                continue
+            problem = _wait_for_engine(model)
+            if problem:
+                print(f"   {problem}", flush=True)
+                results.append((model, {"error": problem}))
+                continue
+            result = _curl("/v1/models/test", {"model": model, "where": "runpod"})
+            for key in ("capability", "where", "target", "ran", "status", "ok",
+                        "seconds", "sample", "detail", "note", "error"):
+                if result.get(key) not in (None, ""):
+                    print(f"   {key:10} {result[key]}", flush=True)
+            for pod in result.get("pods") or []:
+                print(f"   pod        {pod.get('pod')} {pod.get('gpu')} "
+                      f"{pod.get('state')} ${pod.get('usd_so_far')} "
+                      f"serves={len(pod.get('serves') or [])}", flush=True)
+            results.append((model, result))
+            # Put the catalogue back BEFORE the next model, so a crash between
+            # tests cannot leave a model pinned to a pod that no longer exists.
+            shutil.copy(backup, MODELS)
+            os.utime(MODELS, None)
+            if not args.keep_pods:
+                _reap()
+    finally:
+        restore()
+
+    print("\n================ SUMMARY ================", flush=True)
+    for model, result in results:
+        mark = "PASS" if result.get("ok") else "FAIL"
+        note = (result.get("sample") or result.get("detail")
+                or result.get("error") or "")
+        print(f"{mark:5} {model:52.52} {str(result.get('ran', '-')):24.24} "
+              f"{note}"[:190], flush=True)
+    return 0 if all(r.get("ok") for _, r in results) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())