diff --git a/inferencex-e2e/infx/results/agentic/__init__.py b/inferencex-e2e/infx/results/agentic/__init__.py index 638ab24a84..cc9254b4d6 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: @@ -180,6 +180,7 @@ def build_result( "records_dropped_total": 0, "records_warmup_dropped": 0, "records_error_dropped": 0, + "records_profiling_error_dropped": 0, "error_categories": {}, } @@ -232,6 +233,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 dfe8a52bcd..f63b65794c 100644 --- a/inferencex-e2e/infx/results/agentic/artifacts.py +++ b/inferencex-e2e/infx/results/agentic/artifacts.py @@ -47,6 +47,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: @@ -63,6 +64,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..7d28fb2aa0 --- /dev/null +++ b/inferencex-e2e/infx/results/agentic/atom_validity.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import math +from collections.abc import Mapping +from typing import Any + +from .cache_validity import modern_profile_error + +_STEMS = ("full_tokens", "cached_tokens", "offload_tokens", "requests") + + +def _series(metrics: dict[str, Any], stem: str) -> tuple[dict, set]: + names = [ + f"{prefix}prefix_cache_{stem}{suffix}" + for prefix in ("atom:", "atom_") + for suffix in ("", "_total") + ] + found, invalid = {}, set() + for name in names: + if name not in metrics: + continue + 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") + value = stats.get("total") if isinstance(stats, dict) else None + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + or value != int(value) + ): + return None + return value + + +def account_cache( + metrics: dict[str, Any], + *, + endpoint_roles: Mapping[str, str] | None, + 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 finish(status: str, reason: str | None = None) -> tuple: + detail.update(status=status, reasons=[reason] if reason else []) + return counts, detail + + 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 = 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: + 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: {k: s for k, s in group.items() if k[0] in accounted} + for stem, group in series.items() + } + 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() + } + 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 math.isfinite(cached) + or cached > full + or any(values["cached_tokens"][k] > values["full_tokens"][k] for k in keys) + ): + 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(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 cpu_complete or cpu_invalid: + offload = None + counts = (full, cached, offload) + scope = dict(request_scope or {}) + valid_scope = all( + not isinstance(value, bool) + and isinstance(value, (int, float)) + and math.isfinite(value) + and value >= 0 + 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 = ( + valid_scope + and admissions == scope.get("profiling_requests") + and full == scope.get("profiling_prompt_tokens") + and scope.get("error_records") == scope.get("cancelled_records") == 0 + ) + detail["request_scope"] = { + "server_window_admissions": admissions, + "server_window_prompt_tokens": full, + **scope, + "conserved": conserved, + } + if cpu_invalid: + return finish("INVALID_COUNTER", "offload counters invalid or incompatible") + if offload is None: + return finish( + "PARTIAL_COMPONENT", "CPU reuse missing without confirmed absence of CPU tier" + ) + if not conserved: + return finish( + "REQUEST_SCOPE_UNRESOLVED", "counters do not match successful profiling requests" + ) + 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 32543a005b..3c099134d2 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -7,6 +7,7 @@ from collections.abc import Mapping from typing import Any +from infx.results.agentic.atom_validity import account_cache from infx.results.agentic.common import ( gauge_stat, metric_series, @@ -125,6 +126,7 @@ def _cache_counts( class AtomBackend(ServerMetricsBackend): name = "atom" + accepts_context = True def matches(self, metrics: dict[str, dict[str, Any]], framework: str) -> bool: """Return True for atom metric names, or for an empty scrape when ``framework`` is atom.""" @@ -139,11 +141,13 @@ def populate( flat: dict[str, Any], nested: dict[str, Any], *, - endpoint_roles: Mapping[str, str] | None = None, - disaggregated: bool | None = None, - cpu_offload_disabled: bool = False, + context: Mapping[str, Any] | None = None, ) -> None: """Fill token totals, prefix and LMCache hit rates and KV usage from ATOM ``metrics``.""" + context = context or {} + 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: @@ -176,6 +180,21 @@ def populate( full_tokens, cached_tokens, external_tokens = _cache_counts( metrics, endpoint_roles, disaggregated, cpu_offload_disabled ) + counts, detail = account_cache( + metrics, + endpoint_roles=endpoint_roles, + request_scope=context.get("request_scope"), + disagg=disaggregated, + kv_offloading=context.get("kv_offloading"), + collector_export=context.get("collector_export") or {}, + ) + if counts is not None: + full_tokens, cached_tokens, external_tokens = counts + flat["total_prompt_tokens"] = counter_int(full_tokens) + nested["cache_validity"] = detail + sink = context.get("warnings") + if isinstance(sink, list): + 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) @@ -213,6 +232,12 @@ def populate( } ) nested["kv_cache"]["gpu_usage_pct"] = flat["gpu_kv_cache_usage_pct"] + 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/backends/base.py b/inferencex-e2e/infx/results/agentic/backends/base.py index 8aafdc0885..c17639b192 100644 --- a/inferencex-e2e/infx/results/agentic/backends/base.py +++ b/inferencex-e2e/infx/results/agentic/backends/base.py @@ -55,6 +55,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/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 5d176339fe..05dbe94bc2 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -2,19 +2,60 @@ 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, empty_server_metrics, ) +from .cache_validity import summarize_cache_validity 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 endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: if env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": return {} @@ -45,6 +86,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]]: """Return flat and nested server metrics plus warnings from aiperf's ``server_metrics`` export. @@ -66,6 +110,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: @@ -75,19 +120,23 @@ 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 {} adapter.populate( metrics, flat, nested, - endpoint_roles=endpoint_roles_from_env(source_env), - disaggregated={"true": True, "false": False}.get(source_env.get("DISAGG", "").lower()), - cpu_offload_disabled=(source_env.get("KV_OFFLOADING", "").strip().lower() == "none") - and not ( - source_env.get("KV_OFFLOAD_BACKEND") - or source_env.get("KV_OFFLOAD_BACKEND_METADATA") - ), + context={ + "framework": framework, + "endpoint_roles": endpoint_roles + if endpoint_roles is not None + else endpoint_roles_from_env(source_env), + "request_scope": request_scope, + **deployment_from_env(source_env), + **dict(deployment or {}), + "collector_export": server_metrics, + "warnings": warnings, + }, ) else: adapter.populate(metrics, flat, nested) @@ -109,4 +158,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_validity.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py new file mode 100644 index 0000000000..6df1e47335 --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_validity.py @@ -0,0 +1,374 @@ +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, D = "http://prefill/metrics", "http://decode/metrics" +ROLES = {P: "prefill", D: "decode"} +SCOPE = { + "profiling_requests": 5, + "profiling_prompt_tokens": 1000, + "error_records": 0, + "cancelled_records": 0, +} + + +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), + ) + }, + } + + +def _run(export, *, scope=None, offload="dram", roles=ROLES): + flat, nested, _ = compute_server_metrics( + export, + framework="mooncake-atom", + records=[], + endpoint_roles=roles, + request_scope=SCOPE if scope is None else scope, + deployment={"disagg": roles == ROLES, "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("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] + export["metrics"]["atom:prompt_tokens"] = { + "series": [{"endpoint_url": url, "stats": {"total": 1500}} for url in (P, D)] + } + result = _run(export, roles={P: "unified"} if unified else ROLES) + _rates(result, 0.6, 0.1, 0.7) + assert result["tokens"]["prompt_total"] == 1000 + assert result["tokens"]["prompt_by_source"] == { + "gpu_cache_hit": 600, + "cpu_or_external_cache_hit": 100, + "computed": 300, + "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() + 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_class"] == status + assert result["cache_validity"]["pricing_eligible"] is True + + +@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() + 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_class"] == "UNRESOLVED" + assert result["cache_validity"]["pricing_eligible"] is False + + +def test_distinct_label_domains_are_summed_but_duplicate_aliases_are_rejected(): + export = _export() + 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_class"] == "INVALID" + + +@pytest.mark.parametrize( + ("offload", "total", "cpu", "status"), + [ + ("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, total, cpu, status): + export = _export() + 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) + _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( + ("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"), + [ + ("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_requests", "requests", "tokens", "error", "cancel", "missing_cancel"] +) +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 == "missing_cancel": + del scope["cancelled_records"] + else: + 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, 0.6, 0.1, 0.7) + assert result["cache_validity"]["status"] == "REQUEST_SCOPE_UNRESOLVED" + assert result["cache_validity"]["pricing_eligible"] is False + + +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) + export = _export() + export["metrics"]["atom:prefix_cache_requests"]["series"][0]["stats"]["total"] = 1 + assert accounting["records_error_dropped"] == 1 + 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_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 + assert _run(_export(), scope=scope)["cache_validity"]["pricing_eligible"] is False + + +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, + 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"]["pricing_eligible"] is False + + +@pytest.mark.parametrize(("cached", "offload", "gpu"), [(150, 0, None), (50, 60, 0.25)]) +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)), + ("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_class"] == "INVALID" + + +@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) + + +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) + _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: + 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" + ) 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 c736a237db..a32fab31e0 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