From 9e3c706b3661f03dd12fdb0cd4e3e7e226cba96e Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Fri, 16 Oct 2026 05:36:30 +0000 Subject: [PATCH 1/6] fix(agentx): Dynamo frontend total, vLLM P/D roles and SGLang non-device tiers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Dynamo: the frontend cached/input total reaches the nested overall, a measured 0 is kept instead of the router estimate, and overall_source records worker_counters, frontend or router_estimate. - vLLM P/D: when every cache series carries a role and prefill is present, only prefill series count, because decode workers report transferred prompt tokens as external_kv_transfer. role_scope records prefill, mixed or combined. - SGLang: external carries every non-device tier (host and storage), so consumers that price GPU + external no longer drop storage hits. The vLLM role fix lowers P/D rates. Until the AIPerf warmup fix is pinned (SemiAnalysisAI/agentx-harness#49), prefill counters still carry the cold-cache warmup, so this should ship with or after that fix. Dynamo 前端总率进入嵌套 overall，测得的 0 不再被 router 估计值替换，并以 overall_source 标注来源；vLLM P/D 在所有缓存序列都带角色且存在 prefill 时只统计 prefill（decode 把传输来的 prompt 计为 external_kv_transfer），以 role_scope 标注；SGLang 的 external 包含所有非 GPU 层（host 与 storage）。vLLM 的角色修正会降低 P/D 命中率；在 AIPerf warmup 修复（agentx-harness#49）固定版本之前，prefill 计数仍含冷缓存 warmup，因此应与该修复同时或在其后发布。 --- .../infx/results/agentic/backends/base.py | 2 + .../results/agentic/backends/dynamo_vllm.py | 8 +- .../infx/results/agentic/backends/sglang.py | 6 +- .../infx/results/agentic/backends/vllm.py | 49 ++++- inferencex-e2e/infx/results/agentic/common.py | 3 + .../agentic/test_process_agentic_result.py | 2 + .../agentic/test_server_metrics_c2c3.py | 204 ++++++++++++++++++ 7 files changed, 264 insertions(+), 10 deletions(-) create mode 100644 inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py diff --git a/inferencex-e2e/infx/results/agentic/backends/base.py b/inferencex-e2e/infx/results/agentic/backends/base.py index da6e68b463..4ee3c95f12 100644 --- a/inferencex-e2e/infx/results/agentic/backends/base.py +++ b/inferencex-e2e/infx/results/agentic/backends/base.py @@ -78,6 +78,8 @@ def empty_server_metrics(present: bool, metric_count: int) -> dict[str, Any]: "router_shared_cache_hit_rate": None, "frontend_cached_tokens": None, "frontend_input_tokens": None, + "overall_source": None, + "role_scope": None, }, "cache_validity": None, "kv_cache": { diff --git a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py index 273be93f58..e57ad40cb6 100644 --- a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py @@ -92,7 +92,13 @@ def _populate_dynamo( ) ) if flat["server_overall_cache_hit_rate"] is None: - flat["server_overall_cache_hit_rate"] = frontend_hit_rate or router_shared_hit_rate + if frontend_hit_rate is not None: + flat["server_overall_cache_hit_rate"] = frontend_hit_rate + nested["cache"]["overall_source"] = "frontend" + elif router_shared_hit_rate is not None: + flat["server_overall_cache_hit_rate"] = router_shared_hit_rate + nested["cache"]["overall_source"] = "router_estimate" + nested["cache"]["overall_cache_hit_rate"] = flat["server_overall_cache_hit_rate"] nested["cache"].update( { diff --git a/inferencex-e2e/infx/results/agentic/backends/sglang.py b/inferencex-e2e/infx/results/agentic/backends/sglang.py index c49d0ea038..82df923734 100644 --- a/inferencex-e2e/infx/results/agentic/backends/sglang.py +++ b/inferencex-e2e/infx/results/agentic/backends/sglang.py @@ -59,10 +59,14 @@ def populate( device_hits = cached_by_source.get("device") host_hits = cached_by_source.get("host") total_cached = sum(cached_by_source.values()) if cached_by_source else None + non_device = [v for k, v in cached_by_source.items() if k != "device"] flat["server_gpu_cache_hit_rate"] = rate(device_hits, prompt_total) flat["server_cpu_cache_hit_rate"] = rate(host_hits, prompt_total) - flat["server_external_cache_hit_rate"] = flat["server_cpu_cache_hit_rate"] + # Consumers price GPU + external, so external carries every non-device tier (host and storage). + flat["server_external_cache_hit_rate"] = ( + rate(sum(non_device), prompt_total) if non_device else None + ) flat["server_overall_cache_hit_rate"] = rate(total_cached, prompt_total) if flat["server_overall_cache_hit_rate"] is None: diff --git a/inferencex-e2e/infx/results/agentic/backends/vllm.py b/inferencex-e2e/infx/results/agentic/backends/vllm.py index 5baf5a47f2..1bf5035e54 100644 --- a/inferencex-e2e/infx/results/agentic/backends/vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/vllm.py @@ -41,15 +41,18 @@ def populate( flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) + role_scope, role_filter = _cache_role_filter(metrics) prefix_hits = sum_stat( metrics, "vllm:prefix_cache_hits", preferred_keys=("total", "sum", "max", "avg"), + series_filter=role_filter, ) prefix_queries = sum_stat( metrics, "vllm:prefix_cache_queries", preferred_keys=("total", "sum", "max", "avg"), + series_filter=role_filter, ) gpu_rate = rate(prefix_hits, prefix_queries) flat["server_gpu_cache_hit_rate"] = gpu_rate @@ -58,11 +61,13 @@ def populate( metrics, "vllm:external_prefix_cache_hits", preferred_keys=("total", "sum", "max", "avg"), + series_filter=role_filter, ) external_queries = sum_stat( metrics, "vllm:external_prefix_cache_queries", preferred_keys=("total", "sum", "max", "avg"), + series_filter=role_filter, ) external_rate = rate(external_hits, external_queries) flat["server_external_cache_hit_rate"] = external_rate @@ -73,6 +78,7 @@ def populate( "vllm:prompt_tokens_by_source", "source", preferred_keys=("total", "sum", "max", "avg"), + series_filter=role_filter, ) if prompt_by_source: local_cache_hit = prompt_by_source.get("local_cache_hit") @@ -143,6 +149,10 @@ def populate( "prefix_cache_queries": prefix_queries, "external_prefix_cache_hits": external_hits, "external_prefix_cache_queries": external_queries, + "overall_source": "worker_counters" + if flat["server_overall_cache_hit_rate"] is not None + else None, + "role_scope": role_scope, } ) nested["kv_cache"].update( @@ -296,17 +306,40 @@ def series_filter(series: dict, source_id: str = source_id) -> bool: return sources -def _source_id(series: dict[str, Any]) -> str: +_CACHE_METRICS = ( + "vllm:prompt_tokens_by_source", + "vllm:prefix_cache_hits", + "vllm:prefix_cache_queries", + "vllm:external_prefix_cache_hits", + "vllm:external_prefix_cache_queries", +) + + +def _series_role(series: dict[str, Any]) -> str: labels = series.get("labels") if isinstance(series.get("labels"), dict) else {} component = str(labels.get("dynamo_component") or labels.get("component") or "") if component == "prefill": - role = "prefill" - elif component in ("backend", "decode"): - role = "decode" - elif component in ("frontend", "router"): - role = "router" - else: - role = "combined" + return "prefill" + if component in ("backend", "decode"): + return "decode" + if component in ("frontend", "router"): + return "router" + return "combined" + + +def _cache_role_filter(metrics: dict[str, dict[str, Any]]) -> tuple[str, Any]: + # A P/D decode worker counts every transferred prompt token as external_kv_transfer, so only + # prefill series measure reuse; unlabelled series next to prefill ones are kept, not dropped. + roles = {_series_role(s) for name in _CACHE_METRICS for s in metric_series(metrics, name)} + if "prefill" not in roles: + return "combined", None + if "combined" in roles: + return "mixed", None + return "prefill", lambda series: _series_role(series) == "prefill" + + +def _source_id(series: dict[str, Any]) -> str: + role = _series_role(series) parts = [role] endpoint = series.get("endpoint_url") diff --git a/inferencex-e2e/infx/results/agentic/common.py b/inferencex-e2e/infx/results/agentic/common.py index eb498c9fd6..946dee1226 100644 --- a/inferencex-e2e/infx/results/agentic/common.py +++ b/inferencex-e2e/infx/results/agentic/common.py @@ -189,10 +189,13 @@ def sum_by_label( label_keys: str | list[str], *, preferred_keys: tuple[str, ...] = ("total", "sum", "max", "avg"), + series_filter: Callable[[dict[str, Any]], bool] | None = None, ) -> dict[str, float]: keys = [label_keys] if isinstance(label_keys, str) else label_keys out: dict[str, float] = {} for series in metric_series(metrics, metric_names): + if series_filter is not None and not series_filter(series): + continue labels = series.get("labels") if not isinstance(labels, dict): continue 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 664495f3e7..87517b6c9d 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 @@ -98,6 +98,8 @@ "router_shared_cache_hit_rate", "frontend_cached_tokens", "frontend_input_tokens", + "overall_source", + "role_scope", } SERVER_KV_CACHE_KEYS = { "gpu_usage_pct", diff --git a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py new file mode 100644 index 0000000000..b9aeb9d334 --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py @@ -0,0 +1,204 @@ +"""Dynamo frontend overall, vLLM P/D roles and SGLang non-device tiers. + +Synthetic AIPerf 0.12 exports (counter names without ``_total``). These adapters do not assess +phase validity yet (``NOT_EVALUATED``); the cases pin which series each rate is built from. +""" + +from __future__ import annotations + +import pytest + +from infx.results.agentic.server_metrics import compute_server_metrics + +FE = "http://10.0.0.1:8000/metrics" +PRE = "http://10.0.0.2:7500/metrics" +DEC = "http://10.0.0.3:7501/metrics" + + +def _counter(rows: list[tuple[str, dict, float]]) -> dict: + return { + "type": "counter", + "series": [ + {"endpoint_url": e, "labels": labels, "stats": {"total": float(v)}} + for e, labels, v in rows + ], + } + + +def _histogram(endpoint: str, total: float, count: float = 10.0) -> dict: + return { + "type": "histogram", + "series": [ + {"endpoint_url": endpoint, "labels": {}, "stats": {"sum": float(total), "count": count}} + ], + } + + +def _gauge(endpoint: str, value: float) -> dict: + return { + "type": "gauge", + "series": [ + { + "endpoint_url": endpoint, + "labels": {}, + "stats": {"avg": float(value), "max": float(value)}, + } + ], + } + + +def _run(metrics: dict, framework: str) -> tuple[dict, dict]: + flat, nested, _ = compute_server_metrics({"metrics": metrics}, framework=framework, records=[]) + return flat, nested + + +def _vllm_role(component: str, source: str | None = None) -> dict: + return {"dynamo_component": component, **({"source": source} if source else {})} + + +class TestDynamoFrontendOverall: + def test_frontend_total_reaches_nested_overall(self) -> None: + metrics = { + "dynamo_frontend_cached_tokens": _histogram(FE, 800), + "dynamo_frontend_input_sequence_tokens": _histogram(FE, 1000), + } + flat, nested = _run(metrics, "dynamo-vllm") + assert flat["server_overall_cache_hit_rate"] == pytest.approx(0.8) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.8) + assert nested["cache"]["overall_source"] == "frontend" + + def test_measured_zero_frontend_rate_is_kept(self) -> None: + metrics = { + "dynamo_frontend_cached_tokens": _histogram(FE, 0), + "dynamo_frontend_input_sequence_tokens": _histogram(FE, 1000), + "dynamo_component_router_shared_cache_hit_rate": _gauge(FE, 0.7), + } + flat, nested = _run(metrics, "dynamo-vllm") + assert flat["server_overall_cache_hit_rate"] == 0.0 + assert nested["cache"]["overall_cache_hit_rate"] == 0.0 + assert nested["cache"]["overall_source"] == "frontend" + + def test_router_estimate_is_labelled_as_an_estimate(self) -> None: + metrics = {"dynamo_component_router_shared_cache_hit_rate": _gauge(FE, 0.7)} + flat, nested = _run(metrics, "dynamo-vllm") + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.7) + assert nested["cache"]["overall_source"] == "router_estimate" + + def test_worker_counters_take_precedence_over_the_frontend(self) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [ + (PRE, _vllm_role("prefill", "local_cache_hit"), 600), + (PRE, _vllm_role("prefill", "local_compute"), 400), + ] + ), + "dynamo_frontend_cached_tokens": _histogram(FE, 800), + "dynamo_frontend_input_sequence_tokens": _histogram(FE, 1000), + } + _, nested = _run(metrics, "dynamo-vllm") + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + assert nested["cache"]["overall_source"] == "worker_counters" + assert nested["cache"]["frontend_cache_hit_rate"] == pytest.approx(0.8) + + +class TestVllmPrefillDecodeRoles: + def test_decode_kv_transfer_is_not_a_cache_hit(self) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [ + (DEC, _vllm_role("backend", "external_kv_transfer"), 1000), + (DEC, _vllm_role("backend", "local_cache_hit"), 0), + (DEC, _vllm_role("backend", "local_compute"), 0), + (PRE, _vllm_role("prefill", "external_kv_transfer"), 500), + (PRE, _vllm_role("prefill", "local_cache_hit"), 200), + (PRE, _vllm_role("prefill", "local_compute"), 300), + ] + ) + } + _, nested = _run(metrics, "dynamo-vllm") + assert nested["cache"]["gpu_cache_hit_rate"] == pytest.approx(0.2) + assert nested["cache"]["external_cache_hit_rate"] == pytest.approx(0.5) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.7) + assert nested["cache"]["role_scope"] == "prefill" + assert nested["tokens"]["prompt_by_source"]["cpu_or_external_cache_hit"] == pytest.approx( + 500 + ) + + def test_prefix_cache_counters_use_the_prefill_role(self) -> None: + metrics = { + "vllm:prefix_cache_hits": _counter( + [(PRE, _vllm_role("prefill"), 600), (DEC, _vllm_role("backend"), 0)] + ), + "vllm:prefix_cache_queries": _counter( + [(PRE, _vllm_role("prefill"), 1000), (DEC, _vllm_role("backend"), 1000)] + ), + } + _, nested = _run(metrics, "dynamo-vllm") + assert nested["cache"]["gpu_cache_hit_rate"] == pytest.approx(0.6) + assert nested["cache"]["prefix_cache_hits"] == pytest.approx(600) + assert nested["cache"]["prefix_cache_queries"] == pytest.approx(1000) + + def test_unlabelled_engines_keep_the_combined_scope(self) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [(PRE, {"source": "local_cache_hit"}, 300), (PRE, {"source": "local_compute"}, 700)] + ) + } + _, nested = _run(metrics, "vllm") + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.3) + assert nested["cache"]["role_scope"] == "combined" + + +class TestSglangNonDeviceTiers: + @staticmethod + def _metrics(cached: dict[str, float], prompt: float = 1000.0) -> dict: + return { + "sglang:prompt_tokens": _counter([(PRE, {}, prompt)]), + "sglang:cached_tokens": _counter( + [(PRE, {"cache_source": k}, v) for k, v in cached.items()] + ), + } + + def test_storage_tier_is_counted_with_the_other_non_device_tiers(self) -> None: + _, nested = _run(self._metrics({"device": 600, "host": 200, "storage": 100}), "sglang") + cache = nested["cache"] + assert cache["gpu_cache_hit_rate"] == pytest.approx(0.6) + assert cache["cpu_cache_hit_rate"] == pytest.approx(0.2) + assert cache["external_cache_hit_rate"] == pytest.approx(0.3) + assert cache["overall_cache_hit_rate"] == pytest.approx(0.9) + assert cache["gpu_cache_hit_rate"] + cache["external_cache_hit_rate"] == pytest.approx( + cache["overall_cache_hit_rate"] + ) + + def test_device_and_host_only_keep_external_equal_to_host(self) -> None: + _, nested = _run(self._metrics({"device": 600, "host": 200}), "sglang") + assert nested["cache"]["external_cache_hit_rate"] == pytest.approx(0.2) + assert nested["cache"]["cpu_cache_hit_rate"] == pytest.approx(0.2) + + def test_symmetric_prefill_and_decode_sum_keeps_the_ratio(self) -> None: + roles = {"prefill": {"engine_type": "prefill"}, "decode": {"engine_type": "decode"}} + metrics = { + "sglang:prompt_tokens": _counter( + [(PRE, roles["prefill"], 1000), (DEC, roles["decode"], 1000)] + ), + "sglang:cached_tokens": _counter( + [ + (PRE, {**roles["prefill"], "cache_source": "device"}, 900), + (DEC, {**roles["decode"], "cache_source": "device"}, 900), + ] + ), + } + _, nested = _run(metrics, "sglang") + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.9) + + +def test_frontend_only_total_reaches_the_cache_validity_core() -> None: + metrics = { + "dynamo_frontend_cached_tokens": _histogram(FE, 800), + "dynamo_frontend_input_sequence_tokens": _histogram(FE, 1000), + } + _, nested = _run(metrics, "dynamo-vllm") + core = nested["cache_validity"] + assert core["status_class"] == "NOT_EVALUATED" + assert core["published"] is True + assert core["basis"] == "combined" From 3b7fcbc76fb2a2c4b141f2c6a1b825fff4703ac7 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 20:30:25 +0000 Subject: [PATCH 2/6] fix(agentx): isolate decode transfers and router estimates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按部署角色排除 decode 传输，保留未知角色遥测但不授予计价资格；router 估计仅保留诊断，增加角色、来源及 profiling 总量回归。 --- .../results/agentic/backends/dynamo_vllm.py | 2 +- .../infx/results/agentic/backends/sglang.py | 2 +- .../infx/results/agentic/backends/vllm.py | 57 ++++++++----- .../agentic/test_server_metrics_c2c3.py | 80 +++++++++++++++++-- 4 files changed, 113 insertions(+), 28 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py index e57ad40cb6..fd6b222536 100644 --- a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py @@ -94,11 +94,11 @@ def _populate_dynamo( if flat["server_overall_cache_hit_rate"] is None: if frontend_hit_rate is not None: flat["server_overall_cache_hit_rate"] = frontend_hit_rate + nested["cache"]["overall_cache_hit_rate"] = frontend_hit_rate nested["cache"]["overall_source"] = "frontend" elif router_shared_hit_rate is not None: flat["server_overall_cache_hit_rate"] = router_shared_hit_rate nested["cache"]["overall_source"] = "router_estimate" - nested["cache"]["overall_cache_hit_rate"] = flat["server_overall_cache_hit_rate"] nested["cache"].update( { diff --git a/inferencex-e2e/infx/results/agentic/backends/sglang.py b/inferencex-e2e/infx/results/agentic/backends/sglang.py index 82df923734..b6fa20e0da 100644 --- a/inferencex-e2e/infx/results/agentic/backends/sglang.py +++ b/inferencex-e2e/infx/results/agentic/backends/sglang.py @@ -63,7 +63,7 @@ def populate( flat["server_gpu_cache_hit_rate"] = rate(device_hits, prompt_total) flat["server_cpu_cache_hit_rate"] = rate(host_hits, prompt_total) - # Consumers price GPU + external, so external carries every non-device tier (host and storage). + # External reuse includes host and storage; CPU reuse includes host only. flat["server_external_cache_hit_rate"] = ( rate(sum(non_device), prompt_total) if non_device else None ) diff --git a/inferencex-e2e/infx/results/agentic/backends/vllm.py b/inferencex-e2e/infx/results/agentic/backends/vllm.py index 1bf5035e54..2c82a3558c 100644 --- a/inferencex-e2e/infx/results/agentic/backends/vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/vllm.py @@ -41,7 +41,8 @@ def populate( flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) - role_scope, role_filter = _cache_role_filter(metrics) + disaggregated = nested["validity"]["deployment"]["disaggregated"] + role_scope, role_filter = _cache_role_filter(metrics, disaggregated) prefix_hits = sum_stat( metrics, "vllm:prefix_cache_hits", @@ -181,7 +182,7 @@ def populate( "generation_total": flat["total_generation_tokens"], } ) - nested["sources"] = _vllm_sources(metrics) + nested["sources"] = _vllm_sources(metrics, disaggregated) def prompt_generation_totals( self, @@ -260,7 +261,9 @@ def first_counter_total( return None -def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: +def _vllm_sources( + metrics: dict[str, dict[str, Any]], disaggregated: bool | None +) -> list[dict[str, Any]]: source_ids = set() for metric_name in ( "vllm:prompt_tokens", @@ -271,7 +274,7 @@ def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: "vllm:prompt_tokens_by_source", ): for series in metric_series(metrics, metric_name): - source_ids.add(_source_id(series)) + source_ids.add(_source_id(series, disaggregated)) sources: list[dict[str, Any]] = [] for source_id in sorted(source_ids): @@ -279,7 +282,7 @@ def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: continue def series_filter(series: dict, source_id: str = source_id) -> bool: - return _source_id(series) == source_id + return _source_id(series, disaggregated) == source_id prompt_tokens = sum_stat(metrics, "vllm:prompt_tokens", series_filter=series_filter) generation_tokens = sum_stat(metrics, "vllm:generation_tokens", series_filter=series_filter) @@ -315,11 +318,20 @@ def series_filter(series: dict, source_id: str = source_id) -> bool: ) -def _series_role(series: dict[str, Any]) -> str: +def _series_role(series: dict[str, Any], disaggregated: bool | None = None) -> str: labels = series.get("labels") if isinstance(series.get("labels"), dict) else {} + mode = str(labels.get("disaggregation_mode") or "").lower() + if mode in ("prefill", "context"): + return "prefill" + if mode == "decode": + return "decode" + if mode in ("combined", "unified", "prefill_and_decode"): + return "combined" component = str(labels.get("dynamo_component") or labels.get("component") or "") if component == "prefill": return "prefill" + if component == "backend" and disaggregated is False: + return "combined" if component in ("backend", "decode"): return "decode" if component in ("frontend", "router"): @@ -327,19 +339,26 @@ def _series_role(series: dict[str, Any]) -> str: return "combined" -def _cache_role_filter(metrics: dict[str, dict[str, Any]]) -> tuple[str, Any]: - # A P/D decode worker counts every transferred prompt token as external_kv_transfer, so only - # prefill series measure reuse; unlabelled series next to prefill ones are kept, not dropped. - roles = {_series_role(s) for name in _CACHE_METRICS for s in metric_series(metrics, name)} - if "prefill" not in roles: - return "combined", None - if "combined" in roles: - return "mixed", None - return "prefill", lambda series: _series_role(series) == "prefill" - - -def _source_id(series: dict[str, Any]) -> str: - role = _series_role(series) +def _cache_role_filter( + metrics: dict[str, dict[str, Any]], disaggregated: bool | None +) -> tuple[str, Any]: + roles = { + _series_role(s, disaggregated) + for name in _CACHE_METRICS + for s in metric_series(metrics, name) + } + if "prefill" in roles: + scope = "mixed" if "combined" in roles else "prefill" + else: + scope = "unresolved" if "decode" in roles or disaggregated is True else "combined" + if disaggregated is True and "combined" in roles: + scope = "unresolved" + # Known decode counters measure KV transfer, including alongside unlabelled workers. + return scope, lambda series: _series_role(series, disaggregated) not in ("decode", "router") + + +def _source_id(series: dict[str, Any], disaggregated: bool | None) -> str: + role = _series_role(series, disaggregated) parts = [role] endpoint = series.get("endpoint_url") diff --git a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py index b9aeb9d334..3079134fea 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py @@ -1,9 +1,3 @@ -"""Dynamo frontend overall, vLLM P/D roles and SGLang non-device tiers. - -Synthetic AIPerf 0.12 exports (counter names without ``_total``). These adapters do not assess -phase validity yet (``NOT_EVALUATED``); the cases pin which series each rate is built from. -""" - from __future__ import annotations import pytest @@ -81,8 +75,12 @@ def test_measured_zero_frontend_rate_is_kept(self) -> None: def test_router_estimate_is_labelled_as_an_estimate(self) -> None: metrics = {"dynamo_component_router_shared_cache_hit_rate": _gauge(FE, 0.7)} flat, nested = _run(metrics, "dynamo-vllm") - assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.7) + assert flat["server_overall_cache_hit_rate"] == pytest.approx(0.7) + assert nested["cache"]["overall_cache_hit_rate"] is None assert nested["cache"]["overall_source"] == "router_estimate" + assert nested["cache_validity"]["published"] is False + assert nested["cache_validity"]["basis"] is None + assert nested["cache_validity"]["pricing_eligible"] is False def test_worker_counters_take_precedence_over_the_frontend(self) -> None: metrics = { @@ -148,6 +146,59 @@ def test_unlabelled_engines_keep_the_combined_scope(self) -> None: assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.3) assert nested["cache"]["role_scope"] == "combined" + def test_mixed_roles_exclude_known_decode_transfers(self) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [ + (PRE, _vllm_role("prefill", "local_cache_hit"), 600), + (PRE, _vllm_role("prefill", "local_compute"), 400), + (DEC, _vllm_role("backend", "external_kv_transfer"), 1000), + (FE, {"source": "local_compute"}, 100), + ] + ) + } + _, nested = _run(metrics, "dynamo-vllm") + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(600 / 1100) + assert nested["cache"]["role_scope"] == "mixed" + assert nested["tokens"]["prompt_by_source"]["cpu_or_external_cache_hit"] is None + assert nested["cache_validity"]["pricing_eligible"] is False + + def test_decode_only_has_no_reuse_measurement(self) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [(DEC, _vllm_role("backend", "external_kv_transfer"), 1000)] + ) + } + _, nested, _ = compute_server_metrics( + {"metrics": metrics}, framework="dynamo-vllm", records=[], disaggregated=True + ) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache"]["role_scope"] == "unresolved" + assert nested["cache_validity"]["published"] is False + + @pytest.mark.parametrize("disaggregated", [False, True]) + def test_backend_component_respects_deployment_mode(self, disaggregated: bool) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [ + (PRE, _vllm_role("backend", "local_cache_hit"), 300), + (PRE, _vllm_role("backend", "local_compute"), 700), + ] + ) + } + _, nested, _ = compute_server_metrics( + {"metrics": metrics}, + framework="dynamo-vllm", + records=[], + disaggregated=disaggregated, + ) + if disaggregated: + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache"]["role_scope"] == "unresolved" + else: + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.3) + assert nested["cache"]["role_scope"] == "combined" + class TestSglangNonDeviceTiers: @staticmethod @@ -202,3 +253,18 @@ def test_frontend_only_total_reaches_the_cache_validity_core() -> None: assert core["status_class"] == "NOT_EVALUATED" assert core["published"] is True assert core["basis"] == "combined" + + +@pytest.mark.parametrize("framework", ["vllm", "sglang"]) +@pytest.mark.parametrize("records", [[], [{"metrics": {"input_sequence_length": 1000}}]]) +def test_disaggregated_prompt_total_uses_the_profile_cohort( + framework: str, records: list[dict] +) -> None: + metrics = {f"{framework}:prompt_tokens": _counter([(PRE, {}, 1000), (DEC, {}, 1000)])} + flat, nested, _ = compute_server_metrics( + {"metrics": metrics}, framework=framework, records=records, disaggregated=True + ) + expected = 1000 if records else None + assert flat["total_prompt_tokens"] == expected + assert nested["tokens"]["prompt_total"] == expected + assert nested["cache_validity"]["pricing_eligible"] is False From 15bfaea1bcbac4b3cf74c010be94feac3ce573f5 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 20:42:25 +0000 Subject: [PATCH 3/6] fix(agentx): keep explicit unknown vLLM roles unresolved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 显式非法或空角色不能作为无标签聚合回退发布复用率，保留真正无标签的遥测路径并增加行为边界覆盖。 --- .../infx/results/agentic/backends/vllm.py | 6 ++++-- .../results/agentic/test_server_metrics_c2c3.py | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/vllm.py b/inferencex-e2e/infx/results/agentic/backends/vllm.py index 2c82a3558c..797e7daec2 100644 --- a/inferencex-e2e/infx/results/agentic/backends/vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/vllm.py @@ -327,6 +327,8 @@ def _series_role(series: dict[str, Any], disaggregated: bool | None = None) -> s return "decode" if mode in ("combined", "unified", "prefill_and_decode"): return "combined" + if "disaggregation_mode" in labels: + return "unknown" component = str(labels.get("dynamo_component") or labels.get("component") or "") if component == "prefill": return "prefill" @@ -351,10 +353,10 @@ def _cache_role_filter( scope = "mixed" if "combined" in roles else "prefill" else: scope = "unresolved" if "decode" in roles or disaggregated is True else "combined" - if disaggregated is True and "combined" in roles: + if "unknown" in roles or (disaggregated is True and "combined" in roles): scope = "unresolved" # Known decode counters measure KV transfer, including alongside unlabelled workers. - return scope, lambda series: _series_role(series, disaggregated) not in ("decode", "router") + return scope, lambda series: _series_role(series, disaggregated) in ("prefill", "combined") def _source_id(series: dict[str, Any], disaggregated: bool | None) -> str: diff --git a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py index 3079134fea..7b1592a55a 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py @@ -199,6 +199,20 @@ def test_backend_component_respects_deployment_mode(self, disaggregated: bool) - assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.3) assert nested["cache"]["role_scope"] == "combined" + @pytest.mark.parametrize("mode", ["garbage", None]) + def test_explicit_unknown_mode_does_not_become_combined(self, mode: str | None) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [(PRE, {"disaggregation_mode": mode, "source": "external_kv_transfer"}, 1000)] + ) + } + _, nested, _ = compute_server_metrics( + {"metrics": metrics}, framework="vllm", records=[], disaggregated=False + ) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache"]["role_scope"] == "unresolved" + assert nested["cache_validity"]["published"] is False + class TestSglangNonDeviceTiers: @staticmethod From 8751c0821803c3cdf29907885ac7a205fd1533e9 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Sun, 25 Oct 2026 17:25:56 +0000 Subject: [PATCH 4/6] fix(results): apply declared vLLM roles and normalize SGLang storage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将已声明的 endpoint 角色传入 vLLM 与 Dynamo，避免无角色标签的 decode 传输进入命中率；冲突角色保留未知状态。同步 SGLang normalized external token 与 storage 命中率，并补充行为回归。 --- .../results/agentic/backends/dynamo_vllm.py | 4 +- .../infx/results/agentic/backends/sglang.py | 2 +- .../infx/results/agentic/backends/vllm.py | 64 +++++++++++++++---- .../agentic/test_server_metrics_c2c3.py | 51 ++++++++++++++- 4 files changed, 104 insertions(+), 17 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py index fd6b222536..1933be4123 100644 --- a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py @@ -42,8 +42,10 @@ def populate( metrics: dict[str, dict[str, Any]], flat: dict[str, Any], nested: dict[str, Any], + *, + context: dict[str, Any] | None = None, ) -> None: - super().populate(metrics, flat, nested) + super().populate(metrics, flat, nested, context=context) self._populate_dynamo(metrics, flat, nested) def _populate_dynamo( diff --git a/inferencex-e2e/infx/results/agentic/backends/sglang.py b/inferencex-e2e/infx/results/agentic/backends/sglang.py index b6fa20e0da..07bbf3072b 100644 --- a/inferencex-e2e/infx/results/agentic/backends/sglang.py +++ b/inferencex-e2e/infx/results/agentic/backends/sglang.py @@ -138,7 +138,7 @@ def populate( "generation_total": flat["total_generation_tokens"], "prompt_by_source": { "gpu_cache_hit": device_hits, - "cpu_or_external_cache_hit": host_hits, + "cpu_or_external_cache_hit": sum(non_device) if non_device else None, "computed": prefill_compute, "raw": cached_by_source, }, diff --git a/inferencex-e2e/infx/results/agentic/backends/vllm.py b/inferencex-e2e/infx/results/agentic/backends/vllm.py index 797e7daec2..32a67caf44 100644 --- a/inferencex-e2e/infx/results/agentic/backends/vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/vllm.py @@ -3,7 +3,7 @@ from __future__ import annotations import re -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from typing import Any from infx.results.agentic.common import ( @@ -22,6 +22,7 @@ class VllmBackend(ServerMetricsBackend): name = "vllm" + accepts_context = True _ENGINE_TAG_RE = re.compile(r"\((?PEngineCore(?:_DP\d+)?)\s+pid=\d+\)") _GPU_KV_SIZE_RE = re.compile(r"GPU KV cache size:\s*(?P[\d,]+)\s*tokens") @@ -36,13 +37,16 @@ def populate( metrics: dict[str, dict[str, Any]], flat: dict[str, Any], nested: dict[str, Any], + *, + context: dict[str, Any] | None = None, ) -> None: + endpoint_roles = (context or {}).get("endpoint_roles") prompt_total, generation_total = self.prompt_generation_totals(metrics) flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) disaggregated = nested["validity"]["deployment"]["disaggregated"] - role_scope, role_filter = _cache_role_filter(metrics, disaggregated) + role_scope, role_filter = _cache_role_filter(metrics, disaggregated, endpoint_roles) prefix_hits = sum_stat( metrics, "vllm:prefix_cache_hits", @@ -182,7 +186,7 @@ def populate( "generation_total": flat["total_generation_tokens"], } ) - nested["sources"] = _vllm_sources(metrics, disaggregated) + nested["sources"] = _vllm_sources(metrics, disaggregated, endpoint_roles) def prompt_generation_totals( self, @@ -262,7 +266,9 @@ def first_counter_total( def _vllm_sources( - metrics: dict[str, dict[str, Any]], disaggregated: bool | None + metrics: dict[str, dict[str, Any]], + disaggregated: bool | None, + endpoint_roles: Mapping[str, str] | None = None, ) -> list[dict[str, Any]]: source_ids = set() for metric_name in ( @@ -274,7 +280,7 @@ def _vllm_sources( "vllm:prompt_tokens_by_source", ): for series in metric_series(metrics, metric_name): - source_ids.add(_source_id(series, disaggregated)) + source_ids.add(_source_id(series, disaggregated, endpoint_roles)) sources: list[dict[str, Any]] = [] for source_id in sorted(source_ids): @@ -282,7 +288,7 @@ def _vllm_sources( continue def series_filter(series: dict, source_id: str = source_id) -> bool: - return _source_id(series, disaggregated) == source_id + return _source_id(series, disaggregated, endpoint_roles) == source_id prompt_tokens = sum_stat(metrics, "vllm:prompt_tokens", series_filter=series_filter) generation_tokens = sum_stat(metrics, "vllm:generation_tokens", series_filter=series_filter) @@ -318,7 +324,28 @@ def series_filter(series: dict, source_id: str = source_id) -> bool: ) -def _series_role(series: dict[str, Any], disaggregated: bool | None = None) -> str: +def _series_role( + series: dict[str, Any], + disaggregated: bool | None = None, + endpoint_roles: Mapping[str, str] | None = None, +) -> str: + role = _label_role(series, disaggregated) + declared = (endpoint_roles or {}).get(str(series.get("endpoint_url", ""))) + if declared is None: + return role + declared = { + "prefill": "prefill", + "decode": "decode", + "unified": "combined", + "combined": "combined", + }.get(declared, "unknown") + labels = series.get("labels") if isinstance(series.get("labels"), dict) else {} + if any(key in labels for key in ("disaggregation_mode", "dynamo_component", "component")): + return role if role == declared else "unknown" + return declared + + +def _label_role(series: dict[str, Any], disaggregated: bool | None = None) -> str: labels = series.get("labels") if isinstance(series.get("labels"), dict) else {} mode = str(labels.get("disaggregation_mode") or "").lower() if mode in ("prefill", "context"): @@ -342,10 +369,12 @@ def _series_role(series: dict[str, Any], disaggregated: bool | None = None) -> s def _cache_role_filter( - metrics: dict[str, dict[str, Any]], disaggregated: bool | None + metrics: dict[str, dict[str, Any]], + disaggregated: bool | None, + endpoint_roles: Mapping[str, str] | None = None, ) -> tuple[str, Any]: roles = { - _series_role(s, disaggregated) + _series_role(s, disaggregated, endpoint_roles) for name in _CACHE_METRICS for s in metric_series(metrics, name) } @@ -356,11 +385,18 @@ def _cache_role_filter( if "unknown" in roles or (disaggregated is True and "combined" in roles): scope = "unresolved" # Known decode counters measure KV transfer, including alongside unlabelled workers. - return scope, lambda series: _series_role(series, disaggregated) in ("prefill", "combined") - - -def _source_id(series: dict[str, Any], disaggregated: bool | None) -> str: - role = _series_role(series, disaggregated) + return scope, lambda series: _series_role(series, disaggregated, endpoint_roles) in ( + "prefill", + "combined", + ) + + +def _source_id( + series: dict[str, Any], + disaggregated: bool | None, + endpoint_roles: Mapping[str, str] | None = None, +) -> str: + role = _series_role(series, disaggregated, endpoint_roles) parts = [role] endpoint = series.get("endpoint_url") diff --git a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py index 7b1592a55a..c1d267befe 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py @@ -100,6 +100,45 @@ def test_worker_counters_take_precedence_over_the_frontend(self) -> None: class TestVllmPrefillDecodeRoles: + @pytest.mark.parametrize("framework", ["vllm", "dynamo-vllm"]) + def test_declared_endpoint_roles_scope_unlabelled_worker_counters(self, framework) -> None: + metrics = { + "vllm:prompt_tokens_by_source": _counter( + [ + (PRE, {"source": "local_cache_hit"}, 600), + (PRE, {"source": "local_compute"}, 400), + (DEC, {"source": "external_kv_transfer"}, 1000), + ] + ) + } + _, nested, _ = compute_server_metrics( + {"metrics": metrics}, + framework=framework, + records=[], + disaggregated=True, + endpoint_roles={PRE: "prefill", DEC: "decode"}, + ) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + assert nested["cache"]["role_scope"] == "prefill" + assert {source["role"] for source in nested["sources"]} == {"prefill", "decode"} + assert nested["cache_validity"]["pricing_eligible"] is False + + def test_declared_role_cannot_override_a_conflicting_explicit_label(self) -> None: + metrics = { + "vllm:prefix_cache_hits": _counter([(PRE, _vllm_role("prefill"), 600)]), + "vllm:prefix_cache_queries": _counter([(PRE, _vllm_role("prefill"), 1000)]), + } + _, nested, _ = compute_server_metrics( + {"metrics": metrics}, + framework="vllm", + records=[], + disaggregated=True, + endpoint_roles={PRE: "decode"}, + ) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache"]["role_scope"] == "unresolved" + assert nested["cache_validity"]["pricing_eligible"] is False + def test_decode_kv_transfer_is_not_a_cache_hit(self) -> None: metrics = { "vllm:prompt_tokens_by_source": _counter( @@ -225,7 +264,11 @@ def _metrics(cached: dict[str, float], prompt: float = 1000.0) -> dict: } def test_storage_tier_is_counted_with_the_other_non_device_tiers(self) -> None: - _, nested = _run(self._metrics({"device": 600, "host": 200, "storage": 100}), "sglang") + metrics = self._metrics({"device": 600, "host": 200, "storage": 100}) + metrics["sglang:realtime_tokens"] = _counter( + [(PRE, {"mode": "prefill_compute"}, 100)] + ) + _, nested = _run(metrics, "sglang") cache = nested["cache"] assert cache["gpu_cache_hit_rate"] == pytest.approx(0.6) assert cache["cpu_cache_hit_rate"] == pytest.approx(0.2) @@ -234,6 +277,12 @@ def test_storage_tier_is_counted_with_the_other_non_device_tiers(self) -> None: assert cache["gpu_cache_hit_rate"] + cache["external_cache_hit_rate"] == pytest.approx( cache["overall_cache_hit_rate"] ) + assert nested["tokens"]["prompt_by_source"] == { + "gpu_cache_hit": 600, + "cpu_or_external_cache_hit": 300, + "computed": 100, + "raw": {"device": 600, "host": 200, "storage": 100}, + } def test_device_and_host_only_keep_external_equal_to_host(self) -> None: _, nested = _run(self._metrics({"device": 600, "host": 200}), "sglang") From 00b615fcd253faed0b1ea74e2f899ecd126f8db1 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Sun, 25 Oct 2026 17:27:11 +0000 Subject: [PATCH 5/6] style(results): format endpoint role handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：按仓库格式整理端点角色处理及其回归测试。 --- inferencex-e2e/infx/results/agentic/backends/vllm.py | 12 +++++++++--- .../results/agentic/test_server_metrics_c2c3.py | 4 +--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/vllm.py b/inferencex-e2e/infx/results/agentic/backends/vllm.py index 32a67caf44..5db7b0fee6 100644 --- a/inferencex-e2e/infx/results/agentic/backends/vllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/vllm.py @@ -385,9 +385,15 @@ def _cache_role_filter( if "unknown" in roles or (disaggregated is True and "combined" in roles): scope = "unresolved" # Known decode counters measure KV transfer, including alongside unlabelled workers. - return scope, lambda series: _series_role(series, disaggregated, endpoint_roles) in ( - "prefill", - "combined", + return ( + scope, + lambda series: ( + _series_role(series, disaggregated, endpoint_roles) + in ( + "prefill", + "combined", + ) + ), ) diff --git a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py index c1d267befe..523f01a1a3 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py @@ -265,9 +265,7 @@ def _metrics(cached: dict[str, float], prompt: float = 1000.0) -> dict: def test_storage_tier_is_counted_with_the_other_non_device_tiers(self) -> None: metrics = self._metrics({"device": 600, "host": 200, "storage": 100}) - metrics["sglang:realtime_tokens"] = _counter( - [(PRE, {"mode": "prefill_compute"}, 100)] - ) + metrics["sglang:realtime_tokens"] = _counter([(PRE, {"mode": "prefill_compute"}, 100)]) _, nested = _run(metrics, "sglang") cache = nested["cache"] assert cache["gpu_cache_hit_rate"] == pytest.approx(0.6) From 53904d34b268617072c20df60f9337f3919c4af4 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Sun, 25 Oct 2026 17:35:12 +0000 Subject: [PATCH 6/6] test(results): keep router estimates outside measured cache reuse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按既定 review 合同更新上游新增的 Dynamo 回归：frontend 的真实零和 25% 保留；router-only 的 55% 仅作为 flat 诊断，不能进入 nested 实测或 published core。 --- .../agentic/test_process_agentic_result.py | 26 ++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) 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 de40762f5b..d0677f9f92 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 @@ -1669,13 +1669,12 @@ def test_processor_normalizes_dynamo_server_metrics(tmp_path: Path): assert any(source["role"] == "prefill" for source in agg["server_metrics"]["sources"]) -@pytest.mark.parametrize(("cached", "router_shared", "expected"), [ - (0.0, 0.55, 0.0), - (25.0, None, 0.25), - (None, 0.55, 0.55), -]) -def test_dynamo_overall_cache_hit_rate_falls_back_to_frontend_then_router( - cached, router_shared, expected, +@pytest.mark.parametrize( + ("cached", "router_shared", "measured", "diagnostic"), + [(0.0, 0.55, 0.0, 0.0), (25.0, None, 0.25, 0.25), (None, 0.55, None, 0.55)], +) +def test_dynamo_measured_overall_uses_frontend_and_keeps_router_estimate_diagnostic( + cached, router_shared, measured, diagnostic, ): metrics = { "dynamo_frontend_input_sequence_tokens": { @@ -1691,12 +1690,21 @@ def test_dynamo_overall_cache_hit_rate_falls_back_to_frontend_then_router( "type": "gauge", "series": [{"stats": {"avg": router_shared}}], } - _, nested, _ = compute_server_metrics( + flat, nested, _ = compute_server_metrics( {"metrics": metrics}, framework="dynamo-vllm", records=[], ) assert nested["adapter"] == "dynamo-vllm" - assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(expected) + assert flat["server_overall_cache_hit_rate"] == pytest.approx(diagnostic) + if measured is None: + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache"]["overall_source"] == "router_estimate" + assert nested["cache_validity"]["published"] is False + assert nested["cache_validity"]["basis"] is None + else: + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(measured) + assert nested["cache"]["overall_source"] == "frontend" + assert nested["cache_validity"]["published"] is True