Measure queued-request behaviour and reasoning_control before the M2 design

A request pinned to a busy slot receives no bytes until the slot frees.
reasoning_control ends a thinking block on demand; the capped turn is
re-read once on the next request.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

spike/m0.py

@@ -1,7 +1,7 @@
 #!/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 evict alternate)
+Usage: python3 m0.py <check> [...]   (checks: slots throughput convo tools pin progress rewind findtool rewind2 evict alternate queued reasoncap)
 Raw results are appended to out/<check>.jsonl; findings go in docs/inference-contract.md.
 """
 import json, os, random, sys, time, urllib.request
@@ -335,6 +335,92 @@ def check_alternate():
     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"]: