"""intern-decision-serve acceptance: the service-level checks the bench harness does not cover. Stdlib only, so it runs on fv-ml1's host python or on nh3-dev against a live instance. python3 checks.py --url URL --token-file T --out out.json --checks auth,chunking,queue,maxreq python3 checks.py --url URL --token-file T --out oom.json --checks oom auth 401 without / with a wrong token on both POSTs; /health open chunking 20 decisions in one /decide/shared equal the same decisions asked as 16 + 4 queue past MAX_QUEUE requests in progress, the rest get 429 busy (and the admitted all answer) maxreq the largest request the API accepts: MAX_DECISIONS decisions x 16 options, the state grown until each 16-question call sits just under MAX_TOKENS (found from the model's own 422 message). Sent --repeats times; status, tokens and /health's allocator peak recorded. maxone ONE binary question with the state filling MAX_TOKENS: --repeats x 200, one token more is 422 fit under the cap, the largest 16-question call that answers 200 (binary search, in tokens) oom the over-cap request (maxreq's body when it ran) -> 503 out_of_memory, then reserved memory back to its resting value and an ordinary request answered 200. Earlier text: a long request against a deliberately tight cap -> 503 out_of_memory, then reserved memory back to its resting value and an ordinary request answered 200. Every request's status, timing and the fields that prove the point are kept in --out. """ from __future__ import annotations import argparse import json import re import threading import time import urllib.error import urllib.request from pathlib import Path HERE = Path(__file__).resolve().parent YESNO = [{"id": "yes", "description": "Yes"}, {"id": "no", "description": "No"}] def post(url, body, headers, timeout=600): req = urllib.request.Request(url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", **headers}) t = time.perf_counter() try: with urllib.request.urlopen(req, timeout=timeout) as r: status, raw = r.status, r.read() except urllib.error.HTTPError as e: status, raw = e.code, e.read() ms = (time.perf_counter() - t) * 1000 try: return status, json.loads(raw), ms except ValueError: return status, {"raw": raw[:300].decode(errors="replace")}, ms def get(url): with urllib.request.urlopen(url, timeout=30) as r: return r.status, json.loads(r.read()) def criteria(n, options=YESNO, prefix="c"): return [{"id": f"{prefix}{i}", "question": f"Does the evidence mention item number {i}?", "options": options} for i in range(n)] def check_auth(url, h): out = {} body = {"id": "a", "state": "s", "question": "q?", "options": YESNO} for path, b in (("/decide", body), ("/decide/shared", {"state": "s", "decisions": [body]})): out[path] = {"no_token": post(url + path, b, {})[0], "wrong_token": post(url + path, b, {"Authorization": "Bearer " + "x" * 48})[0], "right_token": post(url + path, b, h)[0]} out["health_no_token"] = get(url + "/health")[0] out["pass"] = all(v["no_token"] == 401 and v["wrong_token"] == 401 and v["right_token"] == 200 for k, v in out.items() if k.startswith("/")) and out["health_no_token"] == 200 return out def check_chunking(url, h, state): crits = [{"id": f"d{i}", "question": q, "options": opts} for i, (q, opts) in enumerate([ (f"Is item {i} a {kind}?", [{"id": "yes", "description": "Yes"}, {"id": "no", "description": "No"}, {"id": "unclear", "description": "Cannot tell from the evidence"}]) for i, kind in enumerate(["policy", "exception", "deadline", "person", "number"] * 4)])] s_all, j_all, ms_all = post(url + "/decide/shared", {"state": state, "decisions": crits}, h) s_a, j_a, _ = post(url + "/decide/shared", {"state": state, "decisions": crits[:16]}, h) s_b, j_b, _ = post(url + "/decide/shared", {"state": state, "decisions": crits[16:]}, h) split = (j_a.get("results", []) + j_b.get("results", [])) if s_a == s_b == 200 else [] rows = [] for whole, part in zip(j_all.get("results", []), split): rows.append({"id": whole["id"], "top_whole": whole["top"], "top_split": part["top"], "max_dp": max(abs(x - y) for x, y in zip(whole["probabilities"], part["probabilities"])), "call_whole": whole["call"], "sha_equal": whole["prompt_sha256"] == part["prompt_sha256"]}) t = j_all.get("timing", {}) return {"status": [s_all, s_a, s_b], "timing": t, "rows": rows, "e2e_ms": round(ms_all, 1), "pass": s_all == s_a == s_b == 200 and t.get("calls") == 2 and t.get("questions_per_call") == [16, 4] and len(rows) == 20 and all(r["top_whole"] == r["top_split"] and r["max_dp"] == 0 and r["sha_equal"] for r in rows)} def check_queue(url, h, state, max_queue, extra): n = max_queue + extra body = {"state": state, "decisions": criteria(16)} barrier = threading.Barrier(n) results = [None] * n def one(i): barrier.wait() s, j, ms = post(url + "/decide/shared", body, h) results[i] = (s, (j.get("error") or {}).get("code"), round(ms, 1)) threads = [threading.Thread(target=one, args=(i,)) for i in range(n)] for t in threads: t.start() for t in threads: t.join() statuses = [r[0] for r in results] return {"sent": n, "max_queue": max_queue, "ok": statuses.count(200), "busy_429": statuses.count(429), "codes_429": sorted({r[1] for r in results if r[0] == 429}), "other": [r for r in results if r[0] not in (200, 429)], "pass": statuses.count(429) >= 1 and statuses.count(200) + statuses.count(429) == n and statuses.count(200) <= max_queue} def question_block(i): """Zero-padded numbers: Qwen tokenises digits one by one, so every 16-question call of a 64-decision request has the same token count and each one sits at the limit.""" return {"id": f"q{i:02d}", "question": f"Which of the listed categories best describes clause {i:02d} of the policy?", "options": [{"id": f"k{j:02d}", "description": f"Category {j:02d}: the clause concerns topic number {j:02d}"} for j in range(16)]} def find_max_state(url, h, base_state, max_tokens, n_decisions=16, decisions=None): """Grow the state (repeating base_state) until the request is refused with 422, using the token count the model's own 422 reports; return the longest state the API accepts and its tokens per call. Sized with the WHOLE n_decisions request, so the largest call in it is the one that decides.""" decisions = decisions or [question_block(i) for i in range(n_decisions)] text = (base_state + "\n\n") * 4 def tokens_for(chars): s, j, _ = post(url + "/decide/shared", {"state": text[:chars], "decisions": decisions}, h) if s == 200: return s, j["timing"]["input_tokens"][0] if s == 503: return s, None m = re.search(r"has (\d+) tokens", (j.get("error") or {}).get("message", "")) return s, int(m.group(1)) if m else None lo, hi = 1000, len(text) s, tok = tokens_for(hi) assert s == 422 and tok and tok > max_tokens, (s, tok) best = None while lo < hi: mid = (lo + hi + 1) // 2 s, tok = tokens_for(mid) if s in (200, 503): # accepted by the API (503 = accepted, then refused by the VRAM cap) lo, best = mid, tok elif s == 422: hi = mid - 1 else: raise RuntimeError(f"unexpected {s} while sizing the max request: {tok}") return text[:lo], best MAX_BODY = {} def check_maxreq(url, h, base_state, max_tokens, max_decisions, repeats): state, tokens = find_max_state(url, h, base_state, max_tokens, max_decisions) decisions = [question_block(i) for i in range(max_decisions)] body = {"state": state, "decisions": decisions} MAX_BODY.update(body) size = len(json.dumps(body).encode()) runs = [] for _ in range(repeats): s, j, ms = post(url + "/decide/shared", body, h) health = get(url + "/health")[1]["model"] runs.append({"status": s, "e2e_ms": round(ms, 1), "code": (j.get("error") or {}).get("code"), "calls": (j.get("timing") or {}).get("calls"), "input_tokens": (j.get("timing") or {}).get("input_tokens"), "reserved_gib_after": health.get("reserved_gib"), "max_reserved_gib": health.get("max_reserved_gib"), "t_end": time.time()}) return {"state_chars": len(state), "tokens_per_call": tokens, "decisions": max_decisions, "options_each": 16, "body_bytes": size, "runs": runs} def check_fit(url, h, base_state): """Under the cap: the largest 16-question call (16 options each) that still answers 200, in tokens.""" decisions = [question_block(i) for i in range(16)] text = (base_state + "\n\n") * 4 lo, hi, best, probes = 1000, len(text), None, [] while lo < hi: mid = (lo + hi + 1) // 2 s, j, _ = post(url + "/decide/shared", {"state": text[:mid], "decisions": decisions}, h) probes.append([mid, s, j["timing"]["input_tokens"][0] if s == 200 else None]) if s == 200: lo, best = mid, j["timing"]["input_tokens"][0] else: hi = mid - 1 return {"largest_ok_state_chars": lo, "largest_ok_tokens": best, "probes": probes} def check_maxone(url, h, base_state, max_tokens, repeats): """The other extreme of the largest call: ONE binary question, the state filling the rest of MAX_TOKENS.""" one = [{"id": "only", "question": "Does the policy allow the exception described?", "options": YESNO}] state, tokens = find_max_state(url, h, base_state, max_tokens, decisions=one) over = post(url + "/decide/shared", {"state": state + " extra words", "decisions": one}, h) runs = [] for _ in range(repeats): s, j, ms = post(url + "/decide", {"id": "only", "state": state, **{k: one[0][k] for k in ("question", "options")}}, h) runs.append({"status": s, "e2e_ms": round(ms, 1), "input_tokens": j.get("input_tokens"), "code": (j.get("error") or {}).get("code"), "t_end": time.time()}) return {"state_chars": len(state), "tokens": tokens, "one_more_is": [over[0], (over[1].get("error") or {}).get("message")], "runs": runs, "pass": all(r["status"] == 200 for r in runs) and over[0] == 422} def check_oom(url, h, long_state): """The over-cap request: the largest one (maxreq) if it ran, else 16 criteria over the long state.""" before = get(url + "/health")[1]["model"] body = dict(MAX_BODY) or {"state": long_state, "decisions": criteria(16)} s, j, ms = post(url + "/decide/shared", body, h) after = get(url + "/health")[1]["model"] s2, j2, _ = post(url + "/decide", {"id": "after", "state": "The deploy passed.", "question": "Did it pass?", "options": YESNO}, h) again = get(url + "/health")[1]["model"] return {"oom_status": s, "oom_error": j.get("error"), "oom_ms": round(ms, 1), "reserved_gib": {"before": before.get("reserved_gib"), "after_oom": after.get("reserved_gib"), "after_next_request": again.get("reserved_gib")}, "next_request_status": s2, "next_top": j2.get("top"), "pass": s == 503 and (j.get("error") or {}).get("code") == "out_of_memory" and s2 == 200 and after.get("reserved_gib") is not None and after["reserved_gib"] <= before["reserved_gib"] + 0.5} def main(): ap = argparse.ArgumentParser() ap.add_argument("--url", required=True) ap.add_argument("--token-file", required=True) ap.add_argument("--out", required=True) ap.add_argument("--checks", default="auth,chunking,queue,maxreq") ap.add_argument("--long-state-file", default=str(HERE / "long_state.txt")) ap.add_argument("--max-queue", type=int, default=32) ap.add_argument("--queue-extra", type=int, default=16) ap.add_argument("--repeats", type=int, default=3) args = ap.parse_args() url = args.url.rstrip("/") h = {"Authorization": "Bearer " + Path(args.token_file).read_text().strip()} long_state = Path(args.long_state_file).read_text() health = get(url + "/health")[1] report = {"url": url, "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "health": health} for c in args.checks.split(","): t = time.time() report[c] = {"auth": lambda: check_auth(url, h), "chunking": lambda: check_chunking(url, h, long_state[:4000]), "queue": lambda: check_queue(url, h, long_state, args.max_queue, args.queue_extra), "maxreq": lambda: check_maxreq(url, h, long_state, health["max_tokens"], health["max_decisions"], args.repeats), "fit": lambda: check_fit(url, h, long_state), "maxone": lambda: check_maxone(url, h, long_state, health["max_tokens"], args.repeats), "oom": lambda: check_oom(url, h, long_state)}[c]() report[c]["t_start"], report[c]["t_end"] = t, time.time() print(c, json.dumps({k: v for k, v in report[c].items() if k not in ("rows", "runs")})[:600], flush=True) report["finished_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) Path(args.out).write_text(json.dumps(report, indent=1)) if __name__ == "__main__": main()