diff --git a/docs/index.md b/docs/index.md index bf51ef64..58f00306 100644 --- a/docs/index.md +++ b/docs/index.md @@ -89,9 +89,9 @@ Organized by domain (model line / subsystem / playbook / lesson) instead of by l | Path | TL;DR | | --- | --- | -| `models/deepseek-v2-lite/status.md` | DeepSeek-V2-Lite EP2 status ledger: correctness, direct attribution, lifecycle reliability, and the #466 six-child host-staged/NCCL HTTP SLO report remain separate evidence buckets; no soak, parity, or production claim. | -| `models/deepseek-v2-lite/benchmarking.md` | Verification ladder and commands for DSV2-Lite correctness, direct decode diagnostics, retained host-staged/NCCL HTTP SLO profiles, and the separate soak/production boundary. | -| `models/deepseek-v2-lite/roadmap.md` | Single-node EP2 roadmap: #466 retained SLO reporting is an evidence layer; soak, device KV/attention, long-prefill scheduling, and Stable promotion remain separate gates. | +| `models/deepseek-v2-lite/status.md` | DeepSeek-V2-Lite EP2 status ledger: correctness, direct attribution, lifecycle reliability, #466 HTTP SLO reports, and #465 host-staged/NCCL soak evidence remain separate buckets; no parity or production claim. | +| `models/deepseek-v2-lite/benchmarking.md` | Verification ladder and commands for DSV2-Lite correctness, direct decode diagnostics, retained host-staged/NCCL HTTP SLO profiles, and #465 sustained soak evidence. | +| `models/deepseek-v2-lite/roadmap.md` | Single-node EP2 roadmap: #466 SLO and #465 short-shape soak are retained evidence layers; device KV/attention, long-prefill scheduling, hard budgets, and Stable promotion remain separate gates. | | `models/deepseek-v2-lite/benchmark-artifact-manifest.md` | Issue #467 implemented: the retained DeepSeek-V2-Lite benchmark matrix emits `artifact_manifest.json` and `regression_summary.json`, with CPU-only summarize-only tests. | | `models/deepseek-v2-lite/hf-accuracy-gate.md` | DeepSeek-V2-Lite EP2 HF accuracy gate after PR #149/#150/#274: HF `generate(use_cache=true)`, host-staged EP2, and NCCL EP2 are compared across the committed small case set. | | `models/deepseek-v2-lite/decode-attribution-gate.md` | Direct diagnostic benchmark for DeepSeek-V2-Lite EP2 `Hello`/16-token batch sizes 1/4/8: structured timing/counters and graph probes with no HTTP SLO or production claim. | diff --git a/docs/models/deepseek-v2-lite/benchmarking.md b/docs/models/deepseek-v2-lite/benchmarking.md index 92bf0353..46862b14 100644 --- a/docs/models/deepseek-v2-lite/benchmarking.md +++ b/docs/models/deepseek-v2-lite/benchmarking.md @@ -1,6 +1,6 @@ # DeepSeek-V2-Lite Verification And Benchmarking -> **TL;DR:** Use `e2e_ep2` for correctness, `dsv2_lite_ep2_decode_attribution` for direct decode diagnostics, and `bench_dsv2lite_http_slo.py` for retained HTTP SLO evidence. The #466 follow-up NCCL readiness fix lets the backend discover a compatible Python-wheel NCCL runtime from `PATH`; these artifacts still answer different questions, and none of them alone proves production readiness. +> **TL;DR:** Use `e2e_ep2` for correctness, `dsv2_lite_ep2_decode_attribution` for direct decode diagnostics, `bench_dsv2lite_http_slo.py` for retained HTTP SLO evidence, and `bench_dsv2lite_http_soak.py` for issue #465 sustained HTTP soak evidence. These artifacts still answer different questions, and none of them alone proves production readiness. > > Last touched: 2026-07 @@ -11,7 +11,7 @@ | Correctness / integration | `openinfer-deepseek-v2-lite/tests/e2e_ep2.rs` | EP2 load, host-staged/NCCL generation, request isolation, output tokens/text/hashes, route and collective accounting | Latency, throughput, SLO, soak, production readiness | | Direct diagnostic | `dsv2_lite_ep2_decode_attribution` | Fixed-shape CPU/CUDA section timing, route/collective counters, graph-readiness diagnostics | HTTP behavior, client pressure, serving SLO | | HTTP serving SLO | `scripts/bench_dsv2lite_http_slo.py` over the shared HTTP harness | Streaming TTFT/TPOT/ITL, request/output throughput, failures/timeouts, server trace coverage, output hashes, repeat spread | Direct-kernel attribution, sustained soak, production readiness | -| Soak / production readiness | Separate sustained-run gate | Memory drift, long-duration tails, recovery and deployment limits | Out of scope for issue #466 | +| Soak / production readiness | `scripts/bench_dsv2lite_http_soak.py` | Sustained request completion, first/last-quartile tail and throughput drift, RSS/VRAM drift, terminal reasons, clean follow-up recovery | Direct decode attribution, vLLM parity, multi-node recovery, or production readiness by itself | ## Correctness Gate @@ -132,20 +132,91 @@ Each `sweep_summary.json` records: A cell is `noisy` when repeat spread exceeds 10% of the median for TTFT/TPOT/ITL p95, request throughput, or output-token throughput. +## Sustained HTTP Soak + +Issue #465 uses `scripts/bench_dsv2lite_http_soak.py`. It reuses the generic streaming `/v1/completions` leaf benchmark inside fixed time buckets, then writes a backend-level `soak_summary.json` and an optional host-staged/NCCL combined report. A failed soak is still useful evidence when the JSON keeps the failing leaf artifact, server log pointer, terminal reasons, and clean follow-up result. + +The default retained shape is greedy, ignore-EOS, `prompt_words=64`, `max_tokens=64`, concurrency `4,8`, and no production latency budget. Tune `--duration-s`, `--bucket-s`, and `--num-requests` for the validation host; `--bucket-s` controls the target wall-clock window per bucket, while `--num-requests` controls each leaf chunk launched continuously inside that window. The summary records actual elapsed time, bucket count, and leaf count. + +```bash +# Template: run one backend soak after the server is ready. +python3 scripts/bench_dsv2lite_http_soak.py run \ + --backend BACKEND \ + --base-url http://127.0.0.1:18000 \ + --server-log artifacts/bench/dsv2-lite/RUN_ID/BACKEND/server.log \ + --model-path MODEL_PATH \ + --server-command "SERVER_COMMAND" \ + --commit COMMIT \ + --model-revision MODEL_REVISION \ + --server-binary SERVER_BINARY \ + --backend-runtime-version BACKEND_RUNTIME_VERSION \ + --duration-s 1800 \ + --bucket-s 300 \ + --num-requests 32 \ + --concurrency 4,8 \ + --prompt-words 64 \ + --max-tokens 64 \ + --out-dir artifacts/bench/dsv2-lite/RUN_ID/BACKEND/soak + +# Template: combine host-staged and NCCL backend summaries. +python3 scripts/bench_dsv2lite_http_soak.py combine \ + --summary artifacts/bench/dsv2-lite/RUN_ID/host-staged/soak/soak_summary.json \ + --summary artifacts/bench/dsv2-lite/RUN_ID/nccl/soak/soak_summary.json \ + --out artifacts/bench/dsv2-lite/RUN_ID/retained_soak_report.json +``` + +Each backend summary records: + +- leaf artifact path and SHA-256 for every leaf chunk inside each bucket; +- completed, failed, timeout, terminal-reason, and error counts; +- TTFT, TPOT, ITL, request throughput, and output-token throughput per bucket; +- first/last-quartile drift summaries for tails, throughput, RSS, and VRAM; +- active-set, pending-queue, decode-batch, token-timing, and missing-trace coverage when server traces are available; +- output hash distribution and combined hash; +- process RSS and total device memory samples; +- post-soak clean follow-up result. + +`soak_gate.passed` means every requested concurrency has loaded bucket evidence, leaf commands completed successfully with zero failures/timeouts, optional trace coverage passed for every bucket, and the clean follow-up completed. Numeric drift is reported but not a hard budget until deployment limits are ratified. + +The combined host-staged/NCCL report hard-fails when either backend is missing, a child `soak_gate` fails, the child commits differ, model/server provenance differs, or a backend runtime boundary is missing or generic. This keeps NCCL selector evidence attached to the soak result instead of turning a conservative runtime run into a default-runtime claim. + ## Local Tooling Gate -These commands were run on 2026-07-13: +These commands were run on 2026-07-18: ```bash -python3 -m py_compile scripts/bench_http_common.py scripts/bench_http_serving.py scripts/bench_http_sweep.py scripts/bench_dsv2lite_http_slo.py -python3 -m unittest -v tests/test_bench_http_serving.py tests/test_bench_http_sweep.py tests/test_bench_dsv2lite_http_slo.py +python3 -m py_compile scripts/bench_http_common.py scripts/bench_http_serving.py scripts/bench_http_sweep.py scripts/bench_dsv2lite_http_slo.py scripts/bench_dsv2lite_http_soak.py +python3 -m unittest -v tests/test_bench_http_common.py tests/test_bench_http_serving.py tests/test_bench_http_sweep.py tests/test_bench_dsv2lite_http_slo.py tests/test_bench_dsv2lite_http_soak.py cargo fmt --all --check cargo metadata --locked --no-deps --format-version 1 ``` ## Current-Source Evidence -Regenerate the retained report when its profile, schema, or measurement contract changes. The current retained 2x RTX 5090 evidence completed all six host-staged/NCCL children with zero failures/timeouts and full required trace coverage. The gitignored local copy is under `artifacts/bench/dsv2-lite//`; aggregate SHA-256: `a7e677c63d1ce92ad0c069f83acfcc8b381e07d06bed9364ab899adecef8d317`. +Regenerate retained reports when their profile, schema, or measurement contract changes. The gitignored local copies live under `artifacts/bench/dsv2-lite//`; keep public docs to artifact basenames, hashes, contract shape, and claim boundaries. + +### Issue #465 Retained HTTP Soak + +The retained #465 latest-HEAD rerun used code commit `a5703d0424d917ce99b4bd8691b0b86eecde966f`, model revision `604d5664dddd88a0433dbae533b7fe9472482de0`, 2x RTX 5090, `prompt_words=64`, `max_tokens=64`, greedy sampling, ignore-EOS, concurrency `4,8`, `duration_s=120`, `bucket_s=60`, `num_requests=8`, and full required trace coverage. Later commits may update docs or tooling gates; regenerate the artifacts before claiming evidence for a changed benchmark schema or runtime path. The combined report SHA-256 is `1c06e8825da70888277f1485f54d7f4fb9b2f61d617149d3ac7357cd5a03e7f1`. + +| Artifact basename | Backend | SHA-256 | Result | Boundary | +| --- | --- | --- | --- | --- | +| `soak_summary.json` | host-staged | `3fe4f163024602a51f10cac0c15cc24a5feffd92bbcb93091e175cd63e49bd33` | `completed=112`, `failed=0`, `timeouts=0`, `soak_gate.passed=true`, clean follow-up passed, combined output hash `6912777bed672f57` | Host-staged sustained HTTP soak evidence | +| `soak_summary.json` | NCCL | `52c2a70d895fc0c080233961ea06824097fbf5b7241a0983e41c0ca6646e762f` | `completed=128`, `failed=0`, `timeouts=0`, `soak_gate.passed=true`, clean follow-up passed, combined output hash `24f2db9fc47acc10` | NCCL sustained HTTP soak evidence with NCCL `2.26.2`, `NCCL_IB_DISABLE=1`, `NCCL_P2P_DISABLE=1` | +| `retained_soak_report.json` | combined | `1c06e8825da70888277f1485f54d7f4fb9b2f61d617149d3ac7357cd5a03e7f1` | `coverage_gate.passed=true`, `child_gates={host-staged:true,nccl:true}`, commit/model provenance consistent, runtime boundaries present | Combined host-staged/NCCL #465 report | + +Resource and drift fields are retained for diagnosis, not hard budgets. Host-staged max device memory stayed flat in both concurrency buckets, with output-token throughput drift `-1.5%` at c4 and `-0.8%` at c8; RSS moved `+1.6%` at c4 and `+12.1%` at c8. NCCL max device memory stayed flat, with output-token throughput drift `-1.0%` at c4 and `+1.1%` at c8; RSS moved `+1.6%` at c4 and `+3.8%` at c8. The NCCL server used a conservative single-node runtime selection and reached readiness after the NCCL communicator init path completed; do not turn this row into a default-runtime speed claim. + +Reviewer evidence summary for issue/PR text: + +- Combined artifact basename/SHA: `retained_soak_report.json` / `1c06e8825da70888277f1485f54d7f4fb9b2f61d617149d3ac7357cd5a03e7f1`. +- Child artifact basenames/SHA: host-staged `soak_summary.json` / `3fe4f163024602a51f10cac0c15cc24a5feffd92bbcb93091e175cd63e49bd33`; NCCL `soak_summary.json` / `52c2a70d895fc0c080233961ea06824097fbf5b7241a0983e41c0ca6646e762f`. +- Gate result: host-staged and NCCL child gates passed, combined gate passed, commit/model provenance consistent, runtime boundary retained. +- Claim boundary: short-shape sustained HTTP soak evidence only; no production-readiness, default-runtime NCCL, long/mixed-prompt, or vLLM-parity claim. + +### Issue #466 Retained HTTP SLO Evidence + +The current retained #466 2x RTX 5090 evidence completed all six host-staged/NCCL children with zero failures/timeouts and full required trace coverage. The gitignored local copy is under `artifacts/bench/dsv2-lite//`; aggregate SHA-256: `a7e677c63d1ce92ad0c069f83acfcc8b381e07d06bed9364ab899adecef8d317`. ### Issue #466 Follow-Up NCCL Readiness Smoke diff --git a/docs/models/deepseek-v2-lite/roadmap.md b/docs/models/deepseek-v2-lite/roadmap.md index ba8b2908..a6620bd8 100644 --- a/docs/models/deepseek-v2-lite/roadmap.md +++ b/docs/models/deepseek-v2-lite/roadmap.md @@ -1,6 +1,6 @@ # DeepSeek-V2-Lite Serving Roadmap -> **TL;DR:** DeepSeek-V2-Lite has single-node EP2 correctness, request-lifecycle reliability, direct diagnostics, and retained HTTP SLO reporting. Production readiness still requires sustained soak, bounded device KV ownership, long-prefill scheduling, and explicit deployment limits. +> **TL;DR:** DeepSeek-V2-Lite has single-node EP2 correctness, request-lifecycle reliability, direct diagnostics, retained HTTP SLO reporting, and a retained #465 short-shape soak baseline. Production readiness still requires bounded device KV ownership, long-prefill scheduling, ratified budgets, and explicit deployment limits. > > Last touched: 2026-07 @@ -15,7 +15,7 @@ Evidence stays in four buckets: 3. HTTP serving SLO: fixed `/v1/completions` contracts, tails, throughput, failures/timeouts, trace coverage, hashes, repeat spread. 4. Soak and production readiness: sustained memory/tail drift, recovery, capacity, deployment and support limits. -The first three can be green while the fourth remains open. +The first three can be green while the fourth is only a retained baseline. A passing soak report records sustained behavior for a named contract; it does not define production budgets by itself. ## Current Gates @@ -25,7 +25,7 @@ The first three can be green while the fourth remains open. | Direct decode attribution | Retained | `decode-attribution-gate.md` | Direct diagnostic only | | HTTP lifecycle reliability | Retained | `status.md`, issue #453 | Failure isolation and recovery scenarios, no long-duration claim | | HTTP SLO report | Retained for #466 | `benchmarking.md`, `bench_dsv2lite_http_slo.py` | Fixed host-staged/NCCL HTTP contracts retained; no soak or production claim | -| Sustained soak | Open | issue #465 | Required before Stable promotion | +| Sustained soak | Retained baseline for #465 | `benchmarking.md`, `bench_dsv2lite_http_soak.py` | Fixed short-shape host-staged/NCCL soak retained; runtime/provenance gates are hard, drift is reported, not a ratified production budget | | Long-prefill scheduling | Open | issue #452 | Current long smoke records the boundary; it does not close latency work | | Device attention and KV | Open | issue #635 | Required for bounded device lifetime and stronger scaling | @@ -35,6 +35,14 @@ Issue #466 is an evidence and reporting milestone. It provides named DSV2-Lite p This closes the missing report layer. It does not optimize latency or throughput and does not close issue #465. +## Issue #465 Position + +Issue #465 is the sustained evidence milestone. The retained 2026-07-18 run covered host-staged and NCCL under the fixed short-shape HTTP soak contract with zero failures/timeouts, full required trace coverage, clean follow-up recovery, output hashes, resource samples, and first/last-quartile drift fields. The NCCL row used NCCL `2.26.2` with `NCCL_IB_DISABLE=1` and `NCCL_P2P_DISABLE=1`; keep that runtime boundary attached to the evidence. + +This closes the first retained soak baseline. It does not set hard latency or drift budgets, does not cover long-prefill scheduling, and does not claim production readiness. + +The tooling now guards the evidence boundary directly: backend summaries fail on missing bucket coverage, leaf command errors, missing trace coverage, or failed clean follow-up, and the combined report fails on missing host-staged/NCCL children, failed child gates, provenance drift, commit drift, or missing/generic runtime boundaries. Long-duration and long/mixed-prompt soaks still need separate versioned contracts before Stable promotion. + ## Sequence ### 1. Retain The Serving Contract @@ -44,13 +52,14 @@ This closes the missing report layer. It does not optimize latency or throughput - Fail retained runs on request failures, timeouts, missing traces, or missing active/decode coverage. - Keep startup failures as structured artifacts instead of dropping failed cells. -### 2. Close Sustained Availability +### 2. Extend Sustained Availability Primary issue: #465. -- Run host-staged and NCCL for ratified short and long durations. -- Track first/last-quartile tails and throughput, RSS/VRAM drift, active/pending state, terminal reasons, and clean follow-up recovery. -- Calibrate budgets from retained variance before promoting numeric thresholds to hard gates. +- Keep the retained host-staged/NCCL short-shape soak reproducible after scheduler, frontend, trace, or backend changes. +- Extend from the short-shape baseline into ratified long-duration and long/mixed-prompt soaks only when those contracts have explicit support limits. +- Calibrate latency, throughput, RSS, and VRAM budgets from repeated retained variance before promoting numeric thresholds to hard gates. +- Keep conservative NCCL runtime selectors in the retained evidence until default-runtime startup is separately proven on the supported hardware matrix. ### 3. Move Decode State To The Device @@ -77,6 +86,7 @@ Promotion requires all of the following: - request and KV capacity are bounded and observable; - lifecycle and soak gates recover cleanly with no unexplained drift; - short, mixed, and long retained SLO reports pass on the supported hardware/runtime matrix; +- long-duration and long/mixed-prompt soak profiles have ratified duration, shape, and drift budgets; - backend startup failures give actionable version or configuration errors; - API, topology, context, sampling, and recovery limits are documented; - performance claims use matched repeated HTTP contracts, while direct and profiler data remain diagnostic. diff --git a/docs/models/deepseek-v2-lite/status.md b/docs/models/deepseek-v2-lite/status.md index 2e372986..010d1924 100644 --- a/docs/models/deepseek-v2-lite/status.md +++ b/docs/models/deepseek-v2-lite/status.md @@ -1,6 +1,6 @@ # DeepSeek-V2-Lite Status And Benchmark Ledger -> **TL;DR:** DeepSeek-V2-Lite keeps correctness, direct decode diagnostics, retained HTTP SLO reports, and soak readiness as separate gates. HF/host-staged/NCCL exactness and HTTP lifecycle evidence are retained; issue #466 added fixed host-staged/NCCL SLO artifacts, and the follow-up NCCL readiness fix covers the no-selector short HTTP path without claiming production readiness. +> **TL;DR:** DeepSeek-V2-Lite keeps correctness, direct decode diagnostics, retained HTTP SLO reports, and sustained soak evidence as separate gates. HF/host-staged/NCCL exactness, HTTP lifecycle evidence, #466 SLO artifacts, and #465 host-staged/NCCL soak artifacts are retained without claiming production readiness. Last touched: 2026-07 @@ -24,6 +24,7 @@ Last touched: 2026-07 | HTTP trace and measured MoE throughput optimization | HTTP and direct evidence | Issue #280 logs DeepSeek-V2-Lite `openinfer_http_trace` records and batches same-position decode subgroups. Issue #464 extends phase/decode-step attribution, groups host-staged and NCCL routes by stable `(owner_rank, global_expert)`, and moves the NCCL gate GEMM to a bitwise-matched CUDA logits kernel while keeping host top-k/softmax. Diagnostic serial/host rollback switches remain available for retained A/B and emergency rollback; they are not production tuning knobs. | | HTTP reliability lifecycle gate | Available | Issue #453 adds `scripts/bench_dsv2lite_http_reliability.py`, which drives real streaming `/v1/completions` scenarios for client cancel/disconnect, unsupported params, active-cap overload, mixed short/long prompts with adjacent failures, and clean follow-up recovery. The 2026-07-04 2x RTX 5090 host-staged and NCCL runs both passed with terminal trace coverage, stable output hashes, active/pending/decode maxima, and healthy final scheduler baselines. | | Retained HTTP serving SLO report | Retained HTTP evidence | Issue #466 adds model-owned short/mixed/long profiles on the shared HTTP benchmark scripts. The current retained run covered all six host-staged/NCCL children with zero failures/timeouts and full trace coverage. The #466 follow-up fixed NCCL no-selector readiness by discovering a compatible Python-wheel NCCL runtime from `PATH`; the short NCCL c1 HTTP smoke now reaches readiness and completes without startup failure or layer-1 illegal address. This is HTTP pressure/SLO evidence only; command details and artifact hashes live in `benchmarking.md`. | +| Sustained HTTP soak report | Retained soak evidence | Issue #465 adds `scripts/bench_dsv2lite_http_soak.py`, which reuses the shared HTTP leaf benchmark inside fixed time buckets, then retains backend-level and host-staged/NCCL combined JSON. The 2026-07-18 2x RTX 5090 run covered host-staged and NCCL with zero failures/timeouts, full required trace coverage, clean follow-up recovery, output hashes, resource samples, and first/last-quartile drift fields. The NCCL row used NCCL `2.26.2` with `NCCL_IB_DISABLE=1` and `NCCL_P2P_DISABLE=1`, so this is sustained HTTP soak evidence for that runtime contract, not a default-runtime speed claim or production-readiness claim. Current tooling also fails missing bucket coverage, leaf command errors, failed child gates, provenance drift, and missing/generic runtime boundaries. | | Retained vLLM comparison matrix | Snapshot complete with clean failed setup rows and supplemental validation rows | The retained matrix for tracking issue #279 keeps HF/host/NCCL correctness, OpenInfer direct diagnostic batch, `vllm bench serve` HTTP pressure, OpenInfer trace rows, and failed setup rows separate. The 2026-06-28 clean full matrix passed HF / host-staged / NCCL correctness plus OpenInfer host-staged/NCCL direct, HTTP pressure, and trace rows; stock vLLM TP2 and TP2+EP2 failed during setup on the target FlashInfer SM120 path. A separate FlashInfer #3633-equivalent validation completed vLLM TP2 and TP2+EP2 under the same HTTP client/workload contract. | | vLLM production parity | Not claimed | The vLLM TP2 / TP2+EP2 rows are gap-finding evidence from a documented contract. The supplemental validation run is not serving parity or a stock-install claim. | @@ -67,6 +68,17 @@ The #466 follow-up readiness fix keeps this fail-closed floor but also scans Pyt The current-source #466 aggregate is retained under `artifacts/bench/dsv2-lite//` (gitignored). `benchmarking.md` contains the command, artifact fields, aggregate hash, and claim boundary. +### Issue #465 Retained HTTP Soak Report + +The retained soak layer uses model-owned `scripts/bench_dsv2lite_http_soak.py` over the generic `bench_http_serving.py` leaf runner. The retained 2026-07-18 latest-HEAD rerun used commit `a5703d0424d917ce99b4bd8691b0b86eecde966f`, model revision `604d5664dddd88a0433dbae533b7fe9472482de0`, 2x RTX 5090, `prompt_words=64`, `max_tokens=64`, greedy sampling, ignore-EOS, concurrency `4,8`, 120-second duration, and 60-second buckets. Both backends passed `soak_gate.passed=true`; command details and artifact hashes live in `benchmarking.md`. + +| Backend | Runtime boundary | Completed | Failed/timeouts | Buckets / leafs | Combined output hash | Boundary | +| --- | --- | ---: | ---: | ---: | --- | --- | +| host-staged | `host-staged` | 112 | 0 / 0 | 4 / 14 | `6912777bed672f57` | Sustained HTTP soak evidence for the host-staged oracle path | +| NCCL | NCCL `2.26.2`, `NCCL_IB_DISABLE=1`, `NCCL_P2P_DISABLE=1` | 128 | 0 / 0 | 4 / 16 | `24f2db9fc47acc10` | Sustained HTTP soak evidence for this conservative single-node NCCL runtime | + +The combined retained report passed child gates for both backends, commit consistency, model provenance consistency, and runtime-boundary retention. The soak JSON records first/last-quartile tail, throughput, RSS, and device-memory drift, but numeric drift is still reported evidence rather than a production budget. Treat long-duration and long/mixed-prompt soaks as follow-up contracts, not implied coverage from this short-shape baseline. + ### Retained vLLM TP2/EP2 Matrix The retained matrix lives in `docs/benchmarks/deepseek-v2-lite-vllm-tp2-ep2.md` and tracks [#279](https://github.com/openinfer-project/openinfer/issues/279). It is the current source for OpenInfer host-staged/NCCL versus vLLM TP2/TP2+EP2 under the `prompt_words=64`, `max_tokens=64`, `num_prompts=32`, `max_concurrency=1/4/8`, `temperature=0`, `ignore_eos=true` HTTP pressure contract. Prompt words are a workload-generator input, not a token count. @@ -275,6 +287,7 @@ Use these labels consistently: | `NCCL device logits router slice` | The gate GEMM runs on CUDA with bitwise host-logit coverage; existing host top-k/softmax builds the route plan. | Fully device-resident routing, no routing D2H, general numerical equivalence, or non-NCCL improvement. | | `HTTP reliability lifecycle gate` | `/v1/completions` cancel/disconnect/reject/overload/mixed-failure scenarios have terminal reason counts, trace coverage, active/pending/decode maxima, output hashes, and clean follow-up recovery evidence. | Production EP readiness, soak stability, SLO latency, vLLM parity, throughput improvement, sparse dispatch, or multi-node EP support. | | `retained HTTP serving SLO` | Named short/mixed/long `/v1/completions` contracts report latency percentiles, throughput, outcomes, full trace coverage, hashes, and repeat spread for one backend/hardware/toolchain. | Direct attribution, sustained soak, production readiness, vLLM parity, or performance outside a matched contract. | +| `retained HTTP soak` | Fixed-shape `/v1/completions` contracts run inside time buckets and report completion, failures/timeouts, trace coverage, output hashes, clean follow-up recovery, resource samples, and first/last-quartile drift. | Direct attribution, long/mixed-prompt readiness, production budgets, default-runtime speed, vLLM parity, or multi-node recovery. | | `covered NCCL decode graph probe` | Probe-only batch-1 `Hello` decode step captured, instantiated, replayed, and token-verified under CUDA Graph. | Default serving graph coverage, multi-step graph replay, batch `4/8` graph coverage, or performance improvement. | | `HTTP concurrency pressure` | `vllm bench serve --max-concurrency N` against an HTTP endpoint. | True OpenInfer batch size unless the engine path proves it. | | `vLLM comparison from documented environment` | vLLM TP2 / TP2+EP2 from the retained matrix or the separate FlashInfer-fixed validation. | Stock vLLM install support, OpenInfer serving parity, or production readiness. | @@ -326,7 +339,7 @@ The next implementation should be chosen from measured evidence: 5. Keep the #453 HTTP reliability evidence retained. - rerun the reliability runner for host-staged and NCCL when scheduler or HTTP lifecycle code changes; - keep the JSON artifact hashes and server-log trace coverage in the PR evidence; - - keep #452 long/mixed-prompt latency, #465 soak, the completed #466 SLO report layer, and #467 benchmark manifest separate. + - keep #452 long/mixed-prompt latency, the completed #465 soak baseline, the completed #466 SLO report layer, and #467 benchmark manifest separate. 6. Keep MoE internals readable. - routing, dispatch, expert execution, and combine should remain distinguishable in code and attribution; diff --git a/scripts/bench_dsv2lite_http_soak.py b/scripts/bench_dsv2lite_http_soak.py new file mode 100644 index 00000000..9c416566 --- /dev/null +++ b/scripts/bench_dsv2lite_http_soak.py @@ -0,0 +1,1282 @@ +#!/usr/bin/env python3 +"""Run and combine sustained DeepSeek-V2-Lite HTTP soak evidence.""" + +from __future__ import annotations + +import argparse +import json +import math +import subprocess +import sys +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from bench_http_common import ( + artifact_command, + combined_output_hash, + current_commit, + detect_hardware_toolchain, + model_fingerprint, + numeric_summary, + sha256_file, + value_counts, + write_json, +) + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parent +BENCH = SCRIPT_DIR / "bench_http_serving.py" +BACKENDS = ("host-staged", "nccl") +DEFAULT_CONTRACT_NAME = "dsv2-lite-http-soak" +DEFAULT_CONTRACT_DESCRIPTION = ( + "DeepSeek-V2-Lite sustained HTTP soak contract for EP2 serving." +) +DEFAULT_CLAIM_BOUNDARY = ( + "Sustained HTTP soak evidence for a fixed DeepSeek-V2-Lite EP2 serving " + "contract. This is not direct decode attribution, vLLM parity, multi-node " + "recovery, or production-readiness evidence by itself." +) +RESOURCE_SAMPLE_VERSION = 1 +GENERIC_RUNTIME_BOUNDARIES = {"", "unknown", "n/a", "na", "none"} + + +@dataclass(frozen=True) +class ResourceSample: + wall_s: float + rss_kib: int | None + gpu_memory_used_mib: list[int] + gpu_memory_scope: str + + +def parse_int_list(value: str) -> list[int]: + items = [item.strip() for item in value.split(",") if item.strip()] + if not items: + raise argparse.ArgumentTypeError("list must not be empty") + parsed = [int(item) for item in items] + if any(item <= 0 for item in parsed): + raise argparse.ArgumentTypeError("all values must be positive") + return parsed + + +def normalized_runtime_boundary(value: Any) -> str: + return str(value or "").strip() + + +def runtime_boundary_is_present(value: Any) -> bool: + return normalized_runtime_boundary(value).lower() not in GENERIC_RUNTIME_BOUNDARIES + + +def run_text(command: list[str]) -> str | None: + try: + return subprocess.check_output( + command, + text=True, + stderr=subprocess.DEVNULL, + timeout=5, + ).strip() + except (OSError, subprocess.SubprocessError): + return None + + +def ps_rss_kib(pid: int | None) -> int | None: + if pid is None: + return None + output = run_text(["ps", "-o", "rss=", "-p", str(pid)]) + if not output: + return None + try: + return int(output.splitlines()[0].strip()) + except (IndexError, ValueError): + return None + + +def gpu_memory_used_mib() -> list[int]: + output = run_text( + [ + "nvidia-smi", + "--query-gpu=memory.used", + "--format=csv,noheader,nounits", + ] + ) + if not output: + return [] + values = [] + for line in output.splitlines(): + try: + values.append(int(line.strip())) + except ValueError: + continue + return values + + +def sample_resources(args: argparse.Namespace, started_wall_s: float) -> ResourceSample: + return ResourceSample( + wall_s=time.time() - started_wall_s, + rss_kib=ps_rss_kib(args.server_pid), + gpu_memory_used_mib=gpu_memory_used_mib(), + gpu_memory_scope="device_total", + ) + + +def validate_commit(commit: str) -> None: + if not commit: + raise SystemExit("--commit is required") + try: + head = subprocess.check_output( + ["git", "rev-parse", "--verify", "HEAD^{commit}"], + cwd=REPO_ROOT, + text=True, + stderr=subprocess.DEVNULL, + timeout=5, + ).strip() + except (OSError, subprocess.SubprocessError) as exc: + raise SystemExit("soak runs require a Git worktree with a valid HEAD") from exc + if commit not in {head, head[:12]}: + raise SystemExit(f"--commit {commit} does not match current HEAD {head}") + + +def required_files(args: argparse.Namespace) -> list[Path]: + paths = [ + args.server_log, + Path(args.model_path) / "config.json", + Path(args.model_path) / "model.safetensors.index.json", + ] + if args.server_binary is not None: + paths.append(args.server_binary) + return paths + + +def validate_run_args(args: argparse.Namespace) -> None: + if args.duration_s <= 0.0: + raise SystemExit("--duration-s must be positive") + if args.bucket_s <= 0.0: + raise SystemExit("--bucket-s must be positive") + if args.num_requests <= 0: + raise SystemExit("--num-requests must be positive") + if args.timeout <= 0.0: + raise SystemExit("--timeout must be positive") + if args.required_trace_coverage is not None and not ( + 0.0 < args.required_trace_coverage <= 1.0 + ): + raise SystemExit("--required-trace-coverage must be in (0, 1]") + if args.max_buckets is not None and args.max_buckets <= 0: + raise SystemExit("--max-buckets must be positive") + runtime_boundary = normalized_runtime_boundary(args.backend_runtime_version) + if not runtime_boundary: + raise SystemExit("--backend-runtime-version must describe the runtime boundary") + if args.backend == "nccl" and runtime_boundary.lower() == "nccl": + raise SystemExit( + "--backend-runtime-version for NCCL must include the NCCL version " + "or selector boundary" + ) + args.backend_runtime_version = runtime_boundary + missing = [str(path) for path in required_files(args) if not path.is_file()] + if missing: + raise SystemExit(f"required soak-run files are missing: {', '.join(missing)}") + validate_commit(args.commit) + + +def optional_arg(flag: str, value: Any) -> list[str]: + if value is None: + return [] + return [flag, str(value)] + + +def build_leaf_command( + args: argparse.Namespace, + *, + out: Path, + concurrency: int, + num_requests: int, + prompt_words: int, + max_tokens: int, + contract_name: str, + contract_description: str, +) -> list[str]: + command = [ + sys.executable, + str(BENCH), + "--base-url", + args.base_url, + "--model", + args.model, + "--num-requests", + str(num_requests), + "--concurrency", + str(concurrency), + "--warmup", + "0", + "--prompt-words", + str(prompt_words), + "--max-tokens", + str(max_tokens), + "--temperature", + "0.0", + "--top-k", + "-1", + "--top-p", + "1.0", + "--timeout", + str(args.timeout), + "--server-log", + str(args.server_log), + "--contract-name", + contract_name, + "--contract-description", + contract_description, + "--claim-boundary", + args.claim_boundary, + "--backend", + args.backend, + "--model-path", + args.model_path, + "--server-command", + args.server_command, + "--commit", + args.commit, + "--source-revision", + args.commit, + "--model-revision", + args.model_revision, + "--backend-runtime-version", + args.backend_runtime_version, + "--out", + str(out), + ] + if args.server_binary is not None: + command.extend(["--server-binary", str(args.server_binary)]) + if args.required_trace_coverage is not None: + command.extend( + ["--required-trace-coverage", str(args.required_trace_coverage)] + ) + if args.no_ignore_eos: + command.append("--no-ignore-eos") + return command + + +def read_report(path: Path) -> dict[str, Any] | None: + try: + report = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return report if isinstance(report, dict) else None + + +def trace_max(report: dict[str, Any], field: str) -> int | None: + values = [] + for request in report.get("requests", []): + if not isinstance(request, dict): + continue + trace = request.get("server_trace") + if isinstance(trace, dict) and isinstance(trace.get(field), int): + values.append(int(trace[field])) + if values: + return max(values) + trace = report.get("server_trace") + if isinstance(trace, dict) and isinstance(trace.get(field), int): + return int(trace[field]) + return None + + +def terminal_reason_counts(report: dict[str, Any]) -> dict[str, int]: + reasons = [] + for request in report.get("requests", []): + if not isinstance(request, dict): + continue + trace = request.get("server_trace") + if isinstance(trace, dict) and isinstance(trace.get("terminal_reason"), str): + reasons.append(trace["terminal_reason"]) + elif request.get("ok") is True: + reasons.append("completed_without_terminal_trace") + elif request.get("timed_out") is True: + reasons.append("client_timeout") + else: + reasons.append("client_failure") + return value_counts(reasons) + + +def error_counts(report: dict[str, Any]) -> dict[str, int]: + values = [] + for request in report.get("requests", []): + if not isinstance(request, dict) or request.get("ok") is True: + continue + error = request.get("error") + values.append(str(error)[:160] if error else "unknown") + return value_counts(values) + + +def metric_percentile(report: dict[str, Any], metric: str, percentile: str) -> float | None: + value = ( + ((report.get("metrics") or {}).get(metric) or {}).get(f"{percentile}_ms") + ) + return float(value) if isinstance(value, (int, float)) else None + + +def percentile(sorted_values: list[float], pct: float) -> float | None: + if not sorted_values: + return None + rank = (pct / 100.0) * (len(sorted_values) - 1) + lower = int(rank) + upper = min(lower + 1, len(sorted_values) - 1) + weight = rank - lower + return sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight + + +def metric_percentiles(values: list[float | int | None]) -> dict[str, float | None]: + clean = sorted( + float(value) + for value in values + if isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + ) + return { + percentile_name: percentile(clean, pct) + for percentile_name, pct in (("p50", 50), ("p95", 95), ("p99", 99)) + } + + +def successful_requests(reports: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [ + request + for report in reports + for request in report.get("requests", []) + if isinstance(request, dict) and request.get("ok") is True + ] + + +def measured_requests(reports: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [ + request + for report in reports + for request in report.get("requests", []) + if isinstance(request, dict) + ] + + +def full_server_trace(trace: dict[str, Any] | None) -> bool: + if trace is None: + return False + trace_fields = { + "queued_at_unix_s", + "scheduled_at_unix_s", + "first_token_emit_unix_s", + "prompt_tokens", + "completion_tokens", + "active_set_size", + "active_set_size_max", + "pending_queue_size_max", + "decode_batch_size_max", + "decode_step_count", + "batch_decode_steps", + } + return any(field in trace for field in trace_fields) + + +def aggregate_trace_coverage(requests: list[dict[str, Any]]) -> dict[str, Any]: + total = len(requests) + traces = [ + request.get("server_trace") + for request in requests + if isinstance(request.get("server_trace"), dict) + ] + full_traces = [trace for trace in traces if full_server_trace(trace)] + active_set_values = [ + int(trace["active_set_size"]) + for trace in full_traces + if isinstance(trace.get("active_set_size"), int) + ] + decode_batch_values = [ + int(trace["decode_batch_size_max"]) + for trace in full_traces + if isinstance(trace.get("decode_batch_size_max"), int) + ] + token_timing_requests = [ + request + for request in requests + if request.get("token_timing_valid") is True + ] + return { + "coverage_ratio": len(full_traces) / total if total else 0.0, + "server_record_coverage_ratio": len(traces) / total if total else 0.0, + "active_set_coverage_ratio": len(active_set_values) / total if total else 0.0, + "decode_batch_coverage_ratio": len(decode_batch_values) / total + if total + else 0.0, + "token_timing_coverage_ratio": len(token_timing_requests) / total + if total + else 0.0, + "missing_traces": [ + request.get("request_id") + for request in requests + if not full_server_trace(request.get("server_trace")) + ], + "missing_server_records": [ + request.get("request_id") + for request in requests + if not isinstance(request.get("server_trace"), dict) + ], + } + + +def aggregate_trace_max(requests: list[dict[str, Any]], field: str) -> int | None: + values = [] + for request in requests: + trace = request.get("server_trace") + if isinstance(trace, dict) and isinstance(trace.get(field), int): + values.append(int(trace[field])) + return max(values) if values else None + + +def aggregate_trace_max_any( + requests: list[dict[str, Any]], fields: tuple[str, ...] +) -> int | None: + values = [ + value + for field in fields + if (value := aggregate_trace_max(requests, field)) is not None + ] + return max(values) if values else None + + +def bucket_record( + *, + bucket_index: int, + concurrency: int, + report_path: Path, + report: dict[str, Any] | None, + returncode: int, + resource_before: ResourceSample, + resource_after: ResourceSample, +) -> dict[str, Any]: + digest = sha256_file(report_path) + record: dict[str, Any] = { + "bucket_index": bucket_index, + "concurrency": concurrency, + "artifact": str(report_path), + "sha256": digest, + "benchmark_returncode": returncode, + "resource_before": asdict(resource_before), + "resource_after": asdict(resource_after), + } + if report is None: + record["report_loaded"] = False + return record + summary = report.get("summary") or {} + trace = report.get("server_trace") or {} + record.update( + { + "report_loaded": True, + "completed": int(summary.get("completed") or 0), + "failed": int(summary.get("failed") or 0), + "timeouts": int(summary.get("timeouts") or 0), + "wall_s": summary.get("wall_s"), + "qps": summary.get("qps"), + "input_tokens_per_s": summary.get("input_tokens_per_s"), + "output_tokens_per_s": summary.get("output_tokens_per_s"), + "output_hash_distribution": summary.get("output_hash_distribution") or {}, + "combined_output_hash": summary.get("combined_output_hash"), + "ttft_ms": { + percentile: metric_percentile(report, "ttft", percentile) + for percentile in ("p50", "p95", "p99") + }, + "tpot_ms": { + percentile: metric_percentile(report, "tpot", percentile) + for percentile in ("p50", "p95", "p99") + }, + "itl_ms": { + percentile: metric_percentile(report, "itl", percentile) + for percentile in ("p50", "p95", "p99") + }, + "trace_coverage": { + "coverage_ratio": trace.get("coverage_ratio"), + "server_record_coverage_ratio": trace.get( + "server_record_coverage_ratio" + ), + "active_set_coverage_ratio": trace.get("active_set_coverage_ratio"), + "decode_batch_coverage_ratio": trace.get( + "decode_batch_coverage_ratio" + ), + "token_timing_coverage_ratio": trace.get( + "token_timing_coverage_ratio" + ), + "missing_traces": trace.get("missing_traces") or [], + "missing_server_records": trace.get("missing_server_records") or [], + }, + "active_set_size_max": trace_max(report, "active_set_size_max"), + "pending_queue_size_max": trace_max(report, "pending_queue_size_max"), + "decode_batch_size_max": trace_max(report, "decode_batch_size_max"), + "terminal_reasons": terminal_reason_counts(report), + "errors": error_counts(report), + } + ) + return record + + +def aggregate_bucket_record( + *, + bucket_index: int, + concurrency: int, + leaf_runs: list[dict[str, Any]], + resource_before: ResourceSample, + resource_after: ResourceSample, +) -> dict[str, Any]: + reports = [ + run["report"] + for run in leaf_runs + if isinstance(run.get("report"), dict) + ] + requests = measured_requests(reports) + successes = successful_requests(reports) + wall_s = max(0.0, resource_after.wall_s - resource_before.wall_s) + total_counts = {"completed": 0, "failed": 0, "timeouts": 0} + input_tokens_total = 0 + output_tokens_total = 0 + output_tokens_complete = True + output_hashes = [] + terminal_reasons: dict[str, int] = {} + errors: dict[str, int] = {} + for report in reports: + summary = report.get("summary") or {} + total_counts["completed"] += int(summary.get("completed") or 0) + total_counts["failed"] += int(summary.get("failed") or 0) + total_counts["timeouts"] += int(summary.get("timeouts") or 0) + input_tokens_total += int(summary.get("input_tokens_total") or 0) + if isinstance(summary.get("output_tokens_total"), int): + output_tokens_total += int(summary["output_tokens_total"]) + else: + output_tokens_complete = False + terminal_reasons = add_counts(terminal_reasons, terminal_reason_counts(report)) + errors = add_counts(errors, error_counts(report)) + output_hashes = [ + str(request.get("output_hash")) + for request in successes + if request.get("output_hash") + ] + returncodes = [int(run.get("returncode") or 0) for run in leaf_runs] + leaf_artifacts = [ + { + "artifact": str(run["report_path"]), + "sha256": sha256_file(run["report_path"]), + "benchmark_returncode": run.get("returncode"), + "report_loaded": isinstance(run.get("report"), dict), + } + for run in leaf_runs + ] + return { + "bucket_index": bucket_index, + "concurrency": concurrency, + "artifact": str(leaf_runs[0]["report_path"]) if leaf_runs else None, + "sha256": sha256_file(leaf_runs[0]["report_path"]) if leaf_runs else None, + "leaf_artifacts": leaf_artifacts, + "leaf_count": len(leaf_runs), + "report_loaded": bool(reports), + "benchmark_returncode": 0 if all(code == 0 for code in returncodes) else 1, + "resource_before": asdict(resource_before), + "resource_after": asdict(resource_after), + "wall_s": wall_s, + "completed": total_counts["completed"], + "failed": total_counts["failed"], + "timeouts": total_counts["timeouts"], + "qps": total_counts["completed"] / wall_s if wall_s > 0 else 0.0, + "input_tokens_per_s": input_tokens_total / wall_s if wall_s > 0 else 0.0, + "output_tokens_per_s": ( + output_tokens_total / wall_s + if output_tokens_complete and wall_s > 0 + else None + ), + "output_hash_distribution": value_counts(output_hashes), + "combined_output_hash": combined_output_hash(output_hashes), + "ttft_ms": metric_percentiles( + [request.get("ttft_ms") for request in successes] + ), + "tpot_ms": metric_percentiles( + [request.get("tpot_ms") for request in successes] + ), + "itl_ms": metric_percentiles( + [ + value + for request in successes + for value in (request.get("itl_ms") or []) + ] + ), + "trace_coverage": aggregate_trace_coverage(requests), + "active_set_size_max": aggregate_trace_max_any( + requests, ("active_set_size_max", "active_set_size") + ), + "pending_queue_size_max": aggregate_trace_max( + requests, "pending_queue_size_max" + ), + "decode_batch_size_max": aggregate_trace_max(requests, "decode_batch_size_max"), + "terminal_reasons": terminal_reasons, + "errors": errors, + } + + +def add_counts(left: dict[str, int], right: dict[str, int]) -> dict[str, int]: + result = dict(left) + for key, value in right.items(): + result[key] = result.get(key, 0) + int(value) + return dict(sorted(result.items())) + + +def numeric_bucket_values( + buckets: list[dict[str, Any]], path: tuple[str, ...] +) -> list[float | int | None]: + values = [] + for bucket in buckets: + current: Any = bucket + for key in path: + if not isinstance(current, dict): + current = None + break + current = current.get(key) + values.append(current if isinstance(current, (int, float)) else None) + return values + + +def delta_pct(first: float | int | None, last: float | int | None) -> float | None: + if first is None or last is None or float(first) == 0.0: + return None + return (float(last) - float(first)) / float(first) * 100.0 + + +def quartile_span(buckets: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + if not buckets: + return [], [] + width = max(1, math.ceil(len(buckets) / 4)) + return buckets[:width], buckets[-width:] + + +def drift_summary(buckets: list[dict[str, Any]]) -> dict[str, Any]: + first, last = quartile_span(buckets) + fields = { + "qps": ("qps",), + "output_tokens_per_s": ("output_tokens_per_s",), + "ttft_p95_ms": ("ttft_ms", "p95"), + "tpot_p95_ms": ("tpot_ms", "p95"), + "itl_p95_ms": ("itl_ms", "p95"), + "rss_kib": ("resource_after", "rss_kib"), + "gpu_memory_max_mib": ("resource_after", "gpu_memory_used_mib"), + } + result = {} + for label, path in fields.items(): + if label == "gpu_memory_max_mib": + first_values = [ + max(sample) + for bucket in first + if isinstance( + sample := ( + (bucket.get("resource_after") or {}).get( + "gpu_memory_used_mib" + ) + ), + list, + ) + and sample + ] + last_values = [ + max(sample) + for bucket in last + if isinstance( + sample := ( + (bucket.get("resource_after") or {}).get( + "gpu_memory_used_mib" + ) + ), + list, + ) + and sample + ] + else: + first_values = numeric_bucket_values(first, path) + last_values = numeric_bucket_values(last, path) + first_summary = numeric_summary(first_values) + last_summary = numeric_summary(last_values) + result[label] = { + "first_quartile": first_summary, + "last_quartile": last_summary, + "median_delta_pct": delta_pct( + first_summary.get("median"), last_summary.get("median") + ), + } + return result + + +def resource_summary(samples: list[ResourceSample]) -> dict[str, Any]: + rss = [sample.rss_kib for sample in samples] + gpu_max = [ + max(sample.gpu_memory_used_mib) + for sample in samples + if sample.gpu_memory_used_mib + ] + return { + "schema_version": RESOURCE_SAMPLE_VERSION, + "samples": [asdict(sample) for sample in samples], + "rss_kib": numeric_summary(rss), + "gpu_memory_max_mib": numeric_summary(gpu_max), + "gpu_memory_scope": "device_total", + } + + +def build_clean_followup( + args: argparse.Namespace, started_wall_s: float +) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + if args.skip_clean_followup: + return None, None + out = args.out_dir / "clean_followup.json" + before = sample_resources(args, started_wall_s) + command = build_leaf_command( + args, + out=out, + concurrency=1, + num_requests=1, + prompt_words=min(args.prompt_words), + max_tokens=min(args.max_tokens), + contract_name=f"{args.contract_name}-clean-followup", + contract_description="Post-soak clean follow-up request.", + ) + completed = subprocess.run(command, check=False) + after = sample_resources(args, started_wall_s) + report = read_report(out) + record = bucket_record( + bucket_index=-1, + concurrency=1, + report_path=out, + report=report, + returncode=completed.returncode, + resource_before=before, + resource_after=after, + ) + return report, record + + +def build_summary( + args: argparse.Namespace, + *, + buckets: list[dict[str, Any]], + resources: list[ResourceSample], + clean_followup: dict[str, Any] | None, + run_errors: list[dict[str, Any]], + started_wall_s: float, + ended_wall_s: float, +) -> dict[str, Any]: + by_concurrency = { + str(concurrency): [ + bucket for bucket in buckets if bucket.get("concurrency") == concurrency + ] + for concurrency in args.concurrency + } + total_counts = {"completed": 0, "failed": 0, "timeouts": 0} + terminal_reasons: dict[str, int] = {} + errors: dict[str, int] = {} + output_hashes = [] + for bucket in buckets: + for field in total_counts: + total_counts[field] += int(bucket.get(field) or 0) + terminal_reasons = add_counts( + terminal_reasons, bucket.get("terminal_reasons") or {} + ) + errors = add_counts(errors, bucket.get("errors") or {}) + output_hashes.extend( + output_hash + for output_hash, count in ( + bucket.get("output_hash_distribution") or {} + ).items() + for _ in range(int(count)) + ) + + clean_summary = None + if clean_followup is not None: + clean_summary = { + "passed": ( + clean_followup.get("benchmark_returncode") == 0 + and clean_followup.get("completed") == 1 + and clean_followup.get("failed") == 0 + and clean_followup.get("timeouts") == 0 + ), + "bucket": clean_followup, + } + + required_trace = args.required_trace_coverage + trace_passed = True + if required_trace is not None: + for bucket in buckets: + coverage = bucket.get("trace_coverage") or {} + trace_passed = trace_passed and all( + isinstance(coverage.get(field), (int, float)) + and float(coverage[field]) >= required_trace + for field in ( + "coverage_ratio", + "active_set_coverage_ratio", + "decode_batch_coverage_ratio", + "token_timing_coverage_ratio", + ) + ) + + bucket_coverage_passed = bool(buckets) and all( + by_concurrency[str(concurrency)] for concurrency in args.concurrency + ) + leaf_commands_passed = bucket_coverage_passed and all( + bucket.get("benchmark_returncode") == 0 + and bucket.get("report_loaded") is True + for bucket in buckets + ) + passed = ( + not run_errors + and leaf_commands_passed + and total_counts["failed"] == 0 + and total_counts["timeouts"] == 0 + and trace_passed + and (clean_summary is None or clean_summary["passed"]) + ) + + return { + "schema_version": 1, + "kind": "deepseek_v2_lite_http_soak_backend", + "report_intent": "http_soak", + "model": args.model, + "backend": args.backend, + "metadata": { + "commit": args.commit or current_commit(), + "backend": args.backend, + "contract_name": args.contract_name, + "model_path": args.model_path, + "server_command": args.server_command, + "source_revision": args.commit, + "model_revision": args.model_revision, + "model_fingerprint": model_fingerprint(args.model_path), + "server_binary_sha256": sha256_file(args.server_binary) + if args.server_binary + else None, + "backend_runtime_version": args.backend_runtime_version, + "benchmark_command": artifact_command(sys.argv), + "hardware_toolchain": detect_hardware_toolchain(), + }, + "contract": { + "name": args.contract_name, + "backend": args.backend, + "description": args.contract_description, + "required_trace_coverage_ratio": args.required_trace_coverage, + "claim_boundary": args.claim_boundary, + }, + "workload": { + "duration_s": args.duration_s, + "bucket_s": args.bucket_s, + "num_requests_per_leaf": args.num_requests, + "prompt_words": args.prompt_words, + "max_tokens": args.max_tokens, + "concurrency": args.concurrency, + "temperature": 0.0, + "top_k": -1, + "top_p": 1.0, + "ignore_eos": not args.no_ignore_eos, + "timeout_s": args.timeout, + "timeout_kind": "absolute_request_deadline", + }, + "summary": { + "started_wall_s": started_wall_s, + "ended_wall_s": ended_wall_s, + "elapsed_s": ended_wall_s - started_wall_s, + "bucket_count": len(buckets), + "completed": total_counts["completed"], + "failed": total_counts["failed"], + "timeouts": total_counts["timeouts"], + "terminal_reasons": terminal_reasons, + "errors": errors, + "output_hash_distribution": value_counts(output_hashes), + "combined_output_hash": combined_output_hash(output_hashes), + }, + "soak_gate": { + "passed": passed, + "bucket_coverage_passed": bucket_coverage_passed, + "leaf_commands_passed": leaf_commands_passed, + "zero_failures": total_counts["failed"] == 0, + "zero_timeouts": total_counts["timeouts"] == 0, + "trace_coverage_passed": trace_passed, + "clean_followup_passed": None + if clean_summary is None + else clean_summary["passed"], + "run_errors": run_errors, + "rule": ( + "This gate checks request completion, optional trace coverage, " + "leaf command success, and clean follow-up. Numeric drift is " + "reported but not a hard budget until production limits are ratified." + ), + }, + "resource_summary": resource_summary(resources), + "drift_by_concurrency": { + concurrency: drift_summary(concurrency_buckets) + for concurrency, concurrency_buckets in by_concurrency.items() + }, + "buckets": buckets, + "clean_followup": clean_summary, + "claim_boundary": args.claim_boundary, + } + + +def run_backend(args: argparse.Namespace) -> int: + validate_run_args(args) + args.out_dir.mkdir(parents=True, exist_ok=True) + summary_path = args.out_dir / "soak_summary.json" + summary_path.unlink(missing_ok=True) + for stale in args.out_dir.glob("c*/bucket*.json"): + stale.unlink() + (args.out_dir / "clean_followup.json").unlink(missing_ok=True) + + started_wall_s = time.time() + started_perf = time.perf_counter() + resources: list[ResourceSample] = [sample_resources(args, started_wall_s)] + buckets: list[dict[str, Any]] = [] + run_errors: list[dict[str, Any]] = [] + + for concurrency in args.concurrency: + bucket_index = 0 + concurrency_deadline = time.perf_counter() + args.duration_s + while bucket_index == 0 or time.perf_counter() < concurrency_deadline: + if args.max_buckets is not None and bucket_index >= args.max_buckets: + break + bucket_dir = args.out_dir / f"c{concurrency}" + bucket_started = sample_resources(args, started_wall_s) + resources.append(bucket_started) + bucket_deadline = min( + concurrency_deadline, time.perf_counter() + args.bucket_s + ) + leaf_runs: list[dict[str, Any]] = [] + leaf_index = 0 + while leaf_index == 0 or time.perf_counter() < bucket_deadline: + prompt_index = bucket_index + leaf_index + prompt_words = args.prompt_words[prompt_index % len(args.prompt_words)] + max_tokens = args.max_tokens[prompt_index % len(args.max_tokens)] + out = bucket_dir / f"bucket{bucket_index:04d}_leaf{leaf_index:03d}.json" + before = sample_resources(args, started_wall_s) + resources.append(before) + command = build_leaf_command( + args, + out=out, + concurrency=concurrency, + num_requests=args.num_requests, + prompt_words=prompt_words, + max_tokens=max_tokens, + contract_name=args.contract_name, + contract_description=args.contract_description, + ) + completed = subprocess.run(command, check=False) + after = sample_resources(args, started_wall_s) + resources.append(after) + report = read_report(out) + leaf_runs.append( + { + "report_path": out, + "report": report, + "returncode": completed.returncode, + } + ) + if completed.returncode != 0: + run_errors.append( + { + "phase": "leaf", + "concurrency": concurrency, + "bucket_index": bucket_index, + "leaf_index": leaf_index, + "returncode": completed.returncode, + "artifact": str(out), + } + ) + if args.stop_on_failure: + break + leaf_index += 1 + + bucket_ended = sample_resources(args, started_wall_s) + resources.append(bucket_ended) + record = aggregate_bucket_record( + bucket_index=bucket_index, + concurrency=concurrency, + leaf_runs=leaf_runs, + resource_before=bucket_started, + resource_after=bucket_ended, + ) + buckets.append(record) + if args.stop_on_failure and record.get("benchmark_returncode") != 0: + break + bucket_index += 1 + if time.perf_counter() - started_perf >= args.duration_s * len( + args.concurrency + ): + break + + _followup_report, clean_followup = build_clean_followup(args, started_wall_s) + if clean_followup is not None: + resources.append( + ResourceSample( + wall_s=float( + (clean_followup.get("resource_after") or {}).get("wall_s", 0.0) + ), + rss_kib=(clean_followup.get("resource_after") or {}).get("rss_kib"), + gpu_memory_used_mib=(clean_followup.get("resource_after") or {}).get( + "gpu_memory_used_mib", [] + ), + gpu_memory_scope="device_total", + ) + ) + if clean_followup.get("benchmark_returncode") != 0: + run_errors.append( + { + "phase": "clean_followup", + "returncode": clean_followup.get("benchmark_returncode"), + "artifact": clean_followup.get("artifact"), + } + ) + ended_wall_s = time.time() + summary = build_summary( + args, + buckets=buckets, + resources=resources, + clean_followup=clean_followup, + run_errors=run_errors, + started_wall_s=started_wall_s, + ended_wall_s=ended_wall_s, + ) + print(write_json(summary_path, summary)) + return 0 if summary["soak_gate"]["passed"] else 1 + + +def load_summary(path: Path) -> dict[str, Any]: + report = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(report, dict): + raise SystemExit(f"{path} is not a JSON object") + return report + + +def combined_contract_document(summary: dict[str, Any]) -> str: + contract = summary.get("contract") or {} + workload = summary.get("workload") or {} + comparable_contract = { + "contract": { + "name": contract.get("name"), + "description": contract.get("description"), + "required_trace_coverage_ratio": contract.get( + "required_trace_coverage_ratio" + ), + "claim_boundary": contract.get("claim_boundary"), + }, + "workload": workload, + } + return json.dumps(comparable_contract, sort_keys=True, separators=(",", ":")) + + +def build_combined_report( + model: str, summaries: list[tuple[Path, dict[str, Any]]] +) -> dict[str, Any]: + found: dict[str, tuple[Path, dict[str, Any]]] = {} + invalid = [] + duplicates = [] + for path, summary in summaries: + backend = summary.get("backend") + if ( + summary.get("kind") != "deepseek_v2_lite_http_soak_backend" + or summary.get("report_intent") != "http_soak" + or summary.get("model") != model + or backend not in BACKENDS + ): + invalid.append({"artifact": str(path), "errors": ["identity"]}) + continue + if backend in found: + duplicates.append({"backend": backend, "artifact": str(path)}) + continue + found[str(backend)] = (path, summary) + missing = sorted(set(BACKENDS) - set(found)) + ordered = [found[backend] for backend in sorted(found)] + commits = [ + str((summary.get("metadata") or {}).get("commit")) + for _, summary in ordered + if (summary.get("metadata") or {}).get("commit") + ] + commit_counts = value_counts(commits) + commit_consistent = len(commit_counts) == 1 and len(commits) == len(found) + provenance_documents = [] + for _, summary in ordered: + metadata = summary.get("metadata") or {} + provenance_documents.append( + json.dumps( + { + "source_revision": metadata.get("source_revision"), + "model_revision": metadata.get("model_revision"), + "model_fingerprint": metadata.get("model_fingerprint"), + "model_path": metadata.get("model_path"), + "server_binary_sha256": metadata.get("server_binary_sha256"), + }, + sort_keys=True, + separators=(",", ":"), + ) + ) + provenance_consistent = ( + len(set(provenance_documents)) == 1 + and len(provenance_documents) == len(found) + and bool(provenance_documents) + ) + contract_documents = [ + combined_contract_document(summary) for _, summary in ordered + ] + contract_consistent = ( + len(set(contract_documents)) == 1 + and len(contract_documents) == len(found) + and bool(contract_documents) + ) + child_gates = { + backend: (summary.get("soak_gate") or {}).get("passed") is True + for backend, (_path, summary) in found.items() + } + runtime_boundaries = {} + missing_runtime_boundaries = [] + for backend, (_path, summary) in found.items(): + metadata = summary.get("metadata") or {} + runtime_boundary = normalized_runtime_boundary( + metadata.get("backend_runtime_version") + ) + if runtime_boundary_is_present(runtime_boundary) and not ( + backend == "nccl" and runtime_boundary.lower() == "nccl" + ): + runtime_boundaries[backend] = runtime_boundary + else: + missing_runtime_boundaries.append(backend) + missing_runtime_boundaries = sorted(missing_runtime_boundaries) + runtime_boundaries_present = ( + len(runtime_boundaries) == len(found) + and not missing_runtime_boundaries + and bool(runtime_boundaries) + ) + passed = ( + not missing + and not invalid + and not duplicates + and all(child_gates.values()) + and commit_consistent + and provenance_consistent + and contract_consistent + and runtime_boundaries_present + ) + return { + "schema_version": 1, + "kind": "deepseek_v2_lite_http_soak_report", + "report_intent": "http_soak", + "model": model, + "backends": list(BACKENDS), + "metadata": { + "commit": commits[0] if commit_consistent else None, + "benchmark_command": artifact_command(sys.argv), + "child_commits": commit_counts, + "provenance": json.loads(provenance_documents[0]) + if provenance_consistent + else None, + "contract": json.loads(contract_documents[0]) + if contract_consistent + else None, + "runtime_boundaries": runtime_boundaries, + }, + "coverage_gate": { + "passed": passed, + "required_backends": len(BACKENDS), + "retained_backends": len(found), + "missing": missing, + "invalid_children": invalid, + "duplicates": duplicates, + "child_gates": child_gates, + "commit_consistent": commit_consistent, + "provenance_consistent": provenance_consistent, + "contract_consistent": contract_consistent, + "runtime_boundaries_present": runtime_boundaries_present, + "missing_runtime_boundaries": missing_runtime_boundaries, + }, + "reports": [ + { + "artifact": str(path), + "sha256": sha256_file(path), + "backend": summary["backend"], + "metadata": summary["metadata"], + "workload": summary["workload"], + "summary": summary["summary"], + "soak_gate": summary["soak_gate"], + "resource_summary": summary["resource_summary"], + "drift_by_concurrency": summary["drift_by_concurrency"], + "clean_followup": summary["clean_followup"], + } + for path, summary in ordered + ], + "claim_boundary": DEFAULT_CLAIM_BOUNDARY, + } + + +def combine(args: argparse.Namespace) -> int: + if any(path.resolve() == args.out.resolve() for path in args.summary): + raise SystemExit("--out must not overwrite an input --summary") + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.unlink(missing_ok=True) + summaries = [(path, load_summary(path)) for path in args.summary] + report = build_combined_report(args.model, summaries) + print(write_json(args.out, report)) + return 0 if report["coverage_gate"]["passed"] else 1 + + +def add_run_parser(subparsers: argparse._SubParsersAction[Any]) -> None: + parser = subparsers.add_parser("run", help="Run one backend soak contract.") + parser.add_argument("--backend", choices=BACKENDS, required=True) + parser.add_argument("--base-url", default="http://127.0.0.1:8000") + parser.add_argument("--model", default="DeepSeek-V2-Lite") + parser.add_argument("--model-path", required=True) + parser.add_argument("--server-command", required=True) + parser.add_argument("--server-log", type=Path, required=True) + parser.add_argument("--server-pid", type=int) + parser.add_argument("--commit", required=True) + parser.add_argument("--model-revision", required=True) + parser.add_argument("--server-binary", type=Path) + parser.add_argument("--backend-runtime-version", required=True) + parser.add_argument("--duration-s", type=float, default=1800.0) + parser.add_argument("--bucket-s", type=float, default=300.0) + parser.add_argument("--num-requests", type=int, default=32) + parser.add_argument("--concurrency", type=parse_int_list, default=[4, 8]) + parser.add_argument("--prompt-words", type=parse_int_list, default=[64]) + parser.add_argument("--max-tokens", type=parse_int_list, default=[64]) + parser.add_argument("--timeout", type=float, default=240.0) + parser.add_argument("--max-buckets", type=int) + parser.add_argument("--required-trace-coverage", type=float) + parser.add_argument( + "--contract-name", + default=DEFAULT_CONTRACT_NAME, + help="Stable contract name written into leaf and summary artifacts.", + ) + parser.add_argument( + "--contract-description", + default=DEFAULT_CONTRACT_DESCRIPTION, + ) + parser.add_argument("--claim-boundary", default=DEFAULT_CLAIM_BOUNDARY) + parser.add_argument( + "--ignore-eos", action=argparse.BooleanOptionalAction, default=True + ) + parser.add_argument("--skip-clean-followup", action="store_true") + parser.add_argument("--stop-on-failure", action="store_true") + parser.add_argument("--out-dir", type=Path, required=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + add_run_parser(subparsers) + combine_parser = subparsers.add_parser( + "combine", help="Combine host-staged and NCCL soak summaries." + ) + combine_parser.add_argument("--model", default="DeepSeek-V2-Lite") + combine_parser.add_argument("--summary", action="append", type=Path, required=True) + combine_parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + if args.command == "run": + args.no_ignore_eos = not args.ignore_eos + raise SystemExit(run_backend(args)) + raise SystemExit(combine(args)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_bench_dsv2lite_http_soak.py b/tests/test_bench_dsv2lite_http_soak.py new file mode 100644 index 00000000..9e6a6c12 --- /dev/null +++ b/tests/test_bench_dsv2lite_http_soak.py @@ -0,0 +1,484 @@ +#!/usr/bin/env python3 +"""Regression tests for scripts/bench_dsv2lite_http_soak.py.""" + +from __future__ import annotations + +import importlib.util +import copy +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + + +SCRIPT_PATH = ( + Path(__file__).resolve().parents[1] / "scripts" / "bench_dsv2lite_http_soak.py" +) +SCRIPTS_DIR = SCRIPT_PATH.parent +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) +SPEC = importlib.util.spec_from_file_location("bench_dsv2lite_http_soak", SCRIPT_PATH) +assert SPEC and SPEC.loader +bench_dsv2lite_http_soak = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = bench_dsv2lite_http_soak +SPEC.loader.exec_module(bench_dsv2lite_http_soak) + + +def args(**overrides: object) -> SimpleNamespace: + values = { + "backend": "host-staged", + "base_url": "http://127.0.0.1:18000", + "model": "DeepSeek-V2-Lite", + "model_path": "models/DeepSeek-V2-Lite", + "server_command": "openinfer --model-path models/DeepSeek-V2-Lite", + "server_log": Path("server.log"), + "server_pid": None, + "commit": "a" * 12, + "model_revision": "model-rev", + "server_binary": Path("target/release/openinfer"), + "backend_runtime_version": "host-staged", + "duration_s": 60.0, + "bucket_s": 30.0, + "num_requests": 8, + "concurrency": [4, 8], + "prompt_words": [64], + "max_tokens": [64], + "timeout": 240.0, + "max_buckets": None, + "required_trace_coverage": 1.0, + "contract_name": "dsv2-lite-http-soak", + "contract_description": "soak", + "claim_boundary": bench_dsv2lite_http_soak.DEFAULT_CLAIM_BOUNDARY, + "no_ignore_eos": False, + "skip_clean_followup": False, + "stop_on_failure": False, + "out_dir": Path("artifacts/soak"), + } + values.update(overrides) + return SimpleNamespace(**values) + + +def resource(wall_s: float, rss_kib: int, gpu_mib: list[int]): + return bench_dsv2lite_http_soak.ResourceSample( + wall_s=wall_s, + rss_kib=rss_kib, + gpu_memory_used_mib=gpu_mib, + gpu_memory_scope="device_total", + ) + + +def leaf_report( + *, + completed: int = 8, + failed: int = 0, + timeouts: int = 0, + qps: float = 2.0, + output_tokens_per_s: float = 128.0, +) -> dict[str, object]: + requests = [ + { + "request_id": f"ok-{index}", + "ok": True, + "timed_out": False, + "error": None, + "ttft_ms": 10.0 + index, + "tpot_ms": 1.0, + "itl_ms": [1.0, 1.0], + "output_hash": "hash", + "token_timing_valid": True, + "server_trace": { + "terminal_reason": "completed_length", + "active_set_size": 4, + "active_set_size_max": 4, + "pending_queue_size_max": 1, + "decode_batch_size_max": 3, + "completion_tokens": 2, + }, + } + for index in range(completed) + ] + requests.extend( + { + "request_id": f"failed-{index}", + "ok": False, + "timed_out": index < timeouts, + "error": "boom", + "ttft_ms": None, + "tpot_ms": None, + "itl_ms": [], + "output_hash": "", + "token_timing_valid": False, + "server_trace": { + "terminal_reason": "error", + "active_set_size": 4, + "active_set_size_max": 4, + "pending_queue_size_max": 1, + "decode_batch_size_max": 3, + }, + } + for index in range(failed) + ) + return { + "summary": { + "completed": completed, + "failed": failed, + "timeouts": timeouts, + "wall_s": 4.0, + "qps": qps, + "input_tokens_total": completed * 80, + "output_tokens_total": completed * 2, + "input_tokens_per_s": 512.0, + "output_tokens_per_s": output_tokens_per_s, + "output_hash_distribution": {"hash": completed}, + "combined_output_hash": "combined", + }, + "metrics": { + "ttft": {"p50_ms": 10.0, "p95_ms": 20.0, "p99_ms": 30.0}, + "tpot": {"p50_ms": 1.0, "p95_ms": 2.0, "p99_ms": 3.0}, + "itl": {"p50_ms": 1.0, "p95_ms": 2.0, "p99_ms": 3.0}, + }, + "server_trace": { + "coverage_ratio": 1.0, + "server_record_coverage_ratio": 1.0, + "active_set_coverage_ratio": 1.0, + "decode_batch_coverage_ratio": 1.0, + "token_timing_coverage_ratio": 1.0, + "missing_traces": [], + "missing_server_records": [], + }, + "requests": requests, + } + + +def bucket_record( + *, + concurrency: int = 4, + completed: int = 8, + failed: int = 0, + timeouts: int = 0, + returncode: int = 0, +) -> dict[str, object]: + return bench_dsv2lite_http_soak.bucket_record( + bucket_index=0, + concurrency=concurrency, + report_path=Path(f"bucket-c{concurrency}.json"), + report=leaf_report(completed=completed, failed=failed, timeouts=timeouts), + returncode=returncode, + resource_before=resource(0.0, 100, [1000, 1000]), + resource_after=resource(10.0, 110, [1100, 1100]), + ) + + +def clean_followup_record( + *, completed: int = 1, failed: int = 0, timeouts: int = 0, returncode: int = 0 +) -> dict[str, object]: + return bench_dsv2lite_http_soak.bucket_record( + bucket_index=-1, + concurrency=1, + report_path=Path("clean_followup.json"), + report=leaf_report(completed=completed, failed=failed, timeouts=timeouts), + returncode=returncode, + resource_before=resource(10.0, 110, [1100, 1100]), + resource_after=resource(11.0, 111, [1101, 1101]), + ) + + +def passing_summary( + *, backend: str = "host-staged", runtime: str = "host-staged" +) -> dict[str, object]: + return bench_dsv2lite_http_soak.build_summary( + args( + backend=backend, + backend_runtime_version=runtime, + concurrency=[4], + required_trace_coverage=1.0, + ), + buckets=[bucket_record(concurrency=4)], + resources=[resource(0.0, 100, [1000, 1000])], + clean_followup=clean_followup_record(), + run_errors=[], + started_wall_s=0.0, + ended_wall_s=11.0, + ) + + +class BenchDsv2LiteHttpSoakTests(unittest.TestCase): + def test_leaf_command_locks_soak_workload_and_provenance(self) -> None: + command = bench_dsv2lite_http_soak.build_leaf_command( + args(), + out=Path("bucket.json"), + concurrency=8, + num_requests=32, + prompt_words=64, + max_tokens=64, + contract_name="dsv2-lite-http-soak", + contract_description="soak", + ) + + self.assertEqual(command[command.index("--model") + 1], "DeepSeek-V2-Lite") + self.assertEqual(command[command.index("--num-requests") + 1], "32") + self.assertEqual(command[command.index("--concurrency") + 1], "8") + self.assertEqual(command[command.index("--prompt-words") + 1], "64") + self.assertEqual(command[command.index("--max-tokens") + 1], "64") + self.assertEqual(command[command.index("--temperature") + 1], "0.0") + self.assertEqual(command[command.index("--top-k") + 1], "-1") + self.assertEqual(command[command.index("--top-p") + 1], "1.0") + self.assertIn("--server-command", command) + self.assertIn("--model-revision", command) + self.assertIn("--backend-runtime-version", command) + self.assertIn("--required-trace-coverage", command) + + def test_bucket_record_preserves_trace_resources_and_terminal_reasons(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "bucket.json" + path.write_text('{"ok": true}\n', encoding="utf-8") + + record = bench_dsv2lite_http_soak.bucket_record( + bucket_index=0, + concurrency=4, + report_path=path, + report=leaf_report(), + returncode=0, + resource_before=resource(0.0, 100, [1000, 1001]), + resource_after=resource(4.0, 120, [1100, 1110]), + ) + + self.assertTrue(record["report_loaded"]) + self.assertEqual(record["completed"], 8) + self.assertEqual(record["active_set_size_max"], 4) + self.assertEqual(record["pending_queue_size_max"], 1) + self.assertEqual(record["decode_batch_size_max"], 3) + self.assertEqual(record["terminal_reasons"], {"completed_length": 8}) + self.assertEqual(record["resource_after"]["gpu_memory_used_mib"], [1100, 1110]) + self.assertRegex(record["sha256"], r"^[0-9a-f]{64}$") + + def test_bucket_window_aggregates_multiple_leaf_artifacts(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + first_path = Path(tmp) / "bucket0000_leaf000.json" + second_path = Path(tmp) / "bucket0000_leaf001.json" + first_path.write_text('{"leaf": 0}\n', encoding="utf-8") + second_path.write_text('{"leaf": 1}\n', encoding="utf-8") + + record = bench_dsv2lite_http_soak.aggregate_bucket_record( + bucket_index=0, + concurrency=4, + leaf_runs=[ + { + "report_path": first_path, + "report": leaf_report(), + "returncode": 0, + }, + { + "report_path": second_path, + "report": leaf_report(), + "returncode": 0, + }, + ], + resource_before=resource(0.0, 100, [1000, 1001]), + resource_after=resource(10.0, 120, [1100, 1110]), + ) + + self.assertTrue(record["report_loaded"]) + self.assertEqual(record["leaf_count"], 2) + self.assertEqual(len(record["leaf_artifacts"]), 2) + self.assertEqual(record["completed"], 16) + self.assertEqual(record["failed"], 0) + self.assertEqual(record["timeouts"], 0) + self.assertEqual(record["qps"], 1.6) + self.assertEqual(record["output_tokens_per_s"], 3.2) + self.assertEqual(record["ttft_ms"]["p50"], 13.5) + self.assertEqual(record["trace_coverage"]["coverage_ratio"], 1.0) + self.assertEqual(record["trace_coverage"]["active_set_coverage_ratio"], 1.0) + self.assertEqual(record["trace_coverage"]["decode_batch_coverage_ratio"], 1.0) + self.assertEqual(record["trace_coverage"]["token_timing_coverage_ratio"], 1.0) + self.assertEqual(record["active_set_size_max"], 4) + self.assertEqual(record["pending_queue_size_max"], 1) + self.assertEqual(record["decode_batch_size_max"], 3) + self.assertEqual(record["terminal_reasons"], {"completed_length": 16}) + + def test_backend_summary_reports_drift_and_clean_followup_gate(self) -> None: + first = bench_dsv2lite_http_soak.bucket_record( + bucket_index=0, + concurrency=4, + report_path=Path("bucket0.json"), + report=leaf_report(qps=4.0, output_tokens_per_s=256.0), + returncode=0, + resource_before=resource(0.0, 100, [1000, 1000]), + resource_after=resource(10.0, 110, [1100, 1100]), + ) + last = bench_dsv2lite_http_soak.bucket_record( + bucket_index=1, + concurrency=4, + report_path=Path("bucket1.json"), + report=leaf_report(qps=2.0, output_tokens_per_s=128.0), + returncode=0, + resource_before=resource(10.0, 110, [1100, 1100]), + resource_after=resource(20.0, 120, [1200, 1200]), + ) + followup = bench_dsv2lite_http_soak.bucket_record( + bucket_index=-1, + concurrency=1, + report_path=Path("clean_followup.json"), + report=leaf_report(completed=1, qps=1.0, output_tokens_per_s=64.0), + returncode=0, + resource_before=resource(20.0, 120, [1200, 1200]), + resource_after=resource(21.0, 121, [1201, 1201]), + ) + + summary = bench_dsv2lite_http_soak.build_summary( + args(concurrency=[4], required_trace_coverage=1.0), + buckets=[first, last], + resources=[resource(0.0, 100, [1000, 1000]), resource(20.0, 120, [1200, 1200])], + clean_followup=followup, + run_errors=[], + started_wall_s=100.0, + ended_wall_s=121.0, + ) + + self.assertTrue(summary["soak_gate"]["passed"]) + self.assertTrue(summary["soak_gate"]["bucket_coverage_passed"]) + self.assertTrue(summary["soak_gate"]["leaf_commands_passed"]) + self.assertEqual(summary["summary"]["completed"], 16) + self.assertEqual(summary["summary"]["terminal_reasons"], {"completed_length": 16}) + self.assertEqual( + summary["drift_by_concurrency"]["4"]["qps"]["median_delta_pct"], + -50.0, + ) + self.assertTrue(summary["clean_followup"]["passed"]) + self.assertIn("not direct decode attribution", summary["claim_boundary"]) + + def test_backend_summary_fails_missing_trace_coverage(self) -> None: + bucket = bucket_record() + bucket["trace_coverage"]["token_timing_coverage_ratio"] = 0.5 + + summary = bench_dsv2lite_http_soak.build_summary( + args(concurrency=[4], required_trace_coverage=1.0), + buckets=[bucket], + resources=[], + clean_followup=clean_followup_record(), + run_errors=[], + started_wall_s=0.0, + ended_wall_s=11.0, + ) + + self.assertFalse(summary["soak_gate"]["passed"]) + self.assertFalse(summary["soak_gate"]["trace_coverage_passed"]) + + def test_backend_summary_fails_leaf_command_error(self) -> None: + summary = bench_dsv2lite_http_soak.build_summary( + args(concurrency=[4], required_trace_coverage=1.0), + buckets=[bucket_record(returncode=1)], + resources=[], + clean_followup=clean_followup_record(), + run_errors=[], + started_wall_s=0.0, + ended_wall_s=11.0, + ) + + self.assertFalse(summary["soak_gate"]["passed"]) + self.assertFalse(summary["soak_gate"]["leaf_commands_passed"]) + + def test_backend_summary_fails_empty_or_missing_bucket(self) -> None: + summary = bench_dsv2lite_http_soak.build_summary( + args(concurrency=[4, 8], required_trace_coverage=None), + buckets=[bucket_record(concurrency=4)], + resources=[], + clean_followup=None, + run_errors=[], + started_wall_s=0.0, + ended_wall_s=1.0, + ) + + self.assertFalse(summary["soak_gate"]["passed"]) + self.assertFalse(summary["soak_gate"]["bucket_coverage_passed"]) + + def test_backend_summary_fails_clean_followup_error(self) -> None: + summary = bench_dsv2lite_http_soak.build_summary( + args(concurrency=[4], required_trace_coverage=1.0), + buckets=[bucket_record(concurrency=4)], + resources=[], + clean_followup=clean_followup_record(completed=0, failed=1, returncode=1), + run_errors=[], + started_wall_s=0.0, + ended_wall_s=11.0, + ) + + self.assertFalse(summary["soak_gate"]["passed"]) + self.assertFalse(summary["soak_gate"]["clean_followup_passed"]) + + def test_combined_report_requires_host_and_nccl(self) -> None: + host = passing_summary() + nccl = passing_summary(backend="nccl", runtime="NCCL 2.26.2") + + missing = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host)] + ) + full = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host), (Path("nccl.json"), nccl)] + ) + + self.assertFalse(missing["coverage_gate"]["passed"]) + self.assertEqual(missing["coverage_gate"]["missing"], ["nccl"]) + self.assertTrue(full["coverage_gate"]["passed"]) + self.assertEqual(full["coverage_gate"]["required_backends"], 2) + self.assertEqual( + full["metadata"]["runtime_boundaries"]["nccl"], + "NCCL 2.26.2", + ) + + def test_combined_report_rejects_failed_child_gate(self) -> None: + host = passing_summary() + nccl = passing_summary(backend="nccl", runtime="NCCL 2.26.2") + nccl["soak_gate"]["passed"] = False + + report = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host), (Path("nccl.json"), nccl)] + ) + + self.assertFalse(report["coverage_gate"]["passed"]) + self.assertFalse(report["coverage_gate"]["child_gates"]["nccl"]) + + def test_combined_report_rejects_provenance_drift(self) -> None: + host = passing_summary() + nccl = passing_summary(backend="nccl", runtime="NCCL 2.26.2") + nccl["metadata"]["model_revision"] = "different-model-rev" + + report = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host), (Path("nccl.json"), nccl)] + ) + + self.assertFalse(report["coverage_gate"]["passed"]) + self.assertFalse(report["coverage_gate"]["provenance_consistent"]) + + def test_combined_report_rejects_workload_drift(self) -> None: + host = passing_summary() + nccl = passing_summary(backend="nccl", runtime="NCCL 2.26.2") + nccl = copy.deepcopy(nccl) + nccl["workload"]["duration_s"] = 120.0 + + report = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host), (Path("nccl.json"), nccl)] + ) + + self.assertFalse(report["coverage_gate"]["passed"]) + self.assertFalse(report["coverage_gate"]["contract_consistent"]) + self.assertIsNone(report["metadata"]["contract"]) + + def test_combined_report_rejects_missing_runtime_boundary(self) -> None: + host = passing_summary() + nccl = passing_summary(backend="nccl", runtime="NCCL 2.26.2") + nccl = copy.deepcopy(nccl) + nccl["metadata"]["backend_runtime_version"] = "nccl" + + report = bench_dsv2lite_http_soak.build_combined_report( + "DeepSeek-V2-Lite", [(Path("host.json"), host), (Path("nccl.json"), nccl)] + ) + + self.assertFalse(report["coverage_gate"]["passed"]) + self.assertFalse(report["coverage_gate"]["runtime_boundaries_present"]) + self.assertEqual(report["coverage_gate"]["missing_runtime_boundaries"], ["nccl"]) + + +if __name__ == "__main__": + unittest.main()