From 0e0d2cc58e03200e2348cba2e9ec12332bc3f61e Mon Sep 17 00:00:00 2001 From: Advay Monga Date: Thu, 8 Oct 2026 18:44:47 -0400 Subject: [PATCH] lab: request-trace contract (TRACE.md) and its validator An engine's per-request rows, engine-agnostic: trace_id = the client's X-Trace-Id, arrival_ts epoch seconds, *_s monotonic durations, TPOT = (E2E - TTFT)/(n - 1), terminal states, schema version, and drop counts in a meta table/file. A dir of *.sqlite (requests + meta) or *.jsonl + .meta.json; python -m lab.trace_contract DIR. inference-server's TELEMETRY_DIR conforms natively from inference-server#102; targets name the env var in [engine.telemetry] dir_env. vLLM OTel mapping noted. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 ++ correctness/client.py | 13 ++- correctness/run.py | 4 +- lab/README.md | 4 +- lab/TRACE.md | 99 ++++++++++++++++++ lab/artifacts.py | 10 +- lab/evaltools.py | 3 +- lab/profile.py | 2 +- lab/proftools.py | 4 +- lab/tools.py | 12 ++- lab/trace_contract.py | 172 +++++++++++++++++++++++++++++++ targets/inference-server.toml | 2 +- tests/fake_engine.py | 3 + tests/test_lab_artifacts.py | 9 ++ tests/test_lab_evaltools.py | 11 +- tests/test_lab_profile.py | 4 +- tests/test_lab_proftools.py | 20 +++- tests/test_lab_trace_contract.py | 137 ++++++++++++++++++++++++ 18 files changed, 497 insertions(+), 18 deletions(-) create mode 100644 lab/TRACE.md create mode 100644 lab/trace_contract.py create mode 100644 tests/test_lab_trace_contract.py diff --git a/CLAUDE.md b/CLAUDE.md index d2157bd..fa79279 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,6 +57,12 @@ pytest/ruff config). Each folder with commands has its own `python -m`. measures each regime's run-to-run band from repeated reference runs into `knowledge/noise/.json` (which `submit` reads). Exit 0 only if both hold. Needs a GPU for a real target; tested on fakes. `lab/LEDGER.md` is the record-format spec: the environment's public interface. +- `lab/TRACE.md` + `lab/trace_contract.py` — the request-trace contract: per-request engine rows (trace_id = the + client's `X-Trace-Id`, `arrival_ts` epoch s, `*_s` monotonic durations, TPOT = (E2E − TTFT)/(n − 1)) in a dir of + `*.sqlite` (`requests` + `meta`) or `*.jsonl` + `.meta.json`, with schema version and drop counts; + `python -m lab.trace_contract DIR`. Diagnostics only, never scored. inference-server's `TELEMETRY_DIR` conforms + natively (schema 1, inference-server#102); the target names its env vars in `[engine.telemetry] env` (`{dir}`); serve collects + the files after every bench/equiv/submit and the contract check's report is stored beside them. vLLM OTel span mapping is in the spec, not implemented. - `corpus/` — frozen workload traces (BurstGPT timing, WildChat text), seen/heldout, hashed, `manifest.json`; `build_corpus.py` / `fetch_traces.py` rebuild it; any change is a new `corpus_version`. - `knowledge/` — measured findings (one JSON each) + `evidence/`; seeded into the ledger as `finding` records. diff --git a/correctness/client.py b/correctness/client.py index 872aaef..9739528 100644 --- a/correctness/client.py +++ b/correctness/client.py @@ -35,8 +35,9 @@ class NoLogprobs(RuntimeError): DEFAULT = Api() -def _post(url: str, body: dict, timeout: float = 600) -> dict: - req = urllib.request.Request(url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}) +def _post(url: str, body: dict, timeout: float = 600, trace_id: str | None = None) -> dict: + headers = {"Content-Type": "application/json", **({"X-Trace-Id": trace_id} if trace_id else {})} + req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) @@ -48,9 +49,11 @@ def _chat_body(api: Api, model: str, messages: list[dict], max_tokens: int) -> d return body -def generate(base_url: str, model: str, messages: list[dict], max_tokens: int = 2048, api: Api = DEFAULT) -> dict: - """Greedy. Returns {"text", "completion_tokens", "finish_reason"}.""" - out = _post(f"{base_url.rstrip('/')}/v1/chat/completions", _chat_body(api, model, messages, max_tokens)) +def generate(base_url: str, model: str, messages: list[dict], max_tokens: int = 2048, api: Api = DEFAULT, + trace_id: str | None = None) -> dict: + """Greedy. Returns {"text", "completion_tokens", "finish_reason"}. `trace_id` rides as X-Trace-Id.""" + out = _post(f"{base_url.rstrip('/')}/v1/chat/completions", _chat_body(api, model, messages, max_tokens), + trace_id=trace_id) ch = out["choices"][0] return {"text": ch["message"]["content"] or "", "completion_tokens": (out.get("usage") or {}).get("completion_tokens"), "finish_reason": ch.get("finish_reason")} diff --git a/correctness/run.py b/correctness/run.py index 3622695..9c9333d 100644 --- a/correctness/run.py +++ b/correctness/run.py @@ -50,6 +50,7 @@ def _jsonl_read(path: Path) -> list[dict]: return [json.loads(l) for l in Path(path).read_text().splitlines() if l.strip()] if Path(path).exists() else [] +TRACE_PREFIX = "eq" # X-Trace-Id of a gate request: eq-, so equiv rows join engine telemetry RETRIES = 6 # a 429/503 (the server shedding under the gate's load) is retried with backoff @@ -75,7 +76,8 @@ def collect(url: str, model: str, items: list[dict], out: Path, concurrency: int def one(it): try: - g = _retrying(client.generate, url, model, it["messages"], it.get("max_tokens", 2048), api) + g = _retrying(client.generate, url, model, it["messages"], it.get("max_tokens", 2048), api, + trace_id=f"{TRACE_PREFIX}-{it['id']}") except Exception as e: # a failed request is a recorded fact, not a crash g = {"text": "", "completion_tokens": None, "finish_reason": None, "error": f"{type(e).__name__}: {e}"[:200]} return {"id": it["id"], **g} diff --git a/lab/README.md b/lab/README.md index a192ecd..4b00468 100644 --- a/lab/README.md +++ b/lab/README.md @@ -30,7 +30,9 @@ copy it, change the engine section (and the reference if the model changes), and - Engine instrumentation the lab reads: `TIMELINE_DIR` turns on the event timeline (`src/inference_server/timeline.py`), `TELEMETRY_DIR` the per-request rows. The target names them in `[engine.telemetry] env` with `{dir}`; serve points them at `lab-telemetry/` in the served tree and collects what - appeared there after teardown. Optional: without the section nothing is collected. + appeared there after teardown. Optional: without the section nothing is collected. The rows follow the + engine-agnostic request-trace contract in `lab/TRACE.md` (a dir of `*.sqlite` or `*.jsonl` + meta, keyed by + `X-Trace-Id`); check a dir with `python -m lab.trace_contract DIR`. ## Tools diff --git a/lab/TRACE.md b/lab/TRACE.md new file mode 100644 index 0000000..e2dea76 --- /dev/null +++ b/lab/TRACE.md @@ -0,0 +1,99 @@ +# Request traces: what an engine reports per request + +The engine's own account of each request: one row per request, keyed by the trace id the client sent. The +environment's client rows (`regimes/client.py`) are still the measurement, and the referee never scores these +rows. They exist so the researcher can join "what the client saw" with "where the engine spent the time". An +engine that writes nothing still gets judged. Checked by `lab/trace_contract.py` +(`python -m lab.trace_contract DIR`). + +## How an engine exposes it + +The target names the engine's environment variables in `[engine.telemetry] env`, with `{dir}` where the directory goes (for inference-server, `TELEMETRY_DIR` and `TIMELINE_DIR`). +The lab sets that variable to a fresh, empty directory for each launch. When the engine exits, the lab collects +every file in it. The directory holds only trace files, one or more (one per engine process), in either format: + +- **SQLite** `*.sqlite`: a table `requests` with one column per field below, and a table `meta (key TEXT PRIMARY + KEY, value)`. +- **JSONL** `*.jsonl`: one JSON object per row, plus `.meta.json`, one object with the meta keys. + +The meta keys are all required: + +| key | type | meaning | +|---|---|---| +| `schema_version` | int | the contract version the file follows (now `1`) | +| `rows_written` | int | rows in this file | +| `rows_dropped` | int | rows the engine produced but could not write (queue full, write error, after shutdown) | +| `closed` | int 0/1 | 1 once the engine shut down cleanly, so the counts are final | + +Drops are never silent. A file without `rows_dropped` fails the check. A closed file whose `rows_written` differs +from its row count also fails. + +## Clocks and units + +- One absolute time per row: `arrival_ts`, **epoch seconds** (wall clock, float). Epoch milliseconds or + nanoseconds and monotonic readings (seconds since boot or since the process started) are rejected. +- Every field ending in `_s` is a **duration in seconds**, measured on a monotonic clock. A duration is never the + difference of two wall-clock readings, and never of a wall reading and a monotonic one. +- **Arrival** is when the engine received the HTTP request, before tokenizing. The wall-clock `arrival_ts` and the + monotonic origin of the durations must be one instant: read both clocks together, or back-date one from the other. +- Token counts use the engine's own tokenizer. The client recounts with the reference tokenizer, so the two may + differ. + +## Required fields + +| field | type | rule | +|---|---|---| +| `schema_version` | int | `1` | +| `trace_id` | str | the client's `X-Trace-Id`, verbatim; non-empty. A retry may reuse it, so duplicates are allowed | +| `arrival_ts` | float | epoch seconds at arrival | +| `terminal_state` | str | one of `ok`, `rejected_429`, `rejected_400`, `expired`, `preempted`, `error`, `cancelled` (the client went away), `aborted` (the engine shut down with the request unfinished) | +| `prompt_tokens` | int | ≥ 0 | +| `tokens_out` | int | ≥ 0, real tokens only (no EOS) | +| `total_s` | float | arrival to the request's end (last token, rejection or abort), ≥ 0 | +| `ttft_s` | float or null | arrival to the first real token. Set exactly when `tokens_out ≥ 1`, and ≤ `total_s` | +| `tpot_s` | float or null | `(total_s − ttft_s) / (tokens_out − 1)` when `tokens_out ≥ 2`, else null. The environment's definition, checked to 0.1% | + +## Optional fields + +Typed if present. Null means "not measured". Any other column is allowed and ignored. + +| field | type | meaning | +|---|---|---| +| `session_id`, `config_id`, `prefill_mode` | str | client session; applied config; how the prompt was prefilled | +| `turn_index` | int | position in the session | +| `queue_wait_s`, `prefill_s`, `decode_s` | float | spans. `queue_wait_s + prefill_s ≤ ttft_s` | +| `replica_age_s` | float | seconds since the engine started serving | +| `pending_depth`, `active_size`, `max_batch_size`, `concurrent_sessions` | int | load at arrival | +| `kv_free_blocks` | int | KV-cache blocks free at arrival | +| `kv_free_frac` | float | the same as a fraction in [0, 1] | +| `cache_hit_tokens` | int | prompt tokens served from a prefix cache, ≤ `prompt_tokens`. Null when unknown, never a stand-in 0 | +| `decode_steps`, `decode_batch_width_max`, `preempted` | int | forward passes after prefill; widest batch it decoded in; preemptions | +| `decode_batch_width_mean` | float | mean batch width over its own decode steps | + +## inference-server + +`TELEMETRY_DIR` makes the engine write `.sqlite`, in the SQLite layout above with the same column names +(`src/inference_server/telemetry.py`, schema 1 since inference-server#102), so no adapter is needed. Its +`TIMELINE_DIR` event log (`events.jsonl`: `ts` epoch seconds, `step`, `kind`, `trace_id` on request events) is +engine-specific and outside this contract. It joins to the rows on `trace_id`. + +## vLLM (mapping only, not implemented) + +vLLM's Prometheus metrics (`vllm:time_to_first_token_seconds`, `vllm:e2e_request_latency_seconds`, ...) are +histograms, which can't be split back into rows. Its OpenTelemetry request spans (`--otlp-traces-endpoint`) are +per request and could map like this. The attribute names below are from the `gen_ai.*` semantic conventions; check +them against the vLLM version in use. + +- `trace_id` ← the request id. The client also sends `X-Request-Id` (honoured with `--enable-request-id-headers`), + or the span's W3C trace id when the client sends `traceparent`. +- `arrival_ts` ← the span start time (ns since epoch, divided by 1e9). +- `ttft_s` ← `gen_ai.latency.time_to_first_token`. `total_s` ← `gen_ai.latency.e2e`. +- `queue_wait_s` / `prefill_s` / `decode_s` ← `gen_ai.latency.time_in_queue` / `time_in_model_prefill` / + `time_in_model_decode`. +- `prompt_tokens` / `tokens_out` ← `gen_ai.usage.prompt_tokens` / `completion_tokens`. `tpot_s` is derived. +- `terminal_state` ← `finish_reason` (`stop`/`length` → `ok`, `abort` → `cancelled`). +- Requests rejected before scheduling emit no span. Their rows are missing, so `rows_dropped` can't count them. +- vLLM measures from its own engine arrival, which can come after tokenizing. + +An OTLP collector's file exporter writes the spans as JSON. A small adapter would turn them into the JSONL +layout above. diff --git a/lab/artifacts.py b/lab/artifacts.py index 9c3ac42..6ef4b79 100644 --- a/lab/artifacts.py +++ b/lab/artifacts.py @@ -25,6 +25,12 @@ def store(work: Path, root: Path) -> dict: def blob(name: str, suffix: str = "") -> str | None: p = Path(work) / name return ledger.put_blob(p, root, suffix) if p.exists() else None + tel = Path(work) / "telemetry" + if tel.is_dir() and any(tel.iterdir()): # the request-trace contract check travels with the data (lab/TRACE.md) + from lab.trace_contract import validate + rep = validate(tel) + (tel / "contract.json").write_text(json.dumps({"valid": rep.valid, "errors": rep.errors[:50], + "rows": len(rep.rows), "files": rep.files}, default=str)) out = {"device": blob("device"), "telemetry": blob("telemetry"), "client_rows": blob(ROWS, ".jsonl"), "serve_log": blob("serve.log", ".log")} shutil.rmtree(work, ignore_errors=True) @@ -36,7 +42,7 @@ def telemetry_rows(d: Path | None) -> list[dict]: if d is None: return [] rows = [] - for p in sorted(Path(d).rglob("*.sqlite")): + for p in sorted(Path(d).glob("*.sqlite")): # top level only: the timeline lives in timeline/ con = sqlite3.connect(f"file:{p}?mode=ro", uri=True) con.row_factory = sqlite3.Row try: @@ -45,7 +51,7 @@ def telemetry_rows(d: Path | None) -> list[dict]: pass finally: con.close() - for p in sorted(Path(d).rglob("*.jsonl")): + for p in sorted(Path(d).glob("*.jsonl")): rows += [r for r in map(json.loads, p.read_text().splitlines()) if isinstance(r, dict) and "trace_id" in r] return rows diff --git a/lab/evaltools.py b/lab/evaltools.py index 56d7126..bf1c44f 100644 --- a/lab/evaltools.py +++ b/lab/evaltools.py @@ -287,6 +287,7 @@ def baseline(self, commit: str) -> dict: if have: return have names = list(self.bench_default_regimes) + self._t0 = time.monotonic() try: seen = self.base_seen(target.load(), names) result = {"verdict": "ok", "metrics": {n: seen[n] for n in names}} @@ -294,7 +295,7 @@ def baseline(self, commit: str) -> dict: result = {"verdict": "error", "reason": _why(e)} return ledger.append({"kind": "baseline", "run": self.s.run_id, "session": self.s.session_id, "config": {"split": "seen", "tier": BENCH_TIER, "commit": commit}, - "result": result, "cost": {"usd": 0.0}}, self.s.ledger_root) + "result": result, "cost": self._cost("baseline")}, self.s.ledger_root) def base_heldout(self, t, names): return self._base_measure(t, names, "heldout", "full") diff --git a/lab/profile.py b/lab/profile.py index 43095e0..d458f5d 100644 --- a/lab/profile.py +++ b/lab/profile.py @@ -191,7 +191,7 @@ async def run(backend: InferenceBackend, settings: Settings, prompts: list[list[ device, settings, prompts, max_tokens, t0, t1, failed=done["errors"] + done["empty"], outcomes=done, warmup={"requests": warmup, "concurrent": True, "prompts": "random ids at the window's prompt lengths", "max_tokens": max_tokens, "wall_s": w1 - w0}, - profiler_on=profiler == "torch", profiler=prof_config if profiler == "torch" else {"kind": profiler}, + profiler_on=profiler == "torch", profiler={"kind": profiler, **(prof_config if profiler == "torch" else {})}, rounds=rounds, gpu={"window_start": gpu0, "window_end": gpu1}, timeline=timeline.stats(), workload=source)) return out diff --git a/lab/proftools.py b/lab/proftools.py index 07898db..d58a058 100644 --- a/lab/proftools.py +++ b/lab/proftools.py @@ -53,7 +53,7 @@ def _schema(**extra) -> dict: def _head(path: Path | None) -> str: - return path.read_text(errors="replace")[:SUMMARY_CHARS] if path and path.is_file() else "" + return path.read_text(errors="replace")[:SUMMARY_CHARS] if path and path.is_file() and not path.is_symlink() else "" class ProfTools: @@ -113,6 +113,7 @@ def _instrumented(self, tool: str, args: dict, binary: str, wrap, flags: list[st proc = T.default_profile_runner(tree, argv, read=read) log = proc.stdout + proc.stderr commands = [argv] + T._drop_links(out) if post and proc.returncode == 0: pargv, dest = post(exe, out) commands.append(pargv) @@ -121,6 +122,7 @@ def _instrumented(self, tool: str, args: dict, binary: str, wrap, flags: list[st dest.write_text(p.stdout) log += p.stderr if dest else p.stdout + p.stderr proc = p if p.returncode else proc + T._drop_links(out) found = sorted(out.glob(summary)) head = _head(found[-1] if found else None) blob, visible = self._keep(out) if any(p.is_file() for p in out.rglob("*")) else ("", "") diff --git a/lab/tools.py b/lab/tools.py index ac77de8..001ff1f 100644 --- a/lab/tools.py +++ b/lab/tools.py @@ -17,7 +17,16 @@ from lab.safety import grader # Tools that occupy the GPU: their wall time is charged at the venue's rate. The rest cost $0. -GPU_TOOLS = frozenset({"profile", "trace", "kernel", "hostprof", "bench", "equiv", "submit"}) +GPU_TOOLS = frozenset({"profile", "trace", "kernel", "hostprof", "bench", "equiv", "submit", "baseline"}) + + +def _drop_links(root: Path) -> None: + """Remove symlinks the jailed run left in its output: copied outside the jail they would read anything.""" + if root.is_symlink(): + raise PermissionError(f"{root.name}: output is a symlink") + for p in root.rglob("*"): + if p.is_symlink(): + p.unlink() class Toolbox(EvalTools, ProfTools): @@ -77,6 +86,7 @@ def _pristine(self) -> Path: def _keep(self, src: Path) -> tuple[str, str]: """`src` into the ledger as a blob, and a copy in the workspace: the ledger is outside the jail, so the agent could not read it there. Returns (blob, workspace-relative copy).""" + _drop_links(src) blob = ledger.put_blob(src, self.s.ledger_root) visible = Path("lab") / "runs" / Path(blob).name dest = self.s.workspace.path / visible diff --git a/lab/trace_contract.py b/lab/trace_contract.py new file mode 100644 index 0000000..aae4612 --- /dev/null +++ b/lab/trace_contract.py @@ -0,0 +1,172 @@ +"""The request-trace contract (lab/TRACE.md): load an engine's trace directory and check every row and file against it.""" + +from __future__ import annotations + +import argparse +import json +import math +import sqlite3 +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path + +SCHEMA_VERSION = 1 +TERMINAL_STATES = frozenset({"ok", "rejected_429", "rejected_400", "expired", "preempted", "error", + "cancelled", "aborted"}) +EPOCH_MIN = 946_684_800.0 # 2000-01-01: anything smaller is a monotonic reading, not epoch seconds +MAX_DURATION_S = 86_400.0 # a request longer than a day is a unit error (ms or ns) +TPOT_RTOL = 1e-3 + +REQUIRED = {"schema_version": int, "trace_id": str, "arrival_ts": float, "terminal_state": str, + "prompt_tokens": int, "tokens_out": int, "total_s": float, "ttft_s": float, "tpot_s": float} +NULLABLE = {"ttft_s", "tpot_s"} +OPTIONAL = {"session_id": str, "config_id": str, "prefill_mode": str, "turn_index": int, + "queue_wait_s": float, "prefill_s": float, "decode_s": float, "replica_age_s": float, + "pending_depth": int, "active_size": int, "max_batch_size": int, "concurrent_sessions": int, + "kv_free_blocks": int, "kv_free_frac": float, "cache_hit_tokens": int, "decode_steps": int, + "decode_batch_width_max": int, "preempted": int, "decode_batch_width_mean": float} +DURATIONS = ("total_s", "ttft_s", "tpot_s", "queue_wait_s", "prefill_s", "decode_s") +META_KEYS = ("schema_version", "rows_written", "rows_dropped", "closed") + + +@dataclass +class Report: + rows: list[dict] = field(default_factory=list) + files: dict[str, dict] = field(default_factory=dict) # file name -> its meta + errors: list[str] = field(default_factory=list) + + @property + def valid(self) -> bool: + return not self.errors + + @property + def rows_dropped(self) -> int: + return sum(int(m.get("rows_dropped") or 0) for m in self.files.values()) + + +def _typed(v, t) -> bool: + if isinstance(v, bool): + return False + return isinstance(v, (int, float)) and math.isfinite(v) if t is float else isinstance(v, t) + + +def check_row(row: dict, now: float | None = None) -> list[str]: + """Every way `row` breaks the contract, as plain sentences; empty when it conforms.""" + now = time.time() if now is None else now + errs = [f"missing {k}" for k in REQUIRED if k not in row] + for k, t in {**REQUIRED, **OPTIONAL}.items(): + v = row.get(k) + if v is None: + if k in REQUIRED and k in row and k not in NULLABLE: + errs.append(f"{k} is null") + elif not _typed(v, t): + errs.append(f"{k} is {type(v).__name__}, expected {t.__name__}") + if errs: + return errs + if row["schema_version"] != SCHEMA_VERSION: + errs.append(f"schema_version {row['schema_version']}, expected {SCHEMA_VERSION}") + if not row["trace_id"]: + errs.append("trace_id is empty") + if row["terminal_state"] not in TERMINAL_STATES: + errs.append(f"terminal_state {row['terminal_state']!r} not one of {sorted(TERMINAL_STATES)}") + ts = row["arrival_ts"] + if ts < EPOCH_MIN: + errs.append(f"arrival_ts {ts} is not epoch seconds (a monotonic clock?)") + elif ts > now + MAX_DURATION_S: + errs.append(f"arrival_ts {ts} is in the future: epoch ms or ns, not seconds?") + for k in DURATIONS: + v = row.get(k) + if v is not None and not 0 <= v <= MAX_DURATION_S: + errs.append(f"{k} {v} outside [0, {MAX_DURATION_S:.0f}] s (mixed clocks or units?)") + n, ttft, tpot, total = row["tokens_out"], row["ttft_s"], row["tpot_s"], row["total_s"] + for k in ("prompt_tokens", "tokens_out"): + if row[k] < 0: + errs.append(f"{k} is negative") + if (n >= 1) != (ttft is not None): + errs.append(f"ttft_s must be set exactly when tokens_out >= 1 (tokens_out {n}, ttft_s {ttft})") + if ttft is not None and ttft > total * (1 + 1e-9): + errs.append(f"ttft_s {ttft} > total_s {total} (mixed units?)") + if n >= 2 and ttft is not None: + want = (total - ttft) / (n - 1) + if tpot is None or abs(tpot - want) > TPOT_RTOL * abs(want) + 1e-9: + errs.append(f"tpot_s {tpot} != (total_s - ttft_s) / (tokens_out - 1) = {want:.6g}") + elif tpot is not None: + errs.append(f"tpot_s {tpot} set with tokens_out {n} (< 2)") + qw, pf = row.get("queue_wait_s"), row.get("prefill_s") + if ttft is not None and qw is not None and pf is not None and qw + pf > ttft * (1 + 1e-6) + 1e-9: + errs.append(f"queue_wait_s + prefill_s = {qw + pf:.6g} > ttft_s {ttft} (mixed units?)") + frac = row.get("kv_free_frac") + if frac is not None and not 0 <= frac <= 1: + errs.append(f"kv_free_frac {frac} outside [0, 1]") + hit = row.get("cache_hit_tokens") + if hit is not None and not 0 <= hit <= row["prompt_tokens"]: + errs.append(f"cache_hit_tokens {hit} outside [0, prompt_tokens {row['prompt_tokens']}]") + return errs + + +def _read_sqlite(path: Path) -> tuple[list[dict], dict]: + conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + try: + tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} + rows = [dict(r) for r in conn.execute("SELECT * FROM requests")] if "requests" in tables else None + meta = dict(tuple(r) for r in conn.execute("SELECT key, value FROM meta")) if "meta" in tables else None + finally: + conn.close() + return rows, meta + + +def _read_jsonl(path: Path) -> tuple[list[dict], dict | None]: + rows = [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + meta_path = path.with_name(path.stem + ".meta.json") + return rows, (json.loads(meta_path.read_text()) if meta_path.exists() else None) + + +def validate(directory: str | Path, now: float | None = None) -> Report: + """Load every trace file in `directory` and check its meta and each row.""" + d, rep = Path(directory), Report() + paths = sorted(d.glob("*.sqlite")) + sorted(d.glob("*.jsonl")) + if not paths: + rep.errors.append(f"{d}: no *.sqlite or *.jsonl trace files") + for p in paths: + try: + rows, meta = _read_sqlite(p) if p.suffix == ".sqlite" else _read_jsonl(p) + except (sqlite3.Error, json.JSONDecodeError, UnicodeDecodeError) as e: + rep.errors.append(f"{p.name}: unreadable ({e})") + continue + if rows is None: + rep.errors.append(f"{p.name}: no requests table") + continue + if meta is None: + rep.errors.append(f"{p.name}: no meta (drop counts must be written, never silent)") + meta = {} + for k in META_KEYS: + if k in meta and not _typed(meta[k], int): + rep.errors.append(f"{p.name}: meta {k} is {type(meta[k]).__name__}, expected int") + elif k not in meta and meta: + rep.errors.append(f"{p.name}: meta missing {k}") + if meta.get("schema_version") not in (None, SCHEMA_VERSION): + rep.errors.append(f"{p.name}: meta schema_version {meta['schema_version']}, expected {SCHEMA_VERSION}") + if meta.get("closed") == 1 and meta.get("rows_written") != len(rows): + rep.errors.append(f"{p.name}: meta rows_written {meta.get('rows_written')} but {len(rows)} rows") + rep.files[p.name] = meta + for i, row in enumerate(rows): + rep.errors += [f"{p.name} row {i} ({row.get('trace_id')!r}): {e}" for e in check_row(row, now)] + rep.rows += rows + return rep + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description="Check an engine's request-trace directory against lab/TRACE.md") + ap.add_argument("dir") + rep = validate(ap.parse_args(argv).dir) + for e in rep.errors: + print(e) + print(f"{'ok' if rep.valid else 'INVALID'}: {len(rep.rows)} rows in {len(rep.files)} files, " + f"{rep.rows_dropped} dropped, {len(rep.errors)} errors") + return 0 if rep.valid else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/targets/inference-server.toml b/targets/inference-server.toml index aa702e3..de30fc0 100644 --- a/targets/inference-server.toml +++ b/targets/inference-server.toml @@ -20,7 +20,7 @@ test = "python -m pytest -q -p no:cacheprovider -m 'not heavy and not needs_host lint = "python -m ruff check ." [engine.telemetry] # optional: engine-native files serve collects after teardown -env = { TELEMETRY_DIR = "{dir}", TIMELINE_DIR = "{dir}" } # per-request SQLite rows, events.jsonl +env = { TELEMETRY_DIR = "{dir}", TIMELINE_DIR = "{dir}/timeline" } # request rows (lab/TRACE.md); events.jsonl apart [reference] # the unmodified model the correctness gate compares against api = "vllm" diff --git a/tests/fake_engine.py b/tests/fake_engine.py index bc24c74..2506ebd 100644 --- a/tests/fake_engine.py +++ b/tests/fake_engine.py @@ -54,6 +54,9 @@ def do_POST(self): text = ("42" if "17 + 25" in last else ("The answer is (B)." if wrong else "The answer is (C).") if "pick one" in last else " ".join(f"t{i}" for i in range(n))) time.sleep(token_s * n) + if os.environ.get("TELEMETRY_DIR"): + with lock, open(os.path.join(os.environ["TELEMETRY_DIR"], "requests.jsonl"), "a") as f: + f.write(json.dumps({"trace_id": self.headers.get("X-Trace-Id"), "tokens_out": n}) + "\n") return self._json({"choices": [{"message": {"role": "assistant", "content": text}, "finish_reason": "stop"}], "usage": {"completion_tokens": n}}) with lock: diff --git a/tests/test_lab_artifacts.py b/tests/test_lab_artifacts.py index d8c8bb7..68c761b 100644 --- a/tests/test_lab_artifacts.py +++ b/tests/test_lab_artifacts.py @@ -33,3 +33,12 @@ def test_absent_telemetry_is_none(tmp_path): artifacts.write_rows(work, []) a = artifacts.store(work, tmp_path / "ledger") assert a["telemetry"] is None and a["device"] is None and artifacts.joined({"result": {"artifacts": a}}, tmp_path / "ledger") == [] + + +def test_timeline_events_are_not_read_as_request_rows(tmp_path): + from lab import artifacts + d = tmp_path / "telemetry"; (d / "timeline").mkdir(parents=True) + (d / "requests.jsonl").write_text(json.dumps({"trace_id": "bg-1", "tokens_out": 3}) + "\n") + (d / "timeline" / "events.jsonl").write_text(json.dumps({"kind": "finish", "trace_id": "bg-1", "step": 4}) + "\n") + rows = artifacts.telemetry_rows(d) + assert rows == [{"trace_id": "bg-1", "tokens_out": 3}] diff --git a/tests/test_lab_evaltools.py b/tests/test_lab_evaltools.py index 06d3586..7fc4bce 100644 --- a/tests/test_lab_evaltools.py +++ b/tests/test_lab_evaltools.py @@ -100,12 +100,15 @@ def lab(tmp_path, monkeypatch): tb = Toolbox(s) tb.serve_jailed = False tb.bench_default_regimes = ("single_stream",) - saved, think = dict(suite.TIERS["short"]), suite.THINK_S + from regimes import runner + saved, think, lag = dict(suite.TIERS["short"]), suite.THINK_S, runner.MAX_CLIENT_LAG_S + runner.MAX_CLIENT_LAG_S = 0.05 # wiring tests on a shared laptop; client precision has its own test suite.TIERS["short"].update(probe_s=1.0, final_s=1.5) suite.TIERS["full"].update(probe_s=1.0, final_s=1.5) suite.THINK_S = 0.3 yield tb, s suite.TIERS["short"].update(saved); suite.TIERS["full"].update(probe_s=60.0, final_s=600.0); suite.THINK_S = think + runner.MAX_CLIENT_LAG_S = lag target._cache.clear() @@ -136,6 +139,8 @@ def test_bench_keeps_passive_data_as_blobs(lab): assert all(j["engine"] and j["engine"]["tokens_out"] == 6 for j in joined) # every request has its engine row assert not (s.run_dir / "pristine" / serve.TELEMETRY_SUBDIR).exists() # moved out of the served tree assert not list(s.run_dir.glob("passive-*")) # scratch emptied into blobs + contract = json.loads((s.ledger_root / a["telemetry"] / "contract.json").read_text()) + assert contract["valid"] is False and contract["errors"] # checked on collection; the fake's rows don't conform def test_equiv_judges_against_the_reference(lab): @@ -148,6 +153,9 @@ def test_equiv_judges_against_the_reference(lab): from lab import artifacts rows = artifacts.load(rec, s.ledger_root)["client_rows"] assert len(rows) == 40 and all("text" not in r for r in rows) + from correctness.run import TRACE_PREFIX + joined = artifacts.joined(rec, s.ledger_root, prefix=TRACE_PREFIX) + assert all(j["engine"] for j in joined) # every gate request has its engine row (X-Trace-Id) def test_submit_keeps_heldout_passive_data_out_of_the_ledger(lab): @@ -269,6 +277,7 @@ def test_baseline_is_measured_once_per_run_and_briefed_with_one_ledger_record(la assert calls == [(["single_stream"], "seen")] base = list(ledger.records(cfg.ledger_root, kind="baseline")) assert len(base) == 1 and base[0]["result"]["verdict"] == "ok" + assert base[0]["cost"]["seconds"] > 0 and "rate_known" in base[0]["cost"] assert base[0]["config"] == {"split": "seen", "tier": "short", "commit": session._resolve(cfg.repo, "HEAD")} assert (tmp_path / "runs" / "rb" / "base-seen-short.json").exists() # the cache submit compares against first, second = (sp.prompt for sp in p.specs) diff --git a/tests/test_lab_profile.py b/tests/test_lab_profile.py index 05d5db7..471cd1c 100644 --- a/tests/test_lab_profile.py +++ b/tests/test_lab_profile.py @@ -135,7 +135,7 @@ async def test_cuda_range_window_writes_no_torch_trace(tmp_path, stub_backend_cl out = await profile.run(stub_backend_cls(), Settings(max_batch_size=2), profile.synthetic_prompts(2, 4), max_tokens=2, out=tmp_path, profiler="cuda-range") assert not (out / "trace.json").exists() - assert json.loads((out / "meta.json").read_text())["profiler"] == "cuda-range" + assert json.loads((out / "meta.json").read_text())["profiler"]["kind"] == "cuda-range" FAKE_PYSPY = r'''#!/usr/bin/env python3 @@ -162,4 +162,4 @@ async def test_pyspy_samples_this_process_under_a_repeated_workload(tmp_path, st assert dump == ["dump", "--pid", str(os.getpid()), "--nonblocking"] assert (out / "pyspy.speedscope.json").exists() and "MainThread" in (out / "pyspy-dump.txt").read_text() meta = json.loads((out / "meta.json").read_text()) - assert meta["profiler"] == "pyspy" and meta["rounds"] >= 1 and meta["failed"] == 0 + assert meta["profiler"]["kind"] == "pyspy" and meta["rounds"] >= 1 and meta["failed"] == 0 diff --git a/tests/test_lab_proftools.py b/tests/test_lab_proftools.py index 86eb351..e7df890 100644 --- a/tests/test_lab_proftools.py +++ b/tests/test_lab_proftools.py @@ -188,7 +188,7 @@ def test_gpu_time_is_charged_to_the_budget_and_stops_at_the_cap(tb, monkeypatch) tb.trace({}) assert calls(tb) == [] specs = {t.name: t.description for t in tb.specs()} - assert all("Costs GPU time" in specs[n] and "$3600/h" in specs[n] for n in GPU_TOOLS) + assert all("Costs GPU time" in specs[n] and "$3600/h" in specs[n] for n in GPU_TOOLS if n in specs) assert all("Costs no GPU time ($0)" in specs[n] for n in specs if n not in GPU_TOOLS) @@ -201,3 +201,21 @@ def test_every_instrument_passes_a_workload_source_to_lab_profile(tb): ncu = [c for c in calls(tb) if c[0] == "ncu"][0] wl = after(ncu, "--workload") assert wl.endswith("_harness/workload.json") + + +def test_symlinks_left_by_the_jailed_run_are_not_copied_out(tb, monkeypatch): + secret = tb.s.run_dir / "heldout-private" / "rows.jsonl" + secret.parent.mkdir() + secret.write_text("held-out") + + def runner(tree, argv, read=()): + b = __import__("pathlib").Path(after(argv, "--out")) / "b1" + b.mkdir(parents=True) + (b / "meta.json").write_text("{}") + (b / "leak").symlink_to(secret) + (b / "dir").symlink_to(secret.parent, target_is_directory=True) + return subprocess.CompletedProcess(argv, 0, "", "") + monkeypatch.setattr(labtools, "default_profile_runner", runner) + tb.profile({"synthetic": True}) + copy = tb.s.workspace.path / last(tb, kind="profile")["result"]["workspace_copy"] + assert sorted(p.name for p in copy.iterdir()) == ["meta.json"] diff --git a/tests/test_lab_trace_contract.py b/tests/test_lab_trace_contract.py new file mode 100644 index 0000000..43abb61 --- /dev/null +++ b/tests/test_lab_trace_contract.py @@ -0,0 +1,137 @@ +"""lab.trace_contract: conforming trace dirs pass (JSONL, SQLite, the engine's own writer); bad clocks, ids and units fail.""" + +from __future__ import annotations + +import json +import sqlite3 +import time + +import pytest + +from lab import trace_contract as tc + +NOW = 1_790_000_000.0 + + +def _row(**kw) -> dict: + base = dict(schema_version=1, trace_id="bg-1", arrival_ts=NOW - 60, terminal_state="ok", prompt_tokens=10, + tokens_out=5, total_s=1.0, ttft_s=0.2, tpot_s=0.2, queue_wait_s=0.05, prefill_s=0.1, + session_id="s", cache_hit_tokens=None, prefill_mode="batched") + return {**base, **kw} + + +def _meta(n, **kw) -> dict: + return {"schema_version": 1, "rows_written": n, "rows_dropped": 0, "closed": 1, **kw} + + +def _jsonl(tmp_path, rows, meta="auto", name="run"): + (tmp_path / f"{name}.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows)) + if meta is not None: + (tmp_path / f"{name}.meta.json").write_text(json.dumps(_meta(len(rows)) if meta == "auto" else meta)) + return tmp_path + + +def _errors(row) -> list[str]: + return tc.check_row(row, now=NOW) + + +def test_conforming_jsonl_dir_passes(tmp_path): + rows = [_row(), _row(trace_id="bg-2", tokens_out=1, tpot_s=None), _row(trace_id="bg-3", terminal_state="rejected_429", + tokens_out=0, ttft_s=None, tpot_s=None, total_s=0.0, queue_wait_s=None, prefill_s=None)] + rep = tc.validate(_jsonl(tmp_path, rows), now=NOW) + assert rep.valid, rep.errors + assert len(rep.rows) == 3 and rep.rows_dropped == 0 + + +def test_conforming_sqlite_in_the_engine_layout_passes(tmp_path): + row = _row() + conn = sqlite3.connect(tmp_path / "run-x.sqlite") + conn.execute(f"CREATE TABLE requests ({', '.join(row)})") + conn.execute(f"INSERT INTO requests VALUES ({', '.join('?' * len(row))})", list(row.values())) + conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value)") + conn.executemany("INSERT INTO meta VALUES (?, ?)", list(_meta(1, rows_dropped=2, run_id="run-x").items())) + conn.commit() + conn.close() + rep = tc.validate(tmp_path, now=NOW) + assert rep.valid, rep.errors + assert rep.rows_dropped == 2 and rep.rows[0]["trace_id"] == "bg-1" + + +@pytest.mark.parametrize("row,needle", [ + ({k: v for k, v in _row().items() if k != "trace_id"}, "missing trace_id"), + (_row(trace_id=""), "trace_id is empty"), + (_row(trace_id=None), "trace_id is null"), + (_row(arrival_ts=123_456.7), "not epoch seconds"), # perf_counter / uptime + (_row(arrival_ts=NOW * 1000), "epoch ms or ns"), # epoch milliseconds + (_row(ttft_s=200.0), "ttft_s 200.0 > total_s"), # ttft in ms, total in s + (_row(queue_wait_s=50.0), "queue_wait_s + prefill_s"), # one span in ms + (_row(total_s=-0.5), "outside [0"), # wall minus monotonic + (_row(total_s=1_000_000.0, ttft_s=200.0, tpot_s=199_950.0), "outside [0"), # everything in ms... and a day long + (_row(tpot_s=0.05), "tpot_s 0.05 != (total_s - ttft_s)"), # a different TPOT definition + (_row(tokens_out=3, ttft_s=None, tpot_s=None), "ttft_s must be set exactly"), + (_row(tokens_out=0, tpot_s=None), "ttft_s must be set exactly"), + (_row(tokens_out=1), "set with tokens_out 1"), + (_row(terminal_state="done"), "terminal_state 'done'"), + (_row(schema_version=2), "schema_version 2"), + (_row(prompt_tokens="10"), "prompt_tokens is str"), + (_row(tokens_out=True), "tokens_out is bool"), + (_row(cache_hit_tokens=11), "cache_hit_tokens 11"), + (_row(kv_free_frac=1.5), "kv_free_frac 1.5"), +]) +def test_bad_rows_are_rejected_with_a_reason(row, needle): + errs = _errors(row) + assert any(needle in e for e in errs), errs + + +def test_a_bad_row_fails_the_dir_and_names_the_file_and_trace(tmp_path): + rep = tc.validate(_jsonl(tmp_path, [_row(), _row(trace_id="bg-9", arrival_ts=5.0)]), now=NOW) + assert not rep.valid and len(rep.errors) == 1 + assert rep.errors[0].startswith("run.jsonl row 1 ('bg-9')") + + +def test_drops_must_be_reported(tmp_path): + assert "no meta" in " ".join(tc.validate(_jsonl(tmp_path, [_row()], meta=None), now=NOW).errors) + + +def test_meta_missing_a_count_or_disagreeing_with_the_rows_fails(tmp_path): + a, b = tmp_path / "a", tmp_path / "b" + a.mkdir(), b.mkdir() + no_drops = {k: v for k, v in _meta(1).items() if k != "rows_dropped"} + assert "meta missing rows_dropped" in " ".join(tc.validate(_jsonl(a, [_row()], meta=no_drops), now=NOW).errors) + assert "rows_written 3 but 1 rows" in " ".join(tc.validate(_jsonl(b, [_row()], meta=_meta(3)), now=NOW).errors) + + +def test_open_file_may_lag_its_count(tmp_path): + """closed=0: the engine was killed before final counts; the row count is not held against it.""" + assert tc.validate(_jsonl(tmp_path, [_row()], meta=_meta(0, closed=0)), now=NOW).valid + + +def test_empty_dir_fails_and_cli_exit_codes(tmp_path, capsys): + assert tc.main([str(tmp_path)]) == 1 + _jsonl(tmp_path, [_row(arrival_ts=time.time() - 5)]) + assert tc.main([str(tmp_path)]) == 0 + assert capsys.readouterr().out.strip().endswith("ok: 1 rows in 1 files, 0 dropped, 0 errors") + + +def test_inference_server_rows_conform(tmp_path): + """The engine's own writer, through its own scheduler-facing API, produces a conforming dir.""" + telemetry = pytest.importorskip("inference_server.telemetry", reason="needs the engine", exc_type=ImportError) + if not hasattr(telemetry, "SCHEMA_VERSION"): + pytest.skip("engine predates the trace contract (inference-server#102)") + store = telemetry.RowStore(tmp_path) + t0 = time.perf_counter() + for i, (state, n) in enumerate([("ok", 5), ("ok", 1), ("ok", 0), ("rejected_429", 0), ("cancelled", 2)]): + rec = telemetry.RequestRecord( + trace_id=f"bg-{i}", session_id="s", turn_index=0, arrival_ts=time.time(), pending_depth=0, + active_size=0, max_batch_size=8, kv_free_blocks=10, kv_free_frac=0.5, concurrent_sessions=0, + prompt_tokens=10, replica_age_s=1.0) + admitted = state != "rejected_429" + rec.finish(state, arrival_mono=t0, enqueue_ts=t0 + 0.01, admit_ts=t0 + 0.02 if admitted else 0.0, + first_token_ts=t0 + 0.1 if admitted else 0.0, end_ts=t0 + 0.5, tokens_out=n, + cache_hit_tokens=4 if admitted else None, decode_width_steps=n, decode_width_sum=n, + decode_width_max=1, prefill_mode="batched" if admitted else None) + store.put(rec) + store.close() + rep = tc.validate(tmp_path) + assert rep.valid, rep.errors + assert len(rep.rows) == 5 and list(rep.files) == [store.path.name]