kyle/boxmaker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add M0 measurement spike and findings

Browse Source 
spike/m0.py is throwaway. Findings are in docs/inference-contract.md:
cache reuse and tool parsing pass through chat-completions, the tools
array must stay fixed per epoch, and the shared router cannot meet the
slot contract. Clean throughput and slot pinning are still open because
another session was using the GPU.

docs/decisions.md lists the brief changes this implies as proposals.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
kyle
2026-09-17 00:22:31 -07:00
co-authored by Claude Fable 5.1
parent 96bf688b20
commit 06c05e0023
4 changed files with 491 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spike/m0.py
+314
View File
@@ -0,0 +1,314 @@
#!/usr/bin/env python3
"""THROWAWAY M0 measurement spike for Boxmaker. Not harness code. Stdlib only.
Usage: python3 m0.py <check> [...] (checks: slots throughput convo tools pin progress rewind findtool rewind2)
Raw results are appended to out/<check>.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": "<html>ok</html>",
"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\" & <kind>."],
"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 "<tool_call>" in (m.get("content") or "") or "<tool_call>" in (m.get("reasoning_content") or ""):
why.append("raw <tool_call> 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})
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]()