#!/usr/bin/env python3
"""Find the conversation length at which a chat server stops showing the model its beginning.
Purpose: measure, rather than assume, what a server's context length does to a long
conversation. A badge word is planted in the first message, the conversation is
grown with filler exchanges of constant token size, and at every step the model is
asked for the word back. Two small calibration requests first measure exactly how
many prompt tokens the fixed part and each filler exchange cost, so every step knows
how many tokens it sent; comparing that with the prompt tokens the server reports
shows a server dropping part of the conversation even when nothing errors.
Platform: all (spark, strix, mac, nvidia; also Windows with Python 3). Standard library only.
Minimum memory: 8 GB, for a 4B-class model at the server's own context length. Raising the
server's context is what costs memory; this script only sends requests.
Assumes: an OpenAI-compatible chat server is running and --model is a name it serves:
Ollama http://127.0.0.1:11434/v1, llama-server http://127.0.0.1:8080/v1,
LM Studio http://127.0.0.1:1234/v1. Python 3.8 or later. Writes one JSON line to
--labbook (default labbook.md in the current directory) and nothing else.
Verdicts, one per step:
fits the word came back and the server counted every token it was sent
cut-lost the server counted fewer tokens than it was sent and the word did not come back
cut-kept the server counted fewer tokens than it was sent but the word still came back
model the word was lost although the server counted every token: the model missed it
refused the server answered with an HTTP error instead (it fails loudly)
no-answer empty answer that hit the token limit: thinking or --max-tokens, not context
recall-only the server reported no token count, so only the answer can be judged
Usage: python3 context-truncation-test.py --base-url http://127.0.0.1:11434/v1 \\
--model qwen3:4b-q4_K_M --around 4096 --labbook labbook.md
python3 context-truncation-test.py --base-url http://127.0.0.1:8080/v1 \\
--model qwen3-4b --steps 2048,3072,4096,6144 --no-labbook
"""
import argparse
import datetime
import json
import random
import re
import sys
import time
import urllib.error
import urllib.request
# Ordinary words no model is likely to produce by accident when asked for something else.
MARKER_WORDS = [
"PELICAN", "BASALT", "MARZIPAN", "LANTERN", "OBSIDIAN",
"TAMARIND", "QUARTZITE", "HALYARD", "JUNIPER", "CINNABAR",
]
# One filler exchange. The numbers are zero-padded to four digits so that every exchange
# tokenises to the same length under any tokeniser: that is what makes the calibration exact.
FILLER_USER = (
"Note item {n:04d} for the inventory: {n:04d} crates of dried goods arrived at the north "
"warehouse on day {n:04d}, checked in by the day shift, stacked in aisle {n:04d}, and "
"recorded against the standing order. Reply with only: noted {n:04d}."
)
FILLER_ASSISTANT = "noted {n:04d}"
RECALL_QUESTION = (
"What was the badge word I gave you in my very first message? "
"Answer with that single word and nothing else."
)
AROUND_FRACTIONS = (0.5, 0.8, 0.95, 1.05, 1.25, 1.5, 2.0)
CALIBRATION_PAIRS = (1, 9)
def build_messages(marker, pairs, plant_in):
"""The planted word, `pairs` filler exchanges, then the question."""
if plant_in == "system":
messages = [{"role": "system",
"content": f"The badge word for this conversation is {marker}."}]
else:
messages = [
{"role": "user",
"content": f"Remember this for the whole conversation. The badge word is "
f"{marker}. Reply with only: noted."},
{"role": "assistant", "content": "noted"},
]
for n in range(1, pairs + 1):
messages.append({"role": "user", "content": FILLER_USER.format(n=n)})
messages.append({"role": "assistant", "content": FILLER_ASSISTANT.format(n=n)})
messages.append({"role": "user", "content": RECALL_QUESTION})
return messages
def http_json(url, payload=None, api_key="", timeout=30.0):
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
data = json.dumps(payload).encode("utf-8") if payload is not None else None
req = urllib.request.Request(url, data=data, headers=headers)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
def strip_thinking(text):
"""A thinking block, closed or cut off, is not the answer."""
text = re.sub(r".*?", "", text, flags=re.DOTALL)
return re.sub(r".*", "", text, flags=re.DOTALL).strip()
def ask(args, messages):
payload = {"model": args.model, "messages": messages, "stream": False,
"temperature": 0, "max_tokens": args.max_tokens}
if args.reasoning_effort:
payload["reasoning_effort"] = args.reasoning_effort
started = time.time()
body = http_json(f"{args.base_url.rstrip('/')}/chat/completions", payload,
args.api_key, args.timeout)
choice = (body.get("choices") or [{}])[0]
usage = body.get("usage") or {}
return {
"answer": strip_thinking(str((choice.get("message") or {}).get("content") or "")),
"finish_reason": choice.get("finish_reason"),
"prompt_tokens": usage.get("prompt_tokens"),
"seconds": round(time.time() - started, 2),
}
def server_facts(args):
"""What this particular server says about itself. Best effort; no endpoint is required."""
root = re.sub(r"/v1/?$", "", args.base_url.rstrip("/"))
facts = {}
try: # llama-server
props = http_json(f"{root}/props", api_key=args.api_key, timeout=10)
facts["llama_server_n_ctx_per_slot"] = props["default_generation_settings"]["n_ctx"]
facts["llama_server_total_slots"] = props.get("total_slots")
except (urllib.error.URLError, OSError, ValueError, KeyError, TypeError):
pass
wanted = {args.model, f"{args.model}:latest"} # Ollama adds :latest to untagged names
try: # Ollama
for m in http_json(f"{root}/api/ps", api_key=args.api_key, timeout=10).get("models", []):
if wanted & {m.get("name"), m.get("model")}:
facts["ollama_context_length"] = m.get("context_length")
facts["ollama_size_bytes"] = m.get("size")
facts["ollama_size_vram_bytes"] = m.get("size_vram")
except (urllib.error.URLError, OSError, ValueError, AttributeError):
pass
try: # LM Studio
for m in http_json(f"{root}/api/v1/models", api_key=args.api_key, timeout=10).get("models", []):
for inst in m.get("loaded_instances") or []:
if args.model in (inst.get("id"), m.get("key")):
facts["lmstudio_context_length"] = (inst.get("config") or {}).get("context_length")
facts["lmstudio_parallel"] = (inst.get("config") or {}).get("parallel")
facts["lmstudio_max_context_length"] = m.get("max_context_length")
except (urllib.error.URLError, OSError, ValueError, AttributeError):
pass
return facts
def verdict(recalled, sent, counted, finish_reason, answer):
if not answer and finish_reason == "length":
return "no-answer"
if counted is None:
return "recall-only"
dropped = sent - counted > max(16, 0.02 * sent)
if recalled:
return "cut-kept" if dropped else "fits"
return "cut-lost" if dropped else "model"
def main():
p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--base-url", default="http://127.0.0.1:11434/v1",
help="OpenAI-compatible base URL, ending in /v1")
p.add_argument("--model", required=True, help="model name exactly as the server lists it")
group = p.add_mutually_exclusive_group()
group.add_argument("--around", type=int, metavar="TOKENS",
help="context length to test around: steps at 0.5x to 2x of it")
group.add_argument("--steps", help="comma-separated prompt sizes in tokens")
p.add_argument("--plant-in", choices=("user", "system"), default="user",
help="put the badge word in the first user message or a system message")
p.add_argument("--max-tokens", type=int, default=64)
p.add_argument("--reasoning-effort", default="none",
help="sent as reasoning_effort; 'none' turns thinking off on Ollama and "
"llama-server. Pass an empty string to leave the field out")
p.add_argument("--keep-going", action="store_true",
help="do not stop after two steps in a row that lose the word")
p.add_argument("--api-key", default="", help="bearer token, if the server requires one")
p.add_argument("--timeout", type=float, default=900.0, help="seconds per request")
p.add_argument("--seed", type=int, default=7, help="chooses the badge word")
p.add_argument("--labbook", default="labbook.md")
p.add_argument("--no-labbook", action="store_true", help="print the record instead")
p.add_argument("--note", default="", help="a line of your own stored with the record")
args = p.parse_args()
marker = random.Random(args.seed).choice(MARKER_WORDS)
print(f"badge word: {marker} model: {args.model} planted in: {args.plant_in} message")
# Calibration: two small conversations give the fixed cost and the cost per exchange.
counts = []
for pairs in CALIBRATION_PAIRS:
try:
out = ask(args, build_messages(marker, pairs, args.plant_in))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300]
sys.exit(f"calibration request failed with HTTP {exc.code}: {detail}\n"
"If the server rejects reasoning_effort, rerun with --reasoning-effort ''.")
except (urllib.error.URLError, OSError) as exc:
sys.exit(f"cannot reach {args.base_url}: {exc}")
if not out["answer"] and out["finish_reason"] == "length":
sys.exit(f"calibration with {pairs} exchange(s): an empty answer that hit --max-tokens "
f"({args.max_tokens}). The model spent its budget thinking: keep "
"--reasoning-effort none, or raise --max-tokens for a server that ignores it.")
if marker.lower() not in out["answer"].lower():
sys.exit(f"calibration with {pairs} exchange(s): the answer was {out['answer']!r} "
f"(finish_reason {out['finish_reason']}). The model did not return the word "
"from a short conversation, so no context length can be measured with it: "
"the model, not the window, is the limit. See the page's troubleshooting.")
counts.append(out["prompt_tokens"])
if None in counts:
sys.exit("the server reported no usage.prompt_tokens, so the step sizes cannot be "
"calibrated. Use a server that reports usage (Ollama, llama-server, LM Studio do).")
per_pair = (counts[1] - counts[0]) / (CALIBRATION_PAIRS[1] - CALIBRATION_PAIRS[0])
fixed = counts[0] - per_pair * CALIBRATION_PAIRS[0]
print(f"calibration: {fixed:.0f} fixed tokens + {per_pair:.1f} tokens per filler exchange")
facts = server_facts(args)
for key, value in facts.items():
print(f"server reports {key} = {value}")
if args.around:
targets = [int(args.around * f) for f in AROUND_FRACTIONS]
elif args.steps:
targets = [int(s) for s in args.steps.split(",") if s.strip()]
else:
targets = [1024, 2048, 4096, 8192, 16384, 32768]
print()
print(f"{'target':>8} {'pairs':>6} {'sent':>7} {'counted':>8} {'seconds':>8} "
f"{'recalled':<8} {'verdict':<11} answer")
results, losses = [], 0
for target in targets:
pairs = max(1, round((target - fixed) / per_pair))
sent = int(round(fixed + per_pair * pairs))
row = {"target_tokens": target, "filler_pairs": pairs, "sent_tokens": sent}
try:
out = ask(args, build_messages(marker, pairs, args.plant_in))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300]
try: # llama-server and Ollama put the reason in error.message or error
err = json.loads(detail).get("error")
reason = err.get("message") if isinstance(err, dict) else str(err)
except (ValueError, AttributeError):
reason = detail
print(f"{target:>8} {pairs:>6} {sent:>7} {'-':>8} {'-':>8} {'-':<8} "
f"{'refused':<11} HTTP {exc.code}: {reason}")
row.update({"verdict": "refused", "http_status": exc.code, "detail": detail})
results.append(row)
break
except (urllib.error.URLError, OSError) as exc:
sys.exit(f"cannot reach {args.base_url}: {exc}")
recalled = marker.lower() in out["answer"].lower()
v = verdict(recalled, sent, out["prompt_tokens"], out["finish_reason"], out["answer"])
counted = "-" if out["prompt_tokens"] is None else out["prompt_tokens"]
shown = out["answer"].replace("\n", " ")[:32] or f"(empty, {out['finish_reason']})"
print(f"{target:>8} {pairs:>6} {sent:>7} {counted:>8} {out['seconds']:>8} "
f"{'yes' if recalled else 'NO':<8} {v:<11} {shown}")
row.update({"counted_tokens": out["prompt_tokens"], "seconds": out["seconds"],
"recalled": recalled, "verdict": v, "finish_reason": out["finish_reason"],
"answer": out["answer"][:120]})
results.append(row)
losses = 0 if recalled else losses + 1
if losses >= 2 and not args.keep_going:
print("two steps in a row lost the word; stopping (use --keep-going to continue)")
break
fitted = [r["sent_tokens"] for r in results if r["verdict"] == "fits"]
cuts = [r for r in results if r["verdict"] in ("cut-lost", "cut-kept", "refused")]
misses = [r["sent_tokens"] for r in results if r["verdict"] == "model"]
other = sorted({r["verdict"] for r in results} & {"no-answer", "recall-only"})
print()
print(f"largest conversation that fitted: {max(fitted) if fitted else 'none'} tokens sent")
if cuts:
print(f"first step the server cut or refused: {cuts[0]['sent_tokens']} tokens sent "
f"({cuts[0]['verdict']})")
else:
print("no step was cut or refused: the context is larger than the largest step sent")
if misses:
print(f"steps the model missed with every token in view: {misses}")
if other:
print(f"steps that could not be judged: {', '.join(other)} (see the page's troubleshooting)")
record = {
"lab": "part-07/reality-check-the-default-context-is-enough",
"recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
"base_url": args.base_url, "model": args.model, "marker": marker,
"plant_in": args.plant_in, "reasoning_effort": args.reasoning_effort,
"calibration": {"fixed_tokens": round(fixed, 1), "tokens_per_pair": round(per_pair, 1)},
"server_facts": facts,
"largest_fit_sent_tokens": max(fitted) if fitted else None,
"first_cut": ({"sent_tokens": cuts[0]["sent_tokens"], "verdict": cuts[0]["verdict"]}
if cuts else None),
"model_miss_sent_tokens": misses,
"steps": results, "note": args.note,
}
if args.no_labbook:
print(json.dumps(record))
return
with open(args.labbook, "a", encoding="utf-8") as fh:
fh.write(json.dumps(record) + "\n")
print(f"recorded in {args.labbook}")
if __name__ == "__main__":
main()