Instrument MTP3 reuse and preserve continuation checkpoints in source builds

performance/harnesses/vllm/prefill_checks/.gitattributes

@@ -0,0 +1 @@
+*.py text eol=lf

performance/harnesses/vllm/prefill_checks/continuation_serve_checks.py

@@ -0,0 +1,249 @@
+"""Bounded semantic and cold-prefill checks, gated on completed API warmup."""
+
+import argparse
+import os
+import json
+from pathlib import Path
+import re
+import subprocess
+import time
+import urllib.request
+import uuid
+
+BASE = os.environ.get("BENCH_API_BASE", "http://127.0.0.1:8000")
+MODEL = os.environ.get("BENCH_MODEL", "glm-5.3-flash-spark")
+CONTAINER = os.environ.get("BENCH_RANK0_CONTAINER", "sparkring-model-r0")
+if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*", CONTAINER):
+    raise ValueError("Unexpected container name")
+
+
+def cached_prompt_tokens(usage):
+    """Missing accounting cannot establish cold work or successful cache reuse."""
+    details = usage.get("prompt_tokens_details")
+    cached = details.get("cached_tokens") if isinstance(details, dict) else None
+    if type(cached) is not int or cached < 0:
+        raise ValueError("Usage must provide a nonnegative integer cached_tokens")
+    return cached
+
+
+def get(path):
+    with urllib.request.urlopen(BASE + path, timeout=10) as r:
+        return r.read().decode()
+
+
+def post(path, body):
+    req = urllib.request.Request(
+        BASE + path,
+        data=json.dumps(body).encode(),
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=180) as r:
+        return json.load(r)
+
+
+def ready():
+    status = subprocess.check_output(
+        [
+            "ssh",
+            os.environ.get("BENCH_RANK0_SSH", "rank0"),
+            "docker inspect --format '{{.State.Health.Status}}' " + CONTAINER,
+        ],
+        text=True,
+    ).strip()
+    if status != "healthy":
+        return False
+    metrics = get("/metrics")
+    values = {
+        key: [
+            float(line.rsplit(" ", 1)[-1])
+            for line in metrics.splitlines()
+            if line.startswith("vllm:" + key + "{")
+            or line.startswith("vllm:" + key + " ")
+        ]
+        for key in ("num_requests_running", "num_requests_waiting")
+    }
+    return all(rows and sum(rows) == 0 for rows in values.values())
+
+
+def calibrate(tokens, fact):
+    prefix = (
+        "Test "
+        + uuid.uuid4().hex
+        + ". The project code is "
+        + fact
+        + ". Remember it.\n"
+    )
+    suffix = "\nWhat is the project code? Reply with just the code."
+    words = tokens
+    for _ in range(12):
+        text = (
+            prefix
+            + " ".join(
+                (["alpha", "beta", "gamma", "delta"] * ((words + 3) // 4))[:words]
+            )
+            + suffix
+        )
+        msg = [{"role": "user", "content": text}]
+        count = post(
+            "/tokenize",
+            {"model": MODEL, "messages": msg, "add_generation_prompt": True},
+        )["count"]
+        if count == tokens:
+            return msg
+        words += tokens - count
+    raise ValueError("Exact prompt calibration failed")
+
+
+def semantic(tokens, fact):
+    messages = calibrate(tokens, fact)
+    rows = []
+    for phase in ("fresh", "repeated", "extended"):
+        msg = (
+            messages
+            if phase != "extended"
+            else [
+                {
+                    "role": "user",
+                    "content": messages[0]["content"]
+                    + "\nFinal instruction: give exactly the project code.",
+                }
+            ]
+        )
+        started = time.monotonic()
+        response = post(
+            "/v1/chat/completions",
+            {
+                "model": MODEL,
+                "messages": msg,
+                "max_tokens": 384,
+                "temperature": 0,
+                "top_p": 1,
+            },
+        )
+        choice = response["choices"][0]
+        answer = (choice["message"].get("content") or "").strip()
+        usage = response["usage"]
+        cached = cached_prompt_tokens(usage)
+        row = {
+            "tokens": tokens,
+            "phase": phase,
+            "seconds": time.monotonic() - started,
+            "response": response,
+            "answer_ok": answer == fact and choice["finish_reason"] == "stop",
+            "cache_ok": cached == 0 if phase == "fresh" else cached > 0,
+        }
+        row["pass"] = row["answer_ok"] and row["cache_ok"]
+        if phase == "fresh":
+            row["pass"] = row["pass"] and usage["prompt_tokens"] == tokens
+        rows.append(row)
+        print(
+            json.dumps(
+                {k: v for k, v in row.items() if k != "response"}
+                | {"answer": answer, "cached_tokens": cached}
+            ),
+            flush=True,
+        )
+        yield row
+        if not row["pass"]:
+            raise RuntimeError("Semantic/cache check failed; inspect response")
+
+
+def prefill(tokens):
+    msg = calibrate(tokens, "STONE-7482")
+    body = {
+        "model": MODEL,
+        "messages": msg,
+        "max_tokens": 1,
+        "temperature": 0,
+        "stream": True,
+        "stream_options": {"include_usage": True},
+    }
+    request = urllib.request.Request(
+        BASE + "/v1/chat/completions",
+        data=json.dumps(body).encode(),
+        headers={"Content-Type": "application/json"},
+    )
+    started = time.monotonic()
+    first = None
+    usage = None
+    with urllib.request.urlopen(request, timeout=180) as response:
+        for line in response:
+            if not line.startswith(b"data: "):
+                continue
+            raw = line[6:].strip()
+            if raw == b"[DONE]":
+                break
+            chunk = json.loads(raw)
+            if chunk.get("usage"):
+                usage = chunk["usage"]
+            for choice in chunk.get("choices", []):
+                delta = choice.get("delta", {})
+                if first is None and any(
+                    delta.get(k) for k in ("content", "reasoning", "reasoning_content")
+                ):
+                    first = time.monotonic() - started
+    if first is None or usage is None:
+        raise RuntimeError("No streamed token or final usage")
+    assert usage["prompt_tokens"] == tokens
+    assert cached_prompt_tokens(usage) == 0
+    return {
+        "tokens": tokens,
+        "ttft_seconds": first,
+        "tokens_per_second": tokens / first,
+        "usage": usage,
+    }
+
+
+def main():
+    p = argparse.ArgumentParser()
+    p.add_argument("phase", choices=["semantic", "prefill"])
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--sizes", default="16384,10240,32768,65536,8192")
+    p.add_argument("--samples", type=int, default=3)
+    args = p.parse_args()
+    assert not args.output.exists()
+    deadline = time.monotonic() + 1200
+    announced = False
+    while not ready():
+        if time.monotonic() > deadline:
+            raise TimeoutError("Warmup/idle readiness deadline")
+        if not announced:
+            print(
+                "Waiting for completed warmup and zero running/waiting requests.",
+                flush=True,
+            )
+            announced = True
+        time.sleep(5)
+    print("Warmup complete; server idle. Starting controlled checks.", flush=True)
+    rows = []
+
+    def save():
+        args.output.write_text(
+            json.dumps({"phase": args.phase, "rows": rows}, indent=2), encoding="utf-8"
+        )
+
+    sizes = list(map(int, args.sizes.split(",")))
+    if args.phase == "semantic":
+        for i, tokens in enumerate(sizes):
+            for row in semantic(tokens, "RIVER-" + str(5938 + i)):
+                rows.append(row)
+                save()
+    else:
+        for phase in ("warm", "measured"):
+            for sample in range(1 if phase == "warm" else args.samples):
+                for tokens in sizes:
+                    if not ready():
+                        raise RuntimeError(
+                            "Concurrent activity detected before timing sample"
+                        )
+                    row = prefill(tokens) | {"phase": phase, "sample": sample}
+                    rows.append(row)
+                    save()
+                    print(
+                        json.dumps({k: v for k, v in row.items() if k != "usage"}),
+                        flush=True,
+                    )
+
+
+if __name__ == "__main__":
+    main()

performance/harnesses/vllm/prefill_checks/test_prefill_checks.py

@@ -0,0 +1,137 @@
+"""Exercise request timing and readiness gates without contacting a server."""
+
+import importlib.util
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+HERE = Path(__file__).parent
+
+
+@pytest.fixture(
+    params=sorted(
+        path for path in HERE.glob("*_checks.py") if not path.name.startswith("test_")
+    )
+)
+def harness(request):
+    spec = importlib.util.spec_from_file_location(
+        "prefill_harness_" + request.param.stem, request.param
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.mark.parametrize(
+    "health,running,waiting,expected",
+    [
+        ("healthy", 0, 0, True),
+        ("starting", 0, 0, False),
+        ("healthy", 1, 0, False),
+        ("healthy", 0, 1, False),
+    ],
+)
+def test_readiness_requires_healthy_idle_service(
+    harness, monkeypatch, health, running, waiting, expected
+):
+    monkeypatch.setattr(harness.subprocess, "check_output", lambda *a, **kw: health)
+    monkeypatch.setattr(
+        harness,
+        "get",
+        lambda path: (
+            f'vllm:num_requests_running{{model="m"}} {running}\n'
+            f'vllm:num_requests_waiting{{model="m"}} {waiting}\n'
+        ),
+    )
+    assert harness.ready() is expected
+
+
+def test_missing_metrics_do_not_count_as_zero(harness, monkeypatch):
+    monkeypatch.setattr(harness.subprocess, "check_output", lambda *a, **kw: "healthy")
+    monkeypatch.setattr(harness, "get", lambda path: "")
+    assert harness.ready() is False
+
+
+@pytest.mark.parametrize(
+    "tokens,cached,rejected", [(8192, 0, False), (8191, 0, True), (8192, 1, True)]
+)
+def test_timing_rejects_wrong_prompt_length_or_cached_work(
+    harness, monkeypatch, tokens, cached, rejected
+):
+    clock = iter([10.0, 10.25])
+    clock_name = (
+        "perf_counter"
+        if harness.__name__.endswith("mhc_precise_checks")
+        else "monotonic"
+    )
+    monkeypatch.setattr(harness.time, clock_name, lambda: next(clock))
+    monkeypatch.setattr(
+        harness, "calibrate", lambda *args: [{"role": "user", "content": "test"}]
+    )
+    chunks = [
+        {"choices": [{"delta": {"content": ""}}]},
+        {"choices": [{"delta": {"content": "answer"}}]},
+        {
+            "choices": [],
+            "usage": {
+                "prompt_tokens": tokens,
+                "prompt_tokens_details": {"cached_tokens": cached},
+            },
+        },
+    ]
+    stream = (
+        b"".join(b"data: " + json.dumps(chunk).encode() + b"\n" for chunk in chunks)
+        + b"data: [DONE]\n"
+    )
+    monkeypatch.setattr(
+        harness.urllib.request, "urlopen", lambda *a, **kw: io.BytesIO(stream)
+    )
+    if rejected:
+        with pytest.raises(AssertionError):
+            harness.prefill(8192)
+    else:
+        result = harness.prefill(8192)
+        assert result["ttft_seconds"] == 0.25
+        assert result["tokens_per_second"] == 32768
+
+
+@pytest.mark.parametrize(
+    "details",
+    [
+        "omitted",
+        None,
+        {},
+        {"cached_tokens": False},
+        {"cached_tokens": "0"},
+        {"cached_tokens": -1},
+    ],
+)
+def test_timing_rejects_unproven_cache_accounting(harness, monkeypatch, details):
+    clock = iter([10.0, 10.25])
+    clock_name = (
+        "perf_counter"
+        if harness.__name__.endswith("mhc_precise_checks")
+        else "monotonic"
+    )
+    monkeypatch.setattr(harness.time, clock_name, lambda: next(clock))
+    monkeypatch.setattr(
+        harness, "calibrate", lambda *args: [{"role": "user", "content": "test"}]
+    )
+    usage = {"prompt_tokens": 8192}
+    if details != "omitted":
+        usage["prompt_tokens_details"] = details
+    chunks = [
+        {"choices": [{"delta": {"content": "answer"}}]},
+        {"choices": [], "usage": usage},
+    ]
+    stream = (
+        b"".join(b"data: " + json.dumps(chunk).encode() + b"\n" for chunk in chunks)
+        + b"data: [DONE]\n"
+    )
+    monkeypatch.setattr(
+        harness.urllib.request, "urlopen", lambda *a, **kw: io.BytesIO(stream)
+    )
+    with pytest.raises(ValueError, match="cached_tokens"):
+        harness.prefill(8192)

performance/records/glm53-flash/continuation-checkpoints-20260906/.gitattributes

@@ -0,0 +1 @@
+*.json -text whitespace=cr-at-eol