diff --git a/docs/steady-state-detection.md b/docs/steady-state-detection.md new file mode 100644 index 000000000..5194ed89e --- /dev/null +++ b/docs/steady-state-detection.md @@ -0,0 +1,534 @@ +# Steady-State Metrics Reporting + +# 1 Objective {#1-objective} + +Add a post-processing step that runs after a benchmark completes and reports the +sustained steady-state metrics, rather than the whole-run average, +which is deflated by the ramp-up and drain transients. The step is a pure +function over the durable event log; it also ships as an ad-hoc command-line tool +that ingests an `events.jsonl` file. + +Goals: + +- Emit a `steady_state` block alongside the existing whole-run (`total`) metrics + in the run report, reported as the **official result** with the whole-run + (`total`) metrics retained as supplementary context. +- Detect _when there is no steady state_ (a progressively degrading run) and say + so, instead of reporting an unstable/unsteady result. +- Provide a single implementation reachable two ways: automatically as the + post-processing step at the end of a run, and manually as the same step invoked + ad hoc over any recorded `events.jsonl`. +- Add no cost to the measured run: all work is off the hot path. + +Non-goals: + +- Changing the whole-run (`total`) numbers, the live metrics snapshot cadence, or + the wire schema of the live aggregator. +- Changing how load is issued during a run. A staggered-issuance option is + discussed as complementary (§5.7) but is out of scope for this step. +- Accuracy scoring, submission checking, or any change to the audit path. + +# 2 Background {#2-background} + +A run's reported metrics today are aggregated over the entire measurement window. +Two regions of that window are not steady state: + +- **Ramp-up.** At the start of a run the client raises offered load to its target. + Under a concurrency load pattern the target in-flight population is filled as a + burst; against a finite-rate server the leading requests queue and + time-to-first-token (TTFT) inflates, and the inflation grows with concurrency. +- **Ramp-down (drain).** After issuance stops, in-flight decays below the target + while the last requests finish. Throughput deflates because the wall-clock + denominator keeps advancing while offered load is below steady state. + +Averaging over these transients understates throughput and overstates tail +latency. The magnitude is workload-dependent and can be large for the tail: in +experiments over recorded runs (single-turn concurrency, offline/max-throughput, +Poisson, and multi-turn agentic), the reported p99 TTFT +was dominated by the ramp spike and fell substantially once the ramp was +excluded, while per-token latency (TPOT) was essentially unchanged. + +There is already a precedent in the codebase for exactly this shape of feature: +`src/inference_endpoint/metrics/early_stopping.py` computes MLPerf early-stopping +percentile estimates as a cold-path calculation, and +`scripts/early_stopping_estimate_from_events.py` re-runs the same math ad hoc +from a recorded `events.jsonl` (see `docs/early_stopping.md`). This design follows +the same two-entry-point pattern. There is also a precedent for a post-run +orchestration step in `src/inference_endpoint/commands/audit.py`, which +`src/inference_endpoint/commands/benchmark/cli.py` dispatches after the main +benchmark completes. + +# 3 Assumptions and Risks {#3-assumptions-and-risks} + +- **The durable event log is complete and authoritative.** The step reads the + per-sample event stream, not the live snapshot (which can lag under load). Risk: + a run killed by `SIGKILL`/OOM before the log is flushed yields a truncated log; + the step must degrade to a best-effort result with a status flag rather than + fail (§5.6). +- **Per-token latency is approximately sample-invariant in a healthy run.** The + guarded tail-cut (§5.4) rests on this. It held across the recorded corpora, but + it is a property of a healthy run, not a guarantee, so the cut is _guarded_: it + is applied only when the condition is measured to hold, and otherwise the tail + is retained. +- **Bucketing is by issue order.** The unit of analysis is a fixed-size group of + issued requests (§5.1). Risk: for load modes with no repeated dataset pass, the + group size is a free parameter; experiments show the qualitative verdict is + robust to that choice, but it is called out as tunable (§6). +- **Token counts are available or derivable.** TPOT needs an output-token count + per request. When the run already records one it is used directly; otherwise the + step tokenizes on the cold path (§5.6). Risk: cost on very large logs, bounded + by sampling. + +# 4 Alternatives considered {#4-alternatives-considered} + +- **Fix it live, in the aggregator.** Detect and crop the ramp inside the hot-path + aggregator so `total` is already steady. Rejected: it adds latency-critical work + to the hot path, the live snapshot can lag under load, and it couples a + still-evolving heuristic to the measured numbers. A cold-path step keeps the hot + path untouched and the heuristic revisable. +- **Report a fixed time/percentage crop (e.g. drop the first N seconds).** Simple + but wrong across modes and concurrencies: the ramp length depends on load, and a + fixed crop under- or over-cuts. The adaptive, data-driven window (§5.2, §5.5) + self-sizes. +- **Only exclude the ramp; keep the whole tail.** Leaves the drain in the + throughput denominator, re-deflating it. Defining the window on _issue time_ + (§5.1) excludes the drain from throughput for free, without an end-crop. +- **Trust a single convergence detector.** A lone coefficient-of-variation (CoV) + stopping rule converges even on a slowly drifting series, picking a window far + from the asymptote. Rejected in favor of an ensemble plus a mandatory trend gate + (§5.5). + +# 5 Design {#5-design} + +## 5.1 Overview and definitions + +The step is a pure function from a recorded event stream to a `steady_state` +result. Definitions used throughout: + +- **Healthy server.** A server that can support the maximum load issued by the + client. The window measures the sustained behavior of a healthy server; a + genuinely unhealthy server produces bad-but-real numbers, which the drift + detector (§5.5) distinguishes from transient pollution. +- **Super-pass.** The atomic unit of the analysis: a contiguous block of requests + in _issue order_, sized so each block is a representative full-dataset workload + mix. It is named distinctly from a _dataset pass_ on purpose — it is **not** + always one pass. A low-concurrency run may not issue even a single full pass, and + the block size is a tunable hyperparameter (for example two dataset passes per + super-pass, to reduce per-super-pass variance). In the common case it is exactly + one dataset pass (`dataset_size` requests), so a run issues about + `ceil(N / dataset_size)` super-passes, where `N` is the total number of samples + issued. +- **Long-running sample.** A sample whose output length (OSL) or sample latency is + much larger than the dataset average. +- **Level shift (staircase).** A step change from one flat metric level to a higher + flat level partway through a run — distinct from a gradual drift or the drain tail. + Typically caused by long-running samples triggering KV-cache eviction toward the + end of a long run, or sudden failures in a subset of workers partway or at the end of a run. + It produces two (or more) legitimate plateaus; the window selection (§5.5) reports the first + and flags the shift to be reported as an anomaly. +- **Hairball.** A build-up of long-running samples that grows as more dataset + passes are issued and completed: because datasets are issued without + replacement, fresh copies of a long-running sample are issued before earlier + copies finish, so long-running samples remain in flight long after the rest have + completed. +- **Hairball weight.** At issuance-stop under max-concurrency `C`, the percentage + of in-flight samples that are _not_ among the last `C` issued (the lingering + hairball). Ideally 0 — at stop the `C` in-flight samples would be exactly the + last `C` issued. +- **Relative active concurrency.** In-flight count as a percentage of the + max-concurrency budget (100% at saturation). +- **Issue-time window.** A contiguous range of super-passes `[start, end)`. Its + measured set is every request _issued_ in that range. Throughput uses the + issue-time span `last_issue - first_issue`; latency and per-token metrics use + the full lifetime of that same set. One membership set feeds both families, so + they can never disagree on which requests they measured. A sample enters the + window only if it has at least one logged event (for example a first token) + inside it — only metrics _logged within_ the window are counted, so a sample that + was issued but received no response contributes no logged metric and is naturally + excluded. The drain lives after the last issue, so it never enters the throughput + denominator — no end-crop is needed. +- **Steadiness metrics.** Metrics whose variation reflects _system state_ rather + than workload composition: TTFT (admission / prefill queue) and TPOT (per-token + decode). Both are output-length-independent. End-to-end latency is deliberately + excluded — it is TTFT plus decode time, and decode scales with output length, so + its variation tracks the output-length mix, not steadiness. + +## 5.2 Where the step runs + +The step is invoked automatically after a run completes, and is also runnable +standalone. Both paths call the same pure-function core. + +```text + +--------------------------------------+ + | benchmark run (load gen + workers) | + +--------------------------------------+ + | + | emits events -> durable event log + v + +--------------------------------------+ + | live metrics aggregator (hot path) | + | writes final_snapshot.json [total] | + +--------------------------------------+ + | + | run ends (COMPLETE); cold path begins + v + +--------------------------------------+ + | steady-state post-process | + | reads the event log, off hot path | + +--------------------------------------+ + | + | steady_state block + v + +--------------------------------------+ + | Report { total, steady_state } | + +--------------------------------------+ +``` + +- **Automatic path.** The run-completion path in + `src/inference_endpoint/commands/benchmark/execute.py` (finalize) — or the + dispatch in `src/inference_endpoint/commands/benchmark/cli.py`, mirroring how it + already dispatches `src/inference_endpoint/commands/audit.py` — invokes the + steady-state builder over the durable event log after the live aggregator has + written `final_snapshot.json`. The builder returns a `steady_state` result that + `src/inference_endpoint/metrics/report.py` attaches next to `total`. This is + gated by a new settings field in `src/inference_endpoint/config/schema.py`, + following the existing `early_stopping.enabled` flag. +- **Ad-hoc path.** A new script re-runs the identical core over any recorded + `events.jsonl`, mirroring `scripts/early_stopping_estimate_from_events.py`. This + is the tool used to analyze historical runs and to iterate on parameters + without re-running a benchmark. + +The builder never reads the live snapshot; the durable event log is the source of +truth, consistent with the existing early-stopping recomputation path. + +## 5.3 The analysis pipeline + +The core is a sequence of pure stages over the per-super-pass series. + +```text + event log + | + v + [ ingest -> per-super-pass series ] issue-order bucketing + | + v + [ adaptive warmup crop ] remove the ramp (TPOT-driven band) + | + v + [ guarded drain-tail cut ] only if per-token-invariant + | + v + [ plateau segmentation: CoV + trend ] admissible windows -> plateaus + | + v + [ select first plateau + shift flag ] MSER precision; Pettitt level-shift + | + v + { steady_state metrics + status + anomaly } +``` + +### Adaptive warmup crop + +The ramp is removed by a data-driven crop rather than a fixed count. The steady level of +a driver metric is estimated from the median of the per-super-pass series' back half, and +leading super-passes whose driver value is more than a fractional `band` away from that +level — in _either_ direction — are dropped, capped at half the run so it can never crop +everything. The driver is **TPOT p50**: aggregated per super-pass it is a smooth, monotone +signal (it ramps _up_ as the batch fills and per-token decode contends, then plateaus at +saturation), which is exactly where the system reaches steady state. The band is symmetric +because TPOT ramps up to steady (unlike TTFT, which decays down from an admission spike); +TTFT is deliberately not the driver — its per-super-pass value is far too volatile under +closed-loop admission (empirically CoV ~3 on a high-concurrency reasoning run vs ~0.15 for +the same model under Poisson). The crop self-sizes to the workload: near-zero on +fast-settling runs, but tens of super-passes on a long-output reasoning run whose decode +ramp is genuinely long. A fixed crop count remains available as an override. + +The dataset is issued in full, without replacement, repeating across passes, so a +tail of long-output requests accumulates toward the end of every run; only the +magnitude differs by mode. Under concurrency the tail is exactly the in-flight +population at issuance-stop — a representative issue-time snapshot, small for +uniform-output models. Under offline/max-throughput **every** sample is issued in +one huge burst at `t=0` and it is left to the server to work through the flood, so +from the client's perspective there is no issue-phase/drain boundary at all — under +the normal (issue-time) definition the entire run, or very nearly all of it, is +drain, which loses meaning. The drain here is instead _observed_ from the TPS trend +over time (throughput falls once the server can no longer keep the system +saturated). A more robust definition — the drain begins once the server no longer +has enough in-flight samples to saturate the pipeline — depends on server-side +occupancy that the client cannot see, so it is handwaved for now. Under Poisson, +arrival pacing throttles the pile-up, so the tail is smallest. The windowing +response is therefore mode-specific: concurrency crops the ramp and applies the +guarded tail-cut; offline finds its steady region from the TPS/completion-rate +trend rather than an issue-time window (a client-side tail-cut is not meaningful); +Poisson reuses the concurrency tooling with a single-pass super-pass. + +## 5.4 Guarded drain-tail cut + +Excluding the tail is safe for per-token latency, latency tails, and throughput — +_conditional on the tail sharing the steady per-token distribution_. This is the +condition that makes it safe to ignore dataset-pass boundaries and drop +high-output samples: the reported per-token latency is unbiased by which samples +are included **iff** inter-token latency is approximately invariant across +samples. With `ITL(S_i) = (t_last(S_i) - t_first(S_i)) / (OSL(S_i) - 1)` the mean +inter-token latency of sample `i` (`OSL - 1` because `OSL` output tokens have +`OSL - 1` inter-token gaps), and population mean `mu`, median `m`, and standard +deviation `sigma` over the samples: + +```text + cut the tail <=> sigma / mu <= epsilon AND |mu - m| / mu <= delta +``` + +for small tolerances `epsilon` and `delta`. The first term (low coefficient of variation) is the +necessary-and-sufficient core; the second (low skew) is a robustness guard against +a heavy-tailed distribution and is redundant under low CoV. When the condition +fails — for example a mid-run server anomaly that slows the tail — the tail is +retained and the affected metric is flagged rather than cut. For long-reasoning +workloads the tail is a strong long-output selection, so output-length coverage is +reported alongside the steady metrics so a reader can see the steady set +under-samples the output-length tail. + +## 5.5 Convergence and steady-window selection + +The reported window and the steady/drift verdict come from two per-metric signals — +a coefficient-of-variation (CoV) stopping rule and a mandatory trend gate — combined +under a window-selection rule adapted from the steady-state simulation literature. + +**Metric set.** The convergence and trend tests run on TTFT and TPOT. The **p50 and +p95** percentiles are _gating_: a window is steady only when both plateau and are +within CoV for both metrics. The **p99** (and p99.9 where a super-pass holds enough +samples to estimate it) are carried as _diagnostic warnings_, not hard gates — a tail +percentile that fails to converge is surfaced as a warning rather than voiding the +window, because tail percentiles are estimated from far fewer samples per super-pass +and are the noisiest signal available. End-to-end latency is reported as context +only, never as a convergence signal (§5.1). + +**CoV stopping rule.** The coefficient of variation `CoV = sigma / mu` of a +metric's per-super-pass percentile, over a window of super-passes, is a +scale-free measure of how much the metric is still moving relative to its own +level. A region is a candidate steady state when `CoV < bound` for every gating +metric and percentile. The bound loosens toward the tail (a p95 is estimated from +fewer samples per super-pass, so its sampling-noise floor is higher than a p50's). + +**Trend gate (mandatory), drift up vs down.** A low CoV over a window certifies +local flatness, not that the metric has stopped moving: a slowly drifting series can +sit locally flat while climbing overall. A trend test over the window is therefore +applied on top. Because per-super-pass series are autocorrelated, the primary gate is +the rank-based **Mann–Kendall test with the Hamed–Rao autocorrelation correction** +(so serial correlation does not fake a trend); an OLS slope-vs-scatter check and a +Newey–West (autocorrelation-consistent) slope test corroborate it. The verdict +classifies each metric into one of three states — **Drifting Down**, **Plateau**, +**Drifting Up**: + +- **Drifting Up** — the metric worsens across the window (for example a p99 TTFT tail + that never plateaus). Pathological: there is _no_ steady state to report for that + metric, and the step says so rather than emitting a number. +- **Plateau** — no significant trend: the metric is genuinely steady. Only a Plateau + metric is eligible to contribute a steady value. +- **Drifting Down** — the metric is still settling downward, i.e. the warmup crop was + slightly short. Transparent (not a false alarm); the follow-up is a larger crop. + +**Selection principle (MSER): maximize precision, never the metric.** Once a run may +contain several windows that are both in Plateau and within CoV, the question is +which to report. The governing rule is taken from the Marginal Standard Error Rule +(MSER; White 1997) for steady-state truncation: **select the window by the precision +of the estimate, never by the value of the reported metric.** Choosing the window by +the very quantity being reported — the highest-TPS window, say — is selection bias: +it reports the most favorable noise realization and inflates the number. MSER instead +minimizes the standard error of the mean, `SE = sigma / sqrt(n)`. Because `n` grows +with window length, `SE` is minimized by the _longest_ admissible window unless +extending it drags in non-steady super-passes that inflate `sigma` faster than `n` +grows — the classic bias-versus-variance knee. The operational consequence is simply: +prefer more steady data, chosen by a criterion that never looks at the throughput +level. + +**Admissibility.** A candidate window is _admissible_ when, for every gating metric +and percentile, it is (1) in Plateau (trend gate) and (2) within at least one CoV +bound of the ensemble. These two gates are the explicit form of MSER's implicit +"post-transient" restriction, and they make each rejection interpretable (a window is +excluded for a named reason, not a black-box score). + +**Candidate family — general contiguous windows, not fixed-endpoint truncation.** +MSER's textbook form fixes the window's right edge at the end of the run and moves +only the start: it drops warm-up and keeps everything to the end. That is insufficient +here because of a possible failure mode seen in longer runs — a **staircase**: a flat region +that steps up to a higher flat region, possible if KV-cache eviction is triggered towards +the end of the run, or if workers suddenly crash, causing overall max throughput to degrade +by a constant amount. A staircase contains a legitimate steady plateau that +**ends before the end of the run**, and a fixed-endpoint window cannot +isolate it — it can only capture the final, degraded plateau, or fail the gate on the +jump. The candidate family is therefore **all contiguous super-pass windows** +`[lo, hi)`, searched from longest down to a minimum-length floor (the trend test's +minimum sample count). The floor is essential: minimizing `SE` over unconstrained +contiguous windows is degenerate — a two-point flat window has `SE = 0` — so the floor +together with the Plateau/CoV gates rules out the trivial micro-window solution. + +**Plateau segmentation and the reported window (first plateau).** The trend and CoV +gates _implicitly segment_ the run: a window spanning a staircase jump has high CoV +and reads as a trend, so it is inadmissible, while a window inside a single plateau is +admissible. Growing an admissible window from the first post-warmup super-pass until +admissibility breaks isolates the **first plateau**; resuming past the break isolates +each subsequent plateau. **The first plateau is the reported steady state.** This +deliberately overrides the pure longest/min-`SE` choice, because empirically the later +plateaus of a staircase are generally _degradation_ steps — a skewed long-output workload +building up, or a server going unhealthy — so the first plateau is the representative +healthy steady state and the later steps are anomalies, not the number to report. + +**Level shifts (staircase) are detected and flagged, never hidden.** Reporting the +first plateau must not silently discard the fact that the run degraded. A level-shift +detector runs alongside the segmentation: when two or more disjoint admissible +plateaus have pooled means differing by more than the CoV band, corroborated by a +**Pettitt** change-point test (a nonparametric, rank-based single-change-point test +that pairs naturally with the rank-based trend gate) on the TPOT series, the result +carries an **anomaly flag** — the change-point super-pass and the magnitude and +direction of the shift. The steady result is still reported from the first plateau, +with an explicit, honest note that the run changed regime toward the end; every +plateau is carried in the machine-readable output. + +**Ensemble.** Because no single `(window, bound)` CoV setting fits all metrics and +workloads, the CoV rule is evaluated as an **ensemble** of preset settings; a window +passes CoV if at least one ensemble setting certifies it for every gating metric. +Detector concordance is a corroborating guardrail. The trend gate remains mandatory +and primary; the CoV ensemble is secondary. + +## 5.6 Edge cases and error handling + +The step never hard-fails a run; every input yields a result plus a `status`. + +- **Coverage status.** Before windowing, classify the run by how much steady + signal it supports and window accordingly: + +```text + status condition behavior + -------------------- ------------------------------------ ----------------------------- + windowable >= warmup + 1 full super-passes normal warmup-cropped window + insufficient_passes >= 1 pass, < warmup + 1 super-passes best-effort, low confidence + partial_dataset < 1 full dataset pass best-effort, flagged unreliable +``` + +A batch sweep over many runs then never aborts on a short run: each run yields a +row carrying its status. + +- **No steady state.** If a tracked metric is Drifting Up, report drift for that + metric instead of a point estimate (§5.5). This is a first-class outcome (see the + open question on invalid runs, §6). +- **Truncated / interrupted log.** If the run did not complete cleanly, the step + computes a best-effort result over whatever was logged and marks it as partial, + mirroring how the report already distinguishes interrupted runs. +- **Missing token counts.** If the log carries no per-request output-token count, + TPOT is derived by tokenizing outputs on the cold path. Not every output is + tokenized: for large logs a sample sufficient to estimate the per-super-pass + percentiles is enough, and full re-tokenization of every output is avoided. If + neither a count nor a tokenizer is available, the step falls back to TTFT-only + and records that TPOT was not assessed (a metric with no data is skipped, not + treated as zero, so an absent metric can never fake convergence). +- **TPOT from timestamps.** TPOT derived as `(complete − first_token) / (OSL − 1)` + assumes the request stayed resident in the decode phase for its whole lifetime. + If the server evicts or preempts a request mid-decode (paging it out and back), + the wall-clock span includes queue time that is not per-token decode, inflating + TPOT. The derivation is trustworthy only when the server does not evict in-flight + requests during decode; where it might, a server-reported per-token count is + preferred over the timestamp span. +- **Degenerate modes.** Offline/max-throughput has a degenerate issue time + (everything issued at `t=0`), so there is no client-side issue-time window and no + client-side drain boundary. The step detects the mode and finds the steady region + from the **TPS trend** over time — the plateau before throughput falls off — + rather than an issue-time window; the exact drain-onset (server occupancy dropping + below saturation) is server-side and is handwaved for now (§5.3). Near-saturation + Poisson backlogs like offline and is detected by the completion rate falling below + the offered rate. + +## 5.7 Complementary: staggered ("feathered") issuance + +The ramp exists because the target concurrency `C` is filled as a burst. A +staggered fill flattens the ramp-up spike: issue in steps of `ceil(C/k)`, where `k` +is the number of fill steps used to reach the target concurrency (larger `k` = more, +smaller steps), and, between steps, wait until a sample from the previous step has +completed before issuing the next; begin measurements only once `C` is reached. This helps the p99 TTFT +explosion and the initial server hammering. It _reduces the severity_ of the ramp +(a smaller crop is then enough) but does not shorten it, because the server +admission rate, not the client schedule, bounds how fast the pipe fills. It is +therefore +complementary to the warmup crop (timing vs. issue-order), not a replacement, and +under offline/max-throughput it is largely moot (a saturating burst has no steady +baseline). This is a load-generator change, out of scope for the reporting step, +and would require live validation before adoption; it is recorded here so the two +efforts stay aligned. + +## 5.8 Output and consumption + +The steady-state metrics are reported as the **official result**; the whole-run +(`total`) metrics remain as **supplementary** context (in `report.txt` and the +machine-readable summary produced from the report). Each tracked metric carries +its steady value plus its state (`Plateau` / `Drifting Up` / `Drifting Down`), and +the run carries its coverage `status`. Reported quantities: + +- **The steady window** itself: its super-pass range `[start, end)` and sample count, + so a reader can see where in the run it was drawn from. +- **TPS**, reported two ways because they answer different questions: **per-user TPS** + = `1 / mean(TPOT)` (output tokens/s/user, the interactivity number) and **system + TPS** = total output tokens in the window divided by its wall-clock span + (aggregate throughput). Each carries a confidence interval computed by + **non-overlapping batch means** with super-passes as batches, which accounts for + the per-super-pass autocorrelation rather than assuming independent samples. +- **TTFT and TPOT** percentiles (p50/p90/p95/p99) and histograms, over the pooled raw + samples of the steady window. +- **Anomaly.** When the level-shift detector fires (§5.5), an `anomaly` block records + the change-point super-pass, the shift magnitude and direction, and every detected + plateau — so a degradation toward the end is surfaced next to the (first-plateau) + steady result rather than hidden by it. +- **QPS is dropped** for text/token-based LLM workloads — a request is not a unit + of work when output length varies widely, so QPS is a legacy metric of little + meaning; it is retained only for token-free, uniform-work loads. +- **ISL / OSL** are reported for analytics, not as validation numbers: once the + window is allowed to drop samples (not enforcing full dataset-pass boundaries), + the input/output-length distribution is skewed relative to the constructed + dataset and is no longer a meaningful validation quantity, though it remains + useful for analysis. + +The `total`-vs-`steady_state` divergence is itself surfaced: a large gap indicates +excessive ramp relative to run length, or a run too short to window. Existing +plotting (`src/inference_endpoint/metrics/results_plots.py`, +`scripts/plot_results.py`) can be extended to overlay the window on the +per-super-pass series. + +# 6 Open questions {#6-open-questions} + +- **Default parameters.** The warmup band, trailing-window length, per-percentile + CoV bounds, and the guard tolerances (`epsilon`, `delta`) are set from sweeps + over recorded runs; the defaults should be reviewed on a wider set before they + are locked as the official reporting parameters. +- **Super-pass sizing for modes without dataset repetition.** For multi-turn + agentic and other single-pass workloads there is no natural dataset-pass unit. + Experiments show the steady/drift verdict is robust to the group-size choice, but + a principled default (for example keyed to concurrency) is still to be picked. +- **Guard distance measure.** The exact two-sample statistic and acceptance bound + for the per-token-invariance guard (§5.4) need to be fixed. +- **No-steady-state runs.** When no steady state is found (a tracked metric is + Drifting Up _throughout_, with no admissible plateau at any window), should the run + be reported _invalid_ — analogous to legacy LoadGen's statistical-significance gate + — or reported with the offending metrics flagged as unstable while the rest are + reported steady? The current proposal is the latter (show which metrics were stable + vs unstable). This is distinct from the **staircase** case, which is already decided + (§5.5): a run with a first steady plateau followed by a higher plateau _does_ have a + steady state — the first plateau is reported and the later shift is flagged as an + anomaly, not treated as no-steady-state. +- **Offline drain-onset.** In offline/max-throughput the drain has no client-side + boundary; the steady region is read from the TPS trend, and the robust definition + (server occupancy dropping below saturation) is server-side and currently + handwaved (§5.3, §5.6). Whether a server-side occupancy signal can be plumbed + through, and whether the TPS-trend plateau detector lives in the same core or a + sibling, is open. +- **ISL / OSL reporting basis (task force).** ISL/OSL are reported over the + _included_ (windowed) samples, which skews them relative to the constructed + dataset (§5.8). Whether the official artifact should instead report these over + the full issued set, or carry both, is a policy call deferred to the benchmark + task force. +- **First pass as warmup (task force).** Whether to treat the first full dataset + pass (or first super-pass) as warmup by construction — rather than inferring the + warmup band from the data — and the exact band/window/bound defaults that would + accompany such a rule are deferred to the benchmark task force. +- **Per-benchmark gates and invalidation (task force).** Whether the steady-state + gates (CoV bounds, trend thresholds) are tuned per benchmark, and whether a run + that fails them is declared _invalid_ versus reported-with-flags (see + "No-steady-state runs" above), is a benchmark-task-force decision, not fixed by + this proposal. diff --git a/scripts/steady_state_diagnostics.md b/scripts/steady_state_diagnostics.md new file mode 100644 index 000000000..037767968 --- /dev/null +++ b/scripts/steady_state_diagnostics.md @@ -0,0 +1,135 @@ +# `steady_state_diagnostics.py` + +Post-hoc **steady-state / drift diagnostics** for a benchmark run's `events.jsonl`. +Self-contained (no `inference_endpoint` import) — runs anywhere with a tokenizer via +`uv`. The full methodology lives in +[`docs/steady-state-detection.md`](../docs/steady-state-detection.md); this is the +operator's quick reference. + +## What it does + +1. Buckets performance-tracked samples into **super-passes** by issue order + (`--superpass-size` samples each, default `--dataset-size`). +2. Reconstructs per-sample **TTFT** (`recv_first − issued`) and **TPOT** + (`(complete − recv_first) / tokens(output-after-first-chunk)`); TPOT needs the + `--tokenizer`, so it is required. +3. Finds the **first steady plateau**: grow a window from the start while it stays + _admissible_ — every gated metric (TTFT/TPOT p50 & p95) is trend-steady + (Mann–Kendall + Hamed–Rao) **and** within a CoV bound. A staircase jump breaks the + window, segmenting the run into plateaus. The **first plateau is the reported steady + state** (later plateaus are usually degradation). Selection follows MSER: pick by + estimator precision, never by the throughput value. +4. Summarizes that window (TTFT/TPOT histograms + percentiles, **per-user & system + TPS** with batch-means confidence intervals) and **flags a level shift** toward the + end of the run (multi-plateau + Pettitt change-point) as an `anomaly`, rather than + hiding it. + +## Requirements + +- `uv` (the script declares its deps inline via a PEP 723 header — only `transformers`). +- A tokenizer the run's model uses (HF id or local dir), e.g. `openai/gpt-oss-120b`, + `deepseek-ai/DeepSeek-R1`. + +## Run + +```bash +uv run scripts/steady_state_diagnostics.py /events.jsonl \ + --tokenizer \ + --dataset-size \ + [--superpass-size N] # samples per super-pass (default: --dataset-size) + [--window-sizes 4,6,8] # diagnostic scan sizes, in super-passes (min useful: 4) + [--warmup auto] # "auto" (data-driven crop) or a fixed super-pass count + [--warmup-band 0.05] # auto: crop leading super-passes >5% off the steady level + [--warmup-driver tpot_p50] # auto: metric whose ramp defines the crop + [--cov-bounds 0.03,0.05,0.08] + [--trend-gate mk_hamed_rao] # {mann_kendall,mk_hamed_rao,newey_west,theil_sen,slope_vs_scatter} + [--alpha 0.05] + [--json out.json] # full machine-readable result +``` + +`--dataset-size` is the number of samples in one dataset pass (see the run's +`run_meta.json` / config). + +## Interpreting the output + +### Headline — `STEADY STATE` + +``` +=== STEADY STATE (headline) === + window: super-passes 0..3 (post-warmup), 23519 samples + TPS per-user: 302.3 tok/s/user CI [302.1, 302.5] + TPS system: 40960.9 tok/s CI [39399.3, 40606.8] + TTFT p50 86.26ms p90 156.10ms p95 183.83ms p99 248.51ms mean 97.41ms + TPOT p50 3.29ms p90 3.44ms p95 3.48ms p99 3.56ms mean 3.31ms +``` + +- **window** — the steady plateau, as **post-warmup** super-pass indices `lo..hi`, plus + the pooled sample count it was measured over. +- **TPS per-user** = `1 / mean(TPOT)` — output tokens/s for a single stream + (interactivity). **TPS system** = total output tokens ÷ window wall-clock (aggregate + throughput). Each `CI` is a 95% batch-means interval (super-passes as batches), so it + reflects per-super-pass variability, not a naïve iid interval. +- **TTFT / TPOT** — percentiles and mean over the pooled raw samples of the window. + +If no window qualifies: + +``` + not found: no admissible steady plateau +``` + +means no contiguous run of super-passes was steady enough — the run drifts or is too +short. The per-window diagnostics below show why (which metric failed CoV/trend). + +### `ANOMALY` line + +``` + ANOMALY: level shift at super-pass 6, TPOT +100.0% toward end of run (likely degradation) +``` + +A second, materially different plateau was detected after the first and confirmed by a +Pettitt change-point. The headline steady result is still the **first** plateau; this +line says the run degraded later (e.g. KV-cache eviction, a sick worker). `delta_pct` +is signed (+ = TPOT rose = worse). + +### `WARNING` line + +``` + WARNING: ttft_p95 drifting UP over the rest of the run -- the window is a local + plateau; global steady state is questionable +``` + +The reported window is locally steady, but a gated metric keeps climbing over the +super-passes **after** it (the trend gate over the whole tail, not just the window). +This catches a slow global drift that a short per-window check misses — e.g. TTFT +creeping up several-fold across a long high-concurrency run. Treat the steady number as +a best-effort local plateau, not a clean whole-run steady state. (`drifting_up` in +`--json` lists the affected metrics; distinct from `anomaly`, which is a discrete step.) + +### Diagnostics (below the headline) + +Per `--window-size`: a **CoV steadiness** table (per gated/diagnostic metric, PASS/`fail`/`n/a` +against each CoV bound over the trailing window) and a **whole-run trend** summary per +metric across the algorithms. The full rolling drift scan (every window position) is +only in `--json`. + +### `--json` + +Full structured result: `steady_state` (window, `ttft`/`tpot` summaries + histograms, +`tps`, `anomaly` with every plateau), plus `trajectories`, `cov`, and `drift` (the +rolling scan) for deeper analysis. + +## Caveats + +- **TPOT parity.** Token counts use plain tokenization of the output; the live + aggregator uses the chat-template path for reasoning/tool-call outputs, so absolute + TPOT ms can differ for reasoning models. CoV and the trend tests are scale-invariant, + so the steady/drift **verdicts** are unaffected — only the absolute TPOT magnitude. +- **Window sizes < 4** are useless for drift (the trend test needs ≥ 4 points); they + still contribute to the CoV table. +- Window indices are **post-warmup relative** (add the resolved warmup — shown as + `warmup N [auto|fixed]` in the header — for absolute super-pass numbers). +- **Auto warmup** (default) crops leading super-passes still ramping toward the steady + TPOT level (band around the back-half median). On long-ramp runs (high-concurrency, + long-output) this can be large — e.g. ~24 super-passes on a DeepSeek-R1 c28k run — + which is correct: it moves the window off the ramp shoulder onto the true plateau. Use + a fixed `--warmup N` to override. diff --git a/scripts/steady_state_diagnostics.py b/scripts/steady_state_diagnostics.py new file mode 100644 index 000000000..d0354c0a3 --- /dev/null +++ b/scripts/steady_state_diagnostics.py @@ -0,0 +1,1393 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.12" +# dependencies = ["transformers>=4.40"] +# /// +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Steady-state / drift diagnostics from a benchmark run's ``events.jsonl``. + +Self-contained: no ``inference_endpoint`` import, so it runs anywhere with just a +tokenizer available (``uv run scripts/steady_state_diagnostics.py ...``). The event +wire shapes it parses are defined by the product's ``core/record.py`` (event names, +``EventRecord`` fields) and ``core/types.py`` (``TextModelOutput`` array layout); the +parse here mirrors them and is pinned by tests/unit/scripts/test_steady_state_diagnostics.py. + +What it reconstructs (per performance-tracked sample): + - ttft_ns = recv_first.ts - issued.ts + - tpot_ns = (complete.ts - recv_first.ts) / tokens(text_after_first_chunk) +Token counts use plain tokenization of ``text_after_first_chunk``. The live metrics +aggregator instead tokenizes reasoning/tool-call outputs via the chat-template path +(``apply_chat_template``), so absolute TPOT ms here can differ from a run's report for +reasoning models. CoV and the trend tests are scale-invariant, so the steady/drift +diagnosis is unaffected -- only the absolute TPOT magnitude shifts. + +Samples are bucketed into super-passes by issue order (``--superpass-size`` samples per +super-pass, default = ``--dataset-size``), giving a per-super-pass trajectory for each +metric*percentile. + +The **headline** output is the ``steady_state`` block: the first steady plateau (grow- +from-left segmentation, admissible = trend-steady + within a CoV bound on the gated +metrics), summarized with TTFT/TPOT histograms + percentiles and per-user / system TPS +with batch-means confidence intervals. A staircase level-shift toward the end of the run +(multi-plateau difference corroborated by a Pettitt change-point) is flagged as an +``anomaly`` rather than hidden. See docs/steady-state-detection.md. + +Below the headline, per requested window size, the tool also prints diagnostics: a CoV +pass/fail table and a whole-run trend summary (the full rolling drift scan is in +``--json``). + +Convergence metrics are TTFT/TPOT at p50 + p95; p99 is carried as an optional diagnostic +(shown, not gated). End-to-end latency is intentionally excluded (its variation tracks the +OSL mix, not system steadiness). + +usage: + uv run scripts/steady_state_diagnostics.py \ + --tokenizer --dataset-size \ + [--superpass-size ] [--window-sizes 4,5] [--warmup 1] \ + [--cov-bounds 0.03,0.05,0.08] [--alpha 0.05] [--json ] +""" + +from __future__ import annotations + +import argparse +import json +import math +import sys +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field +from statistics import NormalDist, median, pstdev +from typing import Literal, NamedTuple, TypedDict + +# --------------------------------------------------------------------------- # +# Event wire constants (mirror core/record.py category.value topics) +# --------------------------------------------------------------------------- # +EV_START_TRACKING = "session.start_performance_tracking" +EV_STOP_TRACKING = "session.stop_performance_tracking" +EV_ISSUED = "sample.issued" +EV_RECV_FIRST = "sample.recv_first" +EV_COMPLETE = "sample.complete" + +# Below this many super-passes a trend test is statistically meaningless. +MIN_TREND_N = 4 + +# slope-vs-scatter thresholds (mirror the reference drift detector): a metric drifts +# when the run-length change is a large fraction of its level AND large vs the residual +# scatter around the fitted line. +REL_DRIFT_THRESHOLD = 0.15 +SNR_THRESHOLD = 2.0 + +# z for a two-sided 95% confidence interval (Hamed-Rao autocorrelation significance). +CI_Z_95 = 1.96 +# Floor substituted for a zero median so relative-drift ratios stay finite. +ZERO_MEDIAN_FLOOR = 1e-9 +# Texts buffered before a tokenizer flush during the parse. +TOKENIZE_BATCH_SIZE = 4096 + +# A per-super-pass metric trajectory is classified into one of these states. +Verdict = Literal["up", "down", "steady", "insufficient"] + + +class Anomaly(TypedDict): + detected: bool + change_point_sp: int | None + delta_pct: float # signed % change of the later TPOT level vs the first plateau + pettitt: dict | None + plateaus: list[list[int]] + + +class SteadyWindow(TypedDict): + sp_lo: int # post-warmup super-pass index (inclusive) + sp_hi: int # exclusive + n_super_passes: int + n_samples: int + + +class TpsBlock(TypedDict): + per_user: float # 1e9 / mean(TPOT ns) = output tok/s/user + per_user_ci: list[float] # [lo, hi] + system: float # total output tokens / window wall-clock + system_ci: list[float] + + +class SteadyState(TypedDict): + found: bool + reason: str | None + window: SteadyWindow | None + ttft: dict | None # summarize() output + tpot: dict | None + tps: TpsBlock | None + anomaly: Anomaly + global_trend: dict[ + str, Verdict + ] # gated metric -> trend from the plateau to run end + drifting_up: list[str] # gated metrics Drifting Up over the rest of the run + + +class TrackedMetric(NamedTuple): + key: str # display key, e.g. "ttft_p95" + source_attr: str # SuperPassRollup attribute holding the raw samples + percentile: float + gated: bool # participates in the convergence gate (vs. diagnostic-only) + + +# Metric*percentile trajectories tracked. ``gated`` ones participate in convergence; +# p99 is diagnostic only. +TRACKED_METRICS: tuple[TrackedMetric, ...] = ( + TrackedMetric("ttft_p50", "ttft_ns", 0.50, True), + TrackedMetric("ttft_p95", "ttft_ns", 0.95, True), + TrackedMetric("tpot_p50", "tpot_ns", 0.50, True), + TrackedMetric("tpot_p95", "tpot_ns", 0.95, True), + TrackedMetric("ttft_p99", "ttft_ns", 0.99, False), + TrackedMetric("tpot_p99", "tpot_ns", 0.99, False), +) + +# Metrics that gate admissibility (p50/p95); p99 is diagnostic-only. +GATED_METRICS: tuple[TrackedMetric, ...] = tuple(m for m in TRACKED_METRICS if m.gated) +_METRIC_BY_KEY: dict[str, TrackedMetric] = {m.key: m for m in TRACKED_METRICS} + + +# --------------------------------------------------------------------------- # +# TextModelOutput.text_after_first_chunk, ported to the parsed JSON array +# --------------------------------------------------------------------------- # +def text_after_first_chunk(data: object) -> str: + """Return output text excluding the first streamed chunk (the TPOT numerator). + + ``data`` is the COMPLETE event payload: ``[tag, output, reasoning?, tool_calls?]`` + with trailing defaults omitted (msgspec ``array_like`` + ``omit_defaults``). ``output`` + and ``reasoning`` are each either a string (non-streaming) or a list of chunks + (streaming). Mirrors ``TextModelOutput.text_after_first_chunk`` in core/types.py. + """ + if not isinstance(data, list) or not data: + return "" + output = data[1] if len(data) > 1 else "" + reasoning = data[2] if len(data) > 2 else None + parts: list[str] = [] + if reasoning: + if isinstance(reasoning, list) and len(reasoning) > 1: + parts.extend(reasoning[1:]) + # str reasoning is a single (first) chunk -> skip entirely + if output: + if isinstance(output, str): + # Non-streaming output: keep it only if a first chunk already lived in a + # (streaming) reasoning trace; otherwise the str output IS the first chunk. + if parts or (reasoning and isinstance(reasoning, list)): + parts.append(output) + elif isinstance(output, list): + if parts or reasoning: + parts.extend(output) + elif len(output) > 1: + parts.extend(output[1:]) + # Tool-call reconstruction is intentionally omitted: tool-call samples use a + # chat-template tokenization path this diagnostic does not replicate. + return "".join(parts) + + +# --------------------------------------------------------------------------- # +# Super-pass series +# --------------------------------------------------------------------------- # +@dataclass(slots=True) +class SuperPassRollup: + index: int + n_issued: int = 0 # per-super-pass sample count (coverage / bucketing invariant) + first_issue_ns: int = -1 # earliest issue ts (offered-load span start) + last_issue_ns: int = -1 # latest issue ts (offered-load span end; throughput denom) + last_event_ns: int = -1 # latest event ts incl. completions (drain-inclusive end) + ttft_ns: list[float] = field(default_factory=list) + tpot_ns: list[float] = field(default_factory=list) + out_tokens: int = 0 + + +@dataclass(slots=True) +class _PendingRow: + """In-flight sample state during the parse, keyed by uuid until COMPLETE.""" + + sp_index: int + issue_ns: int + recv_first_ns: int | None = None + + +def build_super_pass_series( + events_path: str, + superpass_size: int, + count_tokens: Callable[[list[str]], list[int]], + flush_size: int = TOKENIZE_BATCH_SIZE, +) -> list[SuperPassRollup]: + """Bucket performance-tracked samples into super-passes by issue order. + + ``count_tokens`` maps a batch of texts to token counts; injected so the parse is + testable without a real tokenizer and the tokenizer is swappable. ``flush_size`` + caps how many output texts are buffered before a tokenizer flush — lower it to bound + peak memory on long reasoning outputs (fewer texts held, smaller tokenizer calls). + """ + if superpass_size <= 0: + raise ValueError("superpass_size must be positive") + series: list[SuperPassRollup] = [] + rows: dict[str, _PendingRow] = {} + tracking = False + issue_counter = 0 + batch_uuids: list[str] = [] + batch_texts: list[str] = [] + pending_tpot: dict[str, tuple[int, float]] = {} + + def _ensure(idx: int) -> SuperPassRollup: + while len(series) <= idx: + series.append(SuperPassRollup(index=len(series))) + return series[idx] + + def flush_tpot() -> None: + if batch_texts: + counts = count_tokens(batch_texts) + for uuid, cnt in zip(batch_uuids, counts, strict=True): + sp_idx, delta = pending_tpot.pop(uuid) + if cnt > 0: + series[sp_idx].tpot_ns.append(delta / cnt) + series[sp_idx].out_tokens += cnt + batch_uuids.clear() + batch_texts.clear() + + with open(events_path) as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue # skip a truncated/partial line (e.g. last line of a killed run) + et = rec.get("event_type") + ts = rec.get("timestamp_ns") # may be absent on a truncated/partial event + if et == EV_START_TRACKING: + tracking = True + elif et == EV_STOP_TRACKING: + tracking = False + elif et == EV_ISSUED: + uuid = rec.get("sample_uuid") + if not tracking or not uuid or ts is None: + continue + existing = rows.get(uuid) + if existing is not None: + existing.issue_ns = ts # retry: refresh issue ts only + sp = series[existing.sp_index] + sp.last_issue_ns = max(sp.last_issue_ns, ts) + sp.last_event_ns = max(sp.last_event_ns, ts) + continue + sp_idx = issue_counter // superpass_size + issue_counter += 1 + rows[uuid] = _PendingRow(sp_index=sp_idx, issue_ns=ts) + sp = _ensure(sp_idx) + sp.n_issued += 1 + if sp.first_issue_ns < 0: + sp.first_issue_ns = ts + sp.last_issue_ns = max(sp.last_issue_ns, ts) + sp.last_event_ns = max(sp.last_event_ns, ts) + elif et == EV_RECV_FIRST: + row = rows.get(rec.get("sample_uuid")) + if row is not None and ts is not None: + series[row.sp_index].last_event_ns = max( + series[row.sp_index].last_event_ns, ts + ) + # First recv_first only: a retried sample re-emits recv_first and + # must not contribute a second TTFT to the super-pass. + if row.recv_first_ns is None: + row.recv_first_ns = ts + series[row.sp_index].ttft_ns.append(float(ts - row.issue_ns)) + elif et == EV_COMPLETE: + uuid = rec.get("sample_uuid") + row = rows.pop(uuid, None) + if row is None or ts is None: + continue + sp = series[row.sp_index] + sp.last_event_ns = max(sp.last_event_ns, ts) + if row.recv_first_ns is None: + continue + text = text_after_first_chunk(rec.get("data")) + if text: + pending_tpot[uuid] = (row.sp_index, float(ts - row.recv_first_ns)) + batch_uuids.append(uuid) + batch_texts.append(text) + if len(batch_texts) >= flush_size: + flush_tpot() + flush_tpot() + return series + + +# --------------------------------------------------------------------------- # +# Numeric helpers +# --------------------------------------------------------------------------- # +def percentile_lower(sorted_values: Sequence[float], p: float) -> float: + n = len(sorted_values) + if n == 0: + raise ValueError("percentile of empty series") + return sorted_values[int(p * (n - 1))] + + +def cov(values: Sequence[float]) -> float: + if len(values) < 2: + return 0.0 + m = sum(values) / len(values) + if m == 0: + return 0.0 + return pstdev(values) / abs(m) + + +def _phi(z: float) -> float: + """Standard-normal CDF.""" + return 0.5 * (1.0 + math.erf(z / math.sqrt(2.0))) + + +def _two_sided_p(stat: float) -> float: + return 2.0 * (1.0 - _phi(abs(stat))) + + +def super_pass_percentile_series( + series: Sequence[SuperPassRollup], source_attr: str, percentile: float +) -> list[float]: + """Per-super-pass percentile trajectory; super-passes with no samples are skipped.""" + out: list[float] = [] + for sp in series: + vals = getattr(sp, source_attr) + if vals: + out.append(percentile_lower(sorted(vals), percentile)) + return out + + +def pooled( + series: Sequence[SuperPassRollup], lo: int, hi: int, source_attr: str +) -> list[float]: + """All raw samples of an attribute pooled across super-passes ``[lo, hi)``.""" + out: list[float] = [] + for sp in series[lo:hi]: + out.extend(getattr(sp, source_attr)) + return out + + +def pooled_out_tokens(series: Sequence[SuperPassRollup], lo: int, hi: int) -> int: + return sum(sp.out_tokens for sp in series[lo:hi]) + + +def window_elapsed_ns(series: Sequence[SuperPassRollup], lo: int, hi: int) -> int: + """Completion span of ``[lo, hi)``: earliest issue to latest event (drain-inclusive).""" + window = series[lo:hi] + firsts = [sp.first_issue_ns for sp in window if sp.first_issue_ns >= 0] + lasts = [sp.last_event_ns for sp in window if sp.last_event_ns >= 0] + if not firsts or not lasts: + return 0 + return max(lasts) - min(firsts) + + +def window_issue_span_ns(series: Sequence[SuperPassRollup], lo: int, hi: int) -> int: + """Offered-load span of ``[lo, hi)``: earliest to latest *issue*. + + This is the throughput denominator (§5.1): the drain lives after the last issue, so + counting to the last completion would inflate the denominator and deflate TPS — + badly so for high-tail workloads (long TTFT + decode). + """ + window = series[lo:hi] + firsts = [sp.first_issue_ns for sp in window if sp.first_issue_ns >= 0] + lasts = [sp.last_issue_ns for sp in window if sp.last_issue_ns >= 0] + if not firsts or not lasts: + return 0 + return max(lasts) - min(firsts) + + +def histogram(values: Sequence[float], nbins: int = 20) -> list[dict]: + """Bin counts over ``[min, max]``; log-spaced edges when strictly positive.""" + lo, hi = min(values), max(values) + if lo == hi: + return [{"lo": lo, "hi": hi, "count": len(values)}] + if lo > 0: + ratio = hi / lo + edges = [lo * ratio ** (i / nbins) for i in range(nbins + 1)] + else: + edges = [lo + (hi - lo) * i / nbins for i in range(nbins + 1)] + counts = [0] * nbins + for v in values: + if v >= hi: + counts[-1] += 1 + continue + for b in range(nbins): + if v < edges[b + 1]: + counts[b] += 1 + break + return [ + {"lo": edges[b], "hi": edges[b + 1], "count": counts[b]} for b in range(nbins) + ] + + +def summarize(values: Sequence[float]) -> dict: + """Count, mean, min/max, p50/p90/p95/p99 (nearest-rank-lower), and a histogram.""" + s = sorted(values) + n = len(s) + return { + "count": n, + "mean": sum(s) / n, + "min": s[0], + "max": s[-1], + "p50": percentile_lower(s, 0.50), + "p90": percentile_lower(s, 0.90), + "p95": percentile_lower(s, 0.95), + "p99": percentile_lower(s, 0.99), + "histogram": histogram(s), + } + + +# --------------------------------------------------------------------------- # +# Estimation: batch-means CI, Pettitt change-point, TPS +# --------------------------------------------------------------------------- # +# Two-sided 95% Student-t critical values by degrees of freedom (df>30 -> ~1.96). +_T_CRIT_95: dict[int, float] = { + 1: 12.706, + 2: 4.303, + 3: 3.182, + 4: 2.776, + 5: 2.571, + 6: 2.447, + 7: 2.365, + 8: 2.306, + 9: 2.262, + 10: 2.228, + 11: 2.201, + 12: 2.179, + 13: 2.160, + 14: 2.145, + 15: 2.131, + 16: 2.120, + 17: 2.110, + 18: 2.101, + 19: 2.093, + 20: 2.086, + 21: 2.080, + 22: 2.074, + 23: 2.069, + 24: 2.064, + 25: 2.060, + 26: 2.056, + 27: 2.052, + 28: 2.048, + 29: 2.045, + 30: 2.042, +} + + +def _t_crit_95(df: int) -> float: + return _T_CRIT_95.get(df, 1.96) + + +def batch_means_ci( + batch_means: Sequence[float], confidence: float = 0.95 +) -> tuple[float, float]: + """Confidence interval for the grand mean from non-overlapping batch means. + + Batches (here: super-passes) are treated as approximately independent, so the + interval accounts for per-super-pass autocorrelation that a raw-sample CI would + ignore. Uses a Student-t critical value (small-sample correct) for 95%. + """ + k = len(batch_means) + if k == 0: + return (0.0, 0.0) + m = sum(batch_means) / k + if k < 2: + return (m, m) + var = sum((b - m) ** 2 for b in batch_means) / (k - 1) + se = math.sqrt(var) / math.sqrt(k) + if confidence == 0.95: + crit = _t_crit_95(k - 1) + else: + crit = NormalDist().inv_cdf(1.0 - (1.0 - confidence) / 2.0) + return (m - crit * se, m + crit * se) + + +def _average_ranks(values: Sequence[float]) -> list[float]: + n = len(values) + order = sorted(range(n), key=lambda i: values[i]) + ranks = [0.0] * n + i = 0 + while i < n: + j = i + while j + 1 < n and values[order[j + 1]] == values[order[i]]: + j += 1 + avg = (i + 1 + j + 1) / 2.0 # average of the tied 1-based ranks + for k in range(i, j + 1): + ranks[order[k]] = avg + i = j + 1 + return ranks + + +def pettitt(values: Sequence[float], alpha: float = 0.05) -> dict: + """Pettitt nonparametric single-change-point test. + + Returns the split index (size of the first segment), the ``K`` statistic, an + approximate p-value, and whether a change point is significant at ``alpha``. + Rank-based, so it pairs with the Mann-Kendall trend gate. + """ + n = len(values) + if n < MIN_TREND_N: + return {"change_point": 0, "k_stat": 0.0, "pvalue": 1.0, "significant": False} + ranks = _average_ranks(values) + cum = 0.0 + k_stat = 0.0 + cp = 0 + for t in range(1, n): # t = size of the first segment + cum += ranks[t - 1] + u = 2.0 * cum - t * (n + 1) + if abs(u) > k_stat: + k_stat = abs(u) + cp = t + pvalue = min(1.0, 2.0 * math.exp(-6.0 * k_stat * k_stat / (n**3 + n**2))) + return { + "change_point": cp, + "k_stat": k_stat, + "pvalue": pvalue, + "significant": pvalue < alpha, + } + + +def per_user_tps(mean_tpot_ns: float) -> float: + """Output tokens/s/user from mean time-per-output-token (ns).""" + return 1e9 / mean_tpot_ns if mean_tpot_ns > 0 else 0.0 + + +def system_tps(out_tokens: int, elapsed_ns: int) -> float: + """Aggregate output tokens/s over a window's wall-clock span.""" + return out_tokens / (elapsed_ns / 1e9) if elapsed_ns > 0 else 0.0 + + +# --------------------------------------------------------------------------- # +# Trend algorithms -- each returns a TrendResult with verdict in +# {"up", "steady", "down", "insufficient"}. +# --------------------------------------------------------------------------- # +@dataclass(frozen=True, slots=True) +class TrendResult: + verdict: Verdict + slope: float = 0.0 + statistic: float = 0.0 # primary test statistic (Mann-Kendall S, Newey-West t) + pvalue: float | None = ( + None # None for effect-size tests (theil_sen, slope_vs_scatter) + ) + variance: float = 0.0 + rel_drift: float = 0.0 # signed total change / median + snr: float = 0.0 # |total change| / residual scatter + + +def _insufficient() -> TrendResult: + return TrendResult("insufficient") + + +def _direction(x: float) -> Verdict: + return "up" if x > 0 else "down" if x < 0 else "steady" + + +def _significant_verdict(effect: float, pvalue: float, alpha: float) -> Verdict: + """up/down when the effect is significant (pvalue < alpha) in that direction.""" + if pvalue < alpha: + if effect > 0: + return "up" + if effect < 0: + return "down" + return "steady" + + +def _median_or_floor(values: Sequence[float]) -> float: + """Median, floored away from zero so relative-drift ratios stay finite.""" + m = median(values) + return m if m else ZERO_MEDIAN_FLOOR + + +def _mk_S(values: Sequence[float]) -> int: + n = len(values) + s = 0 + for i in range(n - 1): + vi = values[i] + for j in range(i + 1, n): + d = values[j] - vi + s += (d > 0) - (d < 0) + return s + + +def _mk_variance(values: Sequence[float]) -> float: + n = len(values) + counts: dict[float, int] = {} + for v in values: + counts[v] = counts.get(v, 0) + 1 + tie_term = sum(t * (t - 1) * (2 * t + 5) for t in counts.values()) + return (n * (n - 1) * (2 * n + 5) - tie_term) / 18.0 + + +def _mk_verdict(s: int, variance: float, alpha: float) -> TrendResult: + if variance <= 0: + return TrendResult( + _direction(s), statistic=float(s), pvalue=0.0, variance=variance + ) + if s > 0: + z = (s - 1) / math.sqrt(variance) + elif s < 0: + z = (s + 1) / math.sqrt(variance) + else: + z = 0.0 + p = _two_sided_p(z) + return TrendResult( + _significant_verdict(s, p, alpha), + statistic=float(s), + pvalue=p, + variance=variance, + ) + + +def mann_kendall(values: Sequence[float], alpha: float = 0.05) -> TrendResult: + if len(values) < MIN_TREND_N: + return _insufficient() + return _mk_verdict(_mk_S(values), _mk_variance(values), alpha) + + +def _autocorr_of_ranks(values: Sequence[float]) -> list[float]: + n = len(values) + order = sorted(range(n), key=lambda i: values[i]) + ranks = [0.0] * n + for rank, idx in enumerate(order, start=1): + ranks[idx] = float(rank) + mean = sum(ranks) / n + dev = [r - mean for r in ranks] + denom = sum(d * d for d in dev) + acf: list[float] = [] + if denom == 0: + return [0.0] * (n - 1) + for k in range(1, n): + num = sum(dev[t] * dev[t - k] for t in range(k, n)) + acf.append(num / denom) + return acf + + +def mann_kendall_hamed_rao(values: Sequence[float], alpha: float = 0.05) -> TrendResult: + """Mann-Kendall with the Hamed-Rao autocorrelation variance correction. + + Inflates (or, for negatively autocorrelated data, deflates) the MK variance by an + effective-sample-size factor computed from the significant autocorrelations of the + data ranks, so serial correlation does not fake significance. + """ + n = len(values) + if n < MIN_TREND_N: + return _insufficient() + s = _mk_S(values) + var0 = _mk_variance(values) + acf = _autocorr_of_ranks(values) + ci = CI_Z_95 / math.sqrt(n) + factor_sum = 0.0 + for k in range(1, n): + r = acf[k - 1] + if abs(r) <= ci: # only statistically significant lags contribute + continue + factor_sum += (n - k) * (n - k - 1) * (n - k - 2) * r + correction = 1.0 + (2.0 / (n * (n - 1) * (n - 2))) * factor_sum + # A non-positive effective-sample correction is degenerate (over-correction under + # strong negative autocorrelation). Fall back to the uncorrected MK variance rather + # than clamping to a sliver, which would collapse the variance and manufacture a + # significant trend from essentially no evidence. + if correction <= 0: + correction = 1.0 + return _mk_verdict(s, var0 * correction, alpha) + + +def theil_sen( + values: Sequence[float], rel_threshold: float = REL_DRIFT_THRESHOLD +) -> TrendResult: + n = len(values) + if n < MIN_TREND_N: + return _insufficient() + slopes = [ + (values[j] - values[i]) / (j - i) for i in range(n - 1) for j in range(i + 1, n) + ] + slope = median(slopes) + rel = slope * (n - 1) / _median_or_floor(values) + verdict: Verdict = "steady" if abs(rel) < rel_threshold else _direction(rel) + return TrendResult(verdict, slope=slope, rel_drift=rel) + + +def _ols(values: Sequence[float]) -> tuple[float, float, list[float]]: + n = len(values) + xbar = (n - 1) / 2.0 + ybar = sum(values) / n + sxx = sum((x - xbar) ** 2 for x in range(n)) + sxy = sum((x - xbar) * (v - ybar) for x, v in enumerate(values)) + slope = sxy / sxx if sxx else 0.0 + intercept = ybar - slope * xbar + resid = [v - (intercept + slope * x) for x, v in enumerate(values)] + return slope, sxx, resid + + +def newey_west( + values: Sequence[float], lag: int | None = None, alpha: float = 0.05 +) -> TrendResult: + """OLS slope significance with a Newey-West (HAC) standard error.""" + n = len(values) + if n < MIN_TREND_N: + return _insufficient() + slope, sxx, resid = _ols(values) + if sxx == 0: + return TrendResult("steady", slope=0.0) + xbar = (n - 1) / 2.0 + u = [(x - xbar) * resid[x] for x in range(n)] + if lag is None: + lag = max(1, int(math.floor(4 * (n / 100.0) ** (2.0 / 9.0)))) + s = sum(ui * ui for ui in u) + for lg in range(1, min(lag, n - 1) + 1): + w = 1.0 - lg / (lag + 1.0) + s += 2.0 * w * sum(u[t] * u[t - lg] for t in range(lg, n)) + var_b = s / (sxx * sxx) + se = math.sqrt(var_b) if var_b > 0 else 0.0 + if se == 0: + return TrendResult(_direction(slope), slope=slope, pvalue=0.0) + t = slope / se + p = _two_sided_p(t) + return TrendResult( + _significant_verdict(slope, p, alpha), slope=slope, statistic=t, pvalue=p + ) + + +def slope_vs_scatter( + values: Sequence[float], + rel_threshold: float = REL_DRIFT_THRESHOLD, + snr_threshold: float = SNR_THRESHOLD, +) -> TrendResult: + n = len(values) + if n < MIN_TREND_N: + return _insufficient() + slope, _sxx, resid = _ols(values) + resid_std = pstdev(resid) if n > 1 else 0.0 + total_change = slope * (n - 1) + rel_drift = total_change / _median_or_floor(values) + snr = abs(total_change) / (resid_std + ZERO_MEDIAN_FLOOR) + drifting = abs(rel_drift) >= rel_threshold and snr >= snr_threshold + verdict: Verdict = _direction(rel_drift) if drifting else "steady" + return TrendResult(verdict, slope=slope, snr=snr, rel_drift=rel_drift) + + +ALGORITHMS: dict[str, Callable[[Sequence[float]], TrendResult]] = { + "mk_hamed_rao": mann_kendall_hamed_rao, + "mann_kendall": mann_kendall, + "newey_west": newey_west, + "theil_sen": theil_sen, + "slope_vs_scatter": slope_vs_scatter, +} + + +# --------------------------------------------------------------------------- # +# Rolling scan + CoV table +# --------------------------------------------------------------------------- # +def rolling_windows(n: int, window: int) -> list[tuple[int, int]]: + if window <= 0 or window > n: + return [] + return [(s, s + window) for s in range(0, n - window + 1)] + + +def cov_pass_row( + values: Sequence[float], bounds: Sequence[float] +) -> dict[float, bool | None]: + # Fewer than 2 points -> CoV is undefined; report inconclusive (None), never PASS, + # so a short/empty window can't masquerade as steady. + if len(values) < 2: + return {b: None for b in bounds} + c = cov(values) + return {b: c <= b for b in bounds} + + +# --------------------------------------------------------------------------- # +# Steady-window selection: admissibility, plateau segmentation, level shift +# --------------------------------------------------------------------------- # +def _window_percentile_series( + series: Sequence[SuperPassRollup], lo: int, hi: int, source_attr: str, pct: float +) -> list[float] | None: + """Per-super-pass percentile over ``[lo, hi)``; None if any super-pass is empty.""" + out: list[float] = [] + for sp in series[lo:hi]: + vals = getattr(sp, source_attr) + if not vals: + return None + out.append(percentile_lower(sorted(vals), pct)) + return out + + +def window_admissible( + series: Sequence[SuperPassRollup], + lo: int, + hi: int, + gate_algo: str, + cov_bounds: Sequence[float], + gated_metrics: Sequence[TrackedMetric] = GATED_METRICS, +) -> bool: + """True iff every gated metric is trend-steady and within the loosest CoV bound.""" + gate = ALGORITHMS[gate_algo] + loosest = max(cov_bounds) + for m in gated_metrics: + traj = _window_percentile_series(series, lo, hi, m.source_attr, m.percentile) + if traj is None or len(traj) < MIN_TREND_N: + return False + if gate(traj).verdict != "steady": + return False + if cov(traj) > loosest: + return False + return True + + +def segment_plateaus( + series: Sequence[SuperPassRollup], + gate_algo: str, + cov_bounds: Sequence[float], + gated_metrics: Sequence[TrackedMetric] = GATED_METRICS, + min_len: int = MIN_TREND_N, +) -> list[tuple[int, int]]: + """Grow-from-left segmentation into maximal admissible plateaus. + + From each start, extend the window until admissibility breaks (a staircase jump + fails the CoV/trend gate); the maximal admissible span is one plateau, then resume + past it. Plateaus shorter than ``min_len`` are impossible by construction. + """ + n = len(series) + plateaus: list[tuple[int, int]] = [] + start = 0 + while start <= n - min_len: + hi: int | None = None + for end in range(start + min_len, n + 1): + if window_admissible( + series, start, end, gate_algo, cov_bounds, gated_metrics + ): + hi = end + else: + break + if hi is not None: + plateaus.append((start, hi)) + start = hi + else: + start += 1 + return plateaus + + +def detect_level_shift( + series: Sequence[SuperPassRollup], + plateaus: Sequence[tuple[int, int]], + cov_band: float = 0.05, +) -> Anomaly: + """Flag a staircase: a later plateau whose TPOT level differs from the first by + more than ``cov_band``, corroborated by a Pettitt change-point on the per-super-pass + TPOT means. ``delta_pct`` > 0 means the later level is worse (TPOT rose).""" + result: Anomaly = { + "detected": False, + "change_point_sp": None, + "delta_pct": 0.0, + "pettitt": None, + "plateaus": [list(p) for p in plateaus], + } + if len(plateaus) < 2: + return result + + def _tpot_mean(lo: int, hi: int) -> float: + vals = pooled(series, lo, hi, "tpot_ns") + return sum(vals) / len(vals) if vals else 0.0 + + first_mean = _tpot_mean(*plateaus[0]) + if first_mean <= 0: + return result + sp_means = [ + (sum(sp.tpot_ns) / len(sp.tpot_ns)) if sp.tpot_ns else 0.0 for sp in series + ] + pet = pettitt(sp_means) + result["pettitt"] = pet + for lo, hi in plateaus[1:]: + rel = (_tpot_mean(lo, hi) - first_mean) / first_mean + if abs(rel) > cov_band and pet["significant"]: + result["detected"] = True + result["change_point_sp"] = pet["change_point"] + result["delta_pct"] = rel * 100.0 + break + return result + + +def global_trend( + series: Sequence[SuperPassRollup], + from_idx: int, + gate_algo: str, + gated_metrics: Sequence[TrackedMetric] = GATED_METRICS, +) -> dict[str, Verdict]: + """Trend verdict per gated metric over ``series[from_idx:]`` (plateau onset to end). + + A window can be locally flat while the metric climbs across the rest of the run + (a slow drift the short per-window gate misses); this whole-tail test catches it. + """ + gate = ALGORITHMS[gate_algo] + out: dict[str, Verdict] = {} + for m in gated_metrics: + traj = super_pass_percentile_series( + series[from_idx:], m.source_attr, m.percentile + ) + out[m.key] = gate(traj).verdict if len(traj) >= MIN_TREND_N else "insufficient" + return out + + +def adaptive_warmup( + series: Sequence[SuperPassRollup], + driver: str = "tpot_p50", + band: float = 0.05, + min_warmup: int = 1, + max_frac: float = 0.5, +) -> int: + """Data-driven warmup crop: drop leading super-passes still off the steady level. + + The driver's steady level is estimated from the median of the series' back half; + leading super-passes whose driver value is more than ``band`` (fractional) away from + it — in *either* direction — are cropped. Symmetric because the natural driver, TPOT, + ramps *up* to steady (unlike TTFT, which decays down). Capped at ``max_frac`` of the + run so it can never crop everything. + """ + m = _METRIC_BY_KEY[driver] + vals = super_pass_percentile_series(series, m.source_attr, m.percentile) + n = len(vals) + if n < MIN_TREND_N: + return min_warmup + steady = median(vals[n // 2 :]) or ZERO_MEDIAN_FLOOR + cap = max(min_warmup, int(n * max_frac)) + w = 0 + while w < cap and abs(vals[w] - steady) / steady > band: + w += 1 + return max(min_warmup, w) + + +def build_steady_state( + series: Sequence[SuperPassRollup], + gate_algo: str = "mk_hamed_rao", + cov_bounds: Sequence[float] = (0.03, 0.05, 0.08), + gated_metrics: Sequence[TrackedMetric] = GATED_METRICS, +) -> SteadyState: + """Select the first steady plateau and summarize it (window, TTFT/TPOT, TPS). + + ``series`` is the post-warmup super-pass series; window indices are relative to it. + """ + plateaus = segment_plateaus(series, gate_algo, cov_bounds, gated_metrics) + anomaly = detect_level_shift(series, plateaus) + if not plateaus: + gt = global_trend(series, 0, gate_algo, gated_metrics) + return { + "found": False, + "reason": "no admissible steady plateau", + "window": None, + "ttft": None, + "tpot": None, + "tps": None, + "anomaly": anomaly, + "global_trend": gt, + "drifting_up": [k for k, v in gt.items() if v == "up"], + } + lo, hi = plateaus[0] # first plateau is the reported steady state + gt = global_trend(series, lo, gate_algo, gated_metrics) + ttft = pooled(series, lo, hi, "ttft_ns") + tpot = pooled(series, lo, hi, "tpot_ns") + mean_tpot = sum(tpot) / len(tpot) if tpot else 0.0 + sp_tpot_means = [ + sum(sp.tpot_ns) / len(sp.tpot_ns) for sp in series[lo:hi] if sp.tpot_ns + ] + tpot_ci = ( + batch_means_ci(sp_tpot_means) + if len(sp_tpot_means) >= 2 + else (mean_tpot, mean_tpot) + ) + # per-user TPS = 1e9/TPOT is monotone-decreasing, so invert the CI bounds. + per_user_ci = [per_user_tps(tpot_ci[1]), per_user_tps(tpot_ci[0])] + # Aggregate tokens / offered-load (issue) span (§5.1) — NOT the completion span, which + # would inflate the denominator with the drain and deflate TPS on high-tail workloads. + # The CI is a batch-means half-width from per-super-pass throughput, centered on the + # point (per-super-pass issue spans exclude inter-super-pass gaps, so their mean would + # not equal the aggregate). + system = system_tps( + pooled_out_tokens(series, lo, hi), window_issue_span_ns(series, lo, hi) + ) + sp_system = [ + system_tps(sp.out_tokens, sp.last_issue_ns - sp.first_issue_ns) + for sp in series[lo:hi] + if sp.last_issue_ns > sp.first_issue_ns >= 0 + ] + if len(sp_system) >= 2: + clo, chi = batch_means_ci(sp_system) + half = (chi - clo) / 2.0 + system_ci = [system - half, system + half] + else: + system_ci = [system, system] + return { + "found": True, + "reason": None, + "window": { + "sp_lo": lo, + "sp_hi": hi, + "n_super_passes": hi - lo, + "n_samples": len(ttft), + }, + "ttft": summarize(ttft) if ttft else None, + "tpot": summarize(tpot) if tpot else None, + "tps": { + "per_user": per_user_tps(mean_tpot), + "per_user_ci": per_user_ci, + "system": system, + "system_ci": system_ci, + }, + "anomaly": anomaly, + "global_trend": gt, + "drifting_up": [k for k, v in gt.items() if v == "up"], + } + + +# --------------------------------------------------------------------------- # +# Top-level orchestration +# --------------------------------------------------------------------------- # +class CovCell(TypedDict): + gated: bool + n: int # super-passes in the trailing window + cov: float | None # None when the window has < 2 points + passes: dict[str, bool | None] # cov-bound (as str) -> pass / fail / inconclusive + + +class RollingCell(TypedDict): + window: list[int] # [lo, hi) super-pass indices + verdicts: dict[str, Verdict] # algorithm name -> verdict + + +class DriftEntry(TypedDict): + whole_run: dict[str, Verdict] # algorithm name -> verdict over the full trajectory + rolling: list[RollingCell] + + +class DiagnosticsResult(TypedDict): + n_super_passes: int + superpass_size: int + warmup: int # resolved super-pass crop count + warmup_mode: str # "auto" (adaptive) or "fixed" + n_post_warmup: int + metrics: list[str] + gated_metrics: list[str] + trajectories: dict[ + str, list[float] + ] # metric key -> per-super-pass percentile series + cov: dict[str, dict[str, CovCell]] # window size (str) -> metric key -> cell + drift: dict[str, dict[str, DriftEntry]] # window size (str) -> metric key -> entry + steady_state: SteadyState # the headline: first steady plateau + TPS + anomaly + per_super_pass: list[dict] # raw post-warmup per-super-pass rollups (for plotting) + alpha: float + + +def _drift_verdicts(trajectory: Sequence[float]) -> dict[str, Verdict]: + return {name: fn(trajectory).verdict for name, fn in ALGORITHMS.items()} + + +def per_super_pass_diagnostics(series: Sequence[SuperPassRollup]) -> list[dict]: + """Raw per-super-pass rollups (timestamps, tokens, percentiles) for plotting.""" + out: list[dict] = [] + for i, sp in enumerate(series): + tt = sorted(sp.ttft_ns) + tp = sorted(sp.tpot_ns) + out.append( + { + "i": i, + "n_issued": sp.n_issued, + "out_tokens": sp.out_tokens, + "first_issue_ns": sp.first_issue_ns, + "last_issue_ns": sp.last_issue_ns, + "last_event_ns": sp.last_event_ns, + "ttft_p50": percentile_lower(tt, 0.50) if tt else None, + "ttft_p95": percentile_lower(tt, 0.95) if tt else None, + "tpot_p50": percentile_lower(tp, 0.50) if tp else None, + "tpot_p95": percentile_lower(tp, 0.95) if tp else None, + } + ) + return out + + +def run( + events_path: str, + superpass_size: int, + count_tokens: Callable[[list[str]], list[int]], + window_sizes: Sequence[int] = (4, 5), + warmup: int | str = "auto", + cov_bounds: Sequence[float] = (0.03, 0.05, 0.08), + alpha: float = 0.05, + trend_gate: str = "mk_hamed_rao", + tokenize_batch_size: int = TOKENIZE_BATCH_SIZE, + warmup_band: float = 0.05, + warmup_driver: str = "tpot_p50", +) -> DiagnosticsResult: + """Build the full diagnostics result (the ``--json`` blob). + + ``window_sizes`` are counts of super-passes; ``superpass_size`` is a count of samples. + ``warmup`` is either ``"auto"`` (data-driven crop via ``adaptive_warmup`` on the + ``warmup_driver`` metric) or a fixed super-pass count. ``trend_gate`` names the trend + algorithm gating admissibility. + """ + if isinstance(warmup, int) and warmup < 0: + raise ValueError(f"warmup must be >= 0, got {warmup}") + series = build_super_pass_series( + events_path, superpass_size, count_tokens, tokenize_batch_size + ) + if warmup == "auto": + resolved_warmup = adaptive_warmup(series, warmup_driver, warmup_band) + warmup_mode = "auto" + else: + resolved_warmup = int(warmup) + warmup_mode = "fixed" + post = series[resolved_warmup:] if resolved_warmup < len(series) else [] + trajectories = { + m.key: super_pass_percentile_series(post, m.source_attr, m.percentile) + for m in TRACKED_METRICS + } + + result: DiagnosticsResult = { + "n_super_passes": len(series), + "superpass_size": superpass_size, + "warmup": resolved_warmup, + "warmup_mode": warmup_mode, + "n_post_warmup": len(post), + "metrics": [m.key for m in TRACKED_METRICS], + "gated_metrics": [m.key for m in TRACKED_METRICS if m.gated], + "trajectories": trajectories, + "cov": {}, + "drift": {}, + "steady_state": build_steady_state(post, trend_gate, cov_bounds), + "per_super_pass": per_super_pass_diagnostics(post), + "alpha": alpha, + } + + for w in window_sizes: + cov_tbl: dict[str, CovCell] = {} + drift_tbl: dict[str, DriftEntry] = {} + for m in TRACKED_METRICS: + traj = trajectories[m.key] + trailing = traj[-w:] if len(traj) >= w else traj + cov_tbl[m.key] = { + "gated": m.gated, + "n": len(trailing), + "cov": cov(trailing) if len(trailing) >= 2 else None, + "passes": { + str(b): v for b, v in cov_pass_row(trailing, cov_bounds).items() + }, + } + rolling: list[RollingCell] = [ + {"window": [lo, hi], "verdicts": _drift_verdicts(traj[lo:hi])} + for lo, hi in rolling_windows(len(traj), w) + ] + drift_tbl[m.key] = {"whole_run": _drift_verdicts(traj), "rolling": rolling} + result["cov"][str(w)] = cov_tbl + result["drift"][str(w)] = drift_tbl + + return result + + +# --------------------------------------------------------------------------- # +# Rendering +# --------------------------------------------------------------------------- # +_VERDICT_GLYPH = { + "up": "^ up", + "down": "v down", + "steady": "= steady", + "insufficient": ". n/a", +} + + +def _pass_glyph(ok: bool | None) -> str: + if ok is None: + return "n/a" + return "PASS" if ok else "fail" + + +def _fmt_ms(ns: float) -> str: + return f"{ns / 1e6:.2f}ms" + + +def _render_steady_state(ss: SteadyState) -> list[str]: + out = ["=== STEADY STATE (headline) ==="] + if not ss["found"]: + out.append(f" not found: {ss['reason']}") + else: + w = ss["window"] + tps = ss["tps"] + assert w is not None and tps is not None + out.append( + f" window: super-passes {w['sp_lo']}..{w['sp_hi'] - 1} (post-warmup), " + f"{w['n_samples']} samples" + ) + out.append( + f" TPS per-user: {tps['per_user']:8.1f} tok/s/user " + f"CI [{tps['per_user_ci'][0]:.1f}, {tps['per_user_ci'][1]:.1f}]" + ) + out.append( + f" TPS system: {tps['system']:8.1f} tok/s " + f"CI [{tps['system_ci'][0]:.1f}, {tps['system_ci'][1]:.1f}]" + ) + for name in ("ttft", "tpot"): + s = ss[name] # type: ignore[literal-required] + if s: + out.append( + f" {name.upper():4} p50 {_fmt_ms(s['p50'])} p90 {_fmt_ms(s['p90'])}" + f" p95 {_fmt_ms(s['p95'])} p99 {_fmt_ms(s['p99'])}" + f" mean {_fmt_ms(s['mean'])}" + ) + if ss["drifting_up"]: + out.append( + f" WARNING: {', '.join(ss['drifting_up'])} drifting UP over the rest of the " + f"run -- the window is a local plateau; global steady state is questionable" + ) + an = ss["anomaly"] + if an["detected"]: + out.append( + f" ANOMALY: level shift at super-pass {an['change_point_sp']}, " + f"TPOT {an['delta_pct']:+.1f}% toward end of run (likely degradation)" + ) + return out + + +def render_text(result: DiagnosticsResult, cov_bounds: Sequence[float]) -> str: + lines: list[str] = [] + lines.append( + f"super-passes: {result['n_super_passes']} " + f"(size {result['superpass_size']}, warmup {result['warmup']} " + f"[{result['warmup_mode']}], post-warmup {result['n_post_warmup']})" + ) + lines.append("") + lines.extend(_render_steady_state(result["steady_state"])) + lines.append("") + lines.append("--- diagnostics (per window size) ---") + gated = set(result["gated_metrics"]) + metrics = result["metrics"] + bound_hdr = " ".join(f"cov<={b}" for b in cov_bounds) + for w in sorted(result["cov"], key=int): + lines.append("") + lines.append(f"=== window size {w} ===") + lines.append("") + lines.append(f"CoV steadiness (trailing {w} super-passes)") + lines.append(f" {'metric':<12} {'gate':<5} {'CoV':>8} {bound_hdr}") + for key in metrics: + cell = result["cov"][w][key] + covv = cell["cov"] + covs = f"{covv:.4f}" if covv is not None else " n/a" + passes = " ".join( + f"{_pass_glyph(cell['passes'][str(b)]):>7}" for b in cov_bounds + ) + tag = "gate" if key in gated else "diag" + lines.append(f" {key:<12} {tag:<5} {covs:>8} {passes}") + + lines.append("") + lines.append("drift (whole-run trend per metric; rolling scan is in --json)") + algos = list(ALGORITHMS) + lines.append(f" {'metric':<12} " + " ".join(f"{a:>16}" for a in algos)) + for key in metrics: + whole = result["drift"][w][key]["whole_run"] + cells = " ".join(f"{_VERDICT_GLYPH[whole[a]]:>16}" for a in algos) + lines.append(f" {key:<12} {cells}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # +def _make_token_counter(tokenizer_id: str) -> Callable[[list[str]], list[int]]: + from transformers import AutoTokenizer + + tok = AutoTokenizer.from_pretrained(tokenizer_id) + + def count(texts: list[str]) -> list[int]: + enc = tok(texts, add_special_tokens=False)["input_ids"] + return [len(ids) for ids in enc] + + return count + + +def _parse_int_list(s: str) -> list[int]: + return [int(x) for x in s.split(",") if x.strip()] + + +def _parse_float_list(s: str) -> list[float]: + return [float(x) for x in s.split(",") if x.strip()] + + +def _warmup_arg(s: str) -> int | str: + return "auto" if s == "auto" else int(s) + + +def main(argv: Sequence[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + ap.add_argument("events", help="path to events.jsonl") + ap.add_argument( + "--tokenizer", required=True, help="HF model dir/id for TTFT+TPOT token counts" + ) + ap.add_argument( + "--dataset-size", type=int, required=True, help="samples per dataset pass" + ) + ap.add_argument( + "--superpass-size", + type=int, + default=None, + help="samples per super-pass (default: --dataset-size)", + ) + ap.add_argument( + "--window-sizes", + type=_parse_int_list, + default=[4, 5], + help="comma-separated window sizes, in super-passes", + ) + ap.add_argument( + "--warmup", + type=_warmup_arg, + default="auto", + help="'auto' (data-driven crop) or a fixed super-pass count", + ) + ap.add_argument( + "--warmup-band", + type=float, + default=0.05, + help="auto warmup: crop leading super-passes >this fraction off the steady level", + ) + ap.add_argument( + "--warmup-driver", + default="tpot_p50", + choices=list(_METRIC_BY_KEY), + help="auto warmup: metric whose ramp defines the crop", + ) + ap.add_argument("--cov-bounds", type=_parse_float_list, default=[0.03, 0.05, 0.08]) + ap.add_argument("--alpha", type=float, default=0.05) + ap.add_argument( + "--trend-gate", + default="mk_hamed_rao", + choices=list(ALGORITHMS), + help="trend algorithm gating plateau admissibility", + ) + ap.add_argument( + "--tokenize-batch-size", + type=int, + default=TOKENIZE_BATCH_SIZE, + help="output texts buffered per tokenizer flush; lower to bound peak memory", + ) + ap.add_argument( + "--json", dest="json_out", default=None, help="write JSON blob here" + ) + args = ap.parse_args(argv) + + superpass_size = args.superpass_size or args.dataset_size + count_tokens = _make_token_counter(args.tokenizer) + result = run( + args.events, + superpass_size=superpass_size, + count_tokens=count_tokens, + window_sizes=args.window_sizes, + warmup=args.warmup, + cov_bounds=args.cov_bounds, + alpha=args.alpha, + trend_gate=args.trend_gate, + tokenize_batch_size=args.tokenize_batch_size, + warmup_band=args.warmup_band, + warmup_driver=args.warmup_driver, + ) + print(render_text(result, args.cov_bounds)) + if args.json_out: + with open(args.json_out, "w") as f: + json.dump(result, f, indent=2) + print(f"\nwrote {args.json_out}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/scripts/test_steady_state_diagnostics.py b/tests/unit/scripts/test_steady_state_diagnostics.py new file mode 100644 index 000000000..3a63e8a56 --- /dev/null +++ b/tests/unit/scripts/test_steady_state_diagnostics.py @@ -0,0 +1,568 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for scripts/steady_state_diagnostics.py. + +The script is self-contained (no ``inference_endpoint`` import), so these tests pin +everything it owns: the plain-JSON event parse (wire shapes referenced from +core/record.py + core/types.py), super-pass bucketing, TTFT/TPOT reconstruction via an +injected token counter, the trend-detection algorithms, and the CoV table. +""" + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.unit + + +def _load_script(): + spec = importlib.util.spec_from_file_location( + "steady_state_diagnostics", + Path("scripts/steady_state_diagnostics.py"), + ) + mod = importlib.util.module_from_spec(spec) + # Register before exec so dataclass field annotations (PEP 563 strings) resolve. + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +mod = _load_script() + + +def _ev(event_type, ts, uuid="", data=None): + return json.dumps( + { + "event_type": event_type, + "timestamp_ns": ts, + "sample_uuid": uuid, + "data": data, + } + ) + + +def _words(texts): + """Fake tokenizer: token count == whitespace word count.""" + return [len(t.split()) for t in texts] + + +# --------------------------------------------------------------------------- # +# text_after_first_chunk — port of TextModelOutput.text_after_first_chunk over +# the parsed JSON array [tag, output, reasoning?, tool_calls?]. +# --------------------------------------------------------------------------- # + + +def test_text_after_first_chunk_streaming_output_drops_first(): + data = ["TextModelOutput", ["hello ", "world ", "again"]] + assert mod.text_after_first_chunk(data) == "world again" + + +def test_text_after_first_chunk_reasoning_first_keeps_all_output(): + # reasoning is a tuple (streaming) -> first chunk lived in reasoning, so all + # output chunks are post-first-chunk and are kept. + data = ["TextModelOutput", ["out1 ", "out2"], ["think1 ", "think2 "]] + assert mod.text_after_first_chunk(data) == "think2 out1 out2" + + +def test_text_after_first_chunk_non_streaming_str_has_no_first_chunk(): + data = ["TextModelOutput", "the whole answer"] + assert mod.text_after_first_chunk(data) == "" + + +# --------------------------------------------------------------------------- # +# Super-pass series construction +# --------------------------------------------------------------------------- # + + +def _write_events(tmp_path, lines): + p = tmp_path / "events.jsonl" + p.write_text("\n".join(lines) + "\n") + return str(p) + + +def test_build_super_pass_series_buckets_by_issue_order(tmp_path): + lines = [ + _ev("session.start_performance_tracking", 0), + # super-pass 0 (superpass_size=2): samples A, B + _ev("sample.issued", 1000, "A"), + _ev("sample.issued", 1100, "B"), + _ev("sample.recv_first", 1300, "B"), + _ev("sample.recv_first", 1500, "A"), + _ev("sample.complete", 2100, "B", ["TextModelOutput", ["x ", "y y y"]]), + _ev("sample.complete", 3000, "A", ["TextModelOutput", ["a ", "b b"]]), + # super-pass 1: sample C + _ev("sample.issued", 1200, "C"), + _ev("sample.recv_first", 1400, "C"), + _ev("sample.complete", 2000, "C", ["TextModelOutput", ["p ", "q"]]), + _ev("session.stop_performance_tracking", 4000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=2, count_tokens=_words) + + assert len(series) == 2 + sp0, sp1 = series + assert sp0.n_issued == 2 + assert sorted(sp0.ttft_ns) == [200.0, 500.0] + # TPOT: A -> "b b" 2 tokens over (3000-1500)=1500 -> 750; B -> "y y y" 3 over + # (2100-1300)=800 -> 266.66... + assert sorted(round(v, 2) for v in sp0.tpot_ns) == [266.67, 750.0] + assert sp0.out_tokens == 5 + assert sp1.n_issued == 1 + assert sp1.ttft_ns == [200.0] + + +def test_build_super_pass_series_ignores_events_outside_tracking(tmp_path): + lines = [ + _ev("sample.issued", 500, "PRE"), # before tracking -> ignored + _ev("session.start_performance_tracking", 900), + _ev("sample.issued", 1000, "A"), + _ev("sample.recv_first", 1500, "A"), + _ev("sample.complete", 3000, "A", ["TextModelOutput", ["a ", "b"]]), + _ev("session.stop_performance_tracking", 4000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=4, count_tokens=_words) + assert len(series) == 1 + assert series[0].n_issued == 1 + + +# --------------------------------------------------------------------------- # +# Small numeric helpers +# --------------------------------------------------------------------------- # + + +def test_percentile_lower_nearest_rank(): + s = [1.0, 2.0, 3.0, 4.0] + assert mod.percentile_lower(s, 0.5) == 2.0 + assert mod.percentile_lower(s, 0.0) == 1.0 + # nearest-rank-lower: int(0.99*3) == 2 -> 3.0 (matches reference percentile_lower) + assert mod.percentile_lower(s, 0.99) == 3.0 + assert mod.percentile_lower(s, 1.0) == 4.0 + + +def test_cov_flat_series_is_zero(): + assert mod.cov([5.0, 5.0, 5.0]) == 0.0 + + +def test_cov_positive_for_varied_series(): + assert mod.cov([1.0, 2.0, 3.0]) > 0.0 + + +# --------------------------------------------------------------------------- # +# Trend algorithms — verdict in {"up", "steady", "down"} +# --------------------------------------------------------------------------- # + +STRONG_UP = [float(i) for i in range(12)] +STRONG_DOWN = [float(11 - i) for i in range(12)] +FLAT = [5.0] * 12 + + +def test_mann_kendall_detects_upward_trend(): + assert mod.mann_kendall(STRONG_UP).verdict == "up" + + +def test_mann_kendall_detects_downward_trend(): + assert mod.mann_kendall(STRONG_DOWN).verdict == "down" + + +def test_mann_kendall_flat_is_steady(): + assert mod.mann_kendall(FLAT).verdict == "steady" + + +def test_hamed_rao_inflates_variance_vs_plain_mk(): + # Autocorrelation correction can only widen (or equal) the MK variance. + plain = mod.mann_kendall(STRONG_UP) + corrected = mod.mann_kendall_hamed_rao(STRONG_UP) + assert corrected.variance >= plain.variance + assert corrected.verdict == "up" + + +def test_hamed_rao_does_not_fabricate_trend_on_oscillation(): + # A sawtooth "climbs" in raw issue order, but that trend is an ordering artifact. + # Naive MK is fooled; the Hamed-Rao autocorrelation correction must inflate the + # variance (never collapse it) and return steady. + saw = [0.0, 20.0, 2.0, 22.0, 4.0, 24.0, 6.0, 26.0, 8.0, 28.0, 10.0, 30.0] + plain = mod.mann_kendall(saw) + hr = mod.mann_kendall_hamed_rao(saw) + assert plain.verdict == "up" # uncorrected MK false-positives + assert hr.verdict == "steady" # corrected test refuses + assert hr.variance >= plain.variance # inflated, not collapsed to a sliver + + +def test_theil_sen_recovers_exact_slope(): + assert mod.theil_sen([0.0, 2.0, 4.0, 6.0, 8.0]).slope == 2.0 + + +def test_theil_sen_flat_is_steady(): + assert mod.theil_sen(FLAT).verdict == "steady" + + +def test_newey_west_detects_trend_and_flat(): + assert mod.newey_west(STRONG_UP).verdict == "up" + assert mod.newey_west(FLAT).verdict == "steady" + + +def test_slope_vs_scatter_matches_reference_formula(): + # Clean line: huge SNR, large rel-drift -> up. + assert mod.slope_vs_scatter(STRONG_UP).verdict == "up" + assert mod.slope_vs_scatter(STRONG_DOWN).verdict == "down" + # Flat -> steady. + assert mod.slope_vs_scatter(FLAT).verdict == "steady" + + +def test_trend_algorithms_report_insufficient_below_min_n(): + for fn in (mod.mann_kendall, mod.theil_sen, mod.newey_west, mod.slope_vs_scatter): + assert fn([1.0, 2.0, 3.0]).verdict == "insufficient" + + +# --------------------------------------------------------------------------- # +# Rolling scan + CoV table +# --------------------------------------------------------------------------- # + + +def test_rolling_windows_enumerates_all_positions(): + # n=6, window=4 -> starts 0,1,2 -> [0,4),[1,5),[2,6) + windows = mod.rolling_windows(n=6, window=4) + assert windows == [(0, 4), (1, 5), (2, 6)] + + +def test_cov_table_pass_fail_against_bounds(): + # A per-super-pass p50 series that is dead flat -> CoV 0 -> passes every bound. + flat_series = [3.0, 3.0, 3.0, 3.0] + row = mod.cov_pass_row(flat_series, bounds=(0.03, 0.05, 0.08)) + assert row == {0.03: True, 0.05: True, 0.08: True} + + # A noisy series whose CoV exceeds 0.03 but not 0.08. + noisy = [3.0, 3.2, 2.9, 3.15] + row2 = mod.cov_pass_row(noisy, bounds=(0.03, 0.5)) + assert row2[0.03] is False + assert row2[0.5] is True + + +# --------------------------------------------------------------------------- # +# Top-level run() + render_text() contract +# --------------------------------------------------------------------------- # + + +def _synthetic_events(tmp_path, n, ttft_ns_fn): + """One sample per super-pass (superpass_size=1), n super-passes.""" + lines = [_ev("session.start_performance_tracking", 0)] + t = 1000 + for i in range(n): + uuid = f"s{i}" + issue = t + recv = issue + int(ttft_ns_fn(i)) + complete = recv + 1000 + lines.append(_ev("sample.issued", issue, uuid)) + lines.append(_ev("sample.recv_first", recv, uuid)) + lines.append( + _ev("sample.complete", complete, uuid, ["TextModelOutput", ["a ", "b b"]]) + ) + t += 10_000 + lines.append(_ev("session.stop_performance_tracking", t)) + return _write_events(tmp_path, lines) + + +def test_run_result_structure(tmp_path): + # ttft climbs steadily -> an upward drift the scan should surface. + path = _synthetic_events(tmp_path, n=8, ttft_ns_fn=lambda i: 100 + 20 * i) + result = mod.run( + path, superpass_size=1, count_tokens=_words, window_sizes=[4], warmup=1 + ) + assert result["n_super_passes"] == 8 + assert result["n_post_warmup"] == 7 + assert "ttft_p50" in result["trajectories"] + # window 4 over 7 post-warmup super-passes -> 4 rolling positions + rolling = result["drift"]["4"]["ttft_p50"]["rolling"] + assert [r["window"] for r in rolling] == [[0, 4], [1, 5], [2, 6], [3, 7]] + # CoV table carries a pass/fail cell per bound and a gate flag + cell = result["cov"]["4"]["ttft_p50"] + assert cell["gated"] is True + assert set(cell["passes"]) == {"0.03", "0.05", "0.08"} + # p99 is present but marked diagnostic (not gated) + assert result["cov"]["4"]["ttft_p99"]["gated"] is False + + +def test_text_after_first_chunk_empty_reasoning_str_output_is_first_chunk(): + # reasoning=[] is falsy -> str output IS the sole first chunk -> excluded. + data = ["TextModelOutput", "the whole answer", []] + assert mod.text_after_first_chunk(data) == "" + + +def test_retried_sample_counts_ttft_once(tmp_path): + lines = [ + _ev("session.start_performance_tracking", 0), + _ev("sample.issued", 1000, "A"), + _ev("sample.recv_first", 1500, "A"), # ttft 500 (the real first token) + _ev("sample.issued", 2000, "A"), # retry: refresh issue ts only + _ev("sample.recv_first", 2600, "A"), # must NOT add a second ttft + _ev("sample.complete", 4000, "A", ["TextModelOutput", ["a ", "b b"]]), + _ev("session.stop_performance_tracking", 5000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=4, count_tokens=_words) + assert series[0].ttft_ns == [500.0] + + +def test_valid_json_line_missing_timestamp_is_skipped(tmp_path): + # A syntactically valid event object missing timestamp_ns must be skipped, not crash. + partial = json.dumps({"event_type": "sample.issued", "sample_uuid": "BAD"}) + lines = [ + _ev("session.start_performance_tracking", 0), + partial, + _ev("sample.issued", 1000, "A"), + _ev("sample.recv_first", 1500, "A"), + _ev("sample.complete", 3000, "A", ["TextModelOutput", ["a ", "b"]]), + _ev("session.stop_performance_tracking", 4000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=4, count_tokens=_words) + assert sum(sp.n_issued for sp in series) == 1 # only the well-formed sample + + +def test_cov_pass_row_insufficient_window_is_inconclusive(): + # Fewer than 2 points -> CoV undefined -> cells inconclusive (None), not True. + assert mod.cov_pass_row([3.0], bounds=(0.03, 0.05)) == {0.03: None, 0.05: None} + + +def test_run_rejects_negative_warmup(tmp_path): + path = _synthetic_events(tmp_path, n=6, ttft_ns_fn=lambda i: 100.0) + with pytest.raises(ValueError): + mod.run(path, superpass_size=1, count_tokens=_words, warmup=-1) + + +def test_super_pass_tracks_timestamps(tmp_path): + lines = [ + _ev("session.start_performance_tracking", 0), + _ev("sample.issued", 1000, "A"), + _ev("sample.recv_first", 1500, "A"), + _ev("sample.complete", 3000, "A", ["TextModelOutput", ["a ", "b b"]]), + _ev("sample.issued", 1100, "B"), + _ev("sample.recv_first", 1400, "B"), + _ev("sample.complete", 2500, "B", ["TextModelOutput", ["c ", "d"]]), + _ev("session.stop_performance_tracking", 4000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=4, count_tokens=_words) + assert series[0].first_issue_ns == 1000 # earliest issue + assert series[0].last_event_ns == 3000 # latest event (A's complete) + + +def test_window_elapsed_and_pooling(tmp_path): + lines = [ + _ev("session.start_performance_tracking", 0), + _ev("sample.issued", 1000, "A"), + _ev("sample.recv_first", 1500, "A"), + _ev("sample.complete", 3000, "A", ["TextModelOutput", ["a ", "b b"]]), + _ev("sample.issued", 11000, "B"), + _ev("sample.recv_first", 11500, "B"), + _ev("sample.complete", 13000, "B", ["TextModelOutput", ["c ", "d d d"]]), + _ev("session.stop_performance_tracking", 20000), + ] + path = _write_events(tmp_path, lines) + series = mod.build_super_pass_series(path, superpass_size=1, count_tokens=_words) + assert len(series) == 2 + # pooled ttft over both super-passes + assert sorted(mod.pooled(series, 0, 2, "ttft_ns")) == [500.0, 500.0] + assert mod.pooled_out_tokens(series, 0, 2) == 5 # 2 + 3 + # elapsed = last_event(sp1) - first_issue(sp0) = 13000 - 1000 + assert mod.window_elapsed_ns(series, 0, 2) == 12000 + + +def test_histogram_bins_sum_to_count(): + vals = [1.0, 2.0, 2.0, 3.0, 5.0, 8.0, 13.0] + hist = mod.histogram(vals, nbins=4) + assert sum(b["count"] for b in hist) == len(vals) + assert hist[0]["lo"] == 1.0 + assert hist[-1]["hi"] == 13.0 + + +def test_histogram_degenerate_single_value(): + hist = mod.histogram([5.0, 5.0, 5.0], nbins=4) + assert sum(b["count"] for b in hist) == 3 + + +def test_summarize_reports_stats_and_histogram(): + vals = [float(i) for i in range(1, 101)] + s = mod.summarize(vals) + assert s["count"] == 100 + assert s["mean"] == 50.5 + assert s["p50"] == 50.0 # nearest-rank-lower + assert s["p99"] == 99.0 + assert sum(b["count"] for b in s["histogram"]) == 100 + + +def test_batch_means_ci_zero_variance_is_point(): + lo, hi = mod.batch_means_ci([10.0, 10.0, 10.0, 10.0]) + assert lo == 10.0 and hi == 10.0 + + +def test_batch_means_ci_brackets_mean(): + lo, hi = mod.batch_means_ci([8.0, 10.0, 12.0, 10.0]) + assert lo < 10.0 < hi + + +def test_pettitt_detects_level_shift(): + series = [0.0] * 8 + [10.0] * 8 + res = mod.pettitt(series) + assert res["significant"] is True + assert res["change_point"] == 8 # first segment has 8 elements + + +def test_pettitt_flat_series_no_change(): + res = mod.pettitt([5.0] * 16) + assert res["significant"] is False + + +def test_tps_formulas(): + # per-user = 1e9 / mean_tpot_ns; 2e6 ns/token -> 500 tok/s/user + assert mod.per_user_tps(2_000_000.0) == 500.0 + # system = out_tokens / (elapsed_ns / 1e9); 5000 tokens over 10s -> 500 tok/s + assert mod.system_tps(5000, 10_000_000_000) == 500.0 + + +def _mk_series(levels, samples=40): + """Build a SuperPassRollup list; ``levels`` = list of (tpot, ttft) per super-pass.""" + series = [] + for i, (tp, tt) in enumerate(levels): + sp = mod.SuperPassRollup(index=i) + sp.tpot_ns = [float(tp)] * samples + sp.ttft_ns = [float(tt)] * samples + sp.out_tokens = samples * 10 + sp.n_issued = samples + sp.first_issue_ns = i * 1000 + sp.last_issue_ns = i * 1000 + 100 # issue span within the super-pass + sp.last_event_ns = i * 1000 + 900 # completions land later (drain) + series.append(sp) + return series + + +def test_window_issue_span_excludes_the_drain(): + series = _mk_series([(100, 50)] * 3) + # issue span: last_issue(SP2)=2100 - first_issue(SP0)=0 = 2100 + assert mod.window_issue_span_ns(series, 0, 3) == 2100 + # completion span is larger (includes the drain): last_event(SP2)=2900 - 0 + assert mod.window_elapsed_ns(series, 0, 3) == 2900 + assert mod.window_issue_span_ns(series, 0, 3) < mod.window_elapsed_ns(series, 0, 3) + + +GATE = "mk_hamed_rao" +BOUNDS = (0.03, 0.05, 0.08) + + +def test_window_admissible_flat_yes_spanning_jump_no(): + series = _mk_series([(100, 50)] * 4 + [(200, 60)] * 4) + assert mod.window_admissible(series, 0, 4, GATE, BOUNDS) is True + # a window straddling the 100->200 jump has high CoV -> inadmissible + assert mod.window_admissible(series, 2, 6, GATE, BOUNDS) is False + + +def test_segment_plateaus_splits_staircase(): + series = _mk_series([(100, 50)] * 6 + [(200, 60)] * 6) + plateaus = mod.segment_plateaus(series, GATE, BOUNDS) + assert plateaus == [(0, 6), (6, 12)] + + +def test_segment_plateaus_single_flat_run(): + series = _mk_series([(100, 50)] * 8) + assert mod.segment_plateaus(series, GATE, BOUNDS) == [(0, 8)] + + +def test_detect_level_shift_flags_staircase(): + series = _mk_series([(100, 50)] * 6 + [(200, 60)] * 6) + plateaus = mod.segment_plateaus(series, GATE, BOUNDS) + shift = mod.detect_level_shift(series, plateaus) + assert shift["detected"] is True + assert shift["change_point_sp"] == 6 + assert shift["delta_pct"] > 0 # degradation (TPOT rose) + + +def test_detect_level_shift_none_on_single_plateau(): + series = _mk_series([(100, 50)] * 8) + plateaus = mod.segment_plateaus(series, GATE, BOUNDS) + assert mod.detect_level_shift(series, plateaus)["detected"] is False + + +def test_build_steady_state_reports_first_plateau_and_anomaly(): + series = _mk_series([(100, 50)] * 6 + [(200, 60)] * 6) + ss = mod.build_steady_state(series, GATE, BOUNDS) + assert ss["found"] is True + assert ss["window"]["sp_lo"] == 0 and ss["window"]["sp_hi"] == 6 # first plateau + assert ss["tps"]["per_user"] > 0 and ss["tps"]["system"] > 0 + assert ss["ttft"]["count"] == 6 * 40 + assert ss["anomaly"]["detected"] is True # the 100->200 step is surfaced + + +def test_adaptive_warmup_crops_tpot_ramp(): + # TPOT ramps up to a flat 50 -> symmetric band drops the leading below-steady ramp. + series = _mk_series([(20, 50), (30, 50), (40, 50)] + [(50, 50)] * 7) + # steady (back-half median) = 50; band 0.05 -> keep from first SP within 5% (48 -> SP3) + assert mod.adaptive_warmup(series, driver="tpot_p50", band=0.05) == 3 + + +def test_adaptive_warmup_flat_run_returns_min(): + series = _mk_series([(50, 50)] * 8) + assert mod.adaptive_warmup(series, driver="tpot_p50", band=0.05) == 1 + + +def test_adaptive_warmup_capped_at_max_frac(): + # A never-settling monotonic ramp is capped so it can't crop the whole run. + series = _mk_series([(10 * (i + 1), 50) for i in range(10)]) + assert mod.adaptive_warmup(series, driver="tpot_p50", band=0.05, max_frac=0.5) == 5 + + +def test_run_auto_warmup_resolves_to_int(tmp_path): + path = _synthetic_events(tmp_path, n=8, ttft_ns_fn=lambda i: 100.0) + result = mod.run(path, superpass_size=1, count_tokens=_words, warmup="auto") + assert isinstance(result["warmup"], int) + assert result["warmup_mode"] == "auto" + + +def test_build_steady_state_flags_global_drift_after_first_plateau(): + # First plateau is flat, but TPOT ramps up for the rest of the run (the C22528 + # pattern): a local plateau exists, yet the metric drifts up globally. + series = _mk_series([(100, 50)] * 6 + [(100 + 25 * i, 50) for i in range(1, 8)]) + ss = mod.build_steady_state(series, GATE, BOUNDS) + assert ss["found"] is True + assert ss["window"]["sp_lo"] == 0 and ss["window"]["sp_hi"] == 6 + assert ss["anomaly"]["detected"] is False # gradual ramp, not a discrete staircase + assert "tpot_p50" in ss["drifting_up"] # global Drifting-Up gate catches it + assert "ttft_p50" not in ss["drifting_up"] # TTFT is flat + + +def test_build_steady_state_no_global_drift_on_flat_run(): + series = _mk_series([(100, 50)] * 8) + ss = mod.build_steady_state(series, GATE, BOUNDS) + assert ss["drifting_up"] == [] + + +def test_build_steady_state_none_when_run_never_settles(): + # per-super-pass TPOT ramps every step -> no length-4 window is within CoV + series = _mk_series([(100 + 20 * i, 50) for i in range(8)]) + ss = mod.build_steady_state(series, GATE, BOUNDS) + assert ss["found"] is False + + +def test_run_result_has_steady_state_block(tmp_path): + path = _synthetic_events(tmp_path, n=8, ttft_ns_fn=lambda i: 100.0) + result = mod.run( + path, superpass_size=1, count_tokens=_words, window_sizes=[4], warmup=1 + ) + assert "steady_state" in result + assert "anomaly" in result["steady_state"] + + +def test_render_text_has_section_headers(tmp_path): + path = _synthetic_events(tmp_path, n=8, ttft_ns_fn=lambda i: 100 + 20 * i) + result = mod.run( + path, superpass_size=1, count_tokens=_words, window_sizes=[4], warmup=1 + ) + text = mod.render_text(result, cov_bounds=[0.03, 0.05, 0.08]) + assert "window size 4" in text + assert "CoV steadiness" in text + assert "drift (whole-run" in text + assert "ttft_p50" in text