#!/usr/bin/env python3 """THROWAWAY M0 measurement spike for Boxmaker. Not harness code. Stdlib only. Usage: python3 m0.py [...] (checks: slots throughput convo tools pin progress rewind findtool rewind2 evict alternate queued reasoncap) Raw results are appended to out/.jsonl; findings go in docs/inference-contract.md. """ import json, os, random, sys, time, urllib.request BASE = os.environ.get("LLAMA_URL", "https://straylight.scylla-hammerhead.ts.net:10000") MODEL = os.environ.get("LLAMA_MODEL", "ornith-1.5-35b-a3b") OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "out") SAMPLING = {"temperature": 0.6, "top_p": 0.95, "top_k": 20} NONCE = "%08x" % random.getrandbits(32) def http(path, body=None, timeout=1800): data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(BASE + path, data=data, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.load(r) def log(check, rec): rec = {"check": check, "nonce": NONCE, "t": time.strftime("%FT%T"), **rec} with open(os.path.join(OUT, check + ".jsonl"), "a") as f: f.write(json.dumps(rec) + "\n") def chat(messages, tools=None, slot=-1, max_tokens=2048, **extra): body = {"model": MODEL, "messages": messages, "id_slot": slot, "cache_prompt": True, "max_tokens": max_tokens, "stream": False, **SAMPLING, **extra} if tools: body["tools"] = tools t0 = time.time() r = http("/v1/chat/completions", body) r["_wall_s"] = round(time.time() - t0, 2) return r def tline(label, r): t = r["timings"] print("%-34s cache_n=%-6d prompt_n=%-6d pp=%7.1f t/s predicted_n=%-5d tg=%5.1f t/s wall=%ss" % ( label, t["cache_n"], t["prompt_n"], t.get("prompt_per_second", 0), t["predicted_n"], t.get("predicted_per_second", 0), r["_wall_s"])) return {"label": label, "timings": t, "wall_s": r["_wall_s"], "finish": r["choices"][0]["finish_reason"]} def ntokens(text): return len(http("/tokenize", {"model": MODEL, "content": text})["tokens"]) def filler(target_tokens, seed): """Pseudo-random English-ish text of about target_tokens tokens, unique per run and seed.""" rng = random.Random(NONCE + str(seed)) words = ("box fragment shell cork glass bone thread map brass feather shard coil lattice drift " "orbit cable vault ledger salt ash lens hinge spool wire amber slate quill tin reed").split() chunk = " ".join(rng.choice(words) + ("." if rng.random() < .1 else "") for _ in range(2000)) per = ntokens(chunk) / 2000 n = int(target_tokens / per) return " ".join(rng.choice(words) + ("." if rng.random() < .1 else "") for _ in range(n)) def tool(name, desc, props, required): return {"type": "function", "function": {"name": name, "description": desc, "parameters": { "type": "object", "properties": props, "required": required}}} TOOLS = [ tool("read_file", "Read a text file and return its contents.", {"path": {"type": "string", "description": "Absolute path"}}, ["path"]), tool("write_file", "Write text to a file, replacing it.", {"path": {"type": "string"}, "content": {"type": "string"}}, ["path", "content"]), tool("shell", "Run a shell command in a sandbox and return stdout and stderr.", {"command": {"type": "string"}, "timeout_s": {"type": "integer"}}, ["command"]), tool("http_fetch", "Fetch a URL with GET and return the body.", {"url": {"type": "string"}, "headers": {"type": "object"}}, ["url"]), ] EXTRA_TOOL = tool("recall", "Search long-term memory notes.", {"query": {"type": "string"}, "k": {"type": "integer"}}, ["query"]) SYSTEM = "You are Boxmaker, a careful personal agent. Use tools when they are needed. Run id %s." % NONCE FAKE_RESULTS = {"read_file": "hostname = straylight\nport = 11434\n", "write_file": "ok, 24 bytes written", "shell": "total 4\n-rw-r--r-- 1 kyle users 31 Sep 17 notes.txt\n", "http_fetch": "ok", "recall": "notes/2026-09-01.md:3 owner prefers metric units"} def assistant_msg(r, keep_reasoning=True): m = r["choices"][0]["message"] out = {"role": "assistant", "content": m.get("content") or ""} if keep_reasoning and m.get("reasoning_content"): out["reasoning_content"] = m["reasoning_content"] if m.get("tool_calls"): out["tool_calls"] = m["tool_calls"] return out def run_turn(label, messages, tools, slot, keep_reasoning, recs, max_iters=4): """One user turn: loop model -> tool results until the model answers in text.""" for i in range(max_iters): r = chat(messages, tools, slot) recs.append(tline("%s req%d" % (label, i + 1), r)) messages.append(assistant_msg(r, keep_reasoning)) calls = r["choices"][0]["message"].get("tool_calls") or [] if not calls: return for c in calls: messages.append({"role": "tool", "tool_call_id": c["id"], "content": FAKE_RESULTS.get(c["function"]["name"], "ok")}) USER_TURNS = ["Read /etc/boxmaker/config.toml and tell me which port is configured.", "Now list the files in /home/kyle/notes using the shell.", "Thanks. In one sentence, what did you learn from both steps?"] def check_convo(): """(b) cache reuse over 3 turns, (e) effect of dropping reasoning, (h) adding a tool mid-session.""" for variant in ("keep_reasoning", "drop_reasoning", "add_tool_turn3"): print("\n== variant:", variant) msgs, recs = [{"role": "system", "content": SYSTEM + " Variant " + variant}], [] for n, u in enumerate(USER_TURNS, 1): tools = TOOLS + [EXTRA_TOOL] if (variant == "add_tool_turn3" and n == 3) else TOOLS msgs.append({"role": "user", "content": u}) run_turn("turn%d" % n, msgs, tools, 0, variant != "drop_reasoning", recs) log("convo", {"variant": variant, "requests": recs}) TRIALS = { "read_file": ["Show me what is in /etc/hosts.", "Open /home/kyle/todo.md and summarise it.", "What does /etc/os-release say?", "Read the file /var/log/boot.log.", "I need the contents of /home/kyle/.gitconfig."], "write_file": ["Save the text 'buy milk' to /home/kyle/shopping.txt.", "Create /tmp/hello.txt containing the single word hello.", "Write a two-line haiku about rain into /home/kyle/haiku.txt.", "Put the JSON {\"a\": 1, \"b\": [2, 3]} into /tmp/data.json exactly.", "Replace /home/kyle/motd with: Stay \"curious\" & ."], "shell": ["How much disk space is free? Use the shell.", "Run uname -a for me.", "Count the lines in /etc/passwd with a shell command.", "Use the shell to find files larger than 1GB under /var, with a 30 second timeout.", "Run: echo \"a && b\" | tr a-z A-Z"], "http_fetch": ["Fetch https://example.com and tell me the title.", "GET https://api.github.com/zen please.", "Download https://example.org/robots.txt.", "Fetch https://example.com/api with the header Accept: application/json.", "What does http://neverssl.com return?"], } def check_tools(): """(c) 20 trials across 4 tools: is the tool call parsed server-side, with valid arguments?""" fails = 0 for want, prompts in TRIALS.items(): required = [t for t in TOOLS if t["function"]["name"] == want][0]["function"]["parameters"]["required"] for p in prompts: r = chat([{"role": "system", "content": SYSTEM}, {"role": "user", "content": p}], TOOLS, 0, 1536) m, why = r["choices"][0]["message"], [] calls = m.get("tool_calls") or [] if not calls: why.append("no tool_calls") for c in calls: try: args = json.loads(c["function"]["arguments"]) why += ["missing arg " + a for a in required if a not in args] except ValueError: why.append("arguments not JSON") if c["function"]["name"] != want: why.append("called " + c["function"]["name"]) if "" in (m.get("content") or "") or "" in (m.get("reasoning_content") or ""): why.append("raw text leaked") if r["choices"][0]["finish_reason"] != "tool_calls": why.append("finish_reason=" + r["choices"][0]["finish_reason"]) fails += bool(why) print("%-10s %-4s %s" % (want, "FAIL" if why else "ok", "; ".join(why) or p[:50])) log("tools", {"want": want, "prompt": p, "problems": why, "message": m, "predicted_n": r["timings"]["predicted_n"]}) print("failures: %d / 20" % fails) def slots(): return http("/slots?model=" + MODEL) def check_slots(): for s in slots(): print({k: s.get(k) for k in ("id", "n_ctx", "is_processing", "id_task")}, "n_past/prompt tokens:", s.get("n_past", s.get("next_token"))) def check_pin(): """(d) slot pinning, and what unpinned traffic does to a pinned session's cache.""" recs = [] a = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": "Session A. " + filler(6000, "A") + "\nReply with the single word: alpha"}] b = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": "Session B. " + filler(6000, "B") + "\nReply with the single word: beta"}] for label, msgs, slot in (("A turn1 slot0", a, 0), ("B turn1 slot1", b, 1)): r = chat(msgs, None, slot, 512) recs.append(tline(label, r)); msgs.append(assistant_msg(r)) a.append({"role": "user", "content": "Again, one word."}) r = chat(a, None, 0, 512); recs.append(tline("A turn2 slot0 (after B)", r)); a.append(assistant_msg(r)) # Two unpinned strangers, as Open WebUI or OpenCode would send. for i in (1, 2): c = [{"role": "user", "content": "Stranger %d. %s\nReply: gamma" % (i, filler(3000, "C%d" % i))}] r = chat(c, None, -1, 512); recs.append(tline("stranger %d unpinned" % i, r)) a.append({"role": "user", "content": "Once more, one word."}) r = chat(a, None, 0, 512); recs.append(tline("A turn3 slot0 (after strangers)", r)) b.append({"role": "user", "content": "Again, one word."}) r = chat(b, None, 1, 512); recs.append(tline("B turn2 slot1 (after strangers)", r)) log("pin", {"requests": recs}) def check_throughput(): """(a) timed requests. pp at depth D = rate for a 2k suffix appended to a cached D-token prefix.""" recs = [] for depth in (0, 32000): msgs = [{"role": "user", "content": "Depth %d. " % depth + (filler(depth, depth) if depth else "")}] if depth: msgs[0]["content"] += "\nReply with: ok" r = chat(msgs, None, 0, 64, chat_template_kwargs={"enable_thinking": False}) recs.append(tline("prefill 0..%d (average)" % depth, r)); msgs.append(assistant_msg(r)) msgs.append({"role": "user", "content": ""}) msgs[-1]["content"] += filler(2000, "s%d" % depth) + "\nNow write about 250 words on tide pools." r = chat(msgs, None, 0, 400, chat_template_kwargs={"enable_thinking": False}) recs.append(tline("pp+tg at depth %d" % depth, r)) log("throughput", {"requests": recs}) def check_progress(): """(i) does return_progress give bytes during prefill? Prints gaps between stream events.""" body = {"model": MODEL, "stream": True, "return_progress": True, "id_slot": 0, "max_tokens": 32, **SAMPLING, "chat_template_kwargs": {"enable_thinking": False}, "messages": [{"role": "user", "content": filler(16000, "p") + "\nReply: ok"}]} req = urllib.request.Request(BASE + "/v1/chat/completions", json.dumps(body).encode(), {"Content-Type": "application/json"}) t0 = last = time.time(); gaps, events = [], [] with urllib.request.urlopen(req, timeout=1800) as r: for line in r: if not line.startswith(b"data: {"): continue now = time.time(); gaps.append(round(now - last, 2)); last = now d = json.loads(line[6:]) if "prompt_progress" in d: events.append(d["prompt_progress"]) print("progress events: %d, first: %s, last: %s" % (len(events), events[:1], events[-1:])) print("max gap between events: %.2fs, total %.1fs" % (max(gaps), time.time() - t0)) log("progress", {"n_events": len(events), "events": events[:3] + events[-2:], "max_gap_s": max(gaps)}) def check_rewind(): """(e2) hybrid-attention rewind: change text 10k tokens into a 24k prompt. How much cache survives?""" f1, f2, f2b, f3 = filler(10000, "r1"), filler(10000, "r2"), filler(10000, "r2b"), filler(4000, "r3") recs, kw = [], {"chat_template_kwargs": {"enable_thinking": False}} for label, mid in (("original 24k", f2), ("same again", f2), ("diverge at ~10k", f2b), ("back to original", f2)): r = chat([{"role": "user", "content": f1 + "\n" + mid + "\n" + f3 + "\nReply: ok"}], None, 0, 16, **kw) recs.append(tline(label, r)) log("rewind", {"requests": recs}) FIND_TOOL = tool("find_tool", "Search for additional tools by keyword. Returns tool schemas.", {"query": {"type": "string"}}, ["query"]) CALL_TOOL = tool("call_tool", "Call a tool that was returned by find_tool. Pass its name and an arguments object.", {"name": {"type": "string"}, "arguments": {"type": "object"}}, ["name", "arguments"]) ASKS = ["What do my memory notes say about units of measurement?", "Search my notes for anything about Mattermost.", "Do I have a note about my dentist? Check memory, top 3 results.", "Look in long-term memory for 'tailnet ACL'.", "Recall what I wrote about coffee."] def check_findtool(): """(h2) progressive disclosure without touching the tools array: schema arrives as a tool result.""" for variant, tools in (("meta call_tool", TOOLS[:2] + [FIND_TOOL, CALL_TOOL]), ("undeclared direct", TOOLS[:2] + [FIND_TOOL])): ok = 0 for ask in ASKS: msgs = [{"role": "system", "content": SYSTEM + " If no listed tool fits, use find_tool first."}, {"role": "user", "content": ask}] r1 = chat(msgs, tools, 0, 1536); m1 = r1["choices"][0]["message"] calls = m1.get("tool_calls") or [] if not calls or calls[0]["function"]["name"] != "find_tool": print(variant, "| step1 did not call find_tool:", (calls[0]["function"]["name"] if calls else m1.get("content", "")[:60])); continue msgs.append(assistant_msg(r1)) hint = "Call it with call_tool." if "meta" in variant else "Call it directly by name." msgs.append({"role": "tool", "tool_call_id": calls[0]["id"], "content": "1 tool found. " + hint + "\n" + json.dumps(EXTRA_TOOL["function"])}) r2 = chat(msgs, tools, 0, 1536); m2 = r2["choices"][0]["message"] c2 = (m2.get("tool_calls") or [None])[0] desc = "no tool call; content=" + repr((m2.get("content") or "")[:120]) if c2: a = json.loads(c2["function"]["arguments"]); desc = c2["function"]["name"] + " " + json.dumps(a) good = (c2["function"]["name"] == "call_tool" and a.get("name") == "recall" and isinstance(a.get("arguments"), dict) and "query" in a["arguments"]) if "meta" in variant else (c2["function"]["name"] == "recall" and "query" in a) ok += good print("%-18s cache_n=%-5d %s" % (variant, r2["timings"]["cache_n"], desc)) log("findtool", {"variant": variant, "ask": ask, "step2": m2, "timings": r2["timings"]}) print("==", variant, "ok %d/5" % ok) def check_rewind2(): """(e3) divergence several turns back in a multi-request conversation: are request-boundary checkpoints reused?""" kw, recs = {"chat_template_kwargs": {"enable_thinking": False}}, [] msgs = [{"role": "system", "content": SYSTEM}] for i in range(1, 6): msgs.append({"role": "user", "content": "Part %d. %s\nReply: ok %d" % (i, filler(1500, "t%d" % i), i)}) r = chat(msgs, None, 0, 16, **kw); recs.append(tline("grow turn %d" % i, r)); msgs.append(assistant_msg(r)) for back in (5, 3, 2): # edit the user message of turn `back`, keep everything else m2 = json.loads(json.dumps(msgs)); idx = 1 + 2 * (back - 1) m2[idx]["content"] = m2[idx]["content"].replace("Reply: ok", "Reply now: ok") r = chat(m2[:-1] if False else m2 + [{"role": "user", "content": "Final. Reply: done"}], None, 0, 16, **kw) recs.append(tline("edit turn %d of 5, then ask" % back, r)) r = chat(msgs + [{"role": "user", "content": "Final. Reply: done"}], None, 0, 16, **kw) recs.append(tline(" restore original, then ask", r)) log("rewind2", {"requests": recs}) def check_evict(): """(d2) certain eviction: a different prompt is pinned onto the session's own slot. Is the session restored?""" recs, kw = [], {"chat_template_kwargs": {"enable_thinking": False}} a = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": "Session A. " + filler(30000, "EA") + "\nReply: alpha"}] r = chat(a, None, 0, 16, **kw); recs.append(tline("A turn1 slot0 (30k)", r)); a.append(assistant_msg(r)) for i in (1, 2, 3): c = [{"role": "user", "content": "Intruder %d. %s\nReply: gamma" % (i, filler(8000, "EI%d" % i))}] r = chat(c, None, 0, 16, **kw); recs.append(tline("intruder %d pinned to slot0 (8k)" % i, r)) a.append({"role": "user", "content": "Again."}) r = chat(a, None, 0, 16, **kw); recs.append(tline("A turn2 slot0 (after 3 intruders)", r)); a.append(assistant_msg(r)) a.append({"role": "user", "content": "Again."}) r = chat(a, None, 1, 16, **kw); recs.append(tline("A turn3 moved to slot1", r)) log("evict", {"requests": recs}) def check_alternate(): """(d3) two harness sessions sharing one baseline prefix take turns on one slot, as Mattermost threads would.""" recs, kw = [], {"chat_template_kwargs": {"enable_thinking": False}} base = SYSTEM + " Shared baseline follows. " + filler(2500, "BASE") sess = {n: [{"role": "system", "content": base}, {"role": "user", "content": "Thread %s. %s\nReply: ok" % (n, filler(10000, "AL" + n))}] for n in "AB"} for rnd in (1, 2, 3): for n in "AB": if rnd > 1: sess[n].append({"role": "user", "content": "Round %d. %s\nReply: ok" % (rnd, filler(300, "ALr%d%s" % (rnd, n)))}) r = chat(sess[n], TOOLS, 0, 16, **kw); recs.append(tline("thread %s round %d slot0" % (n, rnd), r)); sess[n].append(assistant_msg(r)) log("alternate", {"requests": recs}) def stream(body, on_event=None): """POST a streaming chat completion. Returns (events, raw_lines) with a receive time on each.""" import threading # noqa: F401 (used by callers) req = urllib.request.Request(BASE + "/v1/chat/completions", json.dumps(body).encode(), {"Content-Type": "application/json"}) t0, events, raw = time.time(), [], [] with urllib.request.urlopen(req, timeout=1800) as r: for line in r: now = round(time.time() - t0, 3) text = line.decode("utf-8", "replace").rstrip("\n") if text: raw.append((now, text[:80])) if text.startswith("data: {"): ev = json.loads(text[6:]); events.append((now, ev)) if on_event: on_event(now, ev) return events, raw def check_queued(): """(j) what does a pinned request receive while its slot is busy with another request?""" import threading kw = {"model": MODEL, "stream": True, "return_progress": True, "id_slot": 0, **SAMPLING, "chat_template_kwargs": {"enable_thinking": False}} a_first, a_done, out = threading.Event(), {}, {} def run_a(): ev, _ = stream({**kw, "max_tokens": 700, "messages": [{"role": "user", "content": "Write about 600 words on the history of cork. Run %s." % NONCE}]}, lambda t, e: a_first.set() if e.get("choices") and e["choices"][0]["delta"].get("content") else None) a_done["t"] = time.time(); out["a_events"] = len(ev) ta = threading.Thread(target=run_a); ta.start(); a_first.wait(120) b_start = time.time() ev, raw = stream({**kw, "max_tokens": 16, "messages": [{"role": "user", "content": "Reply with: ok. Run %s." % NONCE}]}) ta.join() waited = round(a_done["t"] - b_start, 2) print("B was sent while A was generating; A finished %.2fs after B was sent" % waited) print("B first raw line at %.2fs: %r" % (raw[0][0], raw[0][1])) print("B raw lines before A finished: %d" % sum(1 for t, _ in raw if t < waited)) first_kinds = [("progress" if "prompt_progress" in e else "delta") for _, e in ev[:3]] print("B first events: %s; B total events %d; last at %.2fs" % (first_kinds, len(ev), ev[-1][0])) log("queued", {"a_finished_after_s": waited, "b_first_line_s": raw[0][0], "b_first_line": raw[0][1], "b_lines_before_a_finished": sum(1 for t, _ in raw if t < waited), "b_first_events": first_kinds}) def check_reasoncap(): """(k) end a thinking block early with reasoning_control, then check the next turn still hits the cache.""" import threading cap, state = 150, {"n": 0, "sent": None, "resp": None, "after": 0} msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": "Think very carefully and at length: how many distinct ways can 8 rooks be placed on a " "chessboard so that none attack each other and none is on the main diagonal? Run %s." % NONCE}] body = {"model": MODEL, "stream": True, "return_progress": True, "reasoning_control": True, "id_slot": 0, "max_tokens": 4000, "messages": msgs, **SAMPLING} def control(cid): state["resp"] = http("/v1/chat/completions/control", {"id": cid, "action": "reasoning_end", "model": MODEL}, 60) def on_event(t, e): ch = (e.get("choices") or [{}])[0].get("delta", {}) if ch.get("reasoning_content"): state["n"] += 1 if state["sent"] is not None: state["after"] += 1 if state["n"] == cap and state["sent"] is None: state["sent"] = t threading.Thread(target=control, args=(e["id"],)).start() ev, _ = stream(body, on_event) reasoning = "".join((e["choices"][0]["delta"].get("reasoning_content") or "") for _, e in ev if e.get("choices")) content = "".join((e["choices"][0]["delta"].get("content") or "") for _, e in ev if e.get("choices")) finish = [e["choices"][0].get("finish_reason") for _, e in ev if e.get("choices") and e["choices"][0].get("finish_reason")] timings = [e["timings"] for _, e in ev if e.get("timings")][-1] print("control sent at %.2fs after %d reasoning chunks; response: %s" % (state["sent"], cap, state["resp"])) print("reasoning chunks after the control call: %d; content chars: %d; finish: %s" % (state["after"], len(content), finish)) print("reasoning tail: %r" % reasoning[-120:]) print("turn 1 timings: cache_n=%d prompt_n=%d predicted_n=%d" % (timings["cache_n"], timings["prompt_n"], timings["predicted_n"])) msgs.append({"role": "assistant", "content": content, "reasoning_content": reasoning}) msgs.append({"role": "user", "content": "Thanks. Reply with one word: done"}) r = chat(msgs, None, 0, 64, chat_template_kwargs={"enable_thinking": False}) rec = tline("turn 2 after a forced reasoning end", r) log("reasoncap", {"cap": cap, "control": state["resp"], "chunks_after": state["after"], "finish": finish, "turn1": timings, "turn2": rec}) if __name__ == "__main__": os.makedirs(OUT, exist_ok=True) for name in sys.argv[1:] or ["slots"]: print("\n#### %s (nonce %s)" % (name, NONCE)) globals()["check_" + name]()