lab: request-trace contract (TRACE.md) and validator

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/<regime>.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` + `<stem>.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.

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")}

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-<item id>, 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}

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
 

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 `<stem>.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 `<run_id>.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.

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
 

lab/evaltools.py

@@ -287,14 +287,15 @@ 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}}
         except Exception as e:                      # no GPU or a base that will not serve is itself the fact
             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")

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
 

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 ("", "")

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