diff --git a/AGENTS.md b/AGENTS.md
index bf87ebfec..eb0ee43a1 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -85,22 +85,21 @@ Dataset Manager --> Load Generator --> Endpoint Client --> External Endpoint
 
 ### Key Components
 
-| Component                           | Location                                                                                                           | Purpose                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
-| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| **Load Generator**                  | `src/inference_endpoint/load_generator/`                                                                           | Central orchestrator: `BenchmarkSession` owns the lifecycle, `PhaseIssuer` drives per-phase execution, `TimedIssueStrategy`/`BurstStrategy`/`ConcurrencyStrategy` control timing. Emits `ERROR` before `COMPLETE` for failed queries (metrics aggregator depends on this order).                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
-| **Endpoint Client**                 | `src/inference_endpoint/endpoint_client/`                                                                          | Multi-process HTTP workers communicating via ZMQ IPC. `HTTPEndpointClient` is the main entry point                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
-| **Dataset Manager**                 | `src/inference_endpoint/dataset_manager/`                                                                          | Loads JSONL, HuggingFace, CSV, JSON, Parquet datasets. `Dataset` base class with `load_sample()`/`num_samples()` interface                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
-| **Metrics Aggregator**              | `src/inference_endpoint/async_utils/services/metrics_aggregator/`                                                  | Subprocess. Subscribes to events, aggregates per-sample metrics into a `MetricsRegistry` (counters + HDR-histogram series + raw values), publishes `MetricsSnapshot` over IPC PUB at a configurable cadence (`SessionState`: `INITIALIZE` → `LIVE` → `DRAINING` → {`COMPLETE` \| `INTERRUPTED`}). Final snapshot is atomically written to `final_snapshot.json` as the **primary** Report source; the terminal pub/sub frame is a TUI "run finished" signal only.                                                                                                                                                                                                                                                                                                              |
-| **Report**                          | `src/inference_endpoint/metrics/report.py`                                                                         | `Report.from_snapshot(dict)` — pure-function builder consuming the dict form (`snapshot_to_dict`). Reads `final_snapshot.json` directly via `json.loads` (no Struct decode). Plumbs `complete = (state == "complete" and n_pending_tasks == 0)`; renders an explicit warning for `INTERRUPTED` runs.                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
-| **Config**                          | `src/inference_endpoint/config/`, `endpoint_client/config.py`                                                      | Pydantic-based YAML schema (`schema.py`), `HTTPClientConfig` (single Pydantic model for CLI/YAML/runtime), `RuntimeSettings`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
-| **CLI**                             | `src/inference_endpoint/main.py`, `commands/benchmark/cli.py`                                                      | cyclopts-based, auto-generated from `schema.py` and `HTTPClientConfig` Pydantic models. Flat shorthands via `cyclopts.Parameter(alias=...)`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
-| **Async Utils**                     | `src/inference_endpoint/async_utils/`                                                                              | `LoopManager` (uvloop + eager_task_factory), ZMQ transport layer, generic `MessageCodec[T]`-parametrized pub/sub, event publisher                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
-| **OpenAI/SGLang**                   | `src/inference_endpoint/openai/`, `sglang/`                                                                        | Protocol adapters and response accumulators for different API formats. `openai_completions` adapter (`completions_adapter.py`) sends pre-tokenized token IDs to `/v1/completions`, bypassing the server chat template — required for gpt-oss-120b on vLLM. `sglang` adapter sends to `/generate` via `input_ids`. Both apply `Harmonize()` client-side.                                                                                                                                                                                                                                                                                                                                                                                                                        |
-| **TensorRT-LLM**                    | `src/inference_endpoint/trtllm/`                                                                                   | Adapter for TensorRT-LLM endpoints. `TRTLLMAdapter` sends requests; `TRTLLMSSEAccumulator` handles SSE streaming responses.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
-| **DeepSeek-R1 (MLPerf)**            | `src/inference_endpoint/evaluation/scoring.py` (`LegacyMLPerfDeepSeekR1Scorer`), `examples/07_DeepSeekR1_Example/` | MLPerf DeepSeek-R1 accuracy. TensorRT-LLM is OpenAI-compatible, so it is served via `api_type: openai` / `openai_completions` (no dedicated trtllm adapter). The combined multi-subset eval (`math500`/`aime`/`gpqa`/`mmlu_pro`/`livecodebench`) is the official MLCommons `eval_accuracy.py`, run out-of-process via `uv run --project` against the isolated subproject at `src/inference_endpoint/evaluation/legacy_mlperf_deepseek_r1/` (a uv subproject excluded from the parent wheel; mirrors the VBench pattern). The example feeds the exact MLPerf prompt via pre-tokenized `input_tokens` to `/v1/completions`.                                                                                                                                                      |
-| **VideoGen**                        | `src/inference_endpoint/videogen/`                                                                                 | Adapter for video-generation endpoints (e.g. trtllm-serve `POST /v1/videos/generations`, used by MLPerf WAN2.2-T2V-A14B). Defaults to `response_format=video_path` (server saves video to shared storage and returns path) to avoid large byte payloads. Accuracy mode also runs on `video_path`: the adapter mirrors the path into `response_output` so the event log carries it to `VBenchScorer` (see `evaluation/scoring.py`), which scores videos via VBench from a sibling `uv` subproject at `examples/09_Wan22_VideoGen_Example/accuracy/` (vbench's `transformers==4.33.2` + `numpy<2` pins are incompatible with the parent env, so it runs out-of-process via `uv run --project`). Dataset is ingested via the generic JSONL loader.                                |
-| **Compliance (submission checker)** | `src/inference_endpoint/compliance/checker.py`, `scripts/check_compliance.py`                                      | Validates a completed run's report directory against a registered ruleset. `check_submission(report_dir, ruleset, model)` reads the resolved `config.yaml` plus scorer output (`accuracy/accuracy_results.json` for accuracy, `scores.json` for the agentic perf run) and runs config-lock (deterministic + single-stream), the accuracy gate (`score >= factor x reference`, factor 0.97 for Edge-Agentic), and run validity (0 dropped turns). Server-side launch flags (`--reasoning off`, `--ctx-size`) aren't in client artifacts, so they're surfaced as manual attestations. CLI: `scripts/check_compliance.py REPORT_DIR` (exit 0 = pass).                                                                                                                             |
-| **Compliance (audit tests)**        | `src/inference_endpoint/compliance/`, `commands/audit.py`                                                          | MLPerf compliance audits. `AuditTest` protocol + `AuditRunSpec`/`AuditRunArtifacts` + registry (`compliance/__init__.py`); `OutputCachingAudit` (`compliance/audit_test/output_caching_test.py`, which also owns the QPS-specific `AuditRunStats`) implements MLPerf **TEST04** output-caching detection — reference phase (distinct samples) vs. fixed-sample audit phase, comparing QPS against `threshold`. `commands/audit.py:run_audit` runs phases via `AuditTest.plan_runs`/`validate`, writing `audit_result.json`/`verify_<TEST>.txt` atomically via `compliance/result.py`. Enabled by the `audit:` YAML block; `cli._run` runs it after the main benchmark (upstream MLPerf order: perf run, then TEST04), or standalone with `audit.only: true`. Performance-only. |
+| Component                           | Location                                                                                                           | Purpose                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| **Load Generator**                  | `src/inference_endpoint/load_generator/`                                                                           | Central orchestrator: `BenchmarkSession` owns the lifecycle, `PhaseIssuer` drives per-phase execution, `TimedIssueStrategy`/`BurstStrategy`/`ConcurrencyStrategy` control timing. Emits `ERROR` before `COMPLETE` for failed queries (metrics aggregator depends on this order).                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
+| **Endpoint Client**                 | `src/inference_endpoint/endpoint_client/`                                                                          | Multi-process HTTP workers communicating via ZMQ IPC. `HTTPEndpointClient` is the main entry point                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
+| **Dataset Manager**                 | `src/inference_endpoint/dataset_manager/`                                                                          | Loads JSONL, HuggingFace, CSV, JSON, Parquet datasets. `Dataset` base class with `load_sample()`/`num_samples()` interface                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
+| **Metrics Aggregator**              | `src/inference_endpoint/async_utils/services/metrics_aggregator/`                                                  | Subprocess. Subscribes to events, aggregates per-sample metrics into a `MetricsRegistry` (counters + HDR-histogram series + raw values), publishes `MetricsSnapshot` over IPC PUB at a configurable cadence (`SessionState`: `INITIALIZE` → `LIVE` → `DRAINING` → {`COMPLETE` \| `INTERRUPTED`}). Final snapshot is atomically written to `final_snapshot.json` as the **primary** Report source; the terminal pub/sub frame is a TUI "run finished" signal only.                                                                                                                                                                                                                                                                                                                                                                                                         |
+| **Report**                          | `src/inference_endpoint/metrics/report.py`                                                                         | `Report.from_snapshot(dict)` — pure-function builder consuming the dict form (`snapshot_to_dict`). Reads `final_snapshot.json` directly via `json.loads` (no Struct decode). Plumbs `complete = (state == "complete" and n_pending_tasks == 0)`; renders an explicit warning for `INTERRUPTED` runs.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
+| **Config**                          | `src/inference_endpoint/config/`, `endpoint_client/config.py`                                                      | Pydantic-based YAML schema (`schema.py`), `HTTPClientConfig` (single Pydantic model for CLI/YAML/runtime), `RuntimeSettings`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
+| **CLI**                             | `src/inference_endpoint/main.py`, `commands/benchmark/cli.py`                                                      | cyclopts-based, auto-generated from `schema.py` and `HTTPClientConfig` Pydantic models. Flat shorthands via `cyclopts.Parameter(alias=...)`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
+| **Async Utils**                     | `src/inference_endpoint/async_utils/`                                                                              | `LoopManager` (uvloop + eager_task_factory), ZMQ transport layer, generic `MessageCodec[T]`-parametrized pub/sub, event publisher                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
+| **OpenAI/SGLang**                   | `src/inference_endpoint/openai/`, `sglang/`                                                                        | Protocol adapters and response accumulators for different API formats. `openai_completions` adapter (`completions_adapter.py`) sends pre-tokenized token IDs to `/v1/completions`, bypassing the server chat template — required for gpt-oss-120b on vLLM. `sglang` adapter sends to `/generate` via `input_ids`. Both apply `Harmonize()` client-side.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+| **DeepSeek-R1 (MLPerf)**            | `src/inference_endpoint/evaluation/scoring.py` (`LegacyMLPerfDeepSeekR1Scorer`), `examples/07_DeepSeekR1_Example/` | MLPerf DeepSeek-R1 accuracy. TensorRT-LLM is OpenAI-compatible, so it is served via `api_type: openai` / `openai_completions` (no dedicated trtllm adapter). The combined multi-subset eval (`math500`/`aime`/`gpqa`/`mmlu_pro`/`livecodebench`) is the official MLCommons `eval_accuracy.py`, run out-of-process via `uv run --project` against the isolated subproject at `src/inference_endpoint/evaluation/legacy_mlperf_deepseek_r1/` (a uv subproject excluded from the parent wheel; mirrors the VBench pattern). The example feeds the exact MLPerf prompt via pre-tokenized `input_tokens` to `/v1/completions`.                                                                                                                                                                                                                                                 |
+| **VideoGen**                        | `src/inference_endpoint/videogen/`                                                                                 | Adapter for video-generation endpoints (e.g. trtllm-serve `POST /v1/videos/generations`, used by MLPerf WAN2.2-T2V-A14B). Defaults to `response_format=video_path` (server saves video to shared storage and returns path) to avoid large byte payloads. Accuracy mode also runs on `video_path`: the adapter mirrors the path into `response_output` so the event log carries it to `VBenchScorer` (see `evaluation/scoring.py`), which scores videos via VBench from a sibling `uv` subproject at `examples/09_Wan22_VideoGen_Example/accuracy/` (vbench's `transformers==4.33.2` + `numpy<2` pins are incompatible with the parent env, so it runs out-of-process via `uv run --project`). Dataset is ingested via the generic JSONL loader.                                                                                                                           |
+| **Compliance (submission checker)** | `src/inference_endpoint/compliance/checker.py`, `scripts/check_compliance.py`                                      | Validates a completed run's report directory against a registered ruleset. `check_submission(report_dir, ruleset, model)` reads the resolved `config.yaml` plus scorer output (`accuracy/accuracy_results.json` for accuracy, `scores.json` for the agentic perf run) and runs config-lock (deterministic + single-stream), the accuracy gate (`score >= factor x reference`, factor 0.97 for Edge-Agentic), and run validity (0 dropped turns). Server-side launch flags (`--reasoning off`, `--ctx-size`) aren't in client artifacts, so they're surfaced as manual attestations. CLI: `scripts/check_compliance.py REPORT_DIR` (exit 0 = pass).                                                                                                                                                                                                                        |
+| **Compliance (audit tests)**        | `src/inference_endpoint/compliance/`, `commands/audit.py`                                                          | MLPerf compliance audits. `AuditTest` protocol + `AuditRunSpec`/`AuditRunArtifacts` + registry (`compliance/__init__.py`); `OutputCachingAudit` (`compliance/audit_test/output_caching_test.py`, which also owns the QPS-specific `AuditRunStats`) implements MLPerf **TEST04** output-caching detection — reference phase (distinct samples) vs. fixed-sample audit phase, comparing QPS against `threshold`. `commands/audit.py:run_audit` runs phases via `AuditTest.plan_runs`/`validate`, writing `audit_result.json`/`verify_<TEST>.txt` atomically via `compliance/result.py`. Enabled by the `audit:` YAML block; `cli._run` runs it after the main benchmark (upstream MLPerf order: perf run, then TEST04), or standalone with `audit.only: true`. Perf-only by default (a phase may opt into accuracy via `AuditRunSpec.test_mode`, but this is unused today). |
 
 ### Hot-Path Architecture
 
@@ -162,7 +161,7 @@ Validation is layered:
 
 Orthogonal to the main run: a YAML-only `audit:` block (`show=False`, no CLI flag) on `BenchmarkConfig` selects an `AuditTest`. If `audit:` is set, `cli._run` freezes a shared `report_dir`, runs the main benchmark, then calls `commands/audit.py:run_audit` — the upstream MLPerf order (perf run, then TEST04). Unlike upstream there is no SUT reset between stages; see docs/compliance_audit_plan.md ("Run ordering"). `run_audit`:
 
-1. Builds a performance-only per-phase config (drops accuracy datasets so no phase re-issues or re-scores them) and, after the first phase's dataset loads, calls `AuditTest.validate(cfg, dataset_size, load_pattern)` to bounds-check the test's sample counts/indices and load pattern before any phase issues load (reuses the loaded dataset — no extra load).
+1. Builds a per-phase config that is performance-only by default (drops accuracy datasets so no phase re-issues or re-scores them), overridable per-`AuditRunSpec` via `test_mode` (`ACC`/`BOTH` keeps accuracy datasets — supported but currently unused; every registered audit runs perf-only), and, after the first phase's dataset loads, calls `AuditTest.validate(cfg, dataset_size, load_pattern)` to bounds-check the test's sample counts/indices and load pattern before any phase issues load (reuses the loaded dataset — no extra load).
 2. Runs each `AuditRunSpec` phase (from `AuditTest.plan_runs`) back-to-back under its own `<report_dir>/<label>/` subdir via `setup_benchmark`/`run_benchmark_async`. A phase whose `Report.complete` is `False` (drain timeout / interrupt) aborts with `ExecutionError` — no result on partial data.
 3. Calls `AuditTest.verify(...)` and atomically writes `audit_result.json` (durable record) then `verify_<TEST>.txt` (validator marker).
 
@@ -256,10 +255,6 @@ src/inference_endpoint/
 │   ├── accumulator.py         # Streaming response accumulator (shared by chat + completions)
 │   └── harmony.py             # openai_harmony integration
 ├── sglang/                    # SGLang API adapter (/generate with input_ids)
-├── trtllm/                    # TensorRT-LLM adapter (/v1/chat/completions variant)
-│   ├── types.py               # TRTLLMChatRequest (msgspec Struct)
-│   ├── adapter.py             # TRTLLMAdapter (HttpRequestAdapter)
-│   └── accumulator.py         # TRTLLMAccumulator
 ├── videogen/                  # Video generation adapter (e.g. WAN2.2 T2V workload)
 │   ├── __init__.py
 │   ├── types.py               # Pydantic: VideoPathRequest, VideoPathResponse, VideoPayloadResponse
diff --git a/docs/CLIENT_PERFORMANCE_TUNING.md b/docs/CLIENT_PERFORMANCE_TUNING.md
index 0c985cee2..2ad5902ef 100644
--- a/docs/CLIENT_PERFORMANCE_TUNING.md
+++ b/docs/CLIENT_PERFORMANCE_TUNING.md
@@ -32,12 +32,12 @@ enable_cpu_affinity: true # Auto-compute NUMA-aware plan (default)
 # enable_cpu_affinity: false  # Disabled
 ```
 
-**Auto mode allocation** (default 6 physical cores for loadgen):
+**Auto mode allocation** (default 5 physical cores for loadgen, `DEFAULT_LOADGEN_CORES`):
 
 - 1 core: Session thread (scheduler, busy-wait timing)
 - 1 core: Event loop thread (uvloop, response handling)
-- 4 cores: ZMQ I/O threads
-- Remaining physical cores: Workers (one per core with all SMT siblings)
+- Remaining cores: ZMQ I/O threads (up to 4, sharing the leftover loadgen cores)
+- All other physical cores: Workers (one per core with all SMT siblings)
 
 ## Platform Notes
 
diff --git a/docs/ENDPOINT_CLIENT.md b/docs/ENDPOINT_CLIENT.md
index 8b7982a76..81b358192 100644
--- a/docs/ENDPOINT_CLIENT.md
+++ b/docs/ENDPOINT_CLIENT.md
@@ -200,7 +200,7 @@ The endpoint client uses three core message types defined in the parent project
 | Type          | Direction     | Purpose                                                                 |
 | ------------- | ------------- | ----------------------------------------------------------------------- |
 | `Query`       | Main → Worker | Request payload with `id`, `data` (prompt/params), `headers`            |
-| `StreamChunk` | Worker → Main | Intermediate streaming token with `response_chunk`, `is_complete` flag  |
+| `StreamChunk` | Worker → Main | Intermediate streaming token with `response_chunk` and `metadata`       |
 | `QueryResult` | Worker → Main | Final response with `response_output`, `error`, auto-set `completed_at` |
 | `QueryStatus` | Internal      | Enum: `PENDING`, `RUNNING`, `COMPLETED`, `FAILED`, `CANCELLED`          |
 
@@ -221,7 +221,6 @@ class QueryResult(msgspec.Struct, tag="query_result", kw_only=True, frozen=True,
 class StreamChunk(msgspec.Struct, tag="stream_chunk", frozen=True, kw_only=True, array_like=True, omit_defaults=True, gc=False):
     id: str
     response_chunk: str
-    is_complete: bool
     metadata: dict[str, Any]
 
 class QueryStatus(Enum):
@@ -790,29 +789,29 @@ Each worker maintains its own `ConnectionPool` to its assigned endpoint. The poo
 
 **Idle Connection Validation (`is_stale`):**
 
-When `acquire()` pops a connection from the idle stack, it must verify the server hasn't closed it. `PooledConnection.is_stale()` combines a fast-path skip with a zero-cost kernel probe:
+When `acquire()` pops a connection from the idle stack, it must verify the server hasn't closed it. `PooledConnection.is_stale()` combines a fast-path skip with a zero-cost kernel probe. It uses `poll()` rather than `select()` to avoid the `FD_SETSIZE` limit on high fds, and reuses a persistent per-connection poller (registered lazily on first call, since the fd is stable per connection):
 
 ```python
 def is_stale(self) -> bool:
-    # Fast path: skip select() for recently-used connections.
+    # Fast path: skip the probe for recently-used connections.
     # A server won't close a connection within 1s of last use,
     # so the syscall is unnecessary. Saves ~1µs per acquire.
     if time.monotonic() - self.last_used < 1.0:
         return False
 
-    # Zero-timeout select(): asks the kernel if any data or
-    # errors are pending. A healthy idle socket has neither.
-    readable, _, exceptional = select.select([fd], [], [fd], 0)
-    return bool(readable or exceptional)
+    # Zero-timeout poll() on a persistent per-connection poller:
+    # asks the kernel if any data or errors are pending. A healthy
+    # idle socket has neither. A raised poll event means the server
+    # sent FIN (readable EOF) or the socket errored.
+    return bool(self._stale_poller.poll(0))
 ```
 
-| `select()` Result  | Kernel State                                             | Pool Action                       |
-| ------------------ | -------------------------------------------------------- | --------------------------------- |
-| `readable=True`    | Server sent FIN, indicating it is closing the connection | Discard connection, try next idle |
-| `exceptional=True` | Socket error such as a TCP reset from the server         | Discard connection, try next idle |
-| Both empty         | No pending data or errors on the socket                  | Connection is healthy, use it     |
+| `poll()` Result | Kernel State                                         | Pool Action                       |
+| --------------- | ---------------------------------------------------- | --------------------------------- |
+| Event raised    | Server sent FIN (readable EOF) or socket error/reset | Discard connection, try next idle |
+| No event        | No pending data or errors on the socket              | Connection is healthy, use it     |
 
-The fast-path skip (`< 1.0s`) avoids the `select()` syscall entirely for connections in active rotation — reducing validation from ~1.2µs to ~161ns (see latencies below).
+The fast-path skip (`< 1.0s`) avoids the `poll()` syscall entirely for connections in active rotation — reducing validation from ~1.2µs to ~161ns (see latencies below).
 
 **Operation Latencies:**
 
@@ -839,17 +838,17 @@ Per-operation latency measurements for `ConnectionPool` (localhost TCP, uvloop):
 
 The `_SocketConfig` class (`http.py`) defines socket options applied to all TCP connections created by the connection pool. These options are tuned for low-latency streaming workloads where individual request latency directly impacts benchmark measurements.
 
-| Option             | Value | Effect                                                                                                            | Interaction                                                                                                                                                                          |
-| ------------------ | ----- | ----------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| `TCP_NODELAY`      | 1     | Disables Nagle's algorithm, allowing small packets to be sent immediately rather than being buffered for batching | With `TCP_QUICKACK`: eliminates both send batching (Nagle) and receive-side delayed ACK, removing the two primary sources of TCP-induced latency                                     |
-| `TCP_QUICKACK`     | 1     | Immediately acknowledge received packets instead of delaying acknowledgments (Linux-specific)                     | Not sticky — kernel may revert to delayed ACK mode; re-applied per connection. Together with `TCP_NODELAY`, ensures neither side introduces artificial delays                        |
-| `SO_KEEPALIVE`     | 1     | Enable TCP keepalive probes at the socket level                                                                   | Activates the kernel keepalive mechanism; actual probe timing controlled by `TCP_KEEPIDLE`, `TCP_KEEPCNT`, `TCP_KEEPINTVL`                                                           |
-| `TCP_KEEPIDLE`     | 1s    | Start probing after 1 second idle (Linux-specific)                                                                | With `TCP_KEEPCNT=1` + `TCP_KEEPINTVL=1s`: total dead-connection detection time is 2 seconds (1s idle + 1 probe × 1s interval)                                                       |
-| `TCP_KEEPCNT`      | 1     | 1 failed probe = connection declared dead                                                                         | Aggressive: default is 9. Appropriate here because a failed probe to a local/VPC endpoint indicates genuine failure, not transient packet loss                                       |
-| `TCP_KEEPINTVL`    | 1s    | 1 second between probes                                                                                           | With only 1 probe needed (`TCP_KEEPCNT=1`), a single missed probe triggers connection close                                                                                          |
-| `SO_RCVBUF`        | 4 MB  | Receive buffer size                                                                                               | 4 MB buffer accommodates approximately 1M tokens (at 4 bytes per token) for offline-mode full responses. Prevents the kernel from dropping data when the application is briefly busy |
-| `SO_SNDBUF`        | 4 MB  | Send buffer size                                                                                                  | Sized for large request payloads (long prompts). Allows `write()` to complete without blocking even for large HTTP bodies                                                            |
-| `TCP_USER_TIMEOUT` | 0     | Disabled — no timeout on unacknowledged sent data (Linux-specific)                                                | Keepalive handles dead-connection detection; setting this to 0 avoids interfering with long-running SSE streams where the server may take seconds between chunks                     |
+| Option             | Value  | Effect                                                                                                            | Interaction                                                                                                                                                                                                                                                                                                                                         |
+| ------------------ | ------ | ----------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `TCP_NODELAY`      | 1      | Disables Nagle's algorithm, allowing small packets to be sent immediately rather than being buffered for batching | With `TCP_QUICKACK`: eliminates both send batching (Nagle) and receive-side delayed ACK, removing the two primary sources of TCP-induced latency                                                                                                                                                                                                    |
+| `TCP_QUICKACK`     | 1      | Immediately acknowledge received packets instead of delaying acknowledgments (Linux-specific)                     | Not sticky — kernel may revert to delayed ACK mode; re-applied per connection. Together with `TCP_NODELAY`, ensures neither side introduces artificial delays                                                                                                                                                                                       |
+| `SO_KEEPALIVE`     | 0      | Disabled — kernel TCP keepalive is turned off                                                                     | Dead connections are detected in the pool via `connection_lost`/`eof_received`, not kernel keepalive; keepalive probes produced connection-timeout errors in offline and high-concurrency modes. `_SocketConfig.apply` gates the `TCP_KEEP*` options below behind `if cls.SO_KEEPALIVE`, so they are not applied at all while keepalive is disabled |
+| `TCP_KEEPIDLE`     | 1s     | Start probing after 1 second idle (Linux-specific)                                                                | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled                                                                                                                                                                                                                                                                          |
+| `TCP_KEEPCNT`      | 5      | 5 failed probes = connection declared dead                                                                        | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled                                                                                                                                                                                                                                                                          |
+| `TCP_KEEPINTVL`    | 1s     | 1 second between probes                                                                                           | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled                                                                                                                                                                                                                                                                          |
+| `SO_RCVBUF`        | 128 KB | Receive buffer size                                                                                               | Sliding-window buffer, not a full-message buffer: the event loop reads eagerly, so it only holds data between kernel delivery and application read (~one RTT). Larger responses stream through fine via the TCP sliding window                                                                                                                      |
+| `SO_SNDBUF`        | 128 KB | Send buffer size                                                                                                  | Sliding-window send buffer; large request bodies stream through without blocking `write()`                                                                                                                                                                                                                                                          |
+| `TCP_USER_TIMEOUT` | 0      | Disabled — no timeout on unacknowledged sent data (Linux-specific)                                                | Dead-connection detection is handled in the pool via `connection_lost`/`eof_received` (kernel keepalive is disabled); setting this to 0 avoids interfering with long-running SSE streams where the server may take seconds between chunks                                                                                                           |
 
 **Cross-platform compatibility:** Applied via `_SocketConfig.apply(sock)` with `hasattr()` checks for Linux-specific options (`TCP_KEEPIDLE`, `TCP_QUICKACK`, `TCP_USER_TIMEOUT`). On non-Linux platforms, these options are silently skipped — the system runs with reduced tuning but remains functional.
 
@@ -861,9 +860,9 @@ The `_SocketConfig` class (`http.py`) defines socket options applied to all TCP
 | Idle connection strategy | LIFO stack                                                                                                                                                                        | Random from idle list                  | Lower error rate: high load reuses hot connections; low load (long idle) stale ones sink to bottom                                                                                                                                                                                                                                     |
 | Waiter queue             | FIFO via `OrderedDict`                                                                                                                                                            | List, deque                            | Fair scheduling; O(1) insert/remove                                                                                                                                                                                                                                                                                                    |
 | Connection limiting      | `max_connections` cap; tracks `len(all_connections) + _creating`                                                                                                                  | Unlimited                              | Prevents ephemeral port exhaustion — a real production failure mode at high concurrency (see [§8.6](#86-benchmark-results-vs-aiohttp)). Counting in-progress connections (`_creating`) prevents race where concurrent `acquire()` calls overshoot the limit during TCP handshake                                                       |
-| Staleness detection      | `select()` probe on FD with zero timeout                                                                                                                                          | Timeout-based, `poll()`                | Detects server FIN without I/O. `select()` over `poll()` avoids poll object creation overhead                                                                                                                                                                                                                                          |
-| Preclose skip            | `time.monotonic() - last_used < 1.0` → return not-stale immediately                                                                                                               | Always probe                           | Server unlikely to close within 1s of last use; skips `select()` syscall entirely on hot connections under load                                                                                                                                                                                                                        |
-| Socket tuning            | `TCP_NODELAY` + `TCP_QUICKACK` [7], 4MB buffers                                                                                                                                   | Defaults                               | `TCP_NODELAY` disables Nagle batching; `TCP_QUICKACK` disables delayed ACK; together they eliminate both send and receive latency sources. 4MB buffers sized for offline-mode full responses. See [§8.4](#84-socket-config) for full socket option table                                                                               |
+| Staleness detection      | `poll()` on the FD with zero timeout via a persistent per-connection `_stale_poller`                                                                                              | Timeout-based, `select()`              | Detects server FIN without I/O. A persistent `poll` object registered once per connection avoids rebuilding the fd set on every probe                                                                                                                                                                                                  |
+| Preclose skip            | `time.monotonic() - last_used < 1.0` → return not-stale immediately                                                                                                               | Always probe                           | Server unlikely to close within 1s of last use; skips the `poll()` syscall entirely on hot connections under load                                                                                                                                                                                                                      |
+| Socket tuning            | `TCP_NODELAY` + `TCP_QUICKACK` [7], 128KB buffers                                                                                                                                 | Defaults                               | `TCP_NODELAY` disables Nagle batching; `TCP_QUICKACK` disables delayed ACK; together they eliminate both send and receive latency sources. 128KB sliding-window buffers (not full-message buffers — the event loop reads eagerly). See [§8.4](#84-socket-config) for full socket option table                                          |
 | Ephemeral port detection | Read `/proc/sys/net/ipv4/ip_local_port_range`                                                                                                                                     | Manual configuration                   | Auto-sizes `max_connections` to the port budget (range x distinct endpoints); raises `RuntimeError` if explicit value exceeds it. Live socket occupancy is not subtracted (racy, counts unrelated destinations)                                                                                                                        |
 | Connection warmup        | Auto: 50% of an explicit pool, 25% of the port budget for the full-auto config; establishment paced by a per-pool connect limiter (`max_concurrent_warmup_connects`, default 128) | 0% or 100%, unbounded `gather`         | 100% = SYN flood risk to server; 0% = ~100x cold-start penalty per [§8.3](#83-connection-pool). Pacing bounds in-flight `connect()` for warmup AND runtime pool growth, so bursts can't overflow the server accept queue or exhaust ephemeral ports. `return_exceptions=True` ensures individual failures don't abort the warmup batch |
 | Idle connection discard  | `max_idle_time=4.0s` proactive close                                                                                                                                              | Rely on staleness only                 | Proactive discard avoids keepalive race with server timeout; 4s chosen to be shorter than typical server keepalive (5-60s)                                                                                                                                                                                                             |
@@ -1351,7 +1350,7 @@ Per-message round-trip: **4 context switches** (sender↔IO thread↔kernel↔IO
 | Event loop daemon      | 1                        | uvloop in `HTTPEndpointClient` |
 | Transport I/O          | `io_threads` (default 4) | Background threads for IPC     |
 
-Default `loadgen_cores=2` accommodates the 2 Python threads (Session + Event loop).
+Default `loadgen_cores=5` (`DEFAULT_LOADGEN_CORES` in `cpu_affinity.py`) reserves headroom for the loadgen-side threads (Session + event loop) plus the transport I/O threads that share those cores.
 
 **Performance ranking sources (checked in order):**
 
diff --git a/docs/PERF_ARCHITECTURE.md b/docs/PERF_ARCHITECTURE.md
index c7a31d61a..64efea990 100644
--- a/docs/PERF_ARCHITECTURE.md
+++ b/docs/PERF_ARCHITECTURE.md
@@ -367,10 +367,12 @@ Workers are mostly I/O-bound (waiting on HTTP responses).
 └─────────────────────────────────────────────────────────────────────┘
 ```
 
-| Component   | Allocation                       | Rationale                                                                 |
-| ----------- | -------------------------------- | ------------------------------------------------------------------------- |
-| **LoadGen** | 2 physical cores (4 logical)     | Session thread + event loop thread. Bottleneck - gets fastest cores.      |
-| **Workers** | 1 physical core each (2 logical) | I/O-bound. Full core isolation prevents context switches, reduces jitter. |
+| Component   | Allocation                                 | Rationale                                                                                      |
+| ----------- | ------------------------------------------ | ---------------------------------------------------------------------------------------------- |
+| **LoadGen** | `DEFAULT_LOADGEN_CORES` = 5 physical cores | Session thread + event loop thread + up to 4 ZMQ I/O threads. Bottleneck - gets fastest cores. |
+| **Workers** | 1 physical core each (2 logical)           | I/O-bound. Full core isolation prevents context switches, reduces jitter.                      |
+
+> The core-ordering diagram above is schematic — it shows two `LG` cells for illustration; the actual default reserves 5 physical cores for loadgen (`DEFAULT_LOADGEN_CORES` in `cpu_affinity.py`).
 
 ### Why Both Hyperthreads Per Core
 
diff --git a/docs/async_utils/DESIGN.md b/docs/async_utils/DESIGN.md
index 8b1da236a..9e1020613 100644
--- a/docs/async_utils/DESIGN.md
+++ b/docs/async_utils/DESIGN.md
@@ -69,17 +69,18 @@ short-lived coroutines on the hot path.
 
 ### `EventPublisherService`
 
-Singleton via `SingletonMixin` — after the first construction, subsequent calls return the
-cached instance. The first construction requires a `ManagedZMQContext`. Subscribers receive
-`EventRecord` messages over a ZMQ SUB socket.
+A plain per-instance class wrapping `ZmqMessagePublisher[EventRecord]` with `LoopManager`
+integration and auto-generated socket names. Each construction requires a `ManagedZMQContext`.
+Subscribers receive `EventRecord` messages over a ZMQ SUB socket.
 
 ```python
-class EventPublisherService(SingletonMixin, ZmqEventRecordPublisher):
+class EventPublisherService(ZmqMessagePublisher[EventRecord]):
     def __init__(
         self,
         managed_zmq_context: ManagedZMQContext,
         extra_eager: bool = False,
         isolated_event_loop: bool = False,
+        send_threshold: int = 1000,
     ) -> None
 
     def publish(self, record: EventRecord) -> None
diff --git a/docs/async_utils/services/event_logger/DESIGN.md b/docs/async_utils/services/event_logger/DESIGN.md
index 868b5bbcf..62945ddac 100644
--- a/docs/async_utils/services/event_logger/DESIGN.md
+++ b/docs/async_utils/services/event_logger/DESIGN.md
@@ -212,9 +212,9 @@ usage: uv run python -m inference_endpoint.async_utils.services.event_logger
 | `--socket-name` | Yes      | —       | Socket name within socket-dir                             |
 | `--writers`     | No       | `jsonl` | Writer backends: `jsonl`, `sql`, or both                  |
 
-## Not Yet Wired
+## Wiring
 
-The EventRecord pub/sub infrastructure is ready, but actual `publish(EventRecord(...))`
-calls have not been connected in the load generator or worker processes. Once wired,
-the event logger will receive and persist all session/sample/error events published
-during a benchmark run.
+`publish(EventRecord(...))` calls are connected in the load generator (`BenchmarkSession`
+publishes STARTED / ISSUED / COMPLETE / ERROR events via its `EventPublisher`; the prompt rides on
+the `ISSUED` event's `PromptData` payload rather than a separate event), so the event
+logger receives and persists all session/sample/error events published during a benchmark run.
diff --git a/docs/async_utils/transport/zmq/ready_check_design.md b/docs/async_utils/transport/zmq/ready_check_design.md
index 15e979689..1af35ddd8 100644
--- a/docs/async_utils/transport/zmq/ready_check_design.md
+++ b/docs/async_utils/transport/zmq/ready_check_design.md
@@ -40,7 +40,7 @@ This means one receiver socket handles readiness from all subprocesses.
 - Binds a ZMQ PULL socket on an IPC path
 - `wait(timeout)` blocks until `count` signals arrive
 - Returns list of identities in arrival order
-- Closes socket after all signals received (or on timeout)
+- Closes the socket after all signals are received, but deliberately **not** on timeout (the caller may retry)
 - Timeout is a total deadline, not per-message
 
 ### `send_ready_signal()` (subprocess side)
diff --git a/docs/compliance_audit_plan.md b/docs/compliance_audit_plan.md
index 112990695..4f0625b44 100644
--- a/docs/compliance_audit_plan.md
+++ b/docs/compliance_audit_plan.md
@@ -182,7 +182,7 @@ report is already written).
         ╔═══════════ for each spec (back-to-back) ═════════════╗         │
         ║                    ▼                                  ║         │
         ║   ┌──────────────────────────────────────────┐       ║         │
-        ║   │ phase_cfg = perf-only (accuracy datasets  │       ║         │
+        ║   │ phase_cfg = perf-only* (accuracy datasets │       ║         │
         ║   │   dropped; audit=None; report_dir=<label>)│       ║         │
         ║   │ ctx = setup_benchmark(phase_cfg,          │       ║         │
         ║   │   audit_run_spec)                         │       ║         │
@@ -241,6 +241,11 @@ report is already written).
             └──────────────────────────────────────────┘
 ```
 
+\* Perf-only is the **default**. A phase may set `AuditRunSpec.test_mode` (`ACC`/`BOTH`) to keep
+its accuracy datasets instead of dropping them; the orchestrator reads `spec.test_mode` per phase
+rather than hardcoding perf-only. This override is supported but currently **unused** — every
+registered audit (TEST04) leaves `test_mode` at its `PERF` default.
+
 The first-phase gate calls `AuditTest.validate(cfg, N, load_pattern)` once `N` (the loaded
 dataset size) is known: each audit owns its own preconditions there, so the generic loop
 never encodes a single test's rules. For the output-caching test that means only
@@ -388,8 +393,11 @@ The generic loop never names a specific test:
    phase issues load.
 4. Execute each spec back-to-back via the existing `setup_benchmark` /
    `run_benchmark_async` path (no duplicated report-dir or `config.yaml` logic). Each phase
-   config is performance-only (accuracy datasets dropped so no phase re-issues or re-scores
-   them) and has `audit=None` to prevent re-entry into `run_audit`. If any phase raises
+   config is performance-only **by default** (accuracy datasets dropped so no phase re-issues
+   or re-scores them) and has `audit=None` to prevent re-entry into `run_audit`. A phase can
+   override this per-`AuditRunSpec` via `test_mode` (`ACC`/`BOTH` keeps its accuracy datasets);
+   the orchestrator reads `spec.test_mode` rather than hardcoding perf-only. This override is
+   supported but currently unused — every registered audit (TEST04) runs perf-only. If any phase raises
    (`SetupError` / `ExecutionError`), `run_audit` aborts **without verifying** — a crashed
    phase must never produce a result. A phase that returns but whose `Report.complete` is
    `False` (metrics drain timed out, or the run was interrupted → partial stats) is likewise
diff --git a/docs/core/DESIGN.md b/docs/core/DESIGN.md
index 99062ab99..17a15362b 100644
--- a/docs/core/DESIGN.md
+++ b/docs/core/DESIGN.md
@@ -52,7 +52,6 @@ Represents one SSE delta from a streaming response.
 | ---------------- | ---------------- | ------------------------------ |
 | `id`             | `str`            | Matches originating `Query.id` |
 | `response_chunk` | `str`            | Incremental token text         |
-| `is_complete`    | `bool`           | True for the final chunk       |
 | `metadata`       | `dict[str, Any]` | Per-chunk metadata             |
 
 ### `TextModelOutput`
diff --git a/docs/dataset_manager/DESIGN.md b/docs/dataset_manager/DESIGN.md
index e1c8c6d1a..a8b083db5 100644
--- a/docs/dataset_manager/DESIGN.md
+++ b/docs/dataset_manager/DESIGN.md
@@ -77,8 +77,9 @@ parser/remap config, and dataset name. Format is inferred from file extension wh
 - `.parquet` → `PARQUET`
 - explicit `format=huggingface` → `HF`
 
-Presets (e.g. `"gpqa::Qwen/Qwen3-8B"`) are encoded in `config.name` as a `"::"` split — the
-factory resolves them to a predefined dataset class with a model-specific transform stack.
+Presets (e.g. `"gpqa::gptoss"`) are encoded in `config.name` as a `"::"` split — `<dataset>::<preset>` —
+where the factory resolves the first segment to a predefined dataset class and the second to a named
+preset with its transform stack.
 
 ### `Transform` (abstract base)
 
@@ -116,8 +117,8 @@ configs. Each predefined dataset ships with default transforms for supported mod
 
 ## Preset System
 
-A preset string like `"gpqa::Qwen/Qwen3-8B"` resolves to a predefined dataset with a
-model-specific transform stack pre-applied. This is used by rulesets to ensure consistent
+A preset string like `"gpqa::gptoss"` (`<dataset>::<preset>`) resolves to a predefined dataset with a
+named preset's transform stack pre-applied. This is used by rulesets to ensure consistent
 prompt formatting across submissions.
 
 ## Design Decisions
diff --git a/docs/evaluation/DESIGN.md b/docs/evaluation/DESIGN.md
index 2ec023884..1074c343d 100644
--- a/docs/evaluation/DESIGN.md
+++ b/docs/evaluation/DESIGN.md
@@ -35,11 +35,12 @@ accuracy summary written into benchmark results
 
 ## Files
 
-| File             | Purpose                                                           |
-| ---------------- | ----------------------------------------------------------------- |
-| `extractor.py`   | Extracts model answer from raw text (regex, boxed-answer parsing) |
-| `scoring.py`     | Compares extracted answer to ground truth label                   |
-| `livecodebench/` | LiveCodeBench-specific code execution pipeline                    |
+| File             | Purpose                                                                                                                                      |
+| ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
+| `extractor.py`   | Extracts model answer from raw text (regex, boxed-answer parsing)                                                                            |
+| `scoring.py`     | Compares extracted answer to ground truth label                                                                                              |
+| `bfcl_v4_*.py`   | BFCL v4 function-calling eval: `bfcl_v4_execution.py`, `bfcl_v4_scorer.py`, `bfcl_v4_metrics.py`, and the multi-turn `runner`/`scorer`/`cli` |
+| `livecodebench/` | LiveCodeBench-specific code execution pipeline                                                                                               |
 
 ## LiveCodeBench
 
@@ -60,13 +61,19 @@ Files:
 
 The scorer registry in `evaluation/scoring.py` currently includes:
 
-| Method                | Description                                                    |
-| --------------------- | -------------------------------------------------------------- |
-| `pass_at_1`           | Exact-match style scoring; also used by the LiveCodeBench path |
-| `string_match`        | Whitespace-trimmed string equality                             |
-| `rouge`               | ROUGE-based text generation scoring                            |
-| `code_bench_scorer`   | LiveCodeBench code-execution scoring                           |
-| `shopify_category_f1` | Shopify category F1 evaluation                                 |
+| Method                      | Description                                                    |
+| --------------------------- | -------------------------------------------------------------- |
+| `pass_at_1`                 | Exact-match style scoring; also used by the LiveCodeBench path |
+| `string_match`              | Whitespace-trimmed string equality                             |
+| `rouge`                     | ROUGE-based text generation scoring                            |
+| `code_bench_scorer`         | LiveCodeBench code-execution scoring                           |
+| `shopify_category_f1`       | Shopify category F1 evaluation                                 |
+| `agentic_inference_inline`  | Inline scoring for agentic multi-turn inference                |
+| `vbench`                    | VBench video-generation accuracy (WAN 2.2 T2V)                 |
+| `bfcl_v4`                   | BFCL v4 function-calling accuracy                              |
+| `legacy_mlperf_deepseek_r1` | MLPerf DeepSeek-R1 combined multi-subset accuracy              |
+
+The registry (`Scorer.PREDEFINED`) is auto-populated from `Scorer` subclasses via `__init_subclass__`.
 
 The scoring configuration used by benchmark execution is specified per accuracy dataset under
 `datasets[].accuracy_config`, including `accuracy_config.eval_method`,
diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md
index 1c0fa4435..563fba33c 100644
--- a/docs/load_generator/DESIGN.md
+++ b/docs/load_generator/DESIGN.md
@@ -11,12 +11,14 @@ single-thread, single-event-loop-per-process constraint.
 ```
 src/inference_endpoint/load_generator/
 ├── __init__.py          # Public exports
-├── session.py           # BenchmarkSession, SessionResult
-├── strategy.py          # LoadStrategy protocol, TimedIssueStrategy,
-│                        #   BurstStrategy, ConcurrencyStrategy,
-│                        #   create_load_strategy()
-├── sample_order.py      # SampleOrder, WithoutReplacement, WithReplacement
-└── delay.py             # poisson_delay_fn, uniform_delay_fn
+├── session.py                     # BenchmarkSession, PhaseIssuer, PhaseConfig, SessionResult
+├── strategy.py                    # LoadStrategy protocol, TimedIssueStrategy,
+│                                  #   BurstStrategy, ConcurrencyStrategy,
+│                                  #   create_load_strategy()
+├── agentic_inference_strategy.py  # AgenticInferenceStrategy (multi-turn conversations)
+├── conversation_manager.py        # ConversationManager, ConversationState
+├── sample_order.py                # SampleOrder, WithoutReplacement, WithReplacement
+└── delay.py                       # poisson_delay_fn, make_delay_fn
 ```
 
 ## Architecture
@@ -33,9 +35,9 @@ each producing an independent report.
 BenchmarkSession.run(phases)
     |
     +-- STARTED
-    +-- [warmup]     strategy.execute() → NO drain (keep in-flight saturated)
+    +-- [warmup]     strategy.execute() → drain_after=False (keep in-flight saturated)
     +-- [perf phase 1]   START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING
-    +-- [warmup]     strategy.execute() → NO drain (keep in-flight saturated)
+    +-- [warmup]     strategy.execute() → drain_after=False (keep in-flight saturated)
     +-- [perf phase 2]   START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING
     +-- [accuracy x N]   strategy.execute() → drain (uuid maps collected)
     +-- ENDED
@@ -43,10 +45,17 @@ BenchmarkSession.run(phases)
     +-- return SessionResult { perf_results: [PhaseResult, ...], accuracy_results: [...] }
 ```
 
+**Draining is per-phase, not intrinsic to the phase type.** Each `PhaseConfig` carries a
+`drain_after: bool` (default `True`) and an optional `drain_timeout`; `run_phase` drains in-flight
+requests iff `phase.drain_after`. Warmup phases are simply configured with `drain_after=False` so
+the next phase starts with concurrency already saturated — nothing about `PhaseType.WARMUP` itself
+disables draining.
+
 Each performance phase is bracketed by `START_PERFORMANCE_TRACKING` /
-`STOP_PERFORMANCE_TRACKING` events, which the `MetricsAggregator` uses to
-scope its tracked counters and duration. At the end of each perf phase,
-metrics are snapshotted from the KVStoreReader and a `Report` is built.
+`STOP_PERFORMANCE_TRACKING` events, which the metrics aggregator uses to
+scope its tracked counters and duration. The aggregator folds the event stream into rollups and,
+at session end, atomically writes `final_snapshot.json`; the main process reads that snapshot and
+builds the `Report` via `Report.from_snapshot` (see [Integration Points](#eventpublisher--metricsaggregator)).
 
 > **TODO:** The current `MetricsAggregator` does not support per-phase scoping.
 > It maintains a single set of counters and series across all tracking windows.
@@ -68,15 +77,22 @@ already at the target level. Common uses:
 
 ### Load Strategies
 
-Three load patterns, three implementations — each uses the optimal async primitive
-for its scheduling semantics, validated by benchmarking:
+Four load patterns — each uses the async primitive that fits its scheduling semantics, validated
+by benchmarking:
+
+| LoadPatternType   | Strategy                   | Mechanism                       | Best At                  |
+| ----------------- | -------------------------- | ------------------------------- | ------------------------ |
+| POISSON           | `TimedIssueStrategy`       | `loop.call_at` (default)        | ≤50k QPS                 |
+| POISSON (precise) | `TimedIssueStrategy`       | `run_in_executor(busy_wait)`    | Sub-100μs precision      |
+| MAX_THROUGHPUT    | `BurstStrategy`            | `loop.call_soon`                | Max fire rate            |
+| CONCURRENCY       | `ConcurrencyStrategy`      | `asyncio.Semaphore`             | Fixed concurrency        |
+| AGENTIC_INFERENCE | `AgenticInferenceStrategy` | response-driven next-turn issue | Multi-turn conversations |
 
-| LoadPatternType   | Strategy              | Mechanism                    | Best At             |
-| ----------------- | --------------------- | ---------------------------- | ------------------- |
-| POISSON           | `TimedIssueStrategy`  | `loop.call_at` (default)     | ≤50k QPS            |
-| POISSON (precise) | `TimedIssueStrategy`  | `run_in_executor(busy_wait)` | Sub-100μs precision |
-| MAX_THROUGHPUT    | `BurstStrategy`       | `loop.call_soon`             | Max fire rate       |
-| CONCURRENCY       | `ConcurrencyStrategy` | `asyncio.Semaphore`          | Fixed concurrency   |
+The first three are open-loop over independent samples. `AgenticInferenceStrategy` is different:
+it drives multi-turn **conversations** where each turn depends on the previous response (see
+[AgenticInferenceStrategy](#agenticinferencestrategy) below). It is not selected through
+`create_load_strategy` (that path raises for `AGENTIC_INFERENCE`) — it is wired directly when the
+phase uses an `AgenticInferenceDataset`.
 
 **Default for Poisson is `loop.call_at`:** Sub-millisecond timing precision (600–700μs)
 with zero GIL contention and low response latency (0.6–1.4ms). No thread pool overhead.
@@ -135,6 +151,9 @@ class PhaseConfig:
     runtime_settings: RuntimeSettings
     dataset: Dataset
     phase_type: PhaseType = PhaseType.PERFORMANCE
+    drain_after: bool = True             # drain in-flight at phase end (warmup sets False)
+    drain_timeout: float | None = None
+    strategy: LoadStrategy | None = None
 
 
 class BenchmarkSession:
@@ -158,14 +177,15 @@ class BenchmarkSession:
 3. For each phase:
    a. Create `SampleOrder` and `LoadStrategy` from phase settings
    b. Set `self._current_dataset` to phase dataset
-   c. **WARMUP**: execute strategy, **do not drain** in-flight. No tracking
-   events, no report. Purpose: bring endpoint to steady-state concurrency
-   (e.g., fill KV caches, warm up connection pools). The next phase starts
+   c. **WARMUP** (configured `drain_after=False`): execute strategy, do not drain
+   in-flight. No tracking events, no report. Purpose: bring endpoint to steady-state
+   concurrency (e.g., fill KV caches, warm up connection pools). The next phase starts
    immediately with concurrency already at the target level.
-   d. **PERFORMANCE**: publish `START_PERFORMANCE_TRACKING`, execute strategy,
-   drain in-flight, publish `STOP_PERFORMANCE_TRACKING`. Snapshot metrics
-   from KVStoreReader → build `PhaseResult`.
-   e. **ACCURACY**: execute strategy, drain in-flight. No tracking events.
+   d. **PERFORMANCE** (`drain_after=True`): publish `START_PERFORMANCE_TRACKING`, execute
+   strategy, drain in-flight, publish `STOP_PERFORMANCE_TRACKING`. Build `PhaseResult`
+   (issued count + uuid map); the aggregator scopes its tracked counters/duration to the
+   tracking window, and the `Report` is built once at session end from `final_snapshot.json`.
+   e. **ACCURACY** (`drain_after=True`): execute strategy, drain in-flight. No tracking events.
    UUID map collected for eval scoring.
 4. Publish `SessionEventType.ENDED`
 5. Return `SessionResult` (contains `PhaseResult` per perf phase + accuracy maps)
@@ -257,8 +277,10 @@ The strategy calls `phase_issuer.issue(idx)`. After the phase completes,
 the session reads `phase_issuer.uuid_to_index` and `phase_issuer.issued_count`
 to build the `PhaseResult`.
 
-**UUID generation before Query construction** avoids the old `Sample` catch-22.
-`Query` is a frozen `msgspec.Struct` — all fields set at construction, no mutation.
+**UUID generation before Query construction**: the id is generated first so it can be published on
+the ISSUED event and set as `Query.id` in the same construction. `Query` is a frozen
+`msgspec.Struct` — all fields set at construction, no mutation — so the id cannot be attached after
+the fact.
 
 **`_receive_responses()`** — concurrent coroutine, purely async:
 
@@ -454,6 +476,40 @@ class ConcurrencyStrategy(LoadStrategy):
         self._sem.release()
 ```
 
+### AgenticInferenceStrategy
+
+Handles `LoadPatternType.AGENTIC_INFERENCE` (`agentic_inference_strategy.py`). Unlike the open-loop
+strategies, it is **response-driven**: each conversation is a fixed sequence of turns known upfront
+from the dataset (turn prompts are pre-built from the dataset history, not synthesized from the
+model's replies), and turn _N+1_ is issued only after turn _N_'s response arrives — the response
+gates the _timing_ of the next turn, it does not alter its prompt. Concurrency is measured in
+**active conversations** (`_target_concurrency`), not in-flight requests.
+
+- `execute(phase_issuer)` seeds up to `_target_concurrency` initial conversations, then awaits
+  completion of all of them.
+- On each response, `on_sample_complete(result)` synchronously routes it to its conversation and
+  calls `_issue_next_turn()` → `_issue_turn_now()`, which stores the new `query_id → conversation_id`
+  mapping and issues the turn with zero event-loop delay (or an optional per-turn delay). When a
+  conversation's turn cursor reaches the end of its turn list — or a turn fails — the conversation
+  finishes and a fresh conversation is seeded to hold concurrency steady.
+
+Behavior is configured via `AgenticInferenceConfig` (`config/schema.py`): `turn_timeout_s`,
+`enable_salt`, `inject_tool_delay`, `num_trajectories_to_issue`, and
+`stop_issuing_on_first_user_complete`. The dataset is an `AgenticInferenceDataset` (JSONL with
+`conversation_id` / `turn` / `role` / `content` rows grouped consecutively); its
+`ConversationMetadata` supplies the per-turn pre-built message history and any per-turn delays.
+
+### ConversationManager
+
+`conversation_manager.py` tracks per-conversation progress so the strategy knows when each
+conversation is done. `ConversationState` holds `completed_turns`, `failed_turns`, and
+`expected_client_turns`; `ConversationManager` owns the map of conversation → state and exposes
+`mark_turn_complete()` / `mark_turn_failed()` (both advance `completed_turns`; the latter also bumps
+`failed_turns`). A conversation is complete once its completed turns reach the expected count.
+
+Every `EventRecord` the session publishes in agentic mode carries `conversation_id` and `turn`, so
+the metrics aggregator and event log can attribute each sample to its conversation and turn index.
+
 ### SampleIssuer (Protocol)
 
 ```python
@@ -520,7 +576,7 @@ sequenceDiagram
     participant I as SampleIssuer
     participant W as Worker Process
     participant E as EventPublisher
-    participant M as MetricsAggregator
+    participant M as MetricsAggregatorService
 
     S->>B: issue_fn(sample_index)
     B->>D: load_sample(index)
@@ -562,21 +618,19 @@ sequenceDiagram
     participant C as Caller (execute.py)
     participant B as BenchmarkSession
     participant E as EventPublisher
-    participant M as MetricsAggregator
-    participant K as KVStoreReader
+    participant M as MetricsAggregatorService
 
     C->>B: run(phases)
     B->>E: STARTED
 
     Note over B: === Saturation Phase ===
-    B->>B: execute strategy (untracked, no drain)
+    B->>B: execute strategy (untracked, drain_after=False)
 
     Note over B: === Perf Phase 1 (e.g. QPS=1000) ===
     B->>E: START_PERFORMANCE_TRACKING
     B->>B: execute strategy
     B->>B: drain in-flight
     B->>E: STOP_PERFORMANCE_TRACKING
-    B->>K: snapshot metrics → Report
     Note over B: PhaseResult("perf_qps1k", issued_count=N)
 
     Note over B: === Perf Phase 2 (e.g. QPS=5000) ===
@@ -584,7 +638,6 @@ sequenceDiagram
     B->>B: execute strategy
     B->>B: drain in-flight
     B->>E: STOP_PERFORMANCE_TRACKING
-    B->>K: snapshot metrics → Report
     Note over B: PhaseResult("perf_qps5k", issued_count=N)
 
     Note over B: === Accuracy Phase ===
@@ -593,7 +646,10 @@ sequenceDiagram
     Note over B: PhaseResult("accuracy", uuid_map)
 
     B->>E: ENDED
+    E->>M: (event stream throughout)
+    M->>M: write final_snapshot.json
     B-->>C: SessionResult
+    C->>C: Report.from_snapshot(final_snapshot.json)
 ```
 
 ### Separate Timer Process Data Flow
@@ -647,9 +703,9 @@ graph TD
     subgraph "Worker Process N"
         WN["HTTP → endpoint"]
     end
-    subgraph "MetricsAggregator (subprocess)"
-        MA["ZmqEventRecordSubscriber"]
-        KB["KVStore (mmap)"]
+    subgraph "MetricsAggregatorService (subprocess)"
+        MA["EventRecord subscriber"]
+        KB["MetricsRegistry → MetricsSnapshot PUB + final_snapshot.json"]
         MA --> KB
     end
 
@@ -685,8 +741,8 @@ graph TD
     subgraph "Worker Processes"
         W["HTTP → endpoint"]
     end
-    subgraph "MetricsAggregator"
-        MA["Subscriber → KVStore"]
+    subgraph "MetricsAggregatorService"
+        MA["Subscriber → MetricsRegistry → snapshot"]
     end
 
     T -->|"ZMQ IPC"| R
@@ -719,6 +775,15 @@ def create_load_strategy(
 
         case LoadPatternType.CONCURRENCY:
             return ConcurrencyStrategy(lp.target_concurrency, sample_order)
+
+        case LoadPatternType.AGENTIC_INFERENCE:
+            # Agentic runs are response-driven and need an AgenticInferenceDataset;
+            # they are wired up directly, not through this open-loop factory.
+            raise ValueError(
+                "AGENTIC_INFERENCE load pattern requires an AgenticInferenceDataset; "
+                "use 'inference-endpoint benchmark from-config' with an "
+                "agentic inference dataset"
+            )
 ```
 
 ---
@@ -783,15 +848,18 @@ to use `await shutdown_async()`.
 
 ### EventPublisher / MetricsAggregator
 
-Session publishes `EventRecord` instances via `ZmqEventRecordPublisher`. The publisher
-uses non-blocking ZMQ send with fd-based writer fallback — safe to call from sync
-callbacks (like `call_at` fire functions).
+The session publishes `EventRecord` instances via `EventPublisherService` (a per-instance wrapper
+over `ZmqMessagePublisher[EventRecord]`). The publisher uses non-blocking ZMQ send with an fd-based
+writer fallback — safe to call from sync callbacks (like `call_at` fire functions).
 
-> **Key fix:** `Report.from_kv_reader` currently reads counter keys (`n_samples_issued`,
-> `duration_ns`) that don't match the `MetricCounterKey` enum written by the aggregator
-> (`total_samples_issued`, `tracked_duration_ns`). Must update `from_kv_reader` to use
-> the actual key names. Performance reports should use `tracked_*` counters. A
-> `test_started_at` counter must be added to the aggregator (set on `SessionEventType.STARTED`).
+Those records flow to the [`MetricsAggregatorService`](../async_utils/services/metrics_aggregator/DESIGN.md)
+subprocess, which folds them into a `MetricsRegistry` (counters + HDR/series rollups), publishes
+live `MetricsSnapshot` messages, and — on `ENDED` / terminal state — atomically writes
+`final_snapshot.json`. The main process reads that file (`json.loads` → dict) and builds the report
+with `Report.from_snapshot(dict, run_config=...)`; if the file is missing (aggregator SIGKILLed) it
+falls back to the subscriber's last live snapshot and the report is marked `complete=False`. The
+report reads `tracked_*` counters (`tracked_samples_completed`, `tracked_duration_ns`, …) plus the
+LoadGen "completed" window; see [metrics/report_design.md](../metrics/report_design.md).
 
 ### HttpClientSampleIssuer Migration
 
@@ -844,23 +912,31 @@ phase. The receiver must distinguish stale vs current-phase completions:
 ```python
 def _handle_response(self, resp: QueryResult) -> None:
     query_id = resp.id
-    # Always publish the event (aggregator tracks all samples)
-    self._publisher.publish(EventRecord(
-        event_type=SampleEventType.COMPLETE,
-        timestamp_ns=resp.completed_at,
-        sample_uuid=query_id,
-    ))
-    # Only route to current phase strategy if this is a current-phase query
-    if query_id in self._current_phase_issuer.uuid_to_index:
-        self._current_phase_issuer.inflight -= 1
+    phase_issuer = self._current_phase_issuer
+    # ERROR is published before COMPLETE (the aggregator depends on this order
+    # for tracked-failure counting).
+    if resp.error is not None:
+        self._publisher.publish(EventRecord(event_type=ErrorEventType.GENERIC, ...))
+    # COMPLETE is NOT published for WARMUP phases — warmup samples must not
+    # land in the tracked counters/series.
+    if self._current_phase_type != PhaseType.WARMUP:
+        self._publisher.publish(EventRecord(
+            event_type=SampleEventType.COMPLETE,
+            timestamp_ns=resp.completed_at,
+            sample_uuid=query_id,
+        ))
+    # Only route to the current phase's strategy if this is a current-phase query.
+    # Stale completions (arrived from a prior no-drain phase) fall through here.
+    if phase_issuer is not None and query_id in phase_issuer.uuid_to_index:
+        phase_issuer.mark_inflight_complete()
         if self._current_strategy:
             self._current_strategy.on_query_complete(query_id)
-    # Stale completions: event published but strategy/inflight not affected
 ```
 
-Same guard applies to `StreamChunk` with `is_complete=True` — check
-`uuid_to_index` membership before decrementing inflight. Non-final
-StreamChunks don't affect inflight and can be published unconditionally.
+The `uuid_to_index` membership gate keeps a stale completion (arriving from a prior warmup/no-drain
+phase) from being attributed to the current phase's strategy or inflight count. Streaming responses
+are handled on the same path: intermediate `StreamChunk`s emit `RECV_FIRST`/chunk events, and the
+final response drives the COMPLETE/inflight bookkeeping above.
 
 ### Sync Per-Sample Work in Callbacks
 
@@ -934,7 +1010,9 @@ async def _run_benchmark_async(ctx, loop) -> BenchmarkResult:
             await http_client.shutdown_async()
             publisher.close()
             await asyncio.to_thread(launcher.wait_for_exit, None)
-            report = Report.from_kv_reader(kv_reader)  # after aggregator exits
+            # After the aggregator exits it has written final_snapshot.json:
+            snap = json.loads((metrics_output_dir / "final_snapshot.json").read_bytes())
+            report = Report.from_snapshot(snap, run_config=run_config)
 
     return BenchmarkResult(session=result, collector=collector, report=report, ...)
 ```
@@ -1005,27 +1083,29 @@ to a child process, with the TUI as the foreground process reading periodic repo
 ```
 TUI Process (foreground):
     - Renders live dashboard (throughput, latency, progress)
-    - Reads BasicKVStoreReader for real-time metrics from /dev/shm
+    - Subscribes to the aggregator's MetricsSnapshot PUB stream (metrics SUB)
     - Receives SessionResult via IPC on completion
 
 Benchmark Process (child):
     - Runs BenchmarkSession on its own event loop
-    - Writes metrics via EventPublisher -> MetricsAggregator -> KVStore
+    - Emits events via EventPublisher -> MetricsAggregatorService (which PUBs snapshots)
     - Returns SessionResult to parent via IPC (pickle over pipe / ZMQ)
 ```
 
 This architecture is enabled by the current design's clean separation:
 
-- **KVStore** is already cross-process readable (mmap on /dev/shm)
+- **Metrics are already cross-process** — the aggregator publishes `MetricsSnapshot` frames over a
+  ZMQ PUB socket at the `--publish-interval` cadence, so any subscriber renders live percentiles
+  without touching the benchmark process.
 - **BenchmarkSession** has no UI dependencies — it takes callbacks
 - **SessionResult** is a frozen dataclass, trivially serializable
-- The `on_sample_complete` callback would not be used in TUI mode (no
-  cross-process callback). Instead, the TUI polls KVStoreReader for
-  `tracked_samples_completed` to update the progress display.
+- The `on_sample_complete` callback would not be used in TUI mode (no cross-process callback).
+  Instead, the TUI reads the latest `MetricsSnapshot` (e.g. `tracked_samples_completed`) to update
+  the progress display.
 
-The TUI process can also read the `Report` from the KVStore at any time for
-live intermediate reports (current QPS, latency distribution so far), not
-just the final report.
+The TUI can render live intermediate reports (current QPS, latency distribution so far) from each
+live snapshot, and build the final `Report` from `final_snapshot.json` once the run ends — the same
+`Report.from_snapshot` path used by the CLI.
 
 > **Constraint:** The benchmark child process must be a **non-daemon** OS process
 > (e.g., `subprocess.Popen` or `multiprocessing.Process(daemon=False)`).
diff --git a/docs/metrics/DESIGN.md b/docs/metrics/DESIGN.md
index 94b880bee..cc6e4e144 100644
--- a/docs/metrics/DESIGN.md
+++ b/docs/metrics/DESIGN.md
@@ -1,6 +1,6 @@
 # Metrics — Design Spec
 
-> Records per-sample timing events to SQLite during a run (write path), then aggregates them into QPS, latency percentiles, TTFT, and TPOT after the run (read path).
+> Builds the final benchmark `Report` from the metrics-aggregator's rollup snapshot and renders it (console + `result_summary.json`). Also holds the metric target types, the MLPerf early-stopping percentile math, and the standardized run-artifact plots.
 
 **Component specs:** [async_utils](../async_utils/DESIGN.md) · [commands](../commands/DESIGN.md) · [config](../config/DESIGN.md) · [core](../core/DESIGN.md) · [dataset_manager](../dataset_manager/DESIGN.md) · [endpoint_client](../endpoint_client/DESIGN.md) · [evaluation](../evaluation/DESIGN.md) · [load_generator](../load_generator/DESIGN.md) · **metrics** · [openai](../openai/DESIGN.md) · [plugins](../plugins/DESIGN.md) · [profiling](../profiling/DESIGN.md) · [sglang](../sglang/DESIGN.md) · [testing](../testing/DESIGN.md) · [utils](../utils/DESIGN.md)
 
@@ -8,168 +8,151 @@
 
 ## Overview
 
-`metrics/` records benchmark events during a run and aggregates them into performance metrics
-afterwards. It is split into two parts with a clean boundary: `EventRecorder` writes; nothing
-else does. `MetricsReporter` reads.
+`metrics/` is the **reporting** side of the pipeline: it consumes the rollup snapshot produced by
+the metrics-aggregator subprocess and turns it into a human-readable console report and a
+machine-readable `result_summary.json`.
+
+It does **not** record or aggregate events itself. During a run, events are published over ZMQ by
+the load generator and folded into per-sample rollups by a separate
+[`MetricsAggregatorService`](../async_utils/services/metrics_aggregator/DESIGN.md) subprocess,
+which publishes `MetricsSnapshot` messages and atomically writes the terminal snapshot to
+`final_snapshot.json`. `metrics/` reads that snapshot and builds the `Report`.
 
 ## Responsibilities
 
-- Persist every timing event to SQLite during the run (write path)
-- Aggregate events into QPS, latency percentiles, TTFT, and TPOT after the run (read path)
-- Validate results against metric targets from the active ruleset
-- Produce human-readable console output and machine-readable JSON reports
+- Turn a metrics-aggregator snapshot dict into a `Report` (`report.py`, `Report.from_snapshot`)
+- Derive headline QPS/TPS once, so the serialized report is self-complete
+- Render the report to console (`display`) and to `result_summary.json` (`to_json`)
+- Define the metric **target** types used by rulesets (`metric.py`)
+- Compute MLPerf LoadGen early-stopping percentile estimates (`early_stopping.py`)
+- Produce standardized run-artifact plots (`results_plots.py`)
 
-## Component Map
+## Data Flow
 
 ```
-SampleEventHandler  ──► EventRecorder (SQLite, queue-backed)
-                                │
-                                ▼
-                        MetricsReporter
-                                │
-                    ┌───────────┴──────────┐
-                    ▼                      ▼
-              console output          JSON report
+load_generator  ──►  EventPublisher (events PUB)
+                          │
+                          ▼
+        MetricsAggregatorService (subprocess)
+          folds events → MetricsRegistry
+          publishes MetricsSnapshot (metrics PUB)
+          atomically writes final_snapshot.json  ◄── primary Report source
+                          │
+                          ▼
+        main process (commands/benchmark/execute.py)
+          json.loads(final_snapshot.json)  → dict
+                          │
+                          ▼
+        Report.from_snapshot(dict, run_config=...)
+                          │
+              ┌───────────┴───────────┐
+              ▼                       ▼
+        Report.display(fn)      Report.to_json(result_summary.json)
 ```
 
-## Public Interface
-
-### `EventRecorder`
+The snapshot is consumed as a **dict** (`json.loads`), not decoded back into the wire
+`MetricsSnapshot` Struct — the wire form is `array_like=True` for compact msgpack, and the dict
+form (`snapshot_to_dict`) is the canonical consumer contract. If `final_snapshot.json` is missing
+(e.g. the aggregator was SIGKILLed before its signal handler ran), the caller falls back to the
+subscriber's last live snapshot and the resulting `Report.complete` is `False`.
 
-Only one `EventRecorder` may be actively writing at a time per process. The live instance is
-accessible via the class variable `EventRecorder.LIVE`.
+## Files
 
-```python
-class EventRecorder:
-    LIVE: "EventRecorder | None"   # class variable; set on construction, cleared on close
-
-    @classmethod
-    def record_event(
-        cls,
-        ev_type: Event,
-        timestamp_ns: int,
-        sample_uuid: str = "",
-        force_commit: bool = False,
-        assert_active: bool = True,
-        data: Any = None,
-    ) -> bool
-
-    def wait_for_writes(self, force_commit: bool = True) -> None
-    # Blocks until the background writer thread has flushed all queued events
-
-    @staticmethod
-    def db_path(session_id: str) -> Path
-```
+| File                | Purpose                                                                                                            |
+| ------------------- | ------------------------------------------------------------------------------------------------------------------ |
+| `report.py`         | `Report` Struct + `Report.from_snapshot(dict)`; console `display()` and `to_json()` serializer                     |
+| `metric.py`         | Metric target types (`Throughput`, `QueryLatency`, `TTFT`, `TPOT`) used by rulesets                                |
+| `early_stopping.py` | MLPerf LoadGen early-stopping percentile estimates (pure math); see [docs/early_stopping.md](../early_stopping.md) |
+| `results_plots.py`  | Standardized run-artifact plots (matplotlib-guarded); CLI: `scripts/plot_results.py`                               |
 
-`record_event()` is a **classmethod** and is **non-blocking**: events are placed on a queue and
-written by a background thread. Returns `True` if recorded, `False` if no recorder is active
-(when `assert_active=False`).
-
-### `MetricsReporter`
-
-```python
-class MetricsReporter:
-    def __init__(
-        self,
-        connection_name: os.PathLike,
-        client_type: str = "duckdb",
-    ) -> None
-
-    def create_report(
-        self,
-        tokenizer: Tokenizer | None = None,
-        tpot_reporting_mode: TPOTReportingMode = TPOTReportingMode.REQUEST_WEIGHTED,
-    ) -> Report
-
-    def dump_to_json(self, json_path: Path) -> None
-```
+## Public Interface
 
-`create_report()` executes SQL aggregation over the events database and returns a `Report`
-object. The optional `tokenizer` enables output sequence length (OSL) computation. `Report`
-itself exposes `display(fn=print, summary_only=False)` for console output.
+### `Report` (`report.py`)
 
-### Metric Types (`metric.py`)
+`Report` is a frozen `msgspec.Struct`. It is built by `from_snapshot` and is the single object that
+`display()` and `to_json()` render.
 
 ```python
-class Throughput(Metric):
-    REL_TOL = 0.1                  # ±10% relative tolerance
-    def __init__(self, target_qps: float): ...   # stored as self.target
-
-class QueryLatency(Metric):
-    REL_TOL = 0.1
-    def __init__(self, target_latency_ms: float | None = None,
-                 target_qps: float | None = None): ...
-
-class TTFT(Metric):
-    def __init__(self, max_ttft_latency_ms: float): ...  # hard ceiling
+class Report(msgspec.Struct, frozen=True):
+    version: str
+    git_sha: str | None
+    test_started_at: int
+    n_samples_issued: int
+    n_samples_completed: int
+    n_samples_failed: int
+    duration_ns: int | None
+    state: str            # terminal SessionState string ("complete" / "interrupted")
+    complete: bool        # True iff state == "complete" AND n_pending_tasks == 0
+    ttft: dict            # per-metric rollup dicts (min/max/mean/percentiles/...)
+    tpot: dict
+    latency: dict
+    output_sequence_lengths: dict
+    legacy_loadgen_window_duration_ns: int | None = None
+    qps: float | None = None
+    tps: float | None = None
+    finish_reason_counts: dict[str, int] = {}
+    run_config: dict | None = None          # config (load pattern, warmup, RNG seeds)
+    accuracy: list[dict] = []               # per-dataset accuracy, attached post-scoring
 
-class TPOT(Metric):
-    def __init__(self, max_tpot_latency_ms: float): ...  # hard ceiling
-```
-
-Each metric exposes `is_valid(measurement) -> bool`. The target value is stored as
-`self.target` on the base `Metric` class.
-
-## Data Flow
-
-### Write Path (during run)
+    @classmethod
+    def from_snapshot(
+        cls, snap: dict, *, run_config: dict | None = None,
+        use_legacy_loadgen_qps_metrics: bool = True,
+    ) -> "Report": ...
 
-```
-SampleEventHandler.query_result_complete(result)
-  → EventRecorder.record_event(
-        SampleEvent.COMPLETE,
-        time.monotonic_ns(),
-        sample_uuid=result.id,
-        data={...},
-    )
-  → queue.put(EventRow(...))               # non-blocking
-  → background thread: INSERT INTO events
+    def to_json(self, save_to: os.PathLike | None = None) -> bytes: ...
+    def display(self, ...) -> None: ...
 ```
 
-### Read Path (after run)
-
-```
-MetricsReporter.create_report()
-  → SELECT / GROUP BY on events table (DuckDB)
-  → compute percentiles (p50, p90, p99, p999)
-  → compute TTFT  = time from LOADGEN_ISSUE_CALLED to FIRST_CHUNK
-  → compute TPOT  = (COMPLETE.ts - FIRST_CHUNK.ts) / output_tokens
-  → compute tracked duration from TEST_STARTED / STOP_PERFORMANCE_TRACKING windows
-  → compute QPS   = tracked completed samples / tracked duration
-  → validate each metric against RuntimeSettings.reported_metrics
-```
+**`from_snapshot`** reads counters and per-series rollups out of the snapshot dict, all via
+`.get(...)` with safe defaults so a partial/malformed snapshot yields an honest _incomplete_
+report instead of crashing (missing `state` defaults to `"interrupted"`). It derives headline
+throughput once:
+
+- The snapshot always carries **both** `tracked_duration_ns` (endpoints-native full-run window)
+  and `legacy_loadgen_window_duration_ns` (MLPerf LoadGen "completed" window, poisson only). It is
+  therefore config-agnostic and reinterpretable either way.
+- Which window drives the headline QPS/TPS is decided by the run config
+  (`use_legacy_loadgen_qps_metrics`), not by the snapshot. The legacy window is used only when it
+  is enabled, available, and there are ≥2 completions (`QPS = (completed-1)/window`); otherwise
+  both QPS and TPS fall back to the native window so they always share one window, and
+  `legacy_loadgen_window_duration_ns` is left `None` so the serialized report honestly records
+  which view it holds.
+- `tps` is additionally `None` when no OSL was recorded — i.e. no tokenizer is configured (or the
+  output is empty). OSL is captured on `COMPLETE` regardless of streaming, so non-streaming runs
+  still get a TPS when a tokenizer is available.
+
+**Accuracy** is not part of the metrics snapshot; `from_snapshot` leaves `accuracy` empty and the
+finalizer attaches per-dataset accuracy entries after scoring. `run_config` is supplied by the
+caller (it is config, not a measured metric).
+
+### Metric target types (`metric.py`)
+
+`Throughput`, `QueryLatency`, `TTFT`, and `TPOT` are the metric **targets** a ruleset validates a
+run against — not the measured rollups (those live in the snapshot / `Report`). Each stores its
+target on the base `Metric` and exposes `is_valid(measurement) -> bool`; `Throughput`/`QueryLatency`
+apply a relative tolerance, while `TTFT`/`TPOT` are hard latency ceilings.
 
 ## Design Decisions
 
-**SQLite as the event store**
-
-SQLite gives durable, queryable storage with no external dependencies. The write path uses a
-single background writer thread (SQLite's WAL mode is single-writer) to avoid contention.
-Aggregation uses DuckDB for columnar SQL performance over the file written by SQLite.
-
-**Queue-backed writes to decouple hot path**
-
-The `record_event()` call must not block the load generator thread. Events are placed on a
-`queue.Queue` and consumed by a dedicated writer thread. The queue is unbounded; back-pressure
-is not a concern because write throughput (SQLite) exceeds event rate in all tested scenarios.
-
-**Singleton enforcement**
-
-Only one `EventRecorder` may exist per process. The singleton is enforced at construction time
-with a class-level flag. This prevents double-counting if code accidentally constructs a second
-recorder.
+**Reporting is decoupled from aggregation.** Event recording, tokenization, and rollup live in the
+aggregator subprocess; `metrics/` only consumes the finished snapshot. This keeps the hot path (the
+aggregator) isolated from report formatting and lets the report be rebuilt offline from a persisted
+`final_snapshot.json`.
 
-**TPOT calculation modes**
+**The dict snapshot is the contract.** `from_snapshot` takes the `snapshot_to_dict` shape and never
+decodes the wire Struct — see the data-flow note above. `result_summary.json` is self-complete
+(carries derived QPS/TPS and `run_config`) so a valid run is fully described by its own artifact.
 
-TPOT can be weighted by request (each request contributes equally) or by output token count
-(each token contributes equally). The default is request-weighted. The `TPOTReportingMode` enum
-controls this at report time without re-running the benchmark.
+**Honest incompleteness over crashes.** Every snapshot read defaults safely; drain-timeout and
+interrupted runs produce a `Report` with `complete=False` and an explicit indicator in `display()`
+rather than a partial success that looks clean.
 
 ## Integration Points
 
-| Component                    | Role                                                      |
-| ---------------------------- | --------------------------------------------------------- |
-| `load_generator/sample.py`   | Calls `record_event()` for every state transition         |
-| `load_generator/session.py`  | Calls `create_report()` at run end; saves output          |
-| `config/runtime_settings.py` | `reported_metrics` list drives which metrics are computed |
-| `config/ruleset_base.py`     | Provides `Metric` targets for validation                  |
+| Component                                                                    | Role                                                                                                             |
+| ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- |
+| [`metrics_aggregator`](../async_utils/services/metrics_aggregator/DESIGN.md) | Produces the `MetricsSnapshot` / `final_snapshot.json` that `from_snapshot` consumes                             |
+| `commands/benchmark/execute.py`                                              | Launches the aggregator subprocess, reads `final_snapshot.json`, builds/attaches the `Report`, attaches accuracy |
+| `config/` (`use_legacy_loadgen_qps_metrics`)                                 | Selects which duration window drives headline QPS/TPS                                                            |
+| `scripts/plot_results.py`                                                    | CLI entry point over `results_plots.py`                                                                          |
diff --git a/docs/metrics/report_design.md b/docs/metrics/report_design.md
index 7f663c1a8..fa11f5305 100644
--- a/docs/metrics/report_design.md
+++ b/docs/metrics/report_design.md
@@ -1,106 +1,100 @@
 # Report Design
 
+Builder-level reference for `metrics/report.py`. For where the report sits in the run pipeline,
+see [metrics/DESIGN.md](DESIGN.md); for the snapshot it consumes, see the
+[metrics aggregator spec](../async_utils/services/metrics_aggregator/DESIGN.md).
+
 ## Overview
 
-The report module provides benchmark result summarization, display, and
-serialization. It reads from the KVStore (via `BasicKVStoreReader`) and
-produces a `Report` with rollup statistics, percentiles, and histograms.
+`report.py` provides benchmark result summarization, display, and serialization. Its input is the
+terminal metrics-aggregator snapshot in **dict** form (`snapshot_to_dict`, as persisted to
+`final_snapshot.json`); its output is a `Report` carrying rollup statistics, percentiles, and
+histograms, rendered to console and to `result_summary.json`.
 
 ## Architecture
 
 ```
-BasicKVStoreReader.snapshot()
+final_snapshot.json  (dict form of MetricsSnapshot)
         │
-        ▼
-  build_report(reader)
+        ▼  json.loads(...)  → dict
+  Report.from_snapshot(snap, run_config=..., use_legacy_loadgen_qps_metrics=...)
+        │
+        ├── counters  → n_samples_issued/completed/failed, tracked_duration_ns,
+        │                legacy_loadgen_window_duration_ns, finish-reason counts
         │
-        ├── counters → n_issued, n_completed, n_failed, duration_ns
+        ├── for each series (ttft, tpot, latency, osl):
+        │       _series_to_metric_dict(stat) → rollup dict
         │
-        └── for each series metric:
-              SeriesStats.values → compute_summary() → dict
+        └── derive qps / tps once (window chosen by run config)
         │
         ▼
-     Report
-        ├── .display(fn)     → human-readable output
-        ├── .to_json(path)   → JSON serialization
-        ├── .qps             → computed from n_completed / duration
-        └── .tps             → computed from osl total / duration
+     Report (frozen msgspec.Struct)
+        ├── .display(fn)   → human-readable output with histograms
+        └── .to_json(path) → result_summary.json (QPS/TPS + run_config included)
 ```
 
 ## Design Principles
 
-**No SQL, no UUID tracking, no deduplication.**
-
-The old `MetricsReporter` queried SQLite via duckdb and built `RollupQueryTable`
-objects with UUID-indexed rows, repeat counts, and numpy sorted arrays. None of
-this complexity is needed when the input is a `list[float]` from the KVStore.
-
-The entire rollup is a single function: `compute_summary(values) → dict`.
-It calls numpy for percentiles and histograms. No classes, no state.
-
-**Reports are reproducible from the event log.**
+**The dict snapshot is the sole input.** `from_snapshot` consumes the `snapshot_to_dict` shape
+directly and never decodes the wire `MetricsSnapshot` Struct (which is `array_like=True` for compact
+msgpack). A consumer feeds `json.loads(path.read_bytes())` straight in. This lets a report be
+rebuilt offline from any persisted `final_snapshot.json`.
 
-The KVStore is lossy aggregation — it stores per-metric series, not per-sample
-provenance. The authoritative record of what happened during a run is the event
-log written by the `EventLoggerService`. Every number in a `Report` can be
-recomputed by replaying the event log through the same aggregator logic: if a
-production report shows a TTFT spike, the event log is the ground truth a user
-can mine to attribute the spike to specific samples or time windows.
+**Honest incompleteness over crashes.** Every counter and series read uses `.get(...)` with a safe
+default, so a truncated or partial (INTERRUPTED) snapshot produces an honest _empty_ rollup and a
+`complete=False` report rather than raising. Missing `state` defaults to `"interrupted"`.
 
-New metrics must preserve this property: the aggregator may only derive values
-from event fields, never from out-of-band state. If a metric cannot be rebuilt
-from the event log alone, it does not belong in the KVStore.
+**Self-describing artifacts.** Derived QPS/TPS, the window that produced them
+(`legacy_loadgen_window_duration_ns`), and `run_config` are all serialized, so a valid run is fully
+identified by its own `result_summary.json`.
 
 ## Components
 
-### `compute_summary(values, percentiles, n_histogram_buckets) → dict`
+### `_series_to_metric_dict(stat) → dict`
 
-Takes a `list[float]`, returns a dict with:
+Converts one series-stat dict from the snapshot into the rollup shape `display()` expects:
 
 - `total`, `min`, `max`, `avg`, `std_dev`, `median`
-- `percentiles`: dict of `{str(p): float}` for each requested percentile
+- `percentiles`: `{str(p): float}` for each requested percentile
 - `histogram`: `{"buckets": [(lo, hi), ...], "counts": [int, ...]}`
-
-Empty input returns zeros with empty histogram/percentiles.
-
-### `Report` (frozen dataclass)
-
-Fields:
-
-- `version`, `git_sha`, `test_started_at`
-- `n_samples_issued`, `n_samples_completed`, `n_samples_failed`
-- `duration_ns`
-- `ttft`, `tpot`, `latency`, `output_sequence_lengths` — each a summary dict
-
-Properties:
-
-- `qps`: `n_samples_completed / (duration_ns / 1e9)`, or None
-- `tps`: `osl_total / (duration_ns / 1e9)`, or None
-
-Methods:
-
-- `display(fn, summary_only, newline)` — human-readable output with histograms
-- `to_json(save_to)` — JSON serialization with QPS/TPS included
-
-### `build_report(reader) → Report`
-
-Reads a `BasicKVStoreReader` snapshot and constructs a `Report`. Works
-identically for live metrics (mid-test) and final reports (post-drain) —
-the caller decides when to call it.
-
-Counter keys read: `n_samples_issued`, `n_samples_completed`,
-`n_samples_failed`, `duration_ns`, `test_started_at`.
-
-Series keys summarized: `ttft_ns`, `tpot_ns`, `sample_latency_ns`, `osl`.
-
-## What Was Removed
-
-From the old `metrics/reporter.py`:
-
-- `MetricsReporter` — SQLite/duckdb query engine (replaced by KVStore)
-- `RollupQueryTable` — UUID-indexed rollup table (replaced by `compute_summary`)
-- `MetricRow` — per-row accessor (not needed)
-- `TPOTReportingMode` — niche enum (can be re-added if needed)
-- `SampleUUIDNotFoundError` — UUIDs not relevant in KVStore
-- `output_sequence_from_data` — SQL event data parser (not needed)
-- `dump_to_json` / `dump_all_to_csv` — event log export (handled by EventLoggerService)
+- `early_stopping_percentiles` (optional): MLPerf early-stopping estimate map, placed right after
+  `percentiles`; present only when early stopping is enabled (COMPLETE snapshots for ttft/tpot/latency)
+
+`avg`/`std_dev`/`median` are derived from the cheap rollups + percentiles (integer-exact variance
+form for ns series). `median` falls back to `(min + max) / 2` only for hand-crafted dicts that omit
+p50. A zero-count series returns `{}` (or an all-null early-stopping map if the feature is enabled).
+
+### `Report` (frozen `msgspec.Struct`)
+
+Fields: `version`, `git_sha`, `test_started_at`, `n_samples_issued/completed/failed`,
+`duration_ns`, `state`, `complete`, the four rollup dicts (`ttft`, `tpot`, `latency`,
+`output_sequence_lengths`), `legacy_loadgen_window_duration_ns`, `qps`, `tps`,
+`finish_reason_counts`, `run_config`, and `accuracy`.
+
+- `complete` = `state == "complete" and n_pending_tasks == 0` — `False` marks partial async metrics
+  (drain timeout, interrupt, or a live-tick fallback when no final snapshot was found).
+- `qps`/`tps` are computed once in `from_snapshot` (see below) rather than as live properties, so
+  the serialized report is self-complete.
+- `accuracy` is empty from `from_snapshot`; per-dataset entries are attached after scoring.
+
+Methods: `display(fn, ...)` for console output with histograms; `to_json(save_to)` for
+`result_summary.json`.
+
+### `Report.from_snapshot(snap, *, run_config=None, use_legacy_loadgen_qps_metrics=True) → Report`
+
+Splits the snapshot's `metrics` list into counters and series, then builds the `Report`.
+
+Counter keys read: `tracked_samples_issued`, `tracked_samples_completed`,
+`tracked_samples_failed`, `tracked_duration_ns`, `legacy_loadgen_window_duration_ns`, and the
+`tracked_finish_reason_*` counters. Series read (snapshot key → report field, via
+`SERIES_TO_SUMMARY_FIELD`): `ttft_ns` → `ttft`, `tpot_ns` → `tpot`, `sample_latency_ns` →
+`latency`, and `osl` → `output_sequence_lengths`.
+
+**QPS/TPS window selection.** The snapshot always carries both a native window
+(`tracked_duration_ns`) and the MLPerf LoadGen "completed" window
+(`legacy_loadgen_window_duration_ns`, poisson only), so it stays reinterpretable either way. The
+legacy window drives the headline QPS/TPS only when it is enabled
+(`use_legacy_loadgen_qps_metrics`), available, and there are ≥2 completions
+(`QPS = (completed - 1) / window`). Otherwise both QPS and TPS fall back to the native window so
+they always share one window, and `legacy_loadgen_window_duration_ns` is left `None` so the
+serialized report records which view it holds. `tps` is `None` when no OSL was recorded.
diff --git a/docs/openai/DESIGN.md b/docs/openai/DESIGN.md
index 880ae7b75..a52c0bc7f 100644
--- a/docs/openai/DESIGN.md
+++ b/docs/openai/DESIGN.md
@@ -61,13 +61,13 @@ class HttpRequestAdapter(ABC):
     def decode_response(cls, response_bytes: bytes, query_id: str) -> QueryResult: ...
 
     @classmethod
-    def decode_sse_message(cls, json_bytes: bytes) -> str: ...
+    def decode_sse_message(cls, json_bytes: bytes) -> SSEChoice: ...
 ```
 
 `dataset_transforms()` returns adapter-specific transforms that shape dataset rows into the
 expected `Query.data` schema. `encode_query()` serialises a `Query` to HTTP request bytes.
-`decode_response()` parses a non-streaming response. `decode_sse_message()` extracts the content
-string from a single SSE JSON payload; `parse_sse_chunk()` (concrete, on the base class) iterates
+`decode_response()` parses a non-streaming response. `decode_sse_message()` decodes a single SSE
+JSON payload into an `SSEChoice`; `parse_sse_chunk()` (concrete, on the base class) iterates
 the SSE buffer and calls it repeatedly.
 
 ### `SSEAccumulatorProtocol` (protocol, defined in `endpoint_client/accumulator_protocol.py`)
@@ -87,15 +87,16 @@ than shared across a connection.
 
 ## Key Files
 
-| File                        | Purpose                                                      |
-| --------------------------- | ------------------------------------------------------------ |
-| `openai_msgspec_adapter.py` | Hot-path adapter; uses msgspec for request encoding          |
-| `openai_adapter.py`         | Standard adapter; uses stdlib json                           |
-| `accumulator.py`            | Per-request streaming accumulator for OpenAI SSE deltas      |
-| `types.py`                  | Python type annotations for OpenAI response objects          |
-| `openai_types_gen.py`       | Auto-generated from `openapi.yaml`; do not edit manually     |
-| `harmony.py`                | Optional `openai-harmony` integration for compatibility shim |
-| `openapi.yaml`              | OpenAI API spec snapshot; excluded from pre-commit           |
+| File                        | Purpose                                                                                                                       |
+| --------------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
+| `openai_msgspec_adapter.py` | Hot-path chat adapter; uses msgspec for request encoding                                                                      |
+| `openai_adapter.py`         | Standard chat adapter; uses stdlib json                                                                                       |
+| `completions_adapter.py`    | Pre-tokenized `/v1/completions` adapter (`OpenAITextCompletionsAdapter`) — sends token IDs, bypasses the server chat template |
+| `accumulator.py`            | Per-request streaming accumulator for OpenAI SSE deltas                                                                       |
+| `types.py`                  | Python type annotations for OpenAI response objects                                                                           |
+| `openai_types_gen.py`       | Auto-generated from `openapi.yaml`; do not edit manually                                                                      |
+| `harmony.py`                | Optional `openai-harmony` integration for compatibility shim                                                                  |
+| `openapi.yaml`              | OpenAI API spec snapshot; excluded from pre-commit                                                                            |
 
 ## Design Decisions