TTS-Slots (Piper CPU, Qwen3-TTS auf Ada), M4000 außer Betrieb, Engine-Grenzen im Placement (ADR-0006)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

core/tests/test_artifacts.py

@@ -165,7 +165,7 @@ def test_proxmox_site_consistent_with_registry_and_prometheus():
         assert vm["driver_generation"] in {"legacy", "modern"}
     gpus = [vm["gpu"] for vm in vms.values()]
     assert len(gpus) == len(set(gpus))  # keine GPU doppelt durchgereicht
-    by_vmid = {str(vm["vmid"]): (name, vm) for name, vm in vms.items()}
+    by_vmid = {str(vm["vmid"]): (name, vm) for name, vm in {**vms, **site["containers"]}.items()}
     for slot, spec in reg["slots"].items():
         ops = spec.get("ops")
         if not ops:

core/tests/test_inventory_validate.py

@@ -41,7 +41,14 @@ def test_placement_knaster():
     assert all("P2000" in a.gpu_name for a in by_role["gpu-embed"] + by_role["gpu-small"])
     assert by_role["infra"][0].mode == "cpu"
     assert not p.unplaced
-    assert len(p.free_gpus) == 4 and all("M4000" in g for g in p.free_gpus)
+    # M4000 bewusst außer Betrieb (ADR-0006): weder frei noch belegt
+    assert p.free_gpus == []
+    assert len(p.disabled_gpus) == 4 and all("M4000" in g for g in p.disabled_gpus)
+    tts = {a.role: a for a in p.assignments if a.role.startswith("tts")}
+    assert tts["tts-hq"].mode == "shared" and tts["tts-hq"].node == "pve-ada"
+    assert "qwen3-tts-1.7b" in tts["tts-hq"].models_fitting
+    assert tts["tts-cpu"].node == by_role["infra"][0].node
+    assert tts["tts-cpu"].models_fitting == ["piper-de-thorsten-medium"]
 
 
 def test_placement_other_cluster():
@@ -49,7 +56,30 @@ def test_placement_other_cluster():
     p = plan(load_nodes(REPO_ROOT / "sites/example-single-gpu/inventory.yaml"), roles, models)
     modes = {(a.role, a.mode) for a in p.assignments}
     assert ("gpu-main", "gpu") in modes and ("gpu-embed", "cpu_fallback") in modes
-    assert [u.role for u in p.unplaced] == ["gpu-flex"] and p.unplaced[0].optional
+    assert [u.role for u in p.unplaced] == ["gpu-flex", "tts-hq"] and all(u.optional for u in p.unplaced)
+
+
+def test_pytorch_never_on_legacy_gpus():
+    """Engine-Grenze: PyTorch-Modelle (TTS) landen nie auf Maxwell/Pascal, llama.cpp schon (ADR-0006)."""
+    from knaster.inventory.placement import RoleSpec
+
+    _, models = load_catalog(REPO_ROOT / "catalog")
+    assert {m.name: m.min_cc for m in models}["qwen3-tts-0.6b"] >= 7.5
+    role = RoleSpec(name="tts-gpu", needs={"gpu": True, "engine": "pytorch"}, kinds=["tts"], optional=True)
+    legacy = [NodeProfile(name="old", memory_gb=32, gpus=[
+        {"index": 0, "name": "Quadro M4000", "memory_gb": 8, "compute_cap": 5.2},
+        {"index": 1, "name": "Quadro P2000", "memory_gb": 5, "compute_cap": 6.1}])]
+    p = plan(legacy, [role], models)
+    assert [u.role for u in p.unplaced] == ["tts-gpu"]
+
+
+def test_disabled_gpus_are_ignored():
+    roles, models = load_catalog(REPO_ROOT / "catalog")
+    nodes = [NodeProfile(name="n", memory_gb=32, gpus=[
+        {"index": 0, "name": "Quadro P2000", "memory_gb": 5, "compute_cap": 6.1, "disabled": True}])]
+    p = plan(nodes, roles, models)
+    assert all(a.mode != "gpu" for a in p.assignments)
+    assert p.disabled_gpus == ["n/gpu0 Quadro P2000"]
 
 
 def test_placement_maxwell_never_gets_vllm():

core/tests/test_tts.py

@@ -0,0 +1,116 @@
+import io
+import json
+import struct
+import sys
+import wave
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+
+from knaster.audio import rtf, wav_duration
+from knaster.bench import bench_tts
+from knaster.canary import CanaryProber
+from knaster.llm import OpenAIChatClient
+from knaster.tts_server import CommandBackend, create_tts_app
+
+
+def make_wav(seconds: float, rate: int = 22050) -> bytes:
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(rate)
+        w.writeframes(struct.pack("<h", 0) * int(seconds * rate))
+    return buf.getvalue()
+
+
+FAKE_PIPER = """
+import sys, wave, struct
+args = sys.argv[1:]
+out = args[args.index("-f") + 1]
+model = args[args.index("-m") + 1]
+text = sys.stdin.read()
+if "FEHLER" in text:
+    print("kaputt", file=sys.stderr); sys.exit(3)
+with wave.open(out, "wb") as w:
+    w.setnchannels(1); w.setsampwidth(2); w.setframerate(16000)
+    w.writeframes(struct.pack("<h", 0) * (160 * len(text)))   # 10 ms pro Zeichen
+"""
+
+
+@pytest.fixture
+def backend(tmp_path):
+    script = tmp_path / "fake_piper.py"
+    script.write_text(FAKE_PIPER)
+    return CommandBackend({"thorsten": "/voices/t.onnx", "kerstin": "/voices/k.onnx"},
+                          cmd=f"{sys.executable} {script} -m {{model}} -f {{out}}")
+
+
+def test_wav_helpers():
+    assert wav_duration(make_wav(1.5)) == pytest.approx(1.5)
+    assert wav_duration(b"kein wav") == 0.0
+    assert rtf(3.0, 1.5) == 2.0 and rtf(1.0, 0) is None
+
+
+def test_tts_server(backend):
+    client = TestClient(create_tts_app(backend, token="t"))
+    assert client.get("/health").json()["default"] == "thorsten"
+    H = {"Authorization": "Bearer t"}
+    r = client.post("/v1/audio/speech", json={"input": "Hallo Welt"}, headers=H)
+    assert r.status_code == 200 and r.headers["content-type"] == "audio/wav"
+    assert wav_duration(r.content) == pytest.approx(0.1)
+    assert client.post("/v1/audio/speech", json={"input": "x", "voice": "kerstin"}, headers=H).status_code == 200
+    assert client.post("/v1/audio/speech", json={"input": "x", "voice": "nope"}, headers=H).status_code == 400
+    assert client.post("/v1/audio/speech", json={"input": "FEHLER"}, headers=H).status_code == 502
+    assert client.post("/v1/audio/speech", json={"input": "x", "response_format": "mp3"},
+                       headers=H).status_code == 400
+    assert client.post("/v1/audio/speech", json={"input": "x"}).status_code == 401
+
+
+def test_backend_requires_voice():
+    with pytest.raises(ValueError):
+        CommandBackend({})
+
+
+def tts_transport(seconds: float = 2.0, fail: bool = False):
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/v1/audio/speech"
+        body = json.loads(request.content)
+        if fail or body["model"] == "kaputt":
+            return httpx.Response(503, text="down")
+        return httpx.Response(200, content=make_wav(seconds), headers={"content-type": "audio/wav"})
+
+    return httpx.Client(transport=httpx.MockTransport(handler))
+
+
+def test_bench_tts():
+    client = OpenAIChatClient("http://x", client=tts_transport(2.0))
+    result = bench_tts("http://x", "tts-de", runs=2, client=client)
+    assert len(result["runs"]) == 2 and result["errors"] == []
+    assert result["runs"][0]["audio_seconds"] == 2.0 and result["rtf_median"] is not None
+    bad = bench_tts("http://x", "kaputt", runs=1, client=client)
+    assert bad["runs"] == [] and bad["errors"]
+
+
+def test_canary_probes_tts_slots(registry):
+    from prometheus_client import CollectorRegistry
+
+    from knaster.registry import Registry
+    from knaster.store import Store
+
+    from .conftest import REGISTRY
+
+    data = {"slots": {**REGISTRY["slots"], "tts-de": {
+        "tier": 1, "device": "cpu", "engine": "piper", "litellm_model": "tts-de", "base_model": "piper",
+        "max_context": 5000, "kind": "tts"}}}
+    reg = Registry.from_dict(data, Store("sqlite://"))
+    chat = OpenAIChatClient("http://x", client=tts_transport(0.8))
+    prober = CanaryProber(reg, chat, metrics_registry=CollectorRegistry())
+    tts = next(prober.probe(s) for s in reg.slots() if s.name == "tts-de")
+    assert tts.up and tts.ok
+    silent = OpenAIChatClient("http://x", client=tts_transport(0.05))
+    assert not CanaryProber(reg, silent, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).ok
+    down = OpenAIChatClient("http://x", client=tts_transport(fail=True))
+    assert not CanaryProber(reg, down, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).up
+