diff --git a/utils/agentic/aggregation/request_metrics.py b/utils/agentic/aggregation/request_metrics.py index 26fc49df17..66454fb33c 100644 --- a/utils/agentic/aggregation/request_metrics.py +++ b/utils/agentic/aggregation/request_metrics.py @@ -9,6 +9,7 @@ from typing import Any from .aggregation_common import percentile, stats_for, to_float, to_int +from .stability_metrics import compute_stability_metrics from .trace_metadata import expected_output_lengths @@ -266,6 +267,26 @@ def compute_throughput_stats( return flat, nested +def _aiperf_metric_avg(aggregate: dict[str, Any], metric_name: str) -> float | None: + """Read an AIPerf aggregate metric, accepting the legacy scalar shape.""" + metric = aggregate.get(metric_name) + if isinstance(metric, dict): + metric = metric.get("avg") + return to_float(metric) + + +def _benchmark_duration_s(aggregate: dict[str, Any]) -> float | None: + """Prefer the configured duration over AIPerf's measured wall time.""" + input_config = aggregate.get("input_config") + if isinstance(input_config, dict): + loadgen = input_config.get("loadgen") + if isinstance(loadgen, dict): + configured = to_float(loadgen.get("benchmark_duration")) + if configured is not None and configured > 0: + return configured + return _aiperf_metric_avg(aggregate, "benchmark_duration") + + def _aiperf_percent_metric_as_rate( aggregate: dict[str, Any], metric_name: str, @@ -315,6 +336,10 @@ def compute_request_metrics( workload_flat, workload_nested = compute_workload_stats(records) cache_flat, cache_nested = compute_cache_stats(records, aggregate) throughput_flat, throughput_nested = compute_throughput_stats(records) + stability_nested = compute_stability_metrics( + records, + benchmark_duration_s=_benchmark_duration_s(aggregate), + ) for part in (qps_flat, latency_flat, workload_flat, cache_flat, throughput_flat): flat.update(part) @@ -326,6 +351,7 @@ def compute_request_metrics( "tokens": workload_nested, "throughput": throughput_nested, "cache": cache_nested, + "stability": stability_nested, } ) return flat, nested diff --git a/utils/agentic/aggregation/stability_metrics.py b/utils/agentic/aggregation/stability_metrics.py new file mode 100644 index 0000000000..28ab1c9c1e --- /dev/null +++ b/utils/agentic/aggregation/stability_metrics.py @@ -0,0 +1,298 @@ +"""Within-run stability diagnostics for AgentX request records.""" + +from __future__ import annotations + +import math +from collections import Counter, defaultdict +from typing import Any + +from .aggregation_common import percentile, to_float + + +OBSERVED_WINDOW_SECONDS = 600.0 +CONVERGENCE_CHECKPOINT_SECONDS = 300.0 +CONVERGENCE_TOLERANCE_RATIO = 0.05 +CONVERGENCE_MIN_CONFIRMATION_SECONDS = 1200.0 +_PERCENTILES = (75, 90) + + +def _metric_value(record: dict[str, Any], key: str) -> float | None: + metric = record.get("metrics", {}).get(key) + if isinstance(metric, dict): + metric = metric.get("value") + return to_float(metric) + + +def _timestamp_ns(record: dict[str, Any]) -> int | None: + metadata = record.get("metadata", {}) + value = metadata.get("credit_issued_ns") + if ( + isinstance(value, bool) + or not isinstance(value, int | float) + or not math.isfinite(value) + ): + value = metadata.get("request_start_ns") + if isinstance(value, bool) or not isinstance(value, int | float): + return None + if not math.isfinite(value): + return None + return int(value) + + +def _source_trajectory(record: dict[str, Any]) -> str: + metadata = record.get("metadata", {}) + source = metadata.get("source_trace_id") or metadata.get("conversation_id") + return str(source) if source not in (None, "") else "unknown" + + +def _range(values: list[float]) -> dict[str, float] | None: + if not values: + return None + return {"min": min(values), "max": max(values)} + + +def _latency_ranges( + blocks: dict[int, list[dict[str, Any]]], +) -> dict[str, dict[str, dict[str, float] | None]]: + metric_keys = { + "ttft": "time_to_first_token", + "e2el": "request_latency", + "itl": "inter_token_latency", + } + block_percentiles: dict[str, dict[int, list[float]]] = { + name: {p: [] for p in _PERCENTILES} for name in metric_keys + } + + for records in blocks.values(): + for name, key in metric_keys.items(): + values_ms = [ + value + for record in records + if (value := _metric_value(record, key)) is not None and value > 0 + ] + if not values_ms: + continue + values_s = [value / 1000.0 for value in values_ms] + for p in _PERCENTILES: + block_percentiles[name][p].append(percentile(values_s, p)) + + output: dict[str, dict[str, dict[str, float] | None]] = { + "ttft": {}, + "e2el": {}, + "intvty": {}, + } + for p in _PERCENTILES: + output["ttft"][f"p{p}"] = _range(block_percentiles["ttft"][p]) + output["e2el"][f"p{p}"] = _range(block_percentiles["e2el"][p]) + itl = block_percentiles["itl"][p] + output["intvty"][f"p{p}"] = _range([1.0 / value for value in itl if value > 0]) + return output + + +def _convergence_summary( + checkpoints: list[dict[str, float | int]], + *, + tolerance_ratio: float, + min_confirmation_seconds: float, +) -> dict[str, float | int] | None: + """Return the earliest retrospectively stable suffix of a checkpoint series. + + The ratio test is performed in log space. This makes the tolerance + symmetric under reciprocals: a latency series and its interactivity + reciprocal stabilize at the same checkpoint. + """ + if not checkpoints: + return None + + final_value = float(checkpoints[-1]["value"]) + horizon_seconds = float(checkpoints[-1]["seconds"]) + if not math.isfinite(final_value) or final_value <= 0: + return None + + log_tolerance = math.log1p(tolerance_ratio) + boundary_epsilon = 1e-12 + for index, checkpoint in enumerate(checkpoints): + checkpoint_seconds = float(checkpoint["seconds"]) + if horizon_seconds - checkpoint_seconds < min_confirmation_seconds: + continue + + suffix = checkpoints[index:] + values = [float(item["value"]) for item in suffix] + if any(not math.isfinite(value) or value <= 0 for value in values): + continue + if any( + abs(math.log(value / final_value)) > log_tolerance + boundary_epsilon + for value in values + ): + continue + + return { + "time_seconds": checkpoint["seconds"], + "requests": checkpoint["requests"], + "min": min(values), + "max": max(values), + "max_relative_deviation": max( + abs(value / final_value - 1.0) for value in values + ), + } + return None + + +def _cumulative_convergence( + timed: list[tuple[int, dict[str, Any]]], + *, + horizon_seconds: float, + checkpoint_seconds: float, + tolerance_ratio: float, + min_confirmation_seconds: float, +) -> dict[str, dict[str, dict[str, float | int] | None]]: + """Compute retrospective convergence from cumulative request prefixes.""" + output: dict[str, dict[str, dict[str, float | int] | None]] = { + "ttft": {f"p{p}": None for p in _PERCENTILES}, + "e2el": {f"p{p}": None for p in _PERCENTILES}, + "intvty": {f"p{p}": None for p in _PERCENTILES}, + } + if not timed or horizon_seconds <= 0: + return output + + origin_ns = min(timestamp for timestamp, _ in timed) + sorted_timed = sorted(timed, key=lambda item: item[0]) + metric_keys = { + "ttft": "time_to_first_token", + "e2el": "request_latency", + "itl": "inter_token_latency", + } + values: dict[str, list[float]] = {name: [] for name in metric_keys} + series: dict[str, dict[int, list[dict[str, float | int]]]] = { + "ttft": {p: [] for p in _PERCENTILES}, + "e2el": {p: [] for p in _PERCENTILES}, + "intvty": {p: [] for p in _PERCENTILES}, + } + + cursor = 0 + checkpoint_count = int(horizon_seconds // checkpoint_seconds) + for checkpoint_index in range(1, checkpoint_count + 1): + elapsed_seconds = checkpoint_index * checkpoint_seconds + cutoff_ns = origin_ns + elapsed_seconds * 1e9 + while cursor < len(sorted_timed) and sorted_timed[cursor][0] < cutoff_ns: + timestamp, record = sorted_timed[cursor] + if timestamp >= origin_ns: + for name, key in metric_keys.items(): + value_ms = _metric_value(record, key) + if value_ms is not None and value_ms > 0: + values[name].append(value_ms / 1000.0) + cursor += 1 + + for p in _PERCENTILES: + for name in ("ttft", "e2el"): + if values[name]: + series[name][p].append( + { + "seconds": elapsed_seconds, + "requests": len(values[name]), + "value": percentile(values[name], p), + } + ) + if values["itl"]: + itl_percentile = percentile(values["itl"], p) + if itl_percentile > 0: + series["intvty"][p].append( + { + "seconds": elapsed_seconds, + "requests": len(values["itl"]), + "value": 1.0 / itl_percentile, + } + ) + + for name in output: + for p in _PERCENTILES: + checkpoint_series = series[name][p] + output[name][f"p{p}"] = _convergence_summary( + checkpoint_series, + tolerance_ratio=tolerance_ratio, + min_confirmation_seconds=min_confirmation_seconds, + ) + return output + + +def compute_stability_metrics( + records: list[dict[str, Any]], + *, + benchmark_duration_s: float | None, + window_seconds: float = OBSERVED_WINDOW_SECONDS, + checkpoint_seconds: float = CONVERGENCE_CHECKPOINT_SECONDS, + tolerance_ratio: float = CONVERGENCE_TOLERANCE_RATIO, + min_confirmation_seconds: float = CONVERGENCE_MIN_CONFIRMATION_SECONDS, +) -> dict[str, Any]: + """Summarize workload coverage, window drift, and cumulative convergence. + + These are retrospective diagnostics for one run. They deliberately do not + claim to be confidence intervals or estimates of rerun reproducibility. + """ + if window_seconds <= 0: + raise ValueError("window_seconds must be positive") + if checkpoint_seconds <= 0: + raise ValueError("checkpoint_seconds must be positive") + if tolerance_ratio < 0: + raise ValueError("tolerance_ratio must be non-negative") + if min_confirmation_seconds < 0: + raise ValueError("min_confirmation_seconds must be non-negative") + + timed: list[tuple[int, dict[str, Any]]] = [] + for record in records: + timestamp = _timestamp_ns(record) + if timestamp is not None: + timed.append((timestamp, record)) + trajectory_counts = Counter(_source_trajectory(record) for record in records) + n_requests = sum(trajectory_counts.values()) + squared = sum(count * count for count in trajectory_counts.values()) + kish_effective = n_requests * n_requests / squared if squared > 0 else 0.0 + largest_share = ( + max(trajectory_counts.values(), default=0) / n_requests if n_requests else 0.0 + ) + + configured_duration = to_float(benchmark_duration_s) + expected_windows = ( + int(configured_duration // window_seconds) + if configured_duration is not None and configured_duration > 0 + else 0 + ) + convergence_horizon = ( + math.floor(configured_duration / checkpoint_seconds) * checkpoint_seconds + if configured_duration is not None and configured_duration > 0 + else 0.0 + ) + + blocks: dict[int, list[dict[str, Any]]] = defaultdict(list) + if timed and expected_windows > 0: + origin_ns = min(timestamp for timestamp, _ in timed) + window_ns = window_seconds * 1e9 + for timestamp, record in timed: + block = int((timestamp - origin_ns) // window_ns) + if 0 <= block < expected_windows: + blocks[block].append(record) + + block_sizes = [len(block) for block in blocks.values()] + return { + "window_seconds": window_seconds, + "expected_window_count": expected_windows, + "observed_window_count": len(blocks), + "min_window_requests": min(block_sizes) if block_sizes else 0, + "root_trajectory_count": len(trajectory_counts), + "root_trajectory_kish_effective_count": kish_effective, + "root_trajectory_largest_share": largest_share, + "observed_ranges": _latency_ranges(blocks) if len(blocks) >= 2 else {}, + "convergence": { + "checkpoint_seconds": checkpoint_seconds, + "tolerance_ratio": tolerance_ratio, + "min_confirmation_seconds": min_confirmation_seconds, + "horizon_seconds": convergence_horizon, + "metrics": _cumulative_convergence( + timed, + horizon_seconds=convergence_horizon, + checkpoint_seconds=checkpoint_seconds, + tolerance_ratio=tolerance_ratio, + min_confirmation_seconds=min_confirmation_seconds, + ), + }, + } diff --git a/utils/agentic/aggregation/test_process_agentic_result.py b/utils/agentic/aggregation/test_process_agentic_result.py index 21a7307630..9e5ffa363a 100644 --- a/utils/agentic/aggregation/test_process_agentic_result.py +++ b/utils/agentic/aggregation/test_process_agentic_result.py @@ -122,7 +122,7 @@ "computed", "raw", } -REQUEST_METRICS_KEYS = {"qps", "latency", "tokens", "throughput", "cache"} +REQUEST_METRICS_KEYS = {"qps", "latency", "tokens", "throughput", "cache", "stability"} REQUEST_LATENCY_KEYS = {"ttft", "e2el", "itl", "tpot", "intvty"} REQUEST_TOKEN_KEYS = {"input", "output_actual", "output_expected"} REQUEST_THROUGHPUT_KEYS = { @@ -133,6 +133,24 @@ "per_gpu", } REQUEST_CACHE_KEYS = {"theoretical_cache_hit_rate"} +REQUEST_STABILITY_KEYS = { + "window_seconds", + "expected_window_count", + "observed_window_count", + "min_window_requests", + "root_trajectory_count", + "root_trajectory_kish_effective_count", + "root_trajectory_largest_share", + "observed_ranges", + "convergence", +} +REQUEST_CONVERGENCE_KEYS = { + "checkpoint_seconds", + "tolerance_ratio", + "min_confirmation_seconds", + "horizon_seconds", + "metrics", +} def _assert_stable_server_metrics_schema(agg: dict) -> None: @@ -153,6 +171,8 @@ def _assert_stable_request_metrics_schema(agg: dict) -> None: assert set(request_metrics["tokens"]) == REQUEST_TOKEN_KEYS assert set(request_metrics["throughput"]) == REQUEST_THROUGHPUT_KEYS assert set(request_metrics["cache"]) == REQUEST_CACHE_KEYS + assert set(request_metrics["stability"]) == REQUEST_STABILITY_KEYS + assert set(request_metrics["stability"]["convergence"]) == REQUEST_CONVERGENCE_KEYS def _flat_request_keys(result_dir: Path) -> set[str]: @@ -288,7 +308,7 @@ def _write_fixture(tmp_path: Path) -> Path: json.dump( { "request_count": len(records), - "benchmark_duration": 4.1, + "benchmark_duration": {"avg": 4.1, "unit": "sec"}, "request_latency": {"avg": 1090.0, "unit": "ms"}, "metadata": { "dataset": { @@ -452,6 +472,50 @@ def test_processor_latency_units_are_seconds(tmp_path: Path): assert "p90_itl" not in agg +def test_request_metrics_reads_aiperf_benchmark_duration_shape() -> None: + records = [ + _make_record( + conv_id="trace-A", + turn_index=0, + isl=100, + osl=20, + ttft_ms=100, + e2e_ms=1000, + itl_ms=20, + start_ns=1_000_000_000, + end_ns=2_000_000_000, + ), + _make_record( + conv_id="trace-B", + turn_index=0, + isl=100, + osl=20, + ttft_ms=200, + e2e_ms=2000, + itl_ms=40, + start_ns=602_000_000_000, + end_ns=604_000_000_000, + ), + ] + + _, nested = compute_request_metrics( + records, + { + # The observed wall time can land just below a clean boundary; + # fixed-window count must follow the configured run duration. + "benchmark_duration": {"avg": 1199.9, "unit": "sec"}, + "input_config": {"loadgen": {"benchmark_duration": 1200}}, + }, + ) + + stability = nested["stability"] + assert stability["expected_window_count"] == 2 + assert stability["observed_window_count"] == 2 + assert stability["observed_ranges"]["ttft"]["p90"] == pytest.approx( + {"min": 0.1, "max": 0.2} + ) + + def test_processor_derives_interactivity_from_matching_itl_percentile( tmp_path: Path, ): diff --git a/utils/agentic/aggregation/test_stability_metrics.py b/utils/agentic/aggregation/test_stability_metrics.py new file mode 100644 index 0000000000..37caf2646f --- /dev/null +++ b/utils/agentic/aggregation/test_stability_metrics.py @@ -0,0 +1,281 @@ +"""Tests for descriptive AgentX fixed-window diagnostics.""" + +from __future__ import annotations + +import pytest + +from .stability_metrics import _convergence_summary, compute_stability_metrics + + +def _record( + *, + second: float, + source: str, + ttft_ms: float, + e2e_ms: float, + itl_ms: float, +) -> dict: + return { + "metadata": { + "credit_issued_ns": int(second * 1e9), + "request_start_ns": int(second * 1e9), + "source_trace_id": source, + "conversation_id": f"replay-{source}", + }, + "metrics": { + "time_to_first_token": {"value": ttft_ms}, + "request_latency": {"value": e2e_ms}, + "inter_token_latency": {"value": itl_ms}, + }, + } + + +def test_computes_non_overlapping_ranges_and_root_coverage() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=10, source="root-a", ttft_ms=200, e2e_ms=2000, itl_ms=40), + _record(second=610, source="root-b", ttft_ms=300, e2e_ms=3000, itl_ms=50), + _record(second=620, source="root-b", ttft_ms=400, e2e_ms=4000, itl_ms=100), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=1200) + + assert result["window_seconds"] == 600 + assert result["expected_window_count"] == 2 + assert result["observed_window_count"] == 2 + assert result["min_window_requests"] == 2 + assert result["root_trajectory_count"] == 2 + assert result["root_trajectory_kish_effective_count"] == pytest.approx(2) + assert result["root_trajectory_largest_share"] == pytest.approx(0.5) + assert result["observed_ranges"]["ttft"]["p90"] == pytest.approx( + {"min": 0.19, "max": 0.39} + ) + assert result["observed_ranges"]["e2el"]["p75"] == pytest.approx( + {"min": 1.75, "max": 3.75} + ) + # Per-window p90 ITL is 38 ms and 95 ms. Interactivity inverts each + # latency estimate, so the slower window is the lower bound. + assert result["observed_ranges"]["intvty"]["p90"] == pytest.approx( + {"min": 1 / 0.095, "max": 1 / 0.038} + ) + assert result["convergence"] == { + "checkpoint_seconds": 300, + "tolerance_ratio": 0.05, + "min_confirmation_seconds": 1200, + "horizon_seconds": 1200, + "metrics": { + "ttft": {"p75": None, "p90": None}, + "e2el": {"p75": None, "p90": None}, + "intvty": {"p75": None, "p90": None}, + }, + } + + +def test_subagent_requests_share_their_root_trajectory_cluster() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=1, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=2, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=601, source="root-b", ttft_ms=100, e2e_ms=1000, itl_ms=20), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=1200) + + assert result["root_trajectory_count"] == 2 + assert result["root_trajectory_kish_effective_count"] == pytest.approx(16 / 10) + assert result["root_trajectory_largest_share"] == pytest.approx(0.75) + + +def test_omits_ranges_when_run_has_fewer_than_two_observed_windows() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=10, source="root-a", ttft_ms=200, e2e_ms=2000, itl_ms=40), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=600) + + assert result["expected_window_count"] == 1 + assert result["observed_window_count"] == 1 + assert result["observed_ranges"] == {} + + +def test_excludes_records_beyond_the_configured_measurement_windows() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=601, source="root-b", ttft_ms=200, e2e_ms=2000, itl_ms=40), + _record(second=1201, source="root-c", ttft_ms=9999, e2e_ms=9999, itl_ms=9999), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=1200) + + assert result["expected_window_count"] == 2 + assert result["observed_window_count"] == 2 + assert result["observed_ranges"]["ttft"]["p90"]["max"] == pytest.approx(0.2) + + +def test_falls_back_to_request_start_when_credit_timestamp_is_null() -> None: + first = _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20) + second = _record(second=601, source="root-b", ttft_ms=200, e2e_ms=2000, itl_ms=40) + first["metadata"]["credit_issued_ns"] = None + second["metadata"]["credit_issued_ns"] = None + + result = compute_stability_metrics([first, second], benchmark_duration_s=1200) + + assert result["observed_window_count"] == 2 + assert result["observed_ranges"]["ttft"]["p90"] == pytest.approx( + {"min": 0.1, "max": 0.2} + ) + + +def test_rejects_non_positive_window_size() -> None: + with pytest.raises(ValueError, match="window_seconds must be positive"): + compute_stability_metrics([], benchmark_duration_s=3600, window_seconds=0) + + +def test_reports_insufficient_duration_without_stabilization() -> None: + result = compute_stability_metrics( + [_record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20)], + benchmark_duration_s=1200, + ) + + assert result["convergence"]["horizon_seconds"] == 1200 + assert result["convergence"]["metrics"]["ttft"]["p90"] is None + + +def test_no_checkpoint_qualifies_when_a_late_prefix_leaves_the_band() -> None: + checkpoints = [ + {"seconds": 300, "requests": 10, "value": 100.0}, + {"seconds": 600, "requests": 20, "value": 100.0}, + {"seconds": 900, "requests": 30, "value": 100.0}, + {"seconds": 1200, "requests": 40, "value": 100.0}, + {"seconds": 1500, "requests": 50, "value": 106.0}, + {"seconds": 1800, "requests": 60, "value": 100.0}, + ] + + assert ( + _convergence_summary( + checkpoints, + tolerance_ratio=0.05, + min_confirmation_seconds=1200, + ) + is None + ) + + +@pytest.mark.parametrize("boundary_value", [105.0, 100.0 / 1.05]) +def test_exact_symmetric_five_percent_boundary_qualifies(boundary_value: float) -> None: + checkpoints = [ + {"seconds": 300, "requests": 10, "value": boundary_value}, + {"seconds": 600, "requests": 20, "value": 100.0}, + {"seconds": 900, "requests": 30, "value": 100.0}, + {"seconds": 1200, "requests": 40, "value": 100.0}, + {"seconds": 1500, "requests": 50, "value": 100.0}, + ] + + result = _convergence_summary( + checkpoints, + tolerance_ratio=0.05, + min_confirmation_seconds=1200, + ) + + assert result is not None + assert result["time_seconds"] == 300 + + +def test_confirmation_period_rejects_an_otherwise_stable_late_checkpoint() -> None: + checkpoints = [ + {"seconds": 300, "requests": 10, "value": 80.0}, + {"seconds": 600, "requests": 20, "value": 100.0}, + {"seconds": 900, "requests": 30, "value": 100.0}, + {"seconds": 1200, "requests": 40, "value": 100.0}, + {"seconds": 1500, "requests": 50, "value": 100.0}, + ] + + assert ( + _convergence_summary( + checkpoints, + tolerance_ratio=0.05, + min_confirmation_seconds=1200, + ) + is None + ) + + +def test_reciprocal_series_stabilizes_at_the_same_checkpoint() -> None: + latency = [ + {"seconds": 300, "requests": 10, "value": 0.08}, + {"seconds": 600, "requests": 20, "value": 0.105}, + {"seconds": 900, "requests": 30, "value": 0.102}, + {"seconds": 1200, "requests": 40, "value": 0.101}, + {"seconds": 1500, "requests": 50, "value": 0.1}, + {"seconds": 1800, "requests": 60, "value": 0.1}, + ] + interactivity = [{**item, "value": 1 / float(item["value"])} for item in latency] + + latency_result = _convergence_summary( + latency, tolerance_ratio=0.05, min_confirmation_seconds=1200 + ) + interactivity_result = _convergence_summary( + interactivity, tolerance_ratio=0.05, min_confirmation_seconds=1200 + ) + + assert latency_result is not None + assert interactivity_result is not None + assert latency_result["time_seconds"] == interactivity_result["time_seconds"] == 600 + assert interactivity_result["min"] == pytest.approx( + 1 / float(latency_result["max"]) + ) + assert interactivity_result["max"] == pytest.approx( + 1 / float(latency_result["min"]) + ) + + +def test_excludes_records_at_and_beyond_the_complete_convergence_horizon() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=299, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=300, source="root-b", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=1500, source="root-c", ttft_ms=9999, e2e_ms=9999, itl_ms=9999), + _record(second=1600, source="root-d", ttft_ms=9999, e2e_ms=9999, itl_ms=9999), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=1799) + p90_ttft = result["convergence"]["metrics"]["ttft"]["p90"] + + assert result["convergence"]["horizon_seconds"] == 1500 + assert p90_ttft is not None + assert p90_ttft["time_seconds"] == 300 + assert p90_ttft["requests"] == 2 + assert p90_ttft["min"] == pytest.approx(0.1) + assert p90_ttft["max"] == pytest.approx(0.1) + + +def test_convergence_request_count_only_includes_metric_samples() -> None: + records = [ + _record(second=0, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=10, source="root-a", ttft_ms=100, e2e_ms=1000, itl_ms=0), + _record(second=300, source="root-b", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=600, source="root-c", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=900, source="root-d", ttft_ms=100, e2e_ms=1000, itl_ms=20), + _record(second=1200, source="root-e", ttft_ms=100, e2e_ms=1000, itl_ms=20), + ] + + result = compute_stability_metrics(records, benchmark_duration_s=1500) + p90_intvty = result["convergence"]["metrics"]["intvty"]["p90"] + + assert p90_intvty is not None + assert p90_intvty["time_seconds"] == 300 + assert p90_intvty["requests"] == 1 + + +def test_rejects_invalid_convergence_parameters() -> None: + with pytest.raises(ValueError, match="checkpoint_seconds must be positive"): + compute_stability_metrics([], benchmark_duration_s=3600, checkpoint_seconds=0) + with pytest.raises(ValueError, match="tolerance_ratio must be non-negative"): + compute_stability_metrics([], benchmark_duration_s=3600, tolerance_ratio=-0.01) + with pytest.raises( + ValueError, match="min_confirmation_seconds must be non-negative" + ): + compute_stability_metrics( + [], benchmark_duration_s=3600, min_confirmation_seconds=-1 + )