Qualify ATOM cache reuse against profiling requests / 核对 ATOM 缓存复用的 profiling 请求范围

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:

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

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")

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"],

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,

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