"""Verify the swapped char-rp-fast seat before the gateway alias points at it. The order matters: the seat is proven on its direct port FIRST, and only then does `char-rp-fast` start resolving. That way no consumer ever sees a half-working alias -- which is the reason playbook §4.4 wants a temp port. A temp port was not reachable here (18.26 GiB free against 15.9 GiB of weights plus a 8.5 GiB KV pool), so the substitute is: prove it on :8021 while nothing routes to it, and keep the one-flip rollback to Pfish-6 intact until it passes. Five checks, and each one exists because this seat family has broken in that exact way before: 1. served name + context -- a stale served-name is a silent substitution 2. KV pool -- Pfish-6's 9.114 GB pinning should transfer, because the architecture is identical field for field; if the token count moved, that assumption was wrong 3. prose, non-thinking -- the `<|channel>thought` leak into content, which stacks/gemma4-charrp/README.md warns about and which was measured 3/3 on this recipe without the parser pin 4. vision -- the "vision towers intact" claim, tested rather than inferred from a tensor count 5. tool call (auto) -- the seat advertises gemma4 tool parsing """ import base64 import json import struct import sys import urllib.error import urllib.request import zlib BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8021/v1" MODEL = sys.argv[2] if len(sys.argv) > 2 else None KEY = sys.argv[3] if len(sys.argv) > 3 else None fails = [] def post(path, body, timeout=180): req = urllib.request.Request( BASE + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", **({"Authorization": f"Bearer {KEY}"} if KEY else {})}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.load(r) def get(path, timeout=30): req = urllib.request.Request( BASE + path, headers={**({"Authorization": f"Bearer {KEY}"} if KEY else {})}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.load(r) def png(rgb, w=64, h=64): """Minimal solid-colour PNG, built here so the test needs no asset on disk.""" raw = b"".join(b"\x00" + bytes(rgb) * w for _ in range(h)) def chunk(tag, data): c = tag + data return struct.pack(">I", len(data)) + c + struct.pack(">I", zlib.crc32(c)) return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b"")) # ---- 1. served name + context ------------------------------------------------- print("== 1. served name + context") models = get("/models") ids = [m["id"] for m in models["data"]] mlen = {m["id"]: m.get("max_model_len") for m in models["data"]} print(f" served: {ids}") print(f" max_model_len: {mlen}") if MODEL: if MODEL in ids: print(f" OK '{MODEL}' is served") else: fails.append(f"'{MODEL}' not in served names {ids}") print(f" *** '{MODEL}' NOT SERVED") target = MODEL if MODEL in ids else ids[0] if "Pfish-6" in ids: fails.append("Pfish-6 is STILL served -- the swap did not take") print(" *** Pfish-6 still served") # ---- 3. prose, non-thinking --------------------------------------------------- print("\n== 3. prose, non-thinking (the <|channel>thought leak)") r = post("/chat/completions", { "model": target, "messages": [{"role": "user", "content": "Describe a rain-slicked alley at night in two sentences."}], "max_tokens": 120, }) msg = r["choices"][0]["message"] content = msg.get("content") or "" reasoning = msg.get("reasoning_content") or msg.get("reasoning") print(f" content ({len(content)} chars): {content[:220]!r}") print(f" reasoning_content: {reasoning!r}") if not content.strip(): fails.append("prose: content is empty") print(" *** content EMPTY") elif "<|channel" in content or "channel>thought" in content: fails.append("prose: <|channel>thought prefix leaked into content") print(" *** CHANNEL PREFIX LEAKED into content") elif reasoning: fails.append(f"prose: reasoning_content populated with enable_thinking=false ({len(reasoning)} chars)") print(" *** reasoning_content populated despite enable_thinking=false") else: print(" OK clean prose in content, no reasoning, no channel prefix") # ---- 4. vision ---------------------------------------------------------------- print("\n== 4. vision (towers preserved, tested not inferred)") blue = base64.b64encode(png((30, 60, 200))).decode() try: r = post("/chat/completions", { "model": target, "messages": [{"role": "user", "content": [ {"type": "text", "text": "This image is one flat colour. Name that colour in one word."}, {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{blue}"}}, ]}], "max_tokens": 24, "temperature": 0, }) v = (r["choices"][0]["message"].get("content") or "").strip() print(f" answer: {v!r} (image was solid RGB(30,60,200) = blue)") if "blue" in v.lower(): print(" OK image was decoded and read correctly") elif v: fails.append(f"vision: answered {v!r} for a solid blue image") print(" *** answered, but not blue -- vision path suspect") else: fails.append("vision: empty answer") print(" *** empty answer") except urllib.error.HTTPError as e: body = e.read().decode()[:300] fails.append(f"vision: HTTP {e.code} {body}") print(f" *** HTTP {e.code}: {body}") # ---- 5. tool call ------------------------------------------------------------- print("\n== 5. tool call (auto)") try: r = post("/chat/completions", { "model": target, "messages": [{"role": "user", "content": "What is the weather in Anaheim?"}], "tools": [{"type": "function", "function": { "name": "get_weather", "description": "Get the current weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}}], "tool_choice": "auto", "max_tokens": 120, }) m = r["choices"][0]["message"] tc = m.get("tool_calls") print(f" tool_calls: {json.dumps(tc)[:240] if tc else None}") print(f" content: {(m.get('content') or '')[:120]!r}") if tc and tc[0]["function"]["name"] == "get_weather": args = tc[0]["function"].get("arguments") print(f" OK parsed a get_weather call, arguments={args!r}") else: fails.append("tool call: no parsed get_weather tool_call") print(" *** no parsed tool call (auto tool_choice)") except urllib.error.HTTPError as e: body = e.read().decode()[:300] fails.append(f"tool call: HTTP {e.code} {body}") print(f" *** HTTP {e.code}: {body}") # ---- 6. NaN logits ----------------------------------------------------------- print("\n== 6. logprobs (NaN logits, the router-quant tell)") try: r = post("/completions", { "model": target, "prompt": "Rain on asphalt at midnight.", "max_tokens": 8, "temperature": 0, "logprobs": 1, }) txt = r["choices"][0].get("text") lp = r["choices"][0].get("logprobs") or {} vals = lp.get("token_logprobs") or [] print(f" text: {txt!r}") print(f" token_logprobs: {vals[:5]}") if not (txt or "").strip(): fails.append("logprobs: raw completion decoded to the empty string -- generating, but no text") print(" *** EMPTY raw completion: tokens generated that decode to nothing") elif any(v is None or v != v for v in vals): fails.append("logprobs: NaN/None in token_logprobs") print(" *** NaN in token_logprobs") else: print(" OK finite logprobs, non-empty raw text") except urllib.error.HTTPError as e: body = e.read().decode()[:300] # vLLM cannot serialize NaN, so the 400 IS the positive finding here. if "nan" in body.lower(): fails.append("logprobs: NaN logits -- vLLM refused to serialize them. " "On a MoE this is the router-quantized signature (playbook §3.15)") print(f" *** NaN LOGITS: {body}") else: fails.append(f"logprobs: HTTP {e.code} {body}") print(f" *** HTTP {e.code}: {body}") print("\n" + "=" * 60) if fails: print(f"FAILED ({len(fails)}):") for f in fails: print(f" - {f}") sys.exit(1) print("ALL CHECKS PASSED")