diff --git a/inferencex-e2e/infx/results/agentic/backends/base.py b/inferencex-e2e/infx/results/agentic/backends/base.py
index d0bb6994f4..b94045fb45 100644
--- a/inferencex-e2e/infx/results/agentic/backends/base.py
+++ b/inferencex-e2e/infx/results/agentic/backends/base.py
@@ -72,6 +72,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,
         },
         "kv_cache": {
             "gpu_usage_pct": None,
diff --git a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py
index 5e09446461..03082f2287 100644
--- a/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py
+++ b/inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py
@@ -44,9 +44,11 @@ def populate(
         metrics: dict[str, dict[str, Any]],
         flat: dict[str, Any],
         nested: dict[str, Any],
+        *,
+        context: dict[str, Any] | None = None,
     ) -> None:
-        """Fill the vLLM fields, then the Dynamo frontend and router fields, from ``metrics``."""
-        super().populate(metrics, flat, nested)
+        """Fill the vLLM fields, then the Dynamo frontend and router fields."""
+        super().populate(metrics, flat, nested, context=context)
         self._populate_dynamo(metrics, flat, nested)
 
     def _populate_dynamo(
@@ -96,9 +98,13 @@ def _populate_dynamo(
             )
         )
         if flat["server_overall_cache_hit_rate"] is None:
-            overall = frontend_hit_rate if frontend_hit_rate is not None else router_shared_hit_rate
-            flat["server_overall_cache_hit_rate"] = overall
-            nested["cache"]["overall_cache_hit_rate"] = overall
+            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"].update(
             {
diff --git a/inferencex-e2e/infx/results/agentic/backends/sglang.py b/inferencex-e2e/infx/results/agentic/backends/sglang.py
index a6990a16d0..86d0e07dae 100644
--- a/inferencex-e2e/infx/results/agentic/backends/sglang.py
+++ b/inferencex-e2e/infx/results/agentic/backends/sglang.py
@@ -61,10 +61,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"]
+        # 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
+        )
         flat["server_overall_cache_hit_rate"] = rate(total_cached, prompt_total)
 
         if flat["server_overall_cache_hit_rate"] is None:
@@ -136,7 +140,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 bbafcfd2b8..d4ef2296c5 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"\((?P<tag>EngineCore(?:_DP\d+)?)\s+pid=\d+\)")
     _GPU_KV_SIZE_RE = re.compile(r"GPU KV cache size:\s*(?P<tokens>[\d,]+)\s*tokens")
 
@@ -37,21 +38,28 @@ def populate(
         metrics: dict[str, dict[str, Any]],
         flat: dict[str, Any],
         nested: dict[str, Any],
+        *,
+        context: dict[str, Any] | None = None,
     ) -> None:
-        """Fill token totals, prefix and external hit rates, KV usage, offload stats and sources from ``metrics``."""
+        """Fill token totals, cache rates, KV usage, offload statistics and sources."""
+        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, endpoint_roles)
         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
@@ -60,11 +68,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
@@ -75,6 +85,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")
@@ -145,6 +156,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(
@@ -173,7 +188,7 @@ def populate(
                 "generation_total": flat["total_generation_tokens"],
             }
         )
-        nested["sources"] = _vllm_sources(metrics)
+        nested["sources"] = _vllm_sources(metrics, disaggregated, endpoint_roles)
 
     def prompt_generation_totals(
         self,
@@ -256,8 +271,12 @@ def first_counter_total(
     return None
 
 
-def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
-    """Return per-source token, hit rate and KV usage rows from vLLM ``metrics``, sorted by source id."""
+def _vllm_sources(
+    metrics: dict[str, dict[str, Any]],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> list[dict[str, Any]]:
+    """Return per-source token, hit rate and KV usage rows, sorted by source id."""
     source_ids = set()
     for metric_name in (
         "vllm:prompt_tokens",
@@ -268,7 +287,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, endpoint_roles))
 
     sources: list[dict[str, Any]] = []
     for source_id in sorted(source_ids):
@@ -276,8 +295,8 @@ 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 whether ``series`` belongs to ``source_id``."""
-            return _source_id(series) == source_id
+            """Return whether the series belongs to the selected source."""
+            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)
@@ -304,18 +323,95 @@ def series_filter(series: dict, source_id: str = source_id) -> bool:
     return sources
 
 
-def _source_id(series: dict[str, Any]) -> str:
-    """Build a ``role|endpoint|key=value`` id for ``series`` from its component, endpoint and rank labels."""
+_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],
+    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"):
+        return "prefill"
+    if mode == "decode":
+        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":
-        role = "prefill"
-    elif component in ("backend", "decode"):
-        role = "decode"
-    elif component in ("frontend", "router"):
-        role = "router"
+        return "prefill"
+    if component == "backend" and disaggregated is False:
+        return "combined"
+    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]],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> tuple[str, Any]:
+    roles = {
+        _series_role(s, disaggregated, endpoint_roles)
+        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:
-        role = "combined"
+        scope = "unresolved" if "decode" in roles or disaggregated is True else "combined"
+    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",
+            )
+        ),
+    )
+
+
+def _source_id(
+    series: dict[str, Any],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> str:
+    """Build a role, endpoint and rank identity for a worker series."""
+    role = _series_role(series, disaggregated, endpoint_roles)
 
     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 26bb6ea368..f568805d96 100644
--- a/inferencex-e2e/infx/results/agentic/common.py
+++ b/inferencex-e2e/infx/results/agentic/common.py
@@ -200,11 +200,14 @@ 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]:
     """Sum the preferred stat of ``metric_names`` grouped by the first present label among ``label_keys``."""
     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 6ced7b401c..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
@@ -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",
@@ -1667,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": {
@@ -1689,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
 
 
 
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..523f01a1a3
--- /dev/null
+++ b/inferencex-e2e/infx/tests/results/agentic/test_server_metrics_c2c3.py
@@ -0,0 +1,331 @@
+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 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 = {
+            "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:
+    @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(
+                [
+                    (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"
+
+    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"
+
+    @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
+    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:
+        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)
+        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"]
+        )
+        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")
+        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"
+
+
+@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