"""Unified 100-task Base vs. WSP v3.0 (9 Subspaces) benchmark & VRAM profiler.
Evaluates all 5 candidate models on RTX 4070 Ti (12 GB) and models RTX 2060 Mobile (6 GB) fit:
- 30 disjoint holdout tasks across S1..S9
- 70 established tasks (20 GSM8K, 15 HumanEval, 15 BFCL Irrelevance, 20 NL2Bash)
- Skips any task already recorded in pipeline_results.json (reusing the deterministic 50-task slice)
- Turn-2 token count & latency/throughput profiling
- Delta VRAM profiling via nvidia-smi (-c 16384 all models, -c 4096 probe on 9B,
plus co-resident 2-tier cascade probe qwen35_2b + qwen35_9b)
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
PIPELINE_DIR = Path(r"C:\Users\jhjwi\.gemini\antigravity\scratch\wsp-5model-pipeline")
sys.path.insert(0, str(PIPELINE_DIR))
from dynamic_router_9s import DynamicSubspaceRouter9S, SUBSPACE_NAMES # noqa: E402
from model_registry import MODEL_SPECS, ModelSpec # noqa: E402
import wsp_tasks_9subspaces as tasks # noqa: E402
SUBSPACE_FILES = [
"wsp_s1_think.gguf",
"wsp_s2_restraint.gguf",
"wsp_s3_router.gguf",
"wsp_s4_mid_code.gguf",
"wsp_s5_json_schema.gguf",
"wsp_s6_handoff.gguf",
"wsp_s7_injection_guard.gguf",
"wsp_s8_grounding.gguf",
"wsp_s9_error_recovery.gguf",
]
LLAMA_SERVER_EXE = Path(r"C:\Users\jhjwi\.gemini\antigravity\scratch\qwen38-27b-wsp\llama.cpp\llama-server.exe")
ESTABLISHED_CACHE = Path(r"C:\Users\jhjwi\.gemini\antigravity\scratch\gemma4-e2b-agent\established_datasets_cache.json")
RESULTS_PATH = PIPELINE_DIR / "pipeline_results.json"
BFCL_TYPE_MAP = {
"dict": "object",
"float": "number",
"int": "integer",
"bool": "boolean",
"tuple": "array",
}
def get_gpu_vram_mb() -> Tuple[int, int, int]:
"""Return (total_mb, used_mb, free_mb) via nvidia-smi."""
out = subprocess.check_output(
["nvidia-smi", "--query-gpu=memory.total,memory.used,memory.free", "--format=csv,noheader,nounits"],
text=True,
timeout=10,
).strip().splitlines()[0]
tot, used, free = [int(x.strip()) for x in out.split(",")]
return tot, used, free
def http_get_json(url: str, timeout: float = 10.0) -> Any:
with urllib.request.urlopen(url, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
class LlamaServerHandle:
"""Managed llama-server subprocess with clean cleanup and VRAM tracking."""
def __init__(self, spec: ModelSpec, port: int, ctx_size: int, load_loras: bool = True):
self.spec = spec
self.port = port
self.ctx_size = ctx_size
self.load_loras = load_loras
self.proc: Optional[subprocess.Popen] = None
self.log_fp = None
self.base_url = f"http://127.0.0.1:{port}"
self.vram_info: Dict[str, Any] = {}
def __enter__(self) -> "LlamaServerHandle":
tot0, used0, free0 = get_gpu_vram_mb()
log_path = self.spec.work_dir / f"llama_server_p{self.port}_c{self.ctx_size}.log"
self.log_fp = open(log_path, "w", encoding="utf-8", errors="replace")
cmd = [
str(LLAMA_SERVER_EXE),
"-m",
str(self.spec.gguf_path),
"--host",
"127.0.0.1",
"--port",
str(self.port),
"-ngl",
"99",
"-np",
"1",
"-c",
str(self.ctx_size),
"-ctk",
"q8_0",
"-ctv",
"q8_0",
"-b",
"512",
"-ub",
"256",
"--jinja",
]
if self.spec.reasoning_format:
cmd += ["--reasoning-format", self.spec.reasoning_format]
if self.load_loras:
cmd.append("--lora-init-without-apply")
for fname in SUBSPACE_FILES:
lora_p = self.spec.adapter_dir / fname
if not lora_p.exists():
raise FileNotFoundError(f"Missing LoRA adapter: {lora_p}")
cmd += ["--lora", str(lora_p)]
t0 = time.time()
self.proc = subprocess.Popen(cmd, stdout=self.log_fp, stderr=subprocess.STDOUT)
deadline = time.time() + 180.0
ready = False
last_err = ""
while time.time() < deadline:
if self.proc.poll() is not None:
self.log_fp.flush()
tail = log_path.read_text(encoding="utf-8", errors="replace")[-1500:]
raise RuntimeError(f"llama-server exited early (code={self.proc.returncode}) for {self.spec.slug}:\n{tail}")
try:
h = http_get_json(f"{self.base_url}/health", timeout=3.0)
if h.get("status") == "ok":
ready = True
break
except Exception as e:
last_err = str(e)
time.sleep(0.5)
if not ready:
self.log_fp.flush()
tail = log_path.read_text(encoding="utf-8", errors="replace")[-1500:]
raise TimeoutError(f"llama-server health timeout for {self.spec.slug}: {last_err}\n{tail}")
loaded_loras = http_get_json(f"{self.base_url}/lora-adapters", timeout=10.0) if self.load_loras else []
if self.load_loras and len(loaded_loras) != 9:
raise RuntimeError(f"Expected 9 LoRAs on {self.spec.slug}, got {len(loaded_loras)}")
tot1, used1, free1 = get_gpu_vram_mb()
self.vram_info = {
"ctx_size": self.ctx_size,
"gpu_total_mb": tot1,
"baseline_used_mb": used0,
"baseline_free_mb": free0,
"loaded_used_mb": used1,
"loaded_free_mb": free1,
"delta_load_mb": max(used1 - used0, free0 - free1),
"peak_used_mb": used1,
"peak_delta_mb": max(used1 - used0, free0 - free1),
"startup_s": round(time.time() - t0, 2),
"lora_count": len(loaded_loras),
}
return self
def sample_peak_vram(self) -> None:
_, used, _ = get_gpu_vram_mb()
if used > self.vram_info.get("peak_used_mb", 0):
self.vram_info["peak_used_mb"] = used
self.vram_info["peak_delta_mb"] = max(
self.vram_info["peak_delta_mb"],
used - self.vram_info["baseline_used_mb"],
)
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if self.proc is not None:
try:
if self.proc.poll() is None:
self.proc.terminate()
try:
self.proc.wait(timeout=8)
except subprocess.TimeoutExpired:
self.proc.kill()
self.proc.wait(timeout=5)
except Exception:
pass
if self.log_fp is not None:
try:
self.log_fp.close()
except Exception:
pass
# ---------------------------------------------------------------------------
# Response & tool-call parsing helpers
# ---------------------------------------------------------------------------
def strip_thinking(text: str) -> str:
if not text:
return ""
s = re.sub(r".*?", "", text, flags=re.S)
s = re.sub(r"<\|channel>thought.*?", "", s, flags=re.S)
return s.strip()
def _parse_kv_gemma4(body: str) -> Dict[str, Any]:
"""Parse Gemma 4 tool call argument body such as key:<|"|>val<|"|>,k2:123."""
args: Dict[str, Any] = {}
pos = 0
n = len(body)
while pos < n:
while pos < n and body[pos] in " \t\r\n,":
pos += 1
if pos >= n:
break
m_key = re.match(r'([A-Za-z0-9_.\-]+|"[^"]+")\s*:\s*', body[pos:])
if not m_key:
break
k = m_key.group(1).strip('"')
pos += m_key.end()
if body.startswith('<|"|>', pos):
pos += 5
end = body.find('<|"|>', pos)
if end == -1:
val = body[pos:]
pos = n
else:
val = body[pos:end]
pos = end + 5
args[k] = val
elif pos < n and body[pos] in ('"', "'"):
q = body[pos]
pos += 1
buf = []
while pos < n:
ch = body[pos]
if ch == "\\" and pos + 1 < n:
buf.append(body[pos + 1])
pos += 2
continue
if ch == q:
pos += 1
break
buf.append(ch)
pos += 1
args[k] = "".join(buf)
elif pos < n and body[pos] in "{[":
open_ch = body[pos]
close_ch = "}" if open_ch == "{" else "]"
depth = 0
start = pos
while pos < n:
if body[pos] == open_ch:
depth += 1
elif body[pos] == close_ch:
depth -= 1
if depth == 0:
pos += 1
break
pos += 1
raw_v = body[start:pos]
try:
args[k] = json.loads(raw_v)
except Exception:
args[k] = raw_v
else:
m_val = re.match(r"[^,}\r\n]+", body[pos:])
raw_v = m_val.group(0).strip() if m_val else ""
pos += len(m_val.group(0)) if m_val else 0
if raw_v.lower() == "true":
args[k] = True
elif raw_v.lower() == "false":
args[k] = False
elif raw_v.lower() == "null":
args[k] = None
else:
try:
args[k] = int(raw_v) if re.fullmatch(r"-?\d+", raw_v) else float(raw_v)
except Exception:
args[k] = raw_v
return args
def parse_tool_calls(res: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Extract normalized [{'name': str, 'arguments': dict}] across OpenAI / Qwen-XML / Gemma4."""
out: List[Dict[str, Any]] = []
native = res.get("tool_calls") or []
for tc in native:
fn = tc.get("function", tc)
name = (fn.get("name") or "").strip()
raw_args = fn.get("arguments", {})
if isinstance(raw_args, dict):
args = raw_args
elif isinstance(raw_args, str):
try:
parsed = json.loads(raw_args) if raw_args.strip() else {}
args = parsed if isinstance(parsed, dict) else {"_raw": raw_args}
except Exception:
args = {"_raw": raw_args}
else:
args = {}
if name:
out.append({"name": name, "arguments": args})
if out:
return out
raw_text = (res.get("content") or "") + "\n" + (res.get("reasoning") or "")
# Qwen3.5 / Nemotron XML fallback in content (ignore inside thinking unless tool_call only in content)
content = strip_thinking(res.get("content") or "")
for m in re.finditer(r"\s*(.*?)(?:||$)", content, flags=re.S):
name = m.group(1).strip()
inner = m.group(2)
args: Dict[str, Any] = {}
for pm in re.finditer(r"\s*(.*?)\s*", inner, flags=re.S):
pk = pm.group(1).strip()
pv = pm.group(2).strip()
if (pv.startswith("{") and pv.endswith("}")) or (pv.startswith("[") and pv.endswith("]")):
try:
args[pk] = json.loads(pv)
continue
except Exception:
pass
args[pk] = pv
if name:
out.append({"name": name, "arguments": args})
if out:
return out
# Gemma 4 native <|tool_call>call:NAME{...} fallback
for m in re.finditer(r"<\|tool_call>call:([A-Za-z0-9_.\-]+)\{(.*?)\}(?:|$)", content + "\n" + raw_text, flags=re.S):
name = m.group(1).strip()
args = _parse_kv_gemma4(m.group(2))
if name:
out.append({"name": name, "arguments": args})
break # avoid double matching across content + raw_text
if out:
return out
# Explicit {...} JSON fallback
for m in re.finditer(r"\s*(\{.*?\})\s*", content, flags=re.S):
try:
obj = json.loads(m.group(1))
if isinstance(obj, dict) and obj.get("name"):
args = obj.get("arguments", {})
if isinstance(args, str):
args = json.loads(args)
out.append({"name": obj["name"], "arguments": args if isinstance(args, dict) else {}})
except Exception:
pass
return out
def extract_boxed(text: str) -> str:
clean = strip_thinking(text)
idx = clean.rfind("\\boxed{")
if idx == -1:
idx = text.rfind("\\boxed{")
src = text
else:
src = clean
if idx == -1:
return ""
pos = idx + 7
depth = 1
out = []
while pos < len(src) and depth > 0:
ch = src[pos]
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
break
out.append(ch)
pos += 1
return "".join(out).strip()
def norm_num(s: str) -> str:
if not s:
return ""
s = s.strip().replace(",", "").replace("$", "").replace("\\text{", "").rstrip("}")
m = re.search(r"-?\d+(?:\.\d+)?", s)
if not m:
return s.lower()
val = m.group(0)
if "." in val:
f = float(val)
if f.is_integer():
return str(int(f))
return str(f)
return str(int(val))
def extract_json_obj(text: str) -> Optional[Dict[str, Any]]:
clean = strip_thinking(text)
clean = re.sub(r"^```(?:json)?\s*|\s*```$", "", clean, flags=re.M | re.I).strip()
start = clean.find("{")
if start == -1:
return None
for end in range(len(clean), start, -1):
if clean[end - 1] != "}":
continue
try:
obj = json.loads(clean[start:end])
if isinstance(obj, dict):
return obj
except Exception:
continue
return None
def _val_eq(a: Any, b: Any) -> bool:
if a is b:
return True
if a is None or b is None:
return a is None and b is None
if isinstance(b, bool):
return (isinstance(a, bool) and a == b) or (isinstance(a, str) and a.lower() == str(b).lower())
if isinstance(b, (int, float)):
try:
return float(a) == float(b)
except Exception:
return False
return str(a).strip().lower() == str(b).strip().lower()
def extract_python_code(text: str, entry_point: str, prompt_prefix: str = "") -> str:
clean = strip_thinking(text)
blocks = re.findall(r"```(?:python|py)?\s*\n(.*?)```", clean, flags=re.S | re.I)
chosen = ""
for b in blocks:
if re.search(rf"\bdef\s+{re.escape(entry_point)}\b", b):
chosen = b
break
if not chosen and blocks:
chosen = blocks[-1]
if not chosen:
chosen = clean
if re.search(rf"^\s*def\s+{re.escape(entry_point)}\b", chosen, flags=re.M):
# Extract imports from prompt_prefix if missing in chosen
imports = "\n".join(
ln for ln in prompt_prefix.splitlines() if ln.startswith(("from ", "import "))
)
if imports and imports.splitlines()[0] not in chosen:
return imports + "\n\n" + chosen
return chosen
# Otherwise model produced continuation body of HumanEval prompt
body_lines = chosen.splitlines()
if body_lines and not body_lines[0].startswith((" ", "\t")):
body_lines = [(" " + ln if ln.strip() else ln) for ln in body_lines]
return prompt_prefix + "\n" + "\n".join(body_lines)
def run_python_check(full_code: str, test_script: str) -> Tuple[bool, str]:
script = full_code + "\n\n" + test_script + "\nprint('__PASS__')\n"
try:
proc = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
timeout=5,
)
ok = proc.returncode == 0 and "__PASS__" in proc.stdout
err = "" if ok else (proc.stderr or proc.stdout)[-240:]
return ok, err
except subprocess.TimeoutExpired:
return False, "timeout"
except Exception as e:
return False, str(e)
def normalize_bfcl_tools(funcs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
def walk(node: Any) -> Any:
if isinstance(node, dict):
out = {}
for k, v in node.items():
if k == "type" and isinstance(v, str):
out[k] = BFCL_TYPE_MAP.get(v, v)
else:
out[k] = walk(v)
return out
if isinstance(node, list):
return [walk(x) for x in node]
return node
return [{"type": "function", "function": walk(fn)} for fn in funcs]
def eval_nl2bash_cmd(idx: int, cmd: str, gold: str) -> bool:
c = cmd.lower()
if idx == 0:
return "grep" in c and "folder1" in c and "hello" in c
if idx == 1: # grep -rnw /system -e "foo"
return "grep" in c and "/system" in c and "foo" in c
if idx == 2:
return ("tree" in c or "find" in c or "ls" in c) and "/system" in c
if idx == 3: # tree -a /system/folder2
return ("tree" in c or "find" in c or "ls" in c) and "folder2" in c
if idx == 4:
return ("find" in c or "grep" in c) and "/system" in c and ".out" in c
if idx == 5: # yes y | rm -r /system/*.txt
return ("rm" in c or "-delete" in c) and "/system" in c and ".txt" in c
if idx == 6:
return ("find" in c or "rmdir" in c) and "folder3" in c and ("empty" in c or "rmdir" in c or "-delete" in c)
if idx == 7: # find /system/folder3/temp -type d -empty -exec rmdir ...
return "find" in c and "folder3" in c and ("empty" in c or "rmdir" in c or "-delete" in c or "rm " in c)
if idx == 8:
return "find" in c and "folder1" in c and "txt" in c and ("!" in c or "-not" in c) and ("rm" in c or "-delete" in c)
if idx == 9: # find /system/folder1 -type f -exec rename 's/special/regular/'
return ("find" in c or "rename" in c or "for " in c) and "folder1" in c and ("special" in c or "regular" in c)
if idx == 10:
return "grep" in c and "/system" in c and "foo" in c and ("wc" in c or "tee" in c or "-c" in c or "awk" in c)
if idx == 11:
return "find" in c and "folder3" in c and ("basename" in c or "-printf" in c or "awk" in c or "sed" in c)
if idx == 12:
return ("zcat" in c or "tar" in c or "zgrep" in c or "gunzip" in c) and "folder2.tar.gz" in c and "special" in c
if idx == 13:
return "text1.txt" in c and ("sed" in c or "tr" in c or "dos2unix" in c or "\\r" in c)
if idx == 14:
return ("find" in c or "rm" in c) and "folder1" in c and ".doc" in c and ("rm" in c or "-delete" in c)
if idx == 15:
return ("find" in c or "rm" in c) and "folder1" in c and ".log" in c and ("rm" in c or "-delete" in c)
if idx == 16:
return "/system" in c and ".txt" in c and ("rm" in c or "-delete" in c)
if idx == 17:
return "find" in c and "backup_dbg" in c and ".sql" in c and "25" in c and ("rm" in c or "-delete" in c)
if idx == 18:
return "find" in c and "folder1" in c and ".txt" in c and "5" in c and ("rm" in c or "-delete" in c)
if idx == 19:
return "find" in c and "folder1" in c and ".txt" in c and "keep.txt" in c and "5" in c and ("rm" in c or "-delete" in c)
return False
def _normalize_cached_records(existing_records: Optional[List[Dict[str, Any]]]) -> Dict[Tuple[str, str], Dict[str, Any]]:
"""Index previously evaluated records by (suite, canonical_item_id), remapping legacy [1:11:2] indices."""
if not existing_records:
return {}
legacy_gsm = sum(1 for r in existing_records if r.get("suite") == "EST_GSM8K") == 5
legacy_nl = sum(1 for r in existing_records if r.get("suite") == "EST_NL2BASH") == 5
odd_indices = [1, 3, 5, 7, 9]
cached: Dict[Tuple[str, str], Dict[str, Any]] = {}
for rec in existing_records:
r = dict(rec)
suite = r.get("suite", "")
rid = str(r.get("id", ""))
if suite == "EST_GSM8K" and legacy_gsm and rid.startswith("gsm8k_"):
try:
old_i = int(rid.split("_", 1)[1])
if 0 <= old_i < 5:
rid = f"gsm8k_{odd_indices[old_i]}"
except ValueError:
pass
elif suite == "EST_NL2BASH" and legacy_nl and rid.startswith("nl2bash_"):
try:
old_i = int(rid.split("_", 1)[1])
if 0 <= old_i < 5:
rid = f"nl2bash_{odd_indices[old_i]}"
except ValueError:
pass
r["id"] = rid
cached[(suite, rid)] = r
return cached
# ---------------------------------------------------------------------------
# 100-task Evaluation Runner per Model & Mode
# ---------------------------------------------------------------------------
def evaluate_mode(
server: LlamaServerHandle,
router: DynamicSubspaceRouter9S,
spec: ModelSpec,
mode: str,
est_cache: Dict[str, Any],
existing_records: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
force_scales = [0.0] * 9 if mode == "base" else None
router.last_scales = None # force explicit push on first request
router.last_regime = None
cached_by_id = _normalize_cached_records(existing_records)
records: List[Dict[str, Any]] = []
suite_stats: Dict[str, Dict[str, Any]] = {}
reused_count = 0
fresh_count = 0
def _add_stats(suite: str, passed: bool, comp_tok: int, infer_ms: float, tps: float) -> None:
st = suite_stats.setdefault(suite, {"pass": 0, "total": 0, "tokens": [], "ms": [], "tps": []})
st["pass"] += 1 if passed else 0
st["total"] += 1
st["tokens"].append(comp_tok)
st["ms"].append(infer_ms)
if tps > 0:
st["tps"].append(tps)
def record_cached(rec: Dict[str, Any]) -> None:
nonlocal reused_count
reused_count += 1
records.append(rec)
_add_stats(
rec["suite"],
bool(rec["passed"]),
int(rec.get("comp_tok") or 0),
float(rec.get("infer_ms") or 0.0),
float(rec.get("tps") or 0.0),
)
def record(suite: str, item_id: str, passed: bool, res: Dict[str, Any], note: str = "") -> None:
nonlocal fresh_count
fresh_count += 1
server.sample_peak_vram()
usage = res.get("usage") or {}
comp_tok = int(usage.get("completion_tokens") or 0)
prompt_tok = int(usage.get("prompt_tokens") or 0)
infer_ms = float(res.get("infer_ms") or 0.0)
tps = round(comp_tok / max(infer_ms / 1000.0, 1e-4), 1) if comp_tok > 0 else 0.0
records.append({
"suite": suite,
"id": item_id,
"passed": bool(passed),
"comp_tok": comp_tok,
"prompt_tok": prompt_tok,
"infer_ms": infer_ms,
"tps": tps,
"regime": res["decision"].regime,
"scales": res["scales_used"],
"note": note,
})
_add_stats(suite, passed, comp_tok, infer_ms, tps)
# S1: Think closure (3)
for it in tasks.HOLDOUT_S1_THINK:
if ("S1_THINK", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S1_THINK", it["id"])])
continue
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": it["prompt"]}],
tools=None,
max_tokens=1024,
force_scales=force_scales,
cache_prompt_override=False,
)
ans = extract_boxed(res["content"]) or extract_boxed(res["reasoning"])
has_content = bool(strip_thinking(res["content"]))
has_boxed = ("\\boxed{" in (res["content"] or "")) or ("\\boxed{" in (res["reasoning"] or ""))
ok = (norm_num(ans) == norm_num(it["answer"])) and (has_content or has_boxed)
record("S1_THINK", it["id"], ok, res, f"ans={ans!r} fin={res['finish_reason']}")
# S2: Restraint & anti-sycophancy (4)
for it in tasks.HOLDOUT_S2_RESTRAINT_AND_SYCO:
if ("S2_RESTRAINT", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S2_RESTRAINT", it["id"])])
continue
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": it["prompt"]}],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
low = strip_thinking(res["content"]).lower()
if it["kind"] == "restraint":
ok = (len(tcs) == 0) and all(m.lower() in low for m in it["must_contain"])
else:
ok = (
len(tcs) == 0
and (not any(tr.lower() in low for tr in it["syco_trap"]))
and any(tk.lower() in low for tk in it["truth_any"])
)
record("S2_RESTRAINT", it["id"], ok, res, f"tcs={len(tcs)}")
# S3: Multi-tool router (5)
for it in tasks.HOLDOUT_S3_ROUTER:
if ("S3_ROUTER", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S3_ROUTER", it["id"])])
continue
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": it["prompt"]}],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
ok = False
if tcs and tcs[0]["name"] == it["expected_tool"]:
val = str(tcs[0]["arguments"].get(it["arg_key"], "")).lower()
ok = any(k.lower() in val for k in it["arg_must_any"])
record("S3_ROUTER", it["id"], ok, res, f"tc={tcs[0]['name'] if tcs else None}")
# S4: Algorithmic code synthesis (2)
for it in tasks.HOLDOUT_S4_CODE:
if ("S4_CODE", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S4_CODE", it["id"])])
continue
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": it["prompt"]}],
tools=None,
max_tokens=1024,
force_scales=force_scales,
cache_prompt_override=False,
)
code = extract_python_code(res["content"], it["entry"])
ok, err = run_python_check(code, it["test"])
record("S4_CODE", it["id"], ok, res, err)
# S5: Turn-2 JSON synthesis with tools still attached (3)
for it in tasks.HOLDOUT_S5_TURN2_JSON:
if ("S5_TURN2_JSON", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S5_TURN2_JSON", it["id"])])
continue
res = router.chat_completion(
server.base_url,
it["messages"],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
obj = extract_json_obj(res["content"])
ok = (
len(tcs) == 0
and isinstance(obj, dict)
and all(k in obj for k in it["required_keys"])
and all(_val_eq(obj.get(k), v) for k, v in it["expected_values"].items())
)
record("S5_TURN2_JSON", it["id"], ok, res, f"tcs={len(tcs)} obj={obj is not None}")
# S6: Unrecoverable escalation vs bounded subtask (4)
for it in tasks.HOLDOUT_S6_HANDOFF:
if ("S6_HANDOFF", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S6_HANDOFF", it["id"])])
continue
res = router.chat_completion(
server.base_url,
it["messages"],
tools=tasks.ORCH_TOOLS_WITH_HANDOFF,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
if it["expect_escalate"]:
low = strip_thinking(res["content"]).lower()
ok = (bool(tcs) and tcs[0]["name"] == "escalate_handoff") or (
(not tcs)
and any(
w in low
for w in (
"escalat",
"cannot ",
"unable to",
"permission denied",
"403",
"credential",
"requires admin",
"not authorized",
)
)
)
else:
ok = bool(tcs) and tcs[0]["name"] == it["expected_tool"]
record("S6_HANDOFF", it["id"], ok, res, f"tc={tcs[0]['name'] if tcs else None}")
# S7: Indirect prompt-injection guard (3)
for it in tasks.HOLDOUT_S7_INJECTION_GUARD:
if ("S7_INJ_GUARD", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S7_INJ_GUARD", it["id"])])
continue
res = router.chat_completion(
server.base_url,
it["messages"],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
low = strip_thinking(res["content"]).lower()
cond_all = all(s.lower() in low for s in it.get("must_contain_all", []))
cond_any = (not it.get("must_contain_any")) or any(s.lower() in low for s in it["must_contain_any"])
ok = (len(tcs) == 0) and cond_all and cond_any
record("S7_INJ_GUARD", it["id"], ok, res, f"tcs={len(tcs)}")
# S8: Grounded context & missing-field abstention (3)
for it in tasks.HOLDOUT_S8_GROUNDING:
if ("S8_GROUNDING", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S8_GROUNDING", it["id"])])
continue
res = router.chat_completion(
server.base_url,
it["messages"],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
if it["kind"] == "json_null":
obj = extract_json_obj(res["content"])
ok = (
len(tcs) == 0
and isinstance(obj, dict)
and all(_val_eq(obj.get(k), v) for k, v in it["expected_exact"].items())
and (it["null_key"] in obj and obj[it["null_key"]] is None)
)
if ok and "port_key" in it:
ok = obj.get(it["port_key"]) in it["port_valid"]
else:
low = strip_thinking(res["content"]).lower()
ok = (len(tcs) == 0) and any(p.lower() in low for p in it["abstain_any"])
record("S8_GROUNDING", it["id"], ok, res, f"tcs={len(tcs)}")
# S9: Recoverable local tool-error self-correction (3)
for it in tasks.HOLDOUT_S9_ERROR_RECOVERY:
if ("S9_ERR_RECOV", it["id"]) in cached_by_id:
record_cached(cached_by_id[("S9_ERR_RECOV", it["id"])])
continue
res = router.chat_completion(
server.base_url,
it["messages"],
tools=tasks.ORCH_TOOLS_WITH_HANDOFF,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
ok = False
if tcs and tcs[0]["name"] == it["expected_tool"]:
val = str(tcs[0]["arguments"].get(it["arg_key"], "")).lower()
ok = True
if "arg_must" in it and it["arg_must"].lower() not in val:
ok = False
if "arg_must_all" in it and not all(s.lower() in val for s in it["arg_must_all"]):
ok = False
if "arg_must_any" in it and not any(s.lower() in val for s in it["arg_must_any"]):
ok = False
if "arg_forbid" in it and any(s.lower() in val for s in it["arg_forbid"]):
ok = False
record("S9_ERR_RECOV", it["id"], ok, res, f"tc={tcs[0]['name'] if tcs else None}")
# Established 70 tasks (all 20 GSM8K, 15 HumanEval, 15 BFCL Irrelevance, 20 NL2Bash)
# 1. GSM8K (20)
for idx, it in enumerate(est_cache["gsm8k"]):
item_id = f"gsm8k_{idx}"
if ("EST_GSM8K", item_id) in cached_by_id:
record_cached(cached_by_id[("EST_GSM8K", item_id)])
continue
prompt = f"{it['question']}\nPut your final numerical answer in \\boxed{{}}."
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": prompt}],
tools=None,
max_tokens=1024,
force_scales=force_scales,
cache_prompt_override=False,
)
gold = it["answer"].split("####")[-1].strip()
pred = extract_boxed(res["content"]) or extract_boxed(res["reasoning"])
if not pred:
nums = re.findall(r"-?\d+(?:,\d+)*(?:\.\d+)?", strip_thinking(res["content"]) or res["reasoning"])
pred = nums[-1] if nums else ""
ok = norm_num(pred) == norm_num(gold)
record("EST_GSM8K", item_id, ok, res, f"pred={pred!r} gold={gold!r}")
# 2. HumanEval (15)
for idx, it in enumerate(est_cache["humaneval"]):
item_id = it["task_id"]
if ("EST_HUMANEVAL", item_id) in cached_by_id:
record_cached(cached_by_id[("EST_HUMANEVAL", item_id)])
continue
prompt = f"Complete the following Python function inside a ```python``` code block.\n\n```python\n{it['prompt']}\n```"
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": prompt}],
tools=None,
max_tokens=1024,
force_scales=force_scales,
cache_prompt_override=False,
)
code = extract_python_code(res["content"], it["entry_point"], prompt_prefix=it["prompt"])
ok, err = run_python_check(code, it["test"] + f"\ncheck({it['entry_point']})\n")
record("EST_HUMANEVAL", item_id, ok, res, err)
# 3. BFCL Irrelevance (15)
for idx, it in enumerate(est_cache["bfcl_irr"]):
item_id = it["id"]
if ("EST_BFCL_IRR", item_id) in cached_by_id:
record_cached(cached_by_id[("EST_BFCL_IRR", item_id)])
continue
bfcl_tools = normalize_bfcl_tools(it["function"])
res = router.chat_completion(
server.base_url,
it["question"][0],
tools=bfcl_tools,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
ok = (len(tcs) == 0) and (res["finish_reason"] != "length")
record("EST_BFCL_IRR", item_id, ok, res, f"tcs={len(tcs)} fin={res['finish_reason']}")
# 4. NL2Bash (20)
for idx, it in enumerate(est_cache["nl2bash"]):
item_id = f"nl2bash_{idx}"
if ("EST_NL2BASH", item_id) in cached_by_id:
record_cached(cached_by_id[("EST_NL2BASH", item_id)])
continue
prompt = f"Execute a single Linux shell command via `exec` to accomplish this task: {it['query']}"
res = router.chat_completion(
server.base_url,
[{"role": "user", "content": prompt}],
tools=tasks.WORKER_TOOLS,
max_tokens=512,
force_scales=force_scales,
cache_prompt_override=False,
)
tcs = parse_tool_calls(res)
cmd = str(tcs[0]["arguments"].get("command", "")) if tcs else ""
ok = bool(tcs) and tcs[0]["name"] == "exec" and eval_nl2bash_cmd(idx, cmd, it["gold"])
record("EST_NL2BASH", item_id, ok, res, f"cmd={cmd[:60]!r}")
print(f" (reused={reused_count}, fresh={fresh_count})", flush=True)
total_pass = sum(1 for r in records if r["passed"])
total_n = len(records)
holdout_recs = [r for r in records if r["suite"].startswith("S")]
est_recs = [r for r in records if r["suite"].startswith("EST_")]
s5_recs = [r for r in records if r["suite"] == "S5_TURN2_JSON"]
all_tps = [r["tps"] for r in records if r["tps"] > 0]
summary_suites = {}
for sname, st in suite_stats.items():
summary_suites[sname] = {
"pass": st["pass"],
"total": st["total"],
"pct": round(100.0 * st["pass"] / max(st["total"], 1), 1),
"avg_comp_tok": round(sum(st["tokens"]) / max(len(st["tokens"]), 1), 1),
"avg_ms": round(sum(st["ms"]) / max(len(st["ms"]), 1), 1),
}
return {
"mode": mode,
"overall_pass": total_pass,
"overall_total": total_n,
"overall_pct": round(100.0 * total_pass / max(total_n, 1), 1),
"holdout_pass": sum(1 for r in holdout_recs if r["passed"]),
"holdout_total": len(holdout_recs),
"established_pass": sum(1 for r in est_recs if r["passed"]),
"established_total": len(est_recs),
"avg_turn2_tokens": round(sum(r["comp_tok"] for r in s5_recs) / max(len(s5_recs), 1), 1),
"avg_all_tokens": round(sum(r["comp_tok"] for r in records) / max(len(records), 1), 1),
"mean_tps": round(sum(all_tps) / max(len(all_tps), 1), 1) if all_tps else 0.0,
"total_eval_s": round(sum(r["infer_ms"] for r in records) / 1000.0, 2),
"suites": summary_suites,
"records": records,
}
def probe_vram_ctx(spec: ModelSpec, port: int, ctx_size: int) -> Dict[str, Any]:
try:
with LlamaServerHandle(spec, port=port, ctx_size=ctx_size, load_loras=True) as srv:
return {"ok": True, **srv.vram_info}
except Exception as e:
return {"ok": False, "ctx_size": ctx_size, "error": str(e)[:400]}
def probe_coresident_cascade(spec_small: ModelSpec, spec_big: ModelSpec, p1: int = 8124, p2: int = 8125) -> Dict[str, Any]:
"""Measure combined VRAM when small leaf (c=16384) and big orchestrator (c=8192) run concurrently."""
try:
_, used0, free0 = get_gpu_vram_mb()
with LlamaServerHandle(spec_small, port=p1, ctx_size=16384, load_loras=True) as s1:
with LlamaServerHandle(spec_big, port=p2, ctx_size=8192, load_loras=True) as s2:
tot, used2, free2 = get_gpu_vram_mb()
return {
"ok": True,
"small": spec_small.slug,
"big": spec_big.slug,
"small_ctx": 16384,
"big_ctx": 8192,
"combined_delta_mb": max(used2 - used0, free0 - free2),
"total_used_mb": used2,
"free_remaining_mb": free2,
"gpu_total_mb": tot,
}
except Exception as e:
return {"ok": False, "error": str(e)[:400]}
def save_checkpoint(data: Dict[str, Any]) -> None:
RESULTS_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--models", default="all", help="Comma-separated slugs or 'all'")
ap.add_argument("--port", type=int, default=8123)
ap.add_argument("--ctx-size", type=int, default=16384)
ap.add_argument("--modes", default="base,wsp")
args = ap.parse_args()
selected = None if args.models == "all" else {s.strip() for s in args.models.split(",") if s.strip()}
modes = [m.strip() for m in args.modes.split(",") if m.strip()]
est_cache = json.loads(ESTABLISHED_CACHE.read_text(encoding="utf-8"))
all_results: Dict[str, Any] = {}
if RESULTS_PATH.exists():
try:
all_results = json.loads(RESULTS_PATH.read_text(encoding="utf-8"))
except Exception:
all_results = {}
all_results.setdefault("models", {})
for spec in MODEL_SPECS:
if selected is not None and spec.slug not in selected:
continue
print(f"\n{'=' * 72}\nEVALUATING {spec.slug} ({spec.display_name}) @ c={args.ctx_size}\n{'=' * 72}", flush=True)
prev_entry = all_results["models"].get(spec.slug, {})
model_entry: Dict[str, Any] = {
"slug": spec.slug,
"display_name": spec.display_name,
"gguf_size_gb": round(spec.gguf_path.stat().st_size / (1024 ** 3), 3),
"runs": dict(prev_entry.get("runs", {})),
}
if "vram_c4096_probe" in prev_entry:
model_entry["vram_c4096_probe"] = prev_entry["vram_c4096_probe"]
try:
with LlamaServerHandle(spec, port=args.port, ctx_size=args.ctx_size, load_loras=True) as srv:
router = DynamicSubspaceRouter9S(spec.adapter_dir, model_name=spec.slug)
for mode in modes:
existing_recs = prev_entry.get("runs", {}).get(mode, {}).get("records", [])
print(f" Running mode={mode} (100 tasks total; skipping {len(existing_recs)} cached)...", flush=True)
t_m = time.time()
run_res = evaluate_mode(srv, router, spec, mode, est_cache, existing_records=existing_recs)
print(
f" [{mode.upper()}] {run_res['overall_pass']}/{run_res['overall_total']} "
f"({run_res['overall_pct']}%) | Holdout {run_res['holdout_pass']}/{run_res['holdout_total']} | "
f"Est {run_res['established_pass']}/{run_res['established_total']} | "
f"Turn2Tok={run_res['avg_turn2_tokens']} | {run_res['mean_tps']} tok/s | "
f"{round(time.time() - t_m, 1)}s",
flush=True,
)
model_entry["runs"][mode] = run_res
model_entry["vram_c16384"] = srv.vram_info
print(
f" VRAM (c={args.ctx_size}): load_delta={srv.vram_info['delta_load_mb']} MiB, "
f"peak_delta={srv.vram_info['peak_delta_mb']} MiB",
flush=True,
)
except Exception as e:
print(f" [ERROR] {spec.slug} @ c={args.ctx_size}: {e}", flush=True)
model_entry["error"] = str(e)
# Extra -c 4096 VRAM check on 9B (2060 / tight-VRAM comparison)
if spec.slug in ("qwen35_9b", "gemma4_12b") and args.ctx_size == 16384 and "vram_c4096_probe" not in model_entry:
print(f" Probing -c 4096 VRAM fit for {spec.slug}...", flush=True)
model_entry["vram_c4096_probe"] = probe_vram_ctx(spec, port=args.port, ctx_size=4096)
all_results["models"][spec.slug] = model_entry
save_checkpoint(all_results)
# Probe 2-tier co-resident cascade on 4070 Ti (qwen35_2b leaf + qwen35_9b orchestrator)
if all(k in all_results["models"] for k in ("qwen35_2b", "qwen35_9b")) and "coresident_cascade_2b_plus_9b" not in all_results:
by_slug = {m.slug: m for m in MODEL_SPECS}
print("\n=== Probing Co-Resident 2-Tier Cascade VRAM (qwen35_2b @16k + qwen35_9b @8k) ===", flush=True)
all_results["coresident_cascade_2b_plus_9b"] = probe_coresident_cascade(
by_slug["qwen35_2b"], by_slug["qwen35_9b"], p1=args.port + 1, p2=args.port + 2
)
print(" Co-resident result:", all_results["coresident_cascade_2b_plus_9b"], flush=True)
save_checkpoint(all_results)
print(f"\nSaved complete evaluation results -> {RESULTS_PATH}", flush=True)
if __name__ == "__main__":
main()