From 89fc597c1864735494e1fab3a7d37bffcde34e00 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 16:44:08 +0000 Subject: [PATCH 1/6] fix(results): qualify ATOM cache counters against profiling requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 保留 ATOM 窗口与请求范围合同，复用 counter 修复并隔离诊断数据。 --- .../infx/results/agentic/__init__.py | 4 +- .../infx/results/agentic/artifacts.py | 3 + .../infx/results/agentic/atom_validity.py | 322 ++++++++++++++++ .../infx/results/agentic/backends/atom.py | 51 +++ .../infx/results/agentic/backends/base.py | 1 + .../infx/results/agentic/server_metrics.py | 74 +++- .../agentic/test_atom_cache_validity.py | 350 ++++++++++++++++++ .../agentic/test_process_agentic_result.py | 3 + 8 files changed, 803 insertions(+), 5 deletions(-) create mode 100644 inferencex-e2e/infx/results/agentic/atom_validity.py create mode 100644 inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py diff --git a/inferencex-e2e/infx/results/agentic/__init__.py b/inferencex-e2e/infx/results/agentic/__init__.py index 92a78f8584..cc33acbbba 100644 --- a/inferencex-e2e/infx/results/agentic/__init__.py +++ b/inferencex-e2e/infx/results/agentic/__init__.py @@ -9,7 +9,7 @@ from infx.results.topology import Parallelism, validate_parallelism from .request_metrics import compute_request_metrics -from .server_metrics import compute_server_metrics +from .server_metrics import compute_server_metrics, request_scope_from_records def _env_int(env: Mapping[str, str], name: str, default: int = 0) -> int: @@ -167,6 +167,7 @@ def build_result( "records_dropped_total": 0, "records_warmup_dropped": 0, "records_error_dropped": 0, + "records_profiling_error_dropped": 0, "error_categories": {}, } @@ -219,6 +220,7 @@ def build_result( records=records, server_logs=server_logs, env=env, + request_scope=request_scope_from_records(records, request_accounting), ) if "total_tput_tps" in request_flat and num_gpus > 0: diff --git a/inferencex-e2e/infx/results/agentic/artifacts.py b/inferencex-e2e/infx/results/agentic/artifacts.py index a35ee829ae..ef072d274a 100644 --- a/inferencex-e2e/infx/results/agentic/artifacts.py +++ b/inferencex-e2e/infx/results/agentic/artifacts.py @@ -45,6 +45,7 @@ def load_records_with_accounting( "records_dropped_total": 0, "records_warmup_dropped": 0, "records_error_dropped": 0, + "records_profiling_error_dropped": 0, "error_categories": {}, } with open(path) as f: @@ -61,6 +62,8 @@ def load_records_with_accounting( accounting["records_warmup_dropped"] += 1 if error: accounting["records_error_dropped"] += 1 + if not is_warmup: + accounting["records_profiling_error_dropped"] += 1 category = _error_category(error) categories = accounting["error_categories"] categories[category] = categories.get(category, 0) + 1 diff --git a/inferencex-e2e/infx/results/agentic/atom_validity.py b/inferencex-e2e/infx/results/agentic/atom_validity.py new file mode 100644 index 0000000000..f1f349593e --- /dev/null +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -0,0 +1,322 @@ +"""Qualify ATOM cache counters against the successful profiling cohort.""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from itertools import pairwise +from typing import Any + +from .common import metric_series, series_stat + +FULL = "prefix_cache_full_tokens" +CACHED = "prefix_cache_cached_tokens" +OFFLOAD = "prefix_cache_offload_tokens" +LOADED = "lmcache_loaded_tokens" +REQUESTS = "prefix_cache_requests" +_COUNTERS = (FULL, CACHED, OFFLOAD, LOADED, REQUESTS) +_CORE = (FULL, CACHED, OFFLOAD) + + +def _series(section: Mapping[str, Any], stem: str) -> tuple[dict[tuple, dict], set[str | None]]: + names = [ + f"{prefix}{stem}{suffix}" for prefix in ("atom:", "atom_") for suffix in ("", "_total") + ] + found: dict[tuple, dict] = {} + duplicates: set[str | None] = set() + for series in metric_series(dict(section), names): + url, labels = series.get("endpoint_url"), series.get("labels", {}) + if not isinstance(url, str) or not url or not isinstance(labels, dict): + duplicates.add(None) + continue + key = (url, tuple(sorted((str(k), str(v)) for k, v in labels.items()))) + if key in found: + duplicates.add(url) + found[key] = series + return found, duplicates + + +def _total(series: dict[str, Any]) -> float | None: + stats = series.get("stats") + raw = stats.get("total") if isinstance(stats, dict) else None + value = series_stat(series, preferred_keys=("total",)) + return ( + value + if not isinstance(raw, bool) and value is not None and math.isfinite(value) and value >= 0 + else None + ) + + +def _window_total( + series: dict[str, Any], window: tuple[int, int], warmup: dict[str, Any] | None +) -> tuple[float | None, bool]: + slices = series.get("timeslices") + if not isinstance(slices, list) or not slices: + return None, False + parsed = [] + for item in slices: + if not isinstance(item, dict): + return None, False + lo, hi, value = item.get("start_ns"), item.get("end_ns"), item.get("total") + if ( + type(lo) is not int + or type(hi) is not int + or hi <= lo + or isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + ): + return None, False + parsed.append((lo, hi, value)) + parsed.sort() + if any(b[0] < a[1] for a, b in pairwise(parsed)): + return None, False + start, end = window + covering = [item for item in parsed if item[1] > start and item[0] < end] + if ( + not covering + or covering[0][0] > start + or covering[-1][1] < end + or any(b[0] != a[1] for a, b in pairwise(covering)) + ): + return None, False + before = sum(value for lo, hi, value in parsed if hi <= start) + start_straddle = sum(value for lo, hi, value in parsed if lo < start < hi) + end_straddle = sum(value for lo, hi, value in parsed if lo < end < hi) + warm = _total(warmup) if warmup is not None else None + boundary = end_straddle > 0 or (start_straddle > 0 and before + start_straddle != warm) + return float(sum(value for lo, hi, value in parsed if start <= lo and hi <= end)), boundary + + +def _modern_export(export: Mapping[str, Any], endpoints: list[str]) -> bool: + summary = export.get("summary") or {} + return ( + isinstance(summary, dict) + and export.get("schema_version") == "1.1" + and export.get("aiperf_version") == "1.0.12" + and export.get("metrics_phase") == "profiling" + and all( + isinstance(summary.get(key), list) + and all(isinstance(url, str) for url in summary[key]) + and set(endpoints) <= set(summary[key]) + for key in ("endpoints_configured", "endpoints_successful") + ) + ) + + +def account_cache( + metrics: dict[str, dict[str, Any]], + *, + framework: str, + endpoint_roles: Mapping[str, str] | None, + profiling_window: tuple[int, int] | None, + warmup_metrics: Mapping[str, Any] | None = None, + request_scope: Mapping[str, Any] | None = None, + disagg: bool | None = None, + kv_offloading: str | None = None, + collector_export: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Recover legacy windows or use phase-scoped totals, then prove request conservation. + + Unqualified cumulative telemetry remains the counter adapter's responsibility. The + internal ``replace_telemetry`` flag only selects a proven window or rejects ambiguity. + """ + result: dict[str, Any] = dict.fromkeys( + ("gpu", "cpu", "overall", "full", "cached", "offload", "loaded", "requests") + ) + reasons: list[str] = [] + validity: dict[str, Any] = { + "schema": "atom-cache-validity/1", + "status": None, + "pricing_eligible": False, + "observed_scope": None, + "requested_scope": "profiling_successful_requests", + "cpu_component": None, + "roles_source": None, + "accounted_endpoints": [], + "window": None, + "request_scope": None, + "diagnostics": {}, + "reasons": reasons, + } + result.update(validity=validity, warnings=[], replace_telemetry=False) + + def refuse(status: str, reason: str) -> dict[str, Any]: + validity["status"] = status + reasons.append(reason) + result["warnings"].append(f"atom cache accounting: {reason}") + return result + + series, duplicates = {}, {} + for stem in _COUNTERS: + series[stem], duplicates[stem] = _series(metrics, stem) + observed = {key[0] for stem in _CORE for key in series[stem]} + if not series[FULL]: + return refuse("MISSING_OBSERVATION", "no prefix_cache_full_tokens series") + if endpoint_roles is None and disagg is False and len(observed) == 1: + accounted = sorted(observed) + validity["roles_source"] = "single_endpoint_non_disagg" + elif endpoint_roles: + accounted = sorted( + url for url, role in endpoint_roles.items() if role in {"prefill", "unified"} + ) + if observed - endpoint_roles.keys() or not accounted: + return refuse("ROLE_OR_COVERAGE_UNRESOLVED", f"{framework}: endpoint roles incomplete") + validity["roles_source"] = "launcher_env" + else: + return refuse("ROLE_OR_COVERAGE_UNRESOLVED", f"{framework}: endpoint roles unresolved") + validity["accounted_endpoints"] = accounted + series = { + stem: {key: s for key, s in group.items() if key[0] in accounted} + for stem, group in series.items() + } + duplicates = {stem: bad & {*accounted, None} for stem, bad in duplicates.items()} + if any(duplicates[stem] for stem in (FULL, CACHED)): + result["replace_telemetry"] = True + return refuse("INVALID_COUNTER", "duplicate or unidentifiable cache counter series") + keys = set(series[FULL]) + if {key[0] for key in keys} != set(accounted) or not keys or set(series[CACHED]) != keys: + return refuse("MISSING_OBSERVATION", "full/cached series coverage differs") + scoped = {stem: {key: _total(s) for key, s in group.items()} for stem, group in series.items()} + raw = { + stem: sum(values) if values and all(value is not None for value in values) else None + for stem in _COUNTERS + for values in [list(scoped[stem].values())] + } + validity["diagnostics"]["cumulative_stats_total"] = { + f"atom:{stem}": value for stem, value in raw.items() if value is not None + } + if raw[FULL] is None or raw[CACHED] is None: + result["replace_telemetry"] = True + return refuse("INVALID_COUNTER", "full/cached stats.total missing or invalid") + cpu_missing = not series[OFFLOAD] + cpu_complete = ( + set(series[OFFLOAD]) == keys and raw[OFFLOAD] is not None and not duplicates[OFFLOAD] + ) + cpu_invalid = bool(duplicates[OFFLOAD]) or (bool(series[OFFLOAD]) and raw[OFFLOAD] is None) + if profiling_window is None: + return refuse("WINDOW_UNRESOLVED", "no summary.phase_time_ranges.profiling window") + totals = dict(raw) + boundary = False + modern = _modern_export(collector_export or {}, accounted) + if not modern: + for stem in _COUNTERS: + if raw[stem] is None or (stem == OFFLOAD and not cpu_complete): + totals[stem] = None + continue + warm, warm_duplicate = _series(warmup_metrics or {}, stem) + windows = { + key: _window_total(s, profiling_window, None if warm_duplicate else warm.get(key)) + for key, s in series[stem].items() + } + values = list(windows.values()) + if any(value is None for value, _ in values): + if stem in (FULL, CACHED) or (stem == OFFLOAD and cpu_complete): + return refuse( + "WINDOW_UNRESOLVED", f"{stem}: missing, invalid or gapped timeslices" + ) + totals[stem] = None + else: + scoped[stem] = {key: value for key, (value, _) in windows.items()} + totals[stem] = sum(value for value, _ in values) + if stem in _CORE: + boundary |= any(issue for _, issue in values) + validity["observed_scope"] = ( + "profiling_phase_counter_totals" if modern else "profiling_window_interior_timeslices" + ) + validity["window"] = { + "start_ns": profiling_window[0], + "end_ns": profiling_window[1], + "boundary_issues": ["activity crosses the profiling boundary"] if boundary else [], + } + result["replace_telemetry"] = True + full, cached, offload = totals[FULL], totals[CACHED], totals[OFFLOAD] + if ( + full is None + or not math.isfinite(full) + or full <= 0 + or cached is None + or not 0 <= cached <= full + ): + return refuse("INVALID_COUNTER", "full/cached tokens outside their valid range") + for key in keys: + if scoped[CACHED][key] > scoped[FULL][key]: + return refuse("INVALID_COUNTER", "cached tokens exceed full tokens in a paired series") + if cpu_complete and scoped[CACHED][key] + scoped[OFFLOAD][key] > scoped[FULL][key]: + cpu_invalid = True + if cpu_missing and (kv_offloading or "").lower() == "none": + offload = 0.0 + validity["cpu_component"] = "NO_CPU_TIER_CONFIRMED" + elif cpu_complete and not cpu_invalid and offload is not None and cached + offload <= full: + validity["cpu_component"] = "MEASURED" + else: + offload = None + validity["cpu_component"] = "MISSING" + result.update( + full=full, + cached=cached, + offload=offload, + loaded=totals[LOADED], + requests=totals[REQUESTS], + gpu=cached / full, + ) + if offload is not None: + result.update(cpu=offload / full, overall=(cached + offload) / full) + scope = dict(request_scope or {}) + admissions = ( + totals[REQUESTS] if set(series[REQUESTS]) == keys and not duplicates[REQUESTS] else None + ) + scope_counts = [ + scope.get(key) for key in ("profiling_requests", "profiling_prompt_tokens", "error_records") + ] + if "cancelled_records" in scope: + scope_counts.append(scope["cancelled_records"]) + scope_valid = all( + not isinstance(value, bool) + and isinstance(value, (int, float)) + and math.isfinite(value) + and value >= 0 + and int(value) == value + for value in scope_counts + ) + conserved = ( + scope_valid + and admissions is not None + and scope.get("profiling_prompt_tokens") is not None + and admissions == scope.get("profiling_requests") + and full == scope["profiling_prompt_tokens"] + and scope.get("error_records") == 0 + and not scope.get("cancelled_records") + and (not modern or scope.get("cancelled_records") == 0) + ) + validity["request_scope"] = { + "server_window_admissions": admissions, + "server_window_prompt_tokens": full, + **scope, + "conserved": conserved, + } + if duplicates[LOADED] or totals[LOADED] is None: + result["loaded"] = None + validity["diagnostics"]["loaded_tokens_unavailable"] = True + if boundary: + return refuse( + "WINDOW_UNRESOLVED", "activity crosses the profiling boundary without warmup proof" + ) + if cpu_invalid or (offload is None and cpu_complete): + return refuse( + "INVALID_COUNTER", "offload tokens invalid or incompatible with full/cached tokens" + ) + if offload is None: + return refuse( + "PARTIAL_COMPONENT", "CPU reuse missing and no confirmed absence of a CPU tier" + ) + if not conserved: + return refuse( + "REQUEST_SCOPE_UNRESOLVED", + "server admissions/prompt tokens do not match successful profiling requests", + ) + validity.update( + status="VALID_ZERO" if cached + offload == 0 else "VALID_COMPLETE", pricing_eligible=True + ) + return result diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 7c6071adff..18ab536087 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -7,6 +7,14 @@ from collections.abc import Mapping from typing import Any +from infx.results.agentic.atom_validity import ( + CACHED, + FULL, + LOADED, + OFFLOAD, + REQUESTS, + account_cache, +) from infx.results.agentic.common import ( gauge_stat, metric_series, @@ -100,6 +108,7 @@ def _cache_counts( class AtomBackend(ServerMetricsBackend): name = "atom" + accepts_context = True def matches(self, metrics: dict[str, dict[str, Any]], framework: str) -> bool: metric_names = set(metrics) @@ -113,6 +122,7 @@ def populate( flat: dict[str, Any], nested: dict[str, Any], *, + context: Mapping[str, Any] | None = None, endpoint_roles: Mapping[str, str] | None = None, disaggregated: bool | None = None, ) -> None: @@ -129,9 +139,31 @@ def populate( flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) + context = context or {} + endpoint_roles = context.get("endpoint_roles", endpoint_roles) + disaggregated = context.get("disagg", disaggregated) full_tokens, cached_tokens, external_tokens = _cache_counts( metrics, endpoint_roles, disaggregated ) + accounting = account_cache( + metrics, + framework=str(context.get("framework") or "atom"), + endpoint_roles=endpoint_roles, + profiling_window=context.get("profiling_window"), + warmup_metrics=context.get("warmup_metrics"), + request_scope=context.get("request_scope"), + disagg=disaggregated, + kv_offloading=context.get("kv_offloading"), + collector_export=context.get("collector_export"), + ) + if accounting["replace_telemetry"]: + full_tokens, cached_tokens, external_tokens = ( + accounting[key] for key in ("full", "cached", "offload") + ) + nested["cache_validity"] = accounting["validity"] + sink = context.get("warnings") + if isinstance(sink, list): + sink.extend(accounting["warnings"]) cache_hit_rate = rate(cached_tokens, full_tokens) # Offload is CPU reuse beyond the GPU prefix; loaded tokens measure transfer volume. external_hit_rate = rate(external_tokens, full_tokens) @@ -169,6 +201,25 @@ def populate( } ) nested["kv_cache"]["gpu_usage_pct"] = flat["gpu_kv_cache_usage_pct"] + if accounting["replace_telemetry"]: + by_source = nested["tokens"]["prompt_by_source"] + if full_tokens is not None and external_tokens is not None: + by_source.update( + gpu_cache_hit=cached_tokens, + cpu_or_external_cache_hit=external_tokens, + computed=full_tokens - cached_tokens - external_tokens, + ) + by_source["raw"] = { + f"atom:{stem}": accounting[key] + for stem, key in ( + (FULL, "full"), + (CACHED, "cached"), + (OFFLOAD, "offload"), + (LOADED, "loaded"), + (REQUESTS, "requests"), + ) + if accounting[key] is not None + } nested["tokens"].update( { "prompt_total": flat["total_prompt_tokens"], diff --git a/inferencex-e2e/infx/results/agentic/backends/base.py b/inferencex-e2e/infx/results/agentic/backends/base.py index 427a4c1606..602cdfd6dc 100644 --- a/inferencex-e2e/infx/results/agentic/backends/base.py +++ b/inferencex-e2e/infx/results/agentic/backends/base.py @@ -51,6 +51,7 @@ def empty_server_metrics(present: bool, metric_count: int) -> dict[str, Any]: "present": present, "adapter": "none", "metric_count": metric_count, + "cache_validity": None, "cache": { "gpu_cache_hit_rate": None, "cpu_cache_hit_rate": None, diff --git a/inferencex-e2e/infx/results/agentic/server_metrics.py b/inferencex-e2e/infx/results/agentic/server_metrics.py index e100011642..9b2c690e5a 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -2,11 +2,11 @@ from __future__ import annotations +import math from collections.abc import Iterable, Mapping from typing import Any from .backends import detect_backend -from .backends.atom import AtomBackend from .backends.base import ( SERVER_CACHE_FLAT_FIELDS, apply_profile_totals, @@ -15,6 +15,57 @@ from .common import index_server_metrics +def request_scope_from_records( + records: list[dict[str, Any]], request_accounting: Mapping[str, Any] +) -> dict[str, Any]: + prompt = [] + for record in records: + metric = record.get("metrics", {}).get("usage_prompt_tokens") + value = metric.get("value") if isinstance(metric, dict) else metric + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + or int(value) != value + ): + break + prompt.append(int(value)) + return { + "profiling_requests": len(records), + "profiling_prompt_tokens": sum(prompt) if len(prompt) == len(records) else None, + "error_records": request_accounting.get( + "records_profiling_error_dropped", request_accounting.get("records_error_dropped") + ), + "cancelled_records": sum( + 1 for record in records if (record.get("metadata") or {}).get("was_cancelled") + ), + } + + +def deployment_from_env(env: Mapping[str, str]) -> dict[str, Any]: + offload = (env.get("KV_OFFLOADING") or "").strip().lower() or None + if offload == "none" and ( + env.get("KV_OFFLOAD_BACKEND") or env.get("KV_OFFLOAD_BACKEND_METADATA") + ): + offload = None + return { + "disagg": {"true": True, "false": False}.get((env.get("DISAGG") or "").lower()), + "kv_offloading": offload, + } + + +def _profiling_window(server_metrics: dict[str, Any]) -> tuple[int, int] | None: + summary = server_metrics.get("summary") + ranges = summary.get("phase_time_ranges") if isinstance(summary, dict) else None + profiling = ranges.get("profiling") if isinstance(ranges, dict) else None + if isinstance(profiling, dict): + start, end = profiling.get("start_ns"), profiling.get("end_ns") + if type(start) is int and type(end) is int and start < end: + return start, end + return None + + def endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: if env.get("AIPERF_SERVER_METRICS_URLS") or env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": return {} @@ -43,6 +94,9 @@ def compute_server_metrics( records: list[dict[str, Any]], server_logs: Iterable[str | None] = (), env: Mapping[str, str] | None = None, + endpoint_roles: Mapping[str, str] | None = None, + request_scope: Mapping[str, Any] | None = None, + deployment: Mapping[str, Any] | None = None, ) -> tuple[dict[str, Any], dict[str, Any], list[str]]: metrics = index_server_metrics(server_metrics) flat = SERVER_CACHE_FLAT_FIELDS.copy() @@ -69,14 +123,26 @@ def compute_server_metrics( f"framework={framework.lower()!r}, metric_names=[{preview}]" ) nested["adapter"] = adapter.name - if isinstance(adapter, AtomBackend): + if getattr(adapter, "accepts_context", False): source_env = env or {} + warmup = server_metrics.get("warmup_metrics") adapter.populate( metrics, flat, nested, - endpoint_roles=endpoint_roles_from_env(source_env), - disaggregated={"true": True, "false": False}.get(source_env.get("DISAGG", "").lower()), + context={ + "framework": framework, + "endpoint_roles": endpoint_roles + if endpoint_roles is not None + else endpoint_roles_from_env(source_env), + "profiling_window": _profiling_window(server_metrics), + "warmup_metrics": warmup if isinstance(warmup, dict) else None, + "request_scope": request_scope, + **deployment_from_env(source_env), + **dict(deployment or {}), + "collector_export": server_metrics, + "warnings": warnings, + }, ) else: adapter.populate(metrics, flat, nested) diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py new file mode 100644 index 0000000000..e87c2f2602 --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -0,0 +1,350 @@ +from __future__ import annotations + +import json +from copy import deepcopy + +import pytest + +from infx.results.agentic.artifacts import load_records_with_accounting +from infx.results.agentic.server_metrics import compute_server_metrics, request_scope_from_records + +P = "http://prefill/metrics" +D = "http://decode/metrics" +ROLES = {P: "prefill", D: "decode"} +SCOPE = { + "profiling_requests": 5, + "profiling_prompt_tokens": 1000, + "error_records": 0, + "cancelled_records": 0, +} + + +def _export(*, modern=False): + metrics = {} + for stem, value in ( + ("prefix_cache_full_tokens", 1000), + ("prefix_cache_cached_tokens", 600), + ("prefix_cache_offload_tokens", 100), + ("prefix_cache_requests", 5), + ("lmcache_loaded_tokens", 900), + ): + metrics[f"atom:{stem}"] = { + "series": [ + { + "endpoint_url": url, + "stats": {"total": total}, + "timeslices": [ + {"start_ns": 0, "end_ns": 1000, "total": 0}, + {"start_ns": 1000, "end_ns": 4000, "total": total}, + ], + } + for url, total in ( + (P, value), + (D, 1000 if stem == "prefix_cache_full_tokens" else 0), + ) + ] + } + export = { + "metrics": metrics, + "summary": { + "phase_time_ranges": {"profiling": {"start_ns": 1000, "end_ns": 4000}}, + "endpoints_configured": [P, D], + "endpoints_successful": [P, D], + }, + } + if modern: + export.update(schema_version="1.1", aiperf_version="1.0.12", metrics_phase="profiling") + for entry in metrics.values(): + for series in entry["series"]: + del series["timeslices"] + return export + + +def _run(export, *, scope=None, offload="dram"): + flat, nested, _ = compute_server_metrics( + export, + framework="mooncake-atom", + records=[], + endpoint_roles=ROLES, + request_scope=SCOPE if scope is None else scope, + deployment={"disagg": True, "kv_offloading": offload}, + ) + assert flat["server_gpu_cache_hit_rate"] == nested["cache"]["gpu_cache_hit_rate"] + return nested + + +def _rates(result, gpu, cpu, overall): + for key, expected in (("gpu", gpu), ("cpu", cpu), ("external", cpu), ("overall", overall)): + assert result["cache"][f"{key}_cache_hit_rate"] == ( + pytest.approx(expected) if expected is not None else None + ) + + +@pytest.mark.parametrize("modern", [False, True]) +def test_qualified_pd_preserves_the_old_detail_and_prompt_sources(modern): + result = _run(_export(modern=modern)) + _rates(result, 0.6, 0.1, 0.7) + detail = result["cache_validity"] + assert detail["schema"] == "atom-cache-validity/1" + assert detail["status"] == "VALID_COMPLETE" and detail["pricing_eligible"] is True + assert detail["accounted_endpoints"] == [P] + assert detail["request_scope"]["conserved"] is True + assert result["tokens"]["prompt_by_source"] == { + "gpu_cache_hit": 600, + "cpu_or_external_cache_hit": 100, + "computed": 300, + "raw": { + "atom:prefix_cache_full_tokens": 1000, + "atom:prefix_cache_cached_tokens": 600, + "atom:prefix_cache_offload_tokens": 100, + "atom:lmcache_loaded_tokens": 900, + "atom:prefix_cache_requests": 5, + }, + } + + +@pytest.mark.parametrize( + ("cached", "offload", "status"), [(0, 0, "VALID_ZERO"), (0, 100, "VALID_COMPLETE")] +) +def test_true_zero_and_cpu_only_are_eligible(cached, offload, status): + export = _export(modern=True) + for stem, value in (("cached", cached), ("offload", offload)): + export["metrics"][f"atom:prefix_cache_{stem}_tokens"]["series"][0]["stats"]["total"] = value + result = _run(export) + _rates(result, 0, offload / 1000, offload / 1000) + assert result["cache_validity"]["status"] == status + assert result["cache_validity"]["pricing_eligible"] is True + + +@pytest.mark.parametrize("missing", ["window", "slices", "gap"]) +def test_unproven_legacy_window_keeps_raw_telemetry_but_cannot_be_priced(missing): + export = _export() + series = export["metrics"]["atom:prefix_cache_full_tokens"]["series"][0] + if missing == "window": + del export["summary"] + elif missing == "slices": + del series["timeslices"] + else: + series["timeslices"][1]["start_ns"] = 1001 + result = _run(export) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "WINDOW_UNRESOLVED" + assert result["cache_validity"]["observed_scope"] is None + assert result["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize("proof", ["version", "phase", "endpoint", "window"]) +def test_modern_header_alone_does_not_prove_a_window(proof): + export = _export(modern=True) + if proof == "version": + export["aiperf_version"] = "1.0.11" + elif proof == "phase": + export["metrics_phase"] = "warmup" + elif proof == "endpoint": + export["summary"]["endpoints_successful"] = [D] + else: + export["summary"]["phase_time_ranges"] = {} + result = _run(export) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "WINDOW_UNRESOLVED" + assert result["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize("warmup_proof", [False, True]) +def test_positive_start_slice_requires_matching_warmup_counters(warmup_proof): + export = _export() + warmup = {} + for stem, value in (("full", 100), ("cached", 50), ("offload", 10)): + name = f"atom:prefix_cache_{stem}_tokens" + series = export["metrics"][name]["series"][0] + series["stats"]["total"] += value + series["timeslices"][0].update(end_ns=1500, total=value) + series["timeslices"][1]["start_ns"] = 1500 + warmup[name] = {"series": [{"endpoint_url": P, "stats": {"total": value}}]} + if warmup_proof: + export["warmup_metrics"] = warmup + result = _run(export) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["pricing_eligible"] is warmup_proof + assert result["cache_validity"]["status"] == ( + "VALID_COMPLETE" if warmup_proof else "WINDOW_UNRESOLVED" + ) + + +@pytest.mark.parametrize("problem", ["missing", "gap", "invalid", "duplicate"]) +def test_transfer_diagnostics_never_veto_complete_cache_counters(problem): + export = _export() + entry = export["metrics"]["atom:lmcache_loaded_tokens"] + if problem == "missing": + del export["metrics"]["atom:lmcache_loaded_tokens"] + elif problem == "gap": + entry["series"][0]["timeslices"][1]["start_ns"] = 1001 + elif problem == "invalid": + entry["series"][0]["stats"]["total"] = -1 + else: + export["metrics"]["atom_lmcache_loaded_tokens_total"] = deepcopy(entry) + result = _run(export) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["pricing_eligible"] is True + assert "atom:lmcache_loaded_tokens" not in result["tokens"]["prompt_by_source"]["raw"] + + +def test_distinct_label_domains_are_summed_but_duplicate_aliases_are_rejected(): + export = _export(modern=True) + for entry in export["metrics"].values(): + first, decode = entry["series"] + entry["series"] = [ + { + **first, + "labels": {"engine": str(i)}, + "stats": {"total": first["stats"]["total"] * weight}, + } + for i, weight in enumerate((0.4, 0.6)) + ] + [decode] + assert _run(export)["cache_validity"]["pricing_eligible"] is True + export["metrics"]["atom_prefix_cache_full_tokens_total"] = deepcopy( + export["metrics"]["atom:prefix_cache_full_tokens"] + ) + result = _run(export) + _rates(result, None, None, None) + assert result["cache_validity"]["status"] == "INVALID_COUNTER" + + +@pytest.mark.parametrize("modern", [False, True]) +def test_confirmed_no_cpu_matches_measured_zero_but_invalid_cpu_is_not_zero(modern): + export = _export(modern=modern) + entry = export["metrics"].pop("atom:prefix_cache_offload_tokens") + result = _run(export, offload="none") + _rates(result, 0.6, 0, 0.6) + assert result["cache_validity"]["cpu_component"] == "NO_CPU_TIER_CONFIRMED" + assert result["cache_validity"]["pricing_eligible"] is True + entry["series"][0]["stats"]["total"] = -1 + export["metrics"]["atom:prefix_cache_offload_tokens"] = entry + result = _run(export, offload="none") + _rates(result, 0.6, None, None) + assert result["cache_validity"]["status"] == "INVALID_COUNTER" + + +@pytest.mark.parametrize( + "problem", ["missing_total", "overflow", "missing_requests", "scope", "error", "cancel"] +) +def test_invalid_or_different_cohorts_never_become_pricing_input(problem): + export = _export(modern=True) + scope = dict(SCOPE) + if problem == "missing_total": + del export["metrics"]["atom:prefix_cache_cached_tokens"]["series"][0]["stats"]["total"] + elif problem == "overflow": + export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["stats"]["total"] = 500 + elif problem == "missing_requests": + del export["metrics"]["atom:prefix_cache_requests"] + elif problem == "scope": + scope["profiling_prompt_tokens"] = 999 + else: + scope["error_records" if problem == "error" else "cancelled_records"] = 1 + result = _run(export, scope=scope) + _rates( + result, + None if problem == "missing_total" else 0.6, + None if problem in {"missing_total", "overflow"} else 0.1, + None if problem in {"missing_total", "overflow"} else 0.7, + ) + assert result["cache_validity"]["pricing_eligible"] is False + assert result["cache_validity"]["status"] == ( + "INVALID_COUNTER" + if problem in {"missing_total", "overflow"} + else "REQUEST_SCOPE_UNRESOLVED" + ) + + +def test_warmup_errors_do_not_block_a_complete_profiling_cohort(tmp_path): + records = [ + {"metadata": {"benchmark_phase": "warmup"}, "error": {"type": "Timeout"}}, + { + "metadata": {"benchmark_phase": "profiling"}, + "metrics": {"usage_prompt_tokens": {"value": 1000}}, + }, + ] + path = tmp_path / "profile_export.jsonl" + path.write_text("\n".join(json.dumps(record) for record in records)) + successful, accounting = load_records_with_accounting(path) + scope = request_scope_from_records(successful, accounting) + assert accounting["records_error_dropped"] == 1 + assert accounting["records_profiling_error_dropped"] == 0 + assert scope == { + "profiling_requests": 1, + "profiling_prompt_tokens": 1000, + "error_records": 0, + "cancelled_records": 0, + } + successful[0]["metrics"] = {} + assert request_scope_from_records(successful, accounting)["profiling_prompt_tokens"] is None + + +@pytest.mark.parametrize( + "usage", [(-10, 1010), (True, 999), (1.9, 999), (float("inf"), 1000), (None, 1000)] +) +def test_invalid_usage_cannot_cancel_out_or_be_truncated_into_a_pricing_cohort(usage): + records = [{"metrics": {"usage_prompt_tokens": {"value": value}}} for value in usage] + scope = request_scope_from_records(records, {"records_profiling_error_dropped": 0}) + assert scope["profiling_prompt_tokens"] is None + export = _export(modern=True) + export["metrics"]["atom:prefix_cache_requests"]["series"][0]["stats"]["total"] = 2 + result = _run(export, scope=scope) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" + assert result["cache_validity"]["pricing_eligible"] is False + + +def test_no_cpu_confirmation_cannot_bypass_conflicting_backend_metadata(): + export = _export(modern=True) + del export["metrics"]["atom:prefix_cache_offload_tokens"] + _, result, _ = compute_server_metrics( + export, + framework="mooncake-atom", + records=[], + request_scope=SCOPE, + env={ + "SRT_PREFILL_ENDPOINTS": "prefill", + "SRT_DECODE_ENDPOINTS": "decode", + "DISAGG": "true", + "KV_OFFLOADING": " NONE ", + "KV_OFFLOAD_BACKEND": "mooncake", + }, + ) + _rates(result, 0.6, None, None) + assert result["cache_validity"]["cpu_component"] == "MISSING" + assert result["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize(("cached", "offload", "gpu"), [(150, 0, None), (50, 60, 0.25)]) +def test_a_paired_counter_cannot_hide_behind_another_label_domain(cached, offload, gpu): + export = _export(modern=True) + for stem, values in ( + ("full_tokens", (100, 100)), + ("cached_tokens", (cached, 0)), + ("offload_tokens", (offload, 0)), + ("requests", (1, 1)), + ): + export["metrics"][f"atom:prefix_cache_{stem}"]["series"] = [ + {"endpoint_url": P, "labels": {"engine": str(i)}, "stats": {"total": value}} + for i, value in enumerate(values) + ] + result = _run(export, scope={**SCOPE, "profiling_requests": 2, "profiling_prompt_tokens": 200}) + _rates(result, gpu, None, None) + assert result["cache_validity"]["status"] == "INVALID_COUNTER" + assert result["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize("modern", [False, True]) +def test_only_legacy_callers_may_omit_the_cancelled_count(modern): + scope = {key: value for key, value in SCOPE.items() if key != "cancelled_records"} + result = _run(_export(modern=modern), scope=scope) + assert result["cache_validity"]["pricing_eligible"] is (not modern) + for key, value in ( + ("profiling_requests", True), + ("error_records", False), + ("cancelled_records", None), + ("profiling_prompt_tokens", 1000.1), + ): + result = _run(_export(modern=modern), scope={**SCOPE, key: value}) + assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" diff --git a/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py b/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py index 29aeeeef94..a454208043 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py @@ -67,6 +67,7 @@ "records_dropped_total", "records_warmup_dropped", "records_error_dropped", + "records_profiling_error_dropped", "error_categories", } @@ -74,6 +75,7 @@ "present", "adapter", "metric_count", + "cache_validity", "cache", "kv_cache", "kv_offload", @@ -952,6 +954,7 @@ def test_processor_surfaces_request_accounting(tmp_path: Path, error, category): "records_dropped_total": 2, "records_warmup_dropped": 1, "records_error_dropped": 1, + "records_profiling_error_dropped": 1, "error_categories": {category: 1}, } assert agg["server_metrics"]["tokens"]["requests_completed"] == 1 From e92cfd93731857bd03fd7d3e02f2820827ea8666 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 16:48:46 +0000 Subject: [PATCH 2/6] fix(results): preserve unresolved ATOM endpoint coverage state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 保留缺少声明端点观测时的旧资格状态。 --- inferencex-e2e/infx/results/agentic/atom_validity.py | 4 +++- .../infx/tests/results/agentic/test_atom_cache_counters.py | 1 + 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/inferencex-e2e/infx/results/agentic/atom_validity.py b/inferencex-e2e/infx/results/agentic/atom_validity.py index f1f349593e..028c4fe9b7 100644 --- a/inferencex-e2e/infx/results/agentic/atom_validity.py +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -176,7 +176,9 @@ def refuse(status: str, reason: str) -> dict[str, Any]: result["replace_telemetry"] = True return refuse("INVALID_COUNTER", "duplicate or unidentifiable cache counter series") keys = set(series[FULL]) - if {key[0] for key in keys} != set(accounted) or not keys or set(series[CACHED]) != keys: + if set(accounted) - {key[0] for key in keys}: + return refuse("ROLE_OR_COVERAGE_UNRESOLVED", "accounted endpoints missing full counters") + if not keys or set(series[CACHED]) != keys: return refuse("MISSING_OBSERVATION", "full/cached series coverage differs") scoped = {stem: {key: _total(s) for key, s in group.items()} for stem, group in series.items()} raw = { diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py index 2d0e6f433d..430e9770d2 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py @@ -115,6 +115,7 @@ def test_unknown_conflicting_or_incomplete_roles_do_not_mix_endpoints(env: dict) _export({"p": 100, "d": 100}, {"p": 60, "d": 0}, {"p": 10, "d": 0}), env ) _assert_rates(flat, nested, None, None, None) + assert nested["cache_validity"]["status"] == "ROLE_OR_COVERAGE_UNRESOLVED" def test_single_endpoint_without_explicit_unified_topology_stays_unknown() -> None: From 8ae6ebeaee501102f5e19ab45485be03a28b57eb Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 20:30:34 +0000 Subject: [PATCH 3/6] fix(results): normalize ATOM labels and align scoped token totals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 null labels 作为无标签域处理，继续拒绝非法结构与重复别名。令成功范围的 prompt_total 与缓存来源分解使用相同 full 计数，并删除新增模块概述注释。 --- .../infx/results/agentic/atom_validity.py | 12 ++++------- .../infx/results/agentic/backends/atom.py | 1 + .../agentic/test_atom_cache_validity.py | 21 ++++++++++++++++++- 3 files changed, 25 insertions(+), 9 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/atom_validity.py b/inferencex-e2e/infx/results/agentic/atom_validity.py index 028c4fe9b7..f549ec3ee7 100644 --- a/inferencex-e2e/infx/results/agentic/atom_validity.py +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -1,5 +1,3 @@ -"""Qualify ATOM cache counters against the successful profiling cohort.""" - from __future__ import annotations import math @@ -25,7 +23,9 @@ def _series(section: Mapping[str, Any], stem: str) -> tuple[dict[tuple, dict], s found: dict[tuple, dict] = {} duplicates: set[str | None] = set() for series in metric_series(dict(section), names): - url, labels = series.get("endpoint_url"), series.get("labels", {}) + url, labels = series.get("endpoint_url"), series.get("labels") + if labels is None: + labels = {} if not isinstance(url, str) or not url or not isinstance(labels, dict): duplicates.add(None) continue @@ -117,11 +117,7 @@ def account_cache( kv_offloading: str | None = None, collector_export: Mapping[str, Any] | None = None, ) -> dict[str, Any]: - """Recover legacy windows or use phase-scoped totals, then prove request conservation. - - Unqualified cumulative telemetry remains the counter adapter's responsibility. The - internal ``replace_telemetry`` flag only selects a proven window or rejects ambiguity. - """ + """Qualify counter totals against the profiling window and successful request cohort.""" result: dict[str, Any] = dict.fromkeys( ("gpu", "cpu", "overall", "full", "cached", "offload", "loaded", "requests") ) diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 1c8bde8293..5d465b3747 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -185,6 +185,7 @@ def populate( full_tokens, cached_tokens, external_tokens = ( accounting[key] for key in ("full", "cached", "offload") ) + flat["total_prompt_tokens"] = counter_int(full_tokens) nested["cache_validity"] = accounting["validity"] sink = context.get("warnings") if isinstance(sink, list): diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py index e87c2f2602..427164ac16 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -82,8 +82,13 @@ def _rates(result, gpu, cpu, overall): @pytest.mark.parametrize("modern", [False, True]) def test_qualified_pd_preserves_the_old_detail_and_prompt_sources(modern): - result = _run(_export(modern=modern)) + export = _export(modern=modern) + export["metrics"]["atom:prompt_tokens"] = { + "series": [{"endpoint_url": url, "stats": {"total": 1500}} for url in (P, D)] + } + result = _run(export) _rates(result, 0.6, 0.1, 0.7) + assert result["tokens"]["prompt_total"] == 1000 detail = result["cache_validity"] assert detail["schema"] == "atom-cache-validity/1" assert detail["status"] == "VALID_COMPLETE" and detail["pricing_eligible"] is True @@ -348,3 +353,17 @@ def test_only_legacy_callers_may_omit_the_cancelled_count(modern): ): result = _run(_export(modern=modern), scope={**SCOPE, key: value}) assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" + + +@pytest.mark.parametrize("labels", [None, []]) +def test_null_labels_are_unlabelled_but_malformed_labels_remain_invalid(labels): + export = _export(modern=True) + export["metrics"]["atom:prefix_cache_cached_tokens"]["series"][0]["labels"] = labels + result = _run(export) + if labels is None: + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["pricing_eligible"] is True + else: + _rates(result, None, None, None) + assert result["cache_validity"]["status"] == "INVALID_COUNTER" + assert result["cache_validity"]["pricing_eligible"] is False From 95ffd9517ec1884ca84ccde71e673d736affd500 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 21:18:45 +0000 Subject: [PATCH 4/6] refactor(agentx): qualify only modern ATOM profiling exports MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：删除旧窗口恢复与冗余诊断，保留现代 ATOM 请求守恒与最小共享资格合同。 --- .../infx/results/agentic/atom_validity.py | 386 ++++++------------ .../infx/results/agentic/backends/atom.py | 60 +-- .../infx/results/agentic/cache_validity.py | 81 ++++ .../infx/results/agentic/server_metrics.py | 17 +- .../agentic/test_atom_cache_counters.py | 1 - .../agentic/test_atom_cache_validity.py | 377 ++++++++--------- 6 files changed, 397 insertions(+), 525 deletions(-) create mode 100644 inferencex-e2e/infx/results/agentic/cache_validity.py diff --git a/inferencex-e2e/infx/results/agentic/atom_validity.py b/inferencex-e2e/infx/results/agentic/atom_validity.py index f549ec3ee7..853d8bb185 100644 --- a/inferencex-e2e/infx/results/agentic/atom_validity.py +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -2,319 +2,179 @@ import math from collections.abc import Mapping -from itertools import pairwise from typing import Any -from .common import metric_series, series_stat +from .cache_validity import modern_profile_error -FULL = "prefix_cache_full_tokens" -CACHED = "prefix_cache_cached_tokens" -OFFLOAD = "prefix_cache_offload_tokens" -LOADED = "lmcache_loaded_tokens" -REQUESTS = "prefix_cache_requests" -_COUNTERS = (FULL, CACHED, OFFLOAD, LOADED, REQUESTS) -_CORE = (FULL, CACHED, OFFLOAD) +_STEMS = ("full_tokens", "cached_tokens", "offload_tokens", "requests") -def _series(section: Mapping[str, Any], stem: str) -> tuple[dict[tuple, dict], set[str | None]]: +def _series(metrics: dict[str, Any], stem: str) -> tuple[dict, set]: names = [ - f"{prefix}{stem}{suffix}" for prefix in ("atom:", "atom_") for suffix in ("", "_total") + f"{prefix}prefix_cache_{stem}{suffix}" + for prefix in ("atom:", "atom_") + for suffix in ("", "_total") ] - found: dict[tuple, dict] = {} - duplicates: set[str | None] = set() - for series in metric_series(dict(section), names): - url, labels = series.get("endpoint_url"), series.get("labels") - if labels is None: - labels = {} - if not isinstance(url, str) or not url or not isinstance(labels, dict): - duplicates.add(None) + found, invalid = {}, set() + for name in names: + if name not in metrics: continue - key = (url, tuple(sorted((str(k), str(v)) for k, v in labels.items()))) - if key in found: - duplicates.add(url) - found[key] = series - return found, duplicates + entry = metrics[name] + if ( + not isinstance(entry, dict) + or entry.get("type") != "counter" + or not isinstance(entry.get("series"), list) + ): + invalid.add(None) + continue + for series in entry["series"]: + if not isinstance(series, dict): + invalid.add(None) + continue + endpoint = series.get("endpoint_url") + labels = series.get("labels") + labels = {} if labels is None else labels + if ( + not isinstance(endpoint, str) + or not endpoint + or not isinstance(labels, dict) + or any(not isinstance(k, str) or not isinstance(v, str) for k, v in labels.items()) + ): + invalid.add(endpoint if isinstance(endpoint, str) else None) + continue + key = (endpoint, tuple(sorted(labels.items()))) + if key in found: + invalid.add(endpoint) + found[key] = series + return found, invalid def _total(series: dict[str, Any]) -> float | None: stats = series.get("stats") - raw = stats.get("total") if isinstance(stats, dict) else None - value = series_stat(series, preferred_keys=("total",)) - return ( - value - if not isinstance(raw, bool) and value is not None and math.isfinite(value) and value >= 0 - else None - ) - - -def _window_total( - series: dict[str, Any], window: tuple[int, int], warmup: dict[str, Any] | None -) -> tuple[float | None, bool]: - slices = series.get("timeslices") - if not isinstance(slices, list) or not slices: - return None, False - parsed = [] - for item in slices: - if not isinstance(item, dict): - return None, False - lo, hi, value = item.get("start_ns"), item.get("end_ns"), item.get("total") - if ( - type(lo) is not int - or type(hi) is not int - or hi <= lo - or isinstance(value, bool) - or not isinstance(value, (int, float)) - or not math.isfinite(value) - or value < 0 - ): - return None, False - parsed.append((lo, hi, value)) - parsed.sort() - if any(b[0] < a[1] for a, b in pairwise(parsed)): - return None, False - start, end = window - covering = [item for item in parsed if item[1] > start and item[0] < end] + value = stats.get("total") if isinstance(stats, dict) else None if ( - not covering - or covering[0][0] > start - or covering[-1][1] < end - or any(b[0] != a[1] for a, b in pairwise(covering)) + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + or value != int(value) ): - return None, False - before = sum(value for lo, hi, value in parsed if hi <= start) - start_straddle = sum(value for lo, hi, value in parsed if lo < start < hi) - end_straddle = sum(value for lo, hi, value in parsed if lo < end < hi) - warm = _total(warmup) if warmup is not None else None - boundary = end_straddle > 0 or (start_straddle > 0 and before + start_straddle != warm) - return float(sum(value for lo, hi, value in parsed if start <= lo and hi <= end)), boundary - - -def _modern_export(export: Mapping[str, Any], endpoints: list[str]) -> bool: - summary = export.get("summary") or {} - return ( - isinstance(summary, dict) - and export.get("schema_version") == "1.1" - and export.get("aiperf_version") == "1.0.12" - and export.get("metrics_phase") == "profiling" - and all( - isinstance(summary.get(key), list) - and all(isinstance(url, str) for url in summary[key]) - and set(endpoints) <= set(summary[key]) - for key in ("endpoints_configured", "endpoints_successful") - ) - ) + return None + return value def account_cache( - metrics: dict[str, dict[str, Any]], + metrics: dict[str, Any], *, - framework: str, endpoint_roles: Mapping[str, str] | None, - profiling_window: tuple[int, int] | None, - warmup_metrics: Mapping[str, Any] | None = None, - request_scope: Mapping[str, Any] | None = None, - disagg: bool | None = None, - kv_offloading: str | None = None, - collector_export: Mapping[str, Any] | None = None, -) -> dict[str, Any]: - """Qualify counter totals against the profiling window and successful request cohort.""" - result: dict[str, Any] = dict.fromkeys( - ("gpu", "cpu", "overall", "full", "cached", "offload", "loaded", "requests") - ) - reasons: list[str] = [] - validity: dict[str, Any] = { - "schema": "atom-cache-validity/1", - "status": None, - "pricing_eligible": False, - "observed_scope": None, - "requested_scope": "profiling_successful_requests", - "cpu_component": None, - "roles_source": None, - "accounted_endpoints": [], - "window": None, - "request_scope": None, - "diagnostics": {}, - "reasons": reasons, - } - result.update(validity=validity, warnings=[], replace_telemetry=False) + disagg: bool | None, + kv_offloading: str | None, + collector_export: Mapping[str, Any], + request_scope: Mapping[str, Any] | None, +) -> tuple[tuple[float | None, float | None, float | None] | None, dict[str, Any]]: + detail: dict[str, Any] = {"schema": "atom-cache-validity/1", "pricing_eligible": False} + counts = None - def refuse(status: str, reason: str) -> dict[str, Any]: - validity["status"] = status - reasons.append(reason) - result["warnings"].append(f"atom cache accounting: {reason}") - return result + def finish(status: str, reason: str | None = None) -> tuple: + detail.update(status=status, reasons=[reason] if reason else []) + return counts, detail - series, duplicates = {}, {} - for stem in _COUNTERS: - series[stem], duplicates[stem] = _series(metrics, stem) - observed = {key[0] for stem in _CORE for key in series[stem]} - if not series[FULL]: - return refuse("MISSING_OBSERVATION", "no prefix_cache_full_tokens series") + if reason := modern_profile_error(collector_export): + return finish("WINDOW_UNRESOLVED", reason) + counts = (None, None, None) + series, invalid = {}, {} + for stem in _STEMS: + series[stem], invalid[stem] = _series(metrics, stem) + observed = {key[0] for stem in _STEMS[:3] for key in series[stem]} if endpoint_roles is None and disagg is False and len(observed) == 1: - accounted = sorted(observed) - validity["roles_source"] = "single_endpoint_non_disagg" - elif endpoint_roles: - accounted = sorted( - url for url, role in endpoint_roles.items() if role in {"prefill", "unified"} - ) - if observed - endpoint_roles.keys() or not accounted: - return refuse("ROLE_OR_COVERAGE_UNRESOLVED", f"{framework}: endpoint roles incomplete") - validity["roles_source"] = "launcher_env" + accounted = observed + elif endpoint_roles and not observed - endpoint_roles.keys(): + accounted = {url for url, role in endpoint_roles.items() if role in {"prefill", "unified"}} else: - return refuse("ROLE_OR_COVERAGE_UNRESOLVED", f"{framework}: endpoint roles unresolved") - validity["accounted_endpoints"] = accounted + accounted = set() + if not accounted: + return finish("ROLE_OR_COVERAGE_UNRESOLVED", "cache endpoint roles unresolved") + if reason := modern_profile_error(collector_export, accounted): + counts = None + return finish("WINDOW_UNRESOLVED", reason) series = { - stem: {key: s for key, s in group.items() if key[0] in accounted} + stem: {k: s for k, s in group.items() if k[0] in accounted} for stem, group in series.items() } - duplicates = {stem: bad & {*accounted, None} for stem, bad in duplicates.items()} - if any(duplicates[stem] for stem in (FULL, CACHED)): - result["replace_telemetry"] = True - return refuse("INVALID_COUNTER", "duplicate or unidentifiable cache counter series") - keys = set(series[FULL]) - if set(accounted) - {key[0] for key in keys}: - return refuse("ROLE_OR_COVERAGE_UNRESOLVED", "accounted endpoints missing full counters") - if not keys or set(series[CACHED]) != keys: - return refuse("MISSING_OBSERVATION", "full/cached series coverage differs") - scoped = {stem: {key: _total(s) for key, s in group.items()} for stem, group in series.items()} - raw = { - stem: sum(values) if values and all(value is not None for value in values) else None - for stem in _COUNTERS - for values in [list(scoped[stem].values())] - } - validity["diagnostics"]["cumulative_stats_total"] = { - f"atom:{stem}": value for stem, value in raw.items() if value is not None - } - if raw[FULL] is None or raw[CACHED] is None: - result["replace_telemetry"] = True - return refuse("INVALID_COUNTER", "full/cached stats.total missing or invalid") - cpu_missing = not series[OFFLOAD] - cpu_complete = ( - set(series[OFFLOAD]) == keys and raw[OFFLOAD] is not None and not duplicates[OFFLOAD] - ) - cpu_invalid = bool(duplicates[OFFLOAD]) or (bool(series[OFFLOAD]) and raw[OFFLOAD] is None) - if profiling_window is None: - return refuse("WINDOW_UNRESOLVED", "no summary.phase_time_ranges.profiling window") - totals = dict(raw) - boundary = False - modern = _modern_export(collector_export or {}, accounted) - if not modern: - for stem in _COUNTERS: - if raw[stem] is None or (stem == OFFLOAD and not cpu_complete): - totals[stem] = None - continue - warm, warm_duplicate = _series(warmup_metrics or {}, stem) - windows = { - key: _window_total(s, profiling_window, None if warm_duplicate else warm.get(key)) - for key, s in series[stem].items() - } - values = list(windows.values()) - if any(value is None for value, _ in values): - if stem in (FULL, CACHED) or (stem == OFFLOAD and cpu_complete): - return refuse( - "WINDOW_UNRESOLVED", f"{stem}: missing, invalid or gapped timeslices" - ) - totals[stem] = None - else: - scoped[stem] = {key: value for key, (value, _) in windows.items()} - totals[stem] = sum(value for value, _ in values) - if stem in _CORE: - boundary |= any(issue for _, issue in values) - validity["observed_scope"] = ( - "profiling_phase_counter_totals" if modern else "profiling_window_interior_timeslices" - ) - validity["window"] = { - "start_ns": profiling_window[0], - "end_ns": profiling_window[1], - "boundary_issues": ["activity crosses the profiling boundary"] if boundary else [], + invalid = {stem: bool(bad & {*accounted, None}) for stem, bad in invalid.items()} + if invalid["full_tokens"] or invalid["cached_tokens"]: + return finish("INVALID_COUNTER", "duplicate or malformed full/cached series") + keys = set(series["full_tokens"]) + if not keys or accounted - {key[0] for key in keys} or set(series["cached_tokens"]) != keys: + return finish("MISSING_OBSERVATION", "full/cached series coverage differs") + values = {stem: {k: _total(s) for k, s in group.items()} for stem, group in series.items()} + totals = { + stem: sum(group.values()) if group and None not in group.values() else None + for stem, group in values.items() } - result["replace_telemetry"] = True - full, cached, offload = totals[FULL], totals[CACHED], totals[OFFLOAD] + full, cached, offload, admissions = (totals[stem] for stem in _STEMS) if ( full is None or not math.isfinite(full) or full <= 0 or cached is None - or not 0 <= cached <= full + or not math.isfinite(cached) + or cached > full + or any(values["cached_tokens"][k] > values["full_tokens"][k] for k in keys) ): - return refuse("INVALID_COUNTER", "full/cached tokens outside their valid range") - for key in keys: - if scoped[CACHED][key] > scoped[FULL][key]: - return refuse("INVALID_COUNTER", "cached tokens exceed full tokens in a paired series") - if cpu_complete and scoped[CACHED][key] + scoped[OFFLOAD][key] > scoped[FULL][key]: - cpu_invalid = True - if cpu_missing and (kv_offloading or "").lower() == "none": + return finish("INVALID_COUNTER", "full/cached tokens outside their valid range") + cpu_complete = set(series["offload_tokens"]) == keys + cpu_invalid = invalid["offload_tokens"] or (bool(series["offload_tokens"]) and offload is None) + if cpu_complete and offload is not None: + cpu_invalid |= not math.isfinite(offload) or any( + values["cached_tokens"][k] + values["offload_tokens"][k] > values["full_tokens"][k] + for k in keys + ) + if not series["offload_tokens"] and not cpu_invalid and kv_offloading == "none": offload = 0.0 - validity["cpu_component"] = "NO_CPU_TIER_CONFIRMED" - elif cpu_complete and not cpu_invalid and offload is not None and cached + offload <= full: - validity["cpu_component"] = "MEASURED" - else: + elif not cpu_complete or cpu_invalid: offload = None - validity["cpu_component"] = "MISSING" - result.update( - full=full, - cached=cached, - offload=offload, - loaded=totals[LOADED], - requests=totals[REQUESTS], - gpu=cached / full, - ) - if offload is not None: - result.update(cpu=offload / full, overall=(cached + offload) / full) + counts = (full, cached, offload) scope = dict(request_scope or {}) - admissions = ( - totals[REQUESTS] if set(series[REQUESTS]) == keys and not duplicates[REQUESTS] else None - ) - scope_counts = [ - scope.get(key) for key in ("profiling_requests", "profiling_prompt_tokens", "error_records") - ] - if "cancelled_records" in scope: - scope_counts.append(scope["cancelled_records"]) - scope_valid = all( + valid_scope = all( not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(value) and value >= 0 - and int(value) == value - for value in scope_counts + and value == int(value) + for value in ( + scope.get(k) + for k in ( + "profiling_requests", + "profiling_prompt_tokens", + "error_records", + "cancelled_records", + ) + ) ) + if set(series["requests"]) != keys or invalid["requests"]: + admissions = None conserved = ( - scope_valid - and admissions is not None - and scope.get("profiling_prompt_tokens") is not None + valid_scope and admissions == scope.get("profiling_requests") - and full == scope["profiling_prompt_tokens"] - and scope.get("error_records") == 0 - and not scope.get("cancelled_records") - and (not modern or scope.get("cancelled_records") == 0) + and full == scope.get("profiling_prompt_tokens") + and scope.get("error_records") == scope.get("cancelled_records") == 0 ) - validity["request_scope"] = { + detail["request_scope"] = { "server_window_admissions": admissions, "server_window_prompt_tokens": full, **scope, "conserved": conserved, } - if duplicates[LOADED] or totals[LOADED] is None: - result["loaded"] = None - validity["diagnostics"]["loaded_tokens_unavailable"] = True - if boundary: - return refuse( - "WINDOW_UNRESOLVED", "activity crosses the profiling boundary without warmup proof" - ) - if cpu_invalid or (offload is None and cpu_complete): - return refuse( - "INVALID_COUNTER", "offload tokens invalid or incompatible with full/cached tokens" - ) + if cpu_invalid: + return finish("INVALID_COUNTER", "offload counters invalid or incompatible") if offload is None: - return refuse( - "PARTIAL_COMPONENT", "CPU reuse missing and no confirmed absence of a CPU tier" + return finish( + "PARTIAL_COMPONENT", "CPU reuse missing without confirmed absence of CPU tier" ) if not conserved: - return refuse( - "REQUEST_SCOPE_UNRESOLVED", - "server admissions/prompt tokens do not match successful profiling requests", + return finish( + "REQUEST_SCOPE_UNRESOLVED", "counters do not match successful profiling requests" ) - validity.update( - status="VALID_ZERO" if cached + offload == 0 else "VALID_COMPLETE", pricing_eligible=True - ) - return result + detail["pricing_eligible"] = True + return finish("VALID_ZERO" if cached + offload == 0 else "VALID_COMPLETE") diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 5d465b3747..d77407bb73 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -7,14 +7,7 @@ from collections.abc import Mapping from typing import Any -from infx.results.agentic.atom_validity import ( - CACHED, - FULL, - LOADED, - OFFLOAD, - REQUESTS, - account_cache, -) +from infx.results.agentic.atom_validity import account_cache from infx.results.agentic.common import ( gauge_stat, metric_series, @@ -130,14 +123,11 @@ def populate( nested: dict[str, Any], *, context: Mapping[str, Any] | None = None, - endpoint_roles: Mapping[str, str] | None = None, - disaggregated: bool | None = None, - cpu_offload_disabled: bool = False, ) -> None: context = context or {} - endpoint_roles = context.get("endpoint_roles", endpoint_roles) - disaggregated = context.get("disagg", disaggregated) - cpu_offload_disabled = context.get("kv_offloading") == "none" or cpu_offload_disabled + endpoint_roles = context.get("endpoint_roles") + disaggregated = context.get("disagg") + cpu_offload_disabled = context.get("kv_offloading") == "none" prompt_names = _atom_counter_names("prompt_tokens") prompt_endpoints = {s.get("endpoint_url") for s in metric_series(metrics, prompt_names)} if endpoint_roles is None: @@ -170,26 +160,21 @@ def populate( full_tokens, cached_tokens, external_tokens = _cache_counts( metrics, endpoint_roles, disaggregated, cpu_offload_disabled ) - accounting = account_cache( + counts, detail = account_cache( metrics, - framework=str(context.get("framework") or "atom"), endpoint_roles=endpoint_roles, - profiling_window=context.get("profiling_window"), - warmup_metrics=context.get("warmup_metrics"), request_scope=context.get("request_scope"), disagg=disaggregated, kv_offloading=context.get("kv_offloading"), - collector_export=context.get("collector_export"), + collector_export=context.get("collector_export") or {}, ) - if accounting["replace_telemetry"]: - full_tokens, cached_tokens, external_tokens = ( - accounting[key] for key in ("full", "cached", "offload") - ) + if counts is not None: + full_tokens, cached_tokens, external_tokens = counts flat["total_prompt_tokens"] = counter_int(full_tokens) - nested["cache_validity"] = accounting["validity"] + nested["cache_validity"] = detail sink = context.get("warnings") if isinstance(sink, list): - sink.extend(accounting["warnings"]) + sink.extend(f"atom cache accounting: {reason}" for reason in detail["reasons"]) cache_hit_rate = rate(cached_tokens, full_tokens) # Offload is CPU reuse beyond the GPU prefix; loaded tokens measure transfer volume. external_hit_rate = rate(external_tokens, full_tokens) @@ -227,25 +212,12 @@ def populate( } ) nested["kv_cache"]["gpu_usage_pct"] = flat["gpu_kv_cache_usage_pct"] - if accounting["replace_telemetry"]: - by_source = nested["tokens"]["prompt_by_source"] - if full_tokens is not None and external_tokens is not None: - by_source.update( - gpu_cache_hit=cached_tokens, - cpu_or_external_cache_hit=external_tokens, - computed=full_tokens - cached_tokens - external_tokens, - ) - by_source["raw"] = { - f"atom:{stem}": accounting[key] - for stem, key in ( - (FULL, "full"), - (CACHED, "cached"), - (OFFLOAD, "offload"), - (LOADED, "loaded"), - (REQUESTS, "requests"), - ) - if accounting[key] is not None - } + if counts is not None and full_tokens is not None and external_tokens is not None: + nested["tokens"]["prompt_by_source"].update( + gpu_cache_hit=cached_tokens, + cpu_or_external_cache_hit=external_tokens, + computed=full_tokens - cached_tokens - external_tokens, + ) nested["tokens"].update( { "prompt_total": flat["total_prompt_tokens"], diff --git a/inferencex-e2e/infx/results/agentic/cache_validity.py b/inferencex-e2e/infx/results/agentic/cache_validity.py new file mode 100644 index 0000000000..4a7c53cc26 --- /dev/null +++ b/inferencex-e2e/infx/results/agentic/cache_validity.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from typing import Any + +CORE_SCHEMA = "infx-cache-validity-core/1" +_STATUS_CLASS = { + "VALID_COMPLETE": "VALID_COMPLETE", + "VALID_ZERO": "VALID_ZERO", + "CONDITIONAL_INTERVAL": "VALID_INTERVAL", + "PARTIAL_COMPONENT": "PARTIAL", + "PARTIAL_COVERAGE": "PARTIAL", + "WINDOW_UNRESOLVED": "UNRESOLVED", + "ROLE_OR_COVERAGE_UNRESOLVED": "UNRESOLVED", + "REQUEST_SCOPE_UNRESOLVED": "UNRESOLVED", + "MISSING_OBSERVATION": "MISSING", + "INVALID_COUNTER": "INVALID", + "REJECTED": "INVALID", + "PHASE_CONTAMINATED": "CONTAMINATED", + "NOT_EVALUATED": "NOT_EVALUATED", +} + + +def modern_profile_error(export: Mapping[str, Any], endpoints: Iterable[str] = ()) -> str | None: + if ( + export.get("schema_version") != "1.1" + or export.get("aiperf_version") != "1.0.12" + or export.get("metrics_phase") != "profiling" + ): + return "AgentX 1.0.12 profiling export required" + summary = export.get("summary") + ranges = summary.get("phase_time_ranges") if isinstance(summary, dict) else None + window = ranges.get("profiling") if isinstance(ranges, dict) else None + if not isinstance(window, dict) or not ( + type(window.get("start_ns")) is int + and type(window.get("end_ns")) is int + and 0 <= window["start_ns"] < window["end_ns"] + ): + return "profiling time bounds missing or invalid" + expected = set(endpoints) + for key in ("endpoints_configured", "endpoints_successful"): + covered = summary.get(key) + if ( + not isinstance(covered, list) + or not all(isinstance(url, str) for url in covered) + or not expected.issubset(covered) + ): + return "profiling endpoint coverage incomplete" + return None + + +def summarize_cache_validity(nested: dict[str, Any]) -> dict[str, Any]: + cache = nested["cache"] + tiers = any( + cache.get(f"{tier}_cache_hit_rate") is not None for tier in ("gpu", "cpu", "external") + ) + combined = cache.get("overall_cache_hit_rate") is not None + detail = nested.get("cache_validity") + if not isinstance(detail, dict): + rate = nested.get("validity", {}).get("cache_rate", {}) + detail = { + "status": rate.get("status", "NOT_EVALUATED"), + "pricing_eligible": bool(rate.get("eligible_exact_actual")), + "reasons": list(rate.get("reasons") or []), + } + if not nested.get("present"): + detail = {"status": "MISSING_OBSERVATION", "reasons": ["server_metrics_missing"]} + status = detail.get("status") + core = { + "schema": detail.get("schema", CORE_SCHEMA), + "core_schema": CORE_SCHEMA, + "status": status, + "status_class": _STATUS_CLASS.get(status, "UNRESOLVED"), + "published": tiers or combined, + "pricing_eligible": bool(detail.get("pricing_eligible")), + "basis": "tiers" if tiers else "combined" if combined else None, + "reasons": list(detail.get("reasons") or []), + } + if isinstance(detail.get("request_scope"), dict): + core["request_scope"] = detail["request_scope"] + return core diff --git a/inferencex-e2e/infx/results/agentic/server_metrics.py b/inferencex-e2e/infx/results/agentic/server_metrics.py index df46fa4c98..4e9601e1dc 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -12,6 +12,7 @@ apply_profile_totals, empty_server_metrics, ) +from .cache_validity import summarize_cache_validity from .common import index_server_metrics @@ -55,17 +56,6 @@ def deployment_from_env(env: Mapping[str, str]) -> dict[str, Any]: } -def _profiling_window(server_metrics: dict[str, Any]) -> tuple[int, int] | None: - summary = server_metrics.get("summary") - ranges = summary.get("phase_time_ranges") if isinstance(summary, dict) else None - profiling = ranges.get("profiling") if isinstance(ranges, dict) else None - if isinstance(profiling, dict): - start, end = profiling.get("start_ns"), profiling.get("end_ns") - if type(start) is int and type(end) is int and start < end: - return start, end - return None - - def endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: if env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": return {} @@ -116,6 +106,7 @@ def compute_server_metrics( nested["tokens"]["prompt_total"] = flat["total_prompt_tokens"] nested["tokens"]["generation_total"] = flat["total_generation_tokens"] nested["tokens"]["requests_completed"] = flat["total_requests_completed"] + nested["cache_validity"] = summarize_cache_validity(nested) return flat, nested, warnings if adapter is None: @@ -127,7 +118,6 @@ def compute_server_metrics( nested["adapter"] = adapter.name if getattr(adapter, "accepts_context", False): source_env = env or {} - warmup = server_metrics.get("warmup_metrics") adapter.populate( metrics, flat, @@ -137,8 +127,6 @@ def compute_server_metrics( "endpoint_roles": endpoint_roles if endpoint_roles is not None else endpoint_roles_from_env(source_env), - "profiling_window": _profiling_window(server_metrics), - "warmup_metrics": warmup if isinstance(warmup, dict) else None, "request_scope": request_scope, **deployment_from_env(source_env), **dict(deployment or {}), @@ -166,4 +154,5 @@ def compute_server_metrics( nested["tokens"]["generation_total"] = flat["total_generation_tokens"] nested["tokens"]["requests_completed"] = flat["total_requests_completed"] + nested["cache_validity"] = summarize_cache_validity(nested) return flat, nested, warnings diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py index a6458c2e7a..40ea3d9b89 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py @@ -123,7 +123,6 @@ def test_unknown_conflicting_or_incomplete_roles_do_not_mix_endpoints(env: dict) _export({"p": 100, "d": 100}, {"p": 60, "d": 0}, {"p": 10, "d": 0}), env ) _assert_rates(flat, nested, None, None, None) - assert nested["cache_validity"]["status"] == "ROLE_OR_COVERAGE_UNRESOLVED" def test_single_endpoint_without_explicit_unified_topology_stays_unknown() -> None: diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py index 427164ac16..2326def947 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -8,8 +8,7 @@ from infx.results.agentic.artifacts import load_records_with_accounting from infx.results.agentic.server_metrics import compute_server_metrics, request_scope_from_records -P = "http://prefill/metrics" -D = "http://decode/metrics" +P, D = "http://prefill/metrics", "http://decode/metrics" ROLES = {P: "prefill", D: "decode"} SCOPE = { "profiling_requests": 5, @@ -19,55 +18,42 @@ } -def _export(*, modern=False): - metrics = {} - for stem, value in ( - ("prefix_cache_full_tokens", 1000), - ("prefix_cache_cached_tokens", 600), - ("prefix_cache_offload_tokens", 100), - ("prefix_cache_requests", 5), - ("lmcache_loaded_tokens", 900), - ): - metrics[f"atom:{stem}"] = { - "series": [ - { - "endpoint_url": url, - "stats": {"total": total}, - "timeslices": [ - {"start_ns": 0, "end_ns": 1000, "total": 0}, - {"start_ns": 1000, "end_ns": 4000, "total": total}, - ], - } - for url, total in ( - (P, value), - (D, 1000 if stem == "prefix_cache_full_tokens" else 0), - ) - ] - } - export = { - "metrics": metrics, +def _export(): + return { + "schema_version": "1.1", + "aiperf_version": "1.0.12", + "metrics_phase": "profiling", "summary": { "phase_time_ranges": {"profiling": {"start_ns": 1000, "end_ns": 4000}}, "endpoints_configured": [P, D], "endpoints_successful": [P, D], }, + "metrics": { + f"atom:prefix_cache_{stem}": { + "type": "counter", + "series": [ + {"endpoint_url": url, "stats": {"total": total}} + for url, total in ((P, value), (D, 1000 if stem == "full_tokens" else 0)) + ], + } + for stem, value in ( + ("full_tokens", 1000), + ("cached_tokens", 600), + ("offload_tokens", 100), + ("requests", 5), + ) + }, } - if modern: - export.update(schema_version="1.1", aiperf_version="1.0.12", metrics_phase="profiling") - for entry in metrics.values(): - for series in entry["series"]: - del series["timeslices"] - return export -def _run(export, *, scope=None, offload="dram"): +def _run(export, *, scope=None, offload="dram", roles=ROLES): flat, nested, _ = compute_server_metrics( export, framework="mooncake-atom", records=[], - endpoint_roles=ROLES, + endpoint_roles=roles, request_scope=SCOPE if scope is None else scope, - deployment={"disagg": True, "kv_offloading": offload}, + deployment={"disagg": roles == ROLES, "kv_offloading": offload}, ) assert flat["server_gpu_cache_hit_rate"] == nested["cache"]["gpu_cache_hit_rate"] return nested @@ -80,122 +66,75 @@ def _rates(result, gpu, cpu, overall): ) -@pytest.mark.parametrize("modern", [False, True]) -def test_qualified_pd_preserves_the_old_detail_and_prompt_sources(modern): - export = _export(modern=modern) +@pytest.mark.parametrize("unified", [False, True]) +def test_modern_pd_and_unified_totals_match_the_successful_cohort(unified): + export = _export() + if unified: + for entry in export["metrics"].values(): + entry["series"] = entry["series"][:1] export["metrics"]["atom:prompt_tokens"] = { "series": [{"endpoint_url": url, "stats": {"total": 1500}} for url in (P, D)] } - result = _run(export) + result = _run(export, roles={P: "unified"} if unified else ROLES) _rates(result, 0.6, 0.1, 0.7) assert result["tokens"]["prompt_total"] == 1000 - detail = result["cache_validity"] - assert detail["schema"] == "atom-cache-validity/1" - assert detail["status"] == "VALID_COMPLETE" and detail["pricing_eligible"] is True - assert detail["accounted_endpoints"] == [P] - assert detail["request_scope"]["conserved"] is True assert result["tokens"]["prompt_by_source"] == { "gpu_cache_hit": 600, "cpu_or_external_cache_hit": 100, "computed": 300, - "raw": { - "atom:prefix_cache_full_tokens": 1000, - "atom:prefix_cache_cached_tokens": 600, - "atom:prefix_cache_offload_tokens": 100, - "atom:lmcache_loaded_tokens": 900, - "atom:prefix_cache_requests": 5, - }, + "raw": {}, } + core = result["cache_validity"] + assert core["schema"] == "atom-cache-validity/1" + assert core["core_schema"] == "infx-cache-validity-core/1" + assert core["status_class"] == "VALID_COMPLETE" + assert core["published"] is True and core["basis"] == "tiers" + assert core["pricing_eligible"] is True and core["request_scope"]["conserved"] is True @pytest.mark.parametrize( ("cached", "offload", "status"), [(0, 0, "VALID_ZERO"), (0, 100, "VALID_COMPLETE")] ) def test_true_zero_and_cpu_only_are_eligible(cached, offload, status): - export = _export(modern=True) - for stem, value in (("cached", cached), ("offload", offload)): - export["metrics"][f"atom:prefix_cache_{stem}_tokens"]["series"][0]["stats"]["total"] = value + export = _export() + for stem, value in (("cached_tokens", cached), ("offload_tokens", offload)): + export["metrics"][f"atom:prefix_cache_{stem}"]["series"][0]["stats"]["total"] = value result = _run(export) _rates(result, 0, offload / 1000, offload / 1000) - assert result["cache_validity"]["status"] == status + assert result["cache_validity"]["status_class"] == status assert result["cache_validity"]["pricing_eligible"] is True -@pytest.mark.parametrize("missing", ["window", "slices", "gap"]) -def test_unproven_legacy_window_keeps_raw_telemetry_but_cannot_be_priced(missing): +@pytest.mark.parametrize( + "proof", ["legacy", "version", "future", "phase", "endpoint", "window", "negative_window"] +) +def test_unproven_exports_keep_raw_telemetry_without_pricing_qualification(proof): export = _export() - series = export["metrics"]["atom:prefix_cache_full_tokens"]["series"][0] - if missing == "window": - del export["summary"] - elif missing == "slices": - del series["timeslices"] - else: - series["timeslices"][1]["start_ns"] = 1001 - result = _run(export) - _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["status"] == "WINDOW_UNRESOLVED" - assert result["cache_validity"]["observed_scope"] is None - assert result["cache_validity"]["pricing_eligible"] is False - - -@pytest.mark.parametrize("proof", ["version", "phase", "endpoint", "window"]) -def test_modern_header_alone_does_not_prove_a_window(proof): - export = _export(modern=True) - if proof == "version": - export["aiperf_version"] = "1.0.11" + if proof == "legacy": + del export["metrics_phase"] + for entry in export["metrics"].values(): + for series in entry["series"]: + series["timeslices"] = [ + {"start_ns": 1000, "end_ns": 4000, "total": series["stats"]["total"]} + ] + elif proof in {"version", "future"}: + export["aiperf_version"] = "1.0.11" if proof == "version" else "1.0.13" elif proof == "phase": export["metrics_phase"] = "warmup" elif proof == "endpoint": export["summary"]["endpoints_successful"] = [D] + elif proof == "negative_window": + export["summary"]["phase_time_ranges"]["profiling"]["start_ns"] = -1 else: export["summary"]["phase_time_ranges"] = {} result = _run(export) _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["status"] == "WINDOW_UNRESOLVED" + assert result["cache_validity"]["status_class"] == "UNRESOLVED" assert result["cache_validity"]["pricing_eligible"] is False -@pytest.mark.parametrize("warmup_proof", [False, True]) -def test_positive_start_slice_requires_matching_warmup_counters(warmup_proof): - export = _export() - warmup = {} - for stem, value in (("full", 100), ("cached", 50), ("offload", 10)): - name = f"atom:prefix_cache_{stem}_tokens" - series = export["metrics"][name]["series"][0] - series["stats"]["total"] += value - series["timeslices"][0].update(end_ns=1500, total=value) - series["timeslices"][1]["start_ns"] = 1500 - warmup[name] = {"series": [{"endpoint_url": P, "stats": {"total": value}}]} - if warmup_proof: - export["warmup_metrics"] = warmup - result = _run(export) - _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["pricing_eligible"] is warmup_proof - assert result["cache_validity"]["status"] == ( - "VALID_COMPLETE" if warmup_proof else "WINDOW_UNRESOLVED" - ) - - -@pytest.mark.parametrize("problem", ["missing", "gap", "invalid", "duplicate"]) -def test_transfer_diagnostics_never_veto_complete_cache_counters(problem): - export = _export() - entry = export["metrics"]["atom:lmcache_loaded_tokens"] - if problem == "missing": - del export["metrics"]["atom:lmcache_loaded_tokens"] - elif problem == "gap": - entry["series"][0]["timeslices"][1]["start_ns"] = 1001 - elif problem == "invalid": - entry["series"][0]["stats"]["total"] = -1 - else: - export["metrics"]["atom_lmcache_loaded_tokens_total"] = deepcopy(entry) - result = _run(export) - _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["pricing_eligible"] is True - assert "atom:lmcache_loaded_tokens" not in result["tokens"]["prompt_by_source"]["raw"] - - def test_distinct_label_domains_are_summed_but_duplicate_aliases_are_rejected(): - export = _export(modern=True) + export = _export() for entry in export["metrics"].values(): first, decode = entry["series"] entry["series"] = [ @@ -212,53 +151,69 @@ def test_distinct_label_domains_are_summed_but_duplicate_aliases_are_rejected(): ) result = _run(export) _rates(result, None, None, None) - assert result["cache_validity"]["status"] == "INVALID_COUNTER" + assert result["cache_validity"]["status_class"] == "INVALID" -@pytest.mark.parametrize("modern", [False, True]) -def test_confirmed_no_cpu_matches_measured_zero_but_invalid_cpu_is_not_zero(modern): - export = _export(modern=modern) - entry = export["metrics"].pop("atom:prefix_cache_offload_tokens") - result = _run(export, offload="none") - _rates(result, 0.6, 0, 0.6) - assert result["cache_validity"]["cpu_component"] == "NO_CPU_TIER_CONFIRMED" - assert result["cache_validity"]["pricing_eligible"] is True - entry["series"][0]["stats"]["total"] = -1 - export["metrics"]["atom:prefix_cache_offload_tokens"] = entry - result = _run(export, offload="none") - _rates(result, 0.6, None, None) - assert result["cache_validity"]["status"] == "INVALID_COUNTER" +@pytest.mark.parametrize( + ("offload", "present", "cpu", "status"), + [ + ("none", False, 0, "VALID_COMPLETE"), + ("dram", False, None, "PARTIAL_COMPONENT"), + ("none", True, None, "INVALID_COUNTER"), + ], +) +def test_only_confirmed_absence_of_cpu_can_supply_a_missing_zero(offload, present, cpu, status): + export = _export() + if present: + export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["stats"]["total"] = -1 + else: + del export["metrics"]["atom:prefix_cache_offload_tokens"] + result = _run(export, offload=offload) + _rates(result, 0.6, cpu, 0.6 if cpu == 0 else None) + assert result["cache_validity"]["status"] == status + assert result["cache_validity"]["pricing_eligible"] is (cpu == 0) + + +@pytest.mark.parametrize( + ("stem", "value", "gpu"), + [ + ("cached_tokens", None, None), + ("cached_tokens", True, None), + ("cached_tokens", "600", None), + ("cached_tokens", float("inf"), None), + ("offload_tokens", 500, 0.6), + ], +) +def test_invalid_modern_counter_is_never_published_as_a_valid_rate(stem, value, gpu): + export = _export() + export["metrics"][f"atom:prefix_cache_{stem}"]["series"][0]["stats"]["total"] = value + result = _run(export) + _rates(result, gpu, None, None) + assert result["cache_validity"]["status_class"] == "INVALID" + assert result["cache_validity"]["pricing_eligible"] is False @pytest.mark.parametrize( - "problem", ["missing_total", "overflow", "missing_requests", "scope", "error", "cancel"] + "problem", ["missing_requests", "requests", "tokens", "error", "cancel", "missing_cancel"] ) -def test_invalid_or_different_cohorts_never_become_pricing_input(problem): - export = _export(modern=True) - scope = dict(SCOPE) - if problem == "missing_total": - del export["metrics"]["atom:prefix_cache_cached_tokens"]["series"][0]["stats"]["total"] - elif problem == "overflow": - export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["stats"]["total"] = 500 - elif problem == "missing_requests": +def test_different_or_incomplete_cohorts_are_not_pricing_input(problem): + export, scope = _export(), dict(SCOPE) + if problem == "missing_requests": del export["metrics"]["atom:prefix_cache_requests"] - elif problem == "scope": - scope["profiling_prompt_tokens"] = 999 + elif problem == "missing_cancel": + del scope["cancelled_records"] else: - scope["error_records" if problem == "error" else "cancelled_records"] = 1 + key, value = { + "requests": ("profiling_requests", 4), + "tokens": ("profiling_prompt_tokens", 999), + "error": ("error_records", 1), + "cancel": ("cancelled_records", 1), + }[problem] + scope[key] = value result = _run(export, scope=scope) - _rates( - result, - None if problem == "missing_total" else 0.6, - None if problem in {"missing_total", "overflow"} else 0.1, - None if problem in {"missing_total", "overflow"} else 0.7, - ) + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" assert result["cache_validity"]["pricing_eligible"] is False - assert result["cache_validity"]["status"] == ( - "INVALID_COUNTER" - if problem in {"missing_total", "overflow"} - else "REQUEST_SCOPE_UNRESOLVED" - ) def test_warmup_errors_do_not_block_a_complete_profiling_cohort(tmp_path): @@ -273,35 +228,25 @@ def test_warmup_errors_do_not_block_a_complete_profiling_cohort(tmp_path): path.write_text("\n".join(json.dumps(record) for record in records)) successful, accounting = load_records_with_accounting(path) scope = request_scope_from_records(successful, accounting) + export = _export() + export["metrics"]["atom:prefix_cache_requests"]["series"][0]["stats"]["total"] = 1 assert accounting["records_error_dropped"] == 1 - assert accounting["records_profiling_error_dropped"] == 0 - assert scope == { - "profiling_requests": 1, - "profiling_prompt_tokens": 1000, - "error_records": 0, - "cancelled_records": 0, - } - successful[0]["metrics"] = {} - assert request_scope_from_records(successful, accounting)["profiling_prompt_tokens"] is None + assert scope["error_records"] == 0 + assert _run(export, scope=scope)["cache_validity"]["pricing_eligible"] is True @pytest.mark.parametrize( "usage", [(-10, 1010), (True, 999), (1.9, 999), (float("inf"), 1000), (None, 1000)] ) -def test_invalid_usage_cannot_cancel_out_or_be_truncated_into_a_pricing_cohort(usage): +def test_invalid_usage_cannot_become_a_successful_prompt_cohort(usage): records = [{"metrics": {"usage_prompt_tokens": {"value": value}}} for value in usage] scope = request_scope_from_records(records, {"records_profiling_error_dropped": 0}) assert scope["profiling_prompt_tokens"] is None - export = _export(modern=True) - export["metrics"]["atom:prefix_cache_requests"]["series"][0]["stats"]["total"] = 2 - result = _run(export, scope=scope) - _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" - assert result["cache_validity"]["pricing_eligible"] is False + assert _run(_export(), scope=scope)["cache_validity"]["pricing_eligible"] is False -def test_no_cpu_confirmation_cannot_bypass_conflicting_backend_metadata(): - export = _export(modern=True) +def test_conflicting_offload_metadata_cannot_prove_no_cpu_tier(): + export = _export() del export["metrics"]["atom:prefix_cache_offload_tokens"] _, result, _ = compute_server_metrics( export, @@ -317,13 +262,12 @@ def test_no_cpu_confirmation_cannot_bypass_conflicting_backend_metadata(): }, ) _rates(result, 0.6, None, None) - assert result["cache_validity"]["cpu_component"] == "MISSING" assert result["cache_validity"]["pricing_eligible"] is False @pytest.mark.parametrize(("cached", "offload", "gpu"), [(150, 0, None), (50, 60, 0.25)]) -def test_a_paired_counter_cannot_hide_behind_another_label_domain(cached, offload, gpu): - export = _export(modern=True) +def test_a_paired_counter_cannot_hide_in_another_label_domain(cached, offload, gpu): + export = _export() for stem, values in ( ("full_tokens", (100, 100)), ("cached_tokens", (cached, 0)), @@ -336,34 +280,61 @@ def test_a_paired_counter_cannot_hide_behind_another_label_domain(cached, offloa ] result = _run(export, scope={**SCOPE, "profiling_requests": 2, "profiling_prompt_tokens": 200}) _rates(result, gpu, None, None) - assert result["cache_validity"]["status"] == "INVALID_COUNTER" - assert result["cache_validity"]["pricing_eligible"] is False + assert result["cache_validity"]["status_class"] == "INVALID" -@pytest.mark.parametrize("modern", [False, True]) -def test_only_legacy_callers_may_omit_the_cancelled_count(modern): - scope = {key: value for key, value in SCOPE.items() if key != "cancelled_records"} - result = _run(_export(modern=modern), scope=scope) - assert result["cache_validity"]["pricing_eligible"] is (not modern) - for key, value in ( - ("profiling_requests", True), - ("error_records", False), - ("cancelled_records", None), - ("profiling_prompt_tokens", 1000.1), - ): - result = _run(_export(modern=modern), scope={**SCOPE, key: value}) - assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" +@pytest.mark.parametrize("labels", [None, [], {"rank": 0}]) +def test_null_labels_are_unlabelled_but_malformed_labels_are_invalid(labels): + export = _export() + for entry in export["metrics"].values(): + entry["series"][0]["labels"] = labels + result = _run(export) + _rates( + result, + 0.6 if labels is None else None, + 0.1 if labels is None else None, + 0.7 if labels is None else None, + ) + assert result["cache_validity"]["pricing_eligible"] is (labels is None) -@pytest.mark.parametrize("labels", [None, []]) -def test_null_labels_are_unlabelled_but_malformed_labels_remain_invalid(labels): - export = _export(modern=True) - export["metrics"]["atom:prefix_cache_cached_tokens"]["series"][0]["labels"] = labels +def test_fractional_admissions_cannot_cancel_into_an_integer_cohort(): + export = _export() + for stem, values in ( + ("full_tokens", (500, 500)), + ("cached_tokens", (300, 300)), + ("offload_tokens", (50, 50)), + ("requests", (2.5, 2.5)), + ): + export["metrics"][f"atom:prefix_cache_{stem}"]["series"] = [ + {"endpoint_url": P, "labels": {"engine": str(i)}, "stats": {"total": value}} + for i, value in enumerate(values) + ] result = _run(export) - if labels is None: - _rates(result, 0.6, 0.1, 0.7) - assert result["cache_validity"]["pricing_eligible"] is True + _rates(result, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" + assert result["cache_validity"]["pricing_eligible"] is False + + +def test_absent_metrics_publish_a_missing_ineligible_core(): + _, nested, _ = compute_server_metrics({}, framework="atom", records=[]) + core = nested["cache_validity"] + assert core["schema"] == core["core_schema"] == "infx-cache-validity-core/1" + assert core["status_class"] == "MISSING" and core["basis"] is None + assert core["published"] is False and core["pricing_eligible"] is False + + +@pytest.mark.parametrize("problem", ["gauge", "malformed_series", "partial_requests"]) +def test_non_counter_or_dropped_series_cannot_qualify_a_complete_export(problem): + export = _export() + if problem == "gauge": + export["metrics"]["atom:prefix_cache_full_tokens"]["type"] = "gauge" + elif problem == "malformed_series": + export["metrics"]["atom:prefix_cache_full_tokens"]["series"].append(None) else: - _rates(result, None, None, None) - assert result["cache_validity"]["status"] == "INVALID_COUNTER" - assert result["cache_validity"]["pricing_eligible"] is False + export["metrics"]["atom:prefix_cache_requests"]["series"].append(None) + result = _run(export) + assert result["cache_validity"]["pricing_eligible"] is False + assert result["cache_validity"]["status"] == ( + "REQUEST_SCOPE_UNRESOLVED" if problem == "partial_requests" else "INVALID_COUNTER" + ) From 8c1171bdbf40d6b1d94614ddce13649d05a79111 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 21:19:45 +0000 Subject: [PATCH 5/6] test(agentx): keep transfer diagnostics out of ATOM qualification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：现代 P/D 和 unified 正例包含不完整传输诊断，资格与核心命中率保持有效。 --- .../infx/tests/results/agentic/test_atom_cache_validity.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py index 2326def947..51378cc3d9 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -69,6 +69,9 @@ def _rates(result, gpu, cpu, overall): @pytest.mark.parametrize("unified", [False, True]) def test_modern_pd_and_unified_totals_match_the_successful_cohort(unified): export = _export() + export["metrics"]["atom:lmcache_loaded_tokens"] = { + "type": "counter", "series": [{"endpoint_url": P, "stats": {}}], + } if unified: for entry in export["metrics"].values(): entry["series"] = entry["series"][:1] From b27ca27745739099abb6fda10c86906d7fad6046 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Sun, 25 Oct 2026 17:24:11 +0000 Subject: [PATCH 6/6] fix(agentx): qualify missing CPU counters only for disabled tiers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 确认 CPU offload 关闭时允许已知零值与缺失计数域共存，同时拒绝非零、非法或额外计数域；启用 offload 时继续要求完整覆盖。 --- .../infx/results/agentic/atom_validity.py | 16 ++++--- .../agentic/test_atom_cache_validity.py | 45 ++++++++++++++++--- 2 files changed, 49 insertions(+), 12 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/atom_validity.py b/inferencex-e2e/infx/results/agentic/atom_validity.py index 853d8bb185..7d28fb2aa0 100644 --- a/inferencex-e2e/infx/results/agentic/atom_validity.py +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -124,15 +124,21 @@ def finish(status: str, reason: str | None = None) -> tuple: ): return finish("INVALID_COUNTER", "full/cached tokens outside their valid range") cpu_complete = set(series["offload_tokens"]) == keys - cpu_invalid = invalid["offload_tokens"] or (bool(series["offload_tokens"]) and offload is None) - if cpu_complete and offload is not None: + cpu_invalid = ( + invalid["offload_tokens"] + or bool(set(series["offload_tokens"]) - keys) + or (bool(series["offload_tokens"]) and offload is None) + ) + if kv_offloading == "none" and not cpu_invalid: + cpu_invalid = any(value != 0 for value in values["offload_tokens"].values()) + cpu_complete = True + offload = 0.0 + elif cpu_complete and offload is not None: cpu_invalid |= not math.isfinite(offload) or any( values["cached_tokens"][k] + values["offload_tokens"][k] > values["full_tokens"][k] for k in keys ) - if not series["offload_tokens"] and not cpu_invalid and kv_offloading == "none": - offload = 0.0 - elif not cpu_complete or cpu_invalid: + if not cpu_complete or cpu_invalid: offload = None counts = (full, cached, offload) scope = dict(request_scope or {}) diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py index 51378cc3d9..6df1e47335 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -158,17 +158,18 @@ def test_distinct_label_domains_are_summed_but_duplicate_aliases_are_rejected(): @pytest.mark.parametrize( - ("offload", "present", "cpu", "status"), + ("offload", "total", "cpu", "status"), [ - ("none", False, 0, "VALID_COMPLETE"), - ("dram", False, None, "PARTIAL_COMPONENT"), - ("none", True, None, "INVALID_COUNTER"), + ("none", None, 0, "VALID_COMPLETE"), + ("dram", None, None, "PARTIAL_COMPONENT"), + ("none", -1, None, "INVALID_COUNTER"), + ("none", 100, None, "INVALID_COUNTER"), ], ) -def test_only_confirmed_absence_of_cpu_can_supply_a_missing_zero(offload, present, cpu, status): +def test_only_confirmed_absence_of_cpu_can_supply_a_missing_zero(offload, total, cpu, status): export = _export() - if present: - export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["stats"]["total"] = -1 + if total is not None: + export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["stats"]["total"] = total else: del export["metrics"]["atom:prefix_cache_offload_tokens"] result = _run(export, offload=offload) @@ -177,6 +178,36 @@ def test_only_confirmed_absence_of_cpu_can_supply_a_missing_zero(offload, presen assert result["cache_validity"]["pricing_eligible"] is (cpu == 0) +@pytest.mark.parametrize( + ("offload", "engine", "cpu", "status"), + [ + ("none", "0", 0, "VALID_COMPLETE"), + ("dram", "0", None, "PARTIAL_COMPONENT"), + ("none", "extra", None, "INVALID_COUNTER"), + ], +) +def test_missing_cpu_domains_require_a_disabled_tier_and_matching_present_domains( + offload, engine, cpu, status +): + export = _export() + for stem, values in ( + ("full_tokens", (400, 600)), + ("cached_tokens", (240, 360)), + ("requests", (2, 3)), + ): + export["metrics"][f"atom:prefix_cache_{stem}"]["series"] = [ + {"endpoint_url": P, "labels": {"engine": str(i)}, "stats": {"total": value}} + for i, value in enumerate(values) + ] + export["metrics"]["atom:prefix_cache_offload_tokens"]["series"] = [ + {"endpoint_url": P, "labels": {"engine": engine}, "stats": {"total": 0}} + ] + result = _run(export, offload=offload) + _rates(result, 0.6, cpu, 0.6 if cpu == 0 else None) + assert result["cache_validity"]["status"] == status + assert result["cache_validity"]["pricing_eligible"] is (cpu == 0) + + @pytest.mark.parametrize( ("stem", "value", "gpu"), [