test(sidecar): cover native KV-aware routing in GPU CI

.github/filters.yaml

@@ -353,6 +353,13 @@ sidecar:
   - 'lib/sidecar/**'
   - '.github/workflows/shared-build-sidecar.yml'
   - 'tests/serve/test_sidecar.py'
+  - 'tests/router/common.py'
+  - 'tests/router/helper.py'
+  - 'tests/router/router_process.py'
+  - 'tests/utils/gpu_args.py'
+  - 'tests/utils/prometheus.py'
+  - 'tests/utils/router_logs.py'
+  - 'tests/utils/router_nvext.py'
   - 'Cargo.toml'
   - 'Cargo.lock'
   # Compliance policy and baseline changes can fail the sidecar image build.

lib/sidecar/sglang/launch/agg_kv_router.sh

@@ -3,16 +3,15 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 # Two aggregated SGLang native-gRPC sidecars behind Dynamo's KV-aware router.
-# Requires two GPUs and an SGLang build that exposes KV-event discovery over GetServerInfo.
+# Requires an SGLang build that exposes KV-event discovery over GetServerInfo.
 
 set -e
 
 SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
-export DYNAMO_HOME="${DYNAMO_HOME:-$(readlink -f "$SCRIPT_DIR/../../../..")}"
 # shellcheck disable=SC1091 # Resolved relative to this script at runtime.
-source "$DYNAMO_HOME/examples/common/gpu_utils.sh"
+source "$SCRIPT_DIR/../../../../examples/common/gpu_utils.sh"
 # shellcheck disable=SC1091 # Resolved relative to this script at runtime.
-source "$DYNAMO_HOME/examples/common/launch_utils.sh"
+source "$SCRIPT_DIR/../../../../examples/common/launch_utils.sh"
 
 MODEL="${MODEL:-Qwen/Qwen3-0.6B}"
 
@@ -84,7 +83,7 @@ GPU_MEM_ARGS=$(build_sglang_gpu_mem_args)
 KV_EVENTS_CONFIG_1="{\"publisher\":\"zmq\",\"endpoint\":\"tcp://*:${SGLANG_WORKER1_KV_EVENT_PORT}\",\"topic\":\"\"}"
 KV_EVENTS_CONFIG_2="{\"publisher\":\"zmq\",\"endpoint\":\"tcp://*:${SGLANG_WORKER2_KV_EVENT_PORT}\",\"topic\":\"\"}"
 
-print_launch_banner "Launching SGLang Native-gRPC Sidecars with KV Routing (2 GPUs)" "$MODEL" "$HTTP_PORT" \
+print_launch_banner "Launching SGLang Native-gRPC Sidecars with KV Routing (2 workers)" "$MODEL" "$HTTP_PORT" \
     "Worker 1: GPU ${SGLANG_WORKER1_GPU}, gRPC ${SGLANG_HOST}:${SGLANG_WORKER1_GRPC_PORT}, KV events tcp://*:${SGLANG_WORKER1_KV_EVENT_PORT}" \
     "Worker 2: GPU ${SGLANG_WORKER2_GPU}, gRPC ${SGLANG_HOST}:${SGLANG_WORKER2_GRPC_PORT}, KV events tcp://*:${SGLANG_WORKER2_KV_EVENT_PORT}"
 

lib/sidecar/vllm/launch/agg_kv_router.sh

@@ -3,17 +3,16 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 # Two aggregated vLLM native-gRPC sidecars behind Dynamo's KV-aware router.
-# Requires two GPUs and a vLLM build that exposes KV-event source discovery.
+# Requires a vLLM build that exposes KV-event source discovery.
 # See ../README.md for the validated vLLM/vllm-rs source state.
 
 set -e
 
 SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
-export DYNAMO_HOME="${DYNAMO_HOME:-$(readlink -f "$SCRIPT_DIR/../../../..")}"
 # shellcheck disable=SC1091 # Resolved relative to this script at runtime.
-source "$DYNAMO_HOME/examples/common/gpu_utils.sh"
+source "$SCRIPT_DIR/../../../../examples/common/gpu_utils.sh"
 # shellcheck disable=SC1091 # Resolved relative to this script at runtime.
-source "$DYNAMO_HOME/examples/common/launch_utils.sh"
+source "$SCRIPT_DIR/../../../../examples/common/launch_utils.sh"
 
 MODEL="${MODEL:-Qwen/Qwen3-0.6B}"
 
@@ -92,7 +91,7 @@ KV_EVENTS_CONFIG_1="{\"publisher\":\"zmq\",\"topic\":\"kv-events\",\"endpoint\":
 KV_EVENTS_CONFIG_2="{\"publisher\":\"zmq\",\"topic\":\"kv-events\",\"endpoint\":\"tcp://*:${VLLM_WORKER2_KV_EVENT_PORT}\",\"enable_kv_cache_events\":true}"
 
 HTTP_PORT="${DYN_HTTP_PORT:-8000}"
-print_launch_banner "Launching vLLM Native-gRPC Sidecars with KV Routing (2 GPUs)" "$MODEL" "$HTTP_PORT" \
+print_launch_banner "Launching vLLM Native-gRPC Sidecars with KV Routing (2 workers)" "$MODEL" "$HTTP_PORT" \
     "Worker 1: GPU ${VLLM_WORKER1_GPU}, gRPC ${VLLM_HOST}:${VLLM_WORKER1_GRPC_PORT}, KV events tcp://*:${VLLM_WORKER1_KV_EVENT_PORT}" \
     "Worker 2: GPU ${VLLM_WORKER2_GPU}, gRPC ${VLLM_HOST}:${VLLM_WORKER2_GRPC_PORT}, KV events tcp://*:${VLLM_WORKER2_KV_EVENT_PORT}"
 

lib/sidecar/vllm/src/convert.rs

@@ -352,6 +352,13 @@ fn consume_redundant_nvext(
                 "extra_args.nvext.token_in must be true when present",
             ));
         }
+        // Worker attribution and timing are produced by the frontend.
+        if let Some(serde_json::Value::Array(fields)) = nvext.get_mut("extra_fields") {
+            fields.retain(|field| !matches!(field.as_str(), Some("worker_id" | "timing")));
+            if fields.is_empty() {
+                nvext.remove("extra_fields");
+            }
+        }
         nvext.is_empty()
     };
     if remove_nvext {

lib/sidecar/vllm/src/tests.rs

@@ -1037,6 +1037,33 @@ fn request() -> PreprocessedRequest {
         .expect("request")
 }
 
+#[test]
+fn frontend_router_metadata_does_not_require_engine_support() {
+    let baseline = build_generate_request(
+        request(),
+        "request-1".to_string(),
+        DisaggregationMode::Aggregated,
+    )
+    .unwrap();
+    for (fields, supported) in [
+        (json!(["worker_id", "timing"]), true),
+        (json!(["worker_id", "engine_data"]), false),
+    ] {
+        let mut request = request();
+        request.extra_args.as_mut().unwrap()["nvext"]["extra_fields"] = fields;
+        let result = build_generate_request(
+            request,
+            "request-1".to_string(),
+            DisaggregationMode::Aggregated,
+        );
+        if supported {
+            assert_eq!(result.unwrap(), baseline);
+        } else {
+            assert!(result.is_err());
+        }
+    }
+}
+
 #[test]
 fn skip_special_tokens_is_forwarded_without_compatibility_envelope() {
     let mut request = request();

tests/router/common.py

@@ -20,11 +20,13 @@
 from tests.router.helper import (
     assert_event_dumps_equal,
     get_runtime,
+    get_stored_kv_event_counts,
     managed_runtime,
     parse_sse_json_chunks,
     poll_for_worker_instances,
     send_inflight_requests,
     send_request_via_python_kv_router,
+    send_router_chat_request,
     verify_response_timing,
     wait_for_frontend_ready,
     wait_for_indexer_workers_active,
@@ -36,6 +38,7 @@
     select_kv_event_diagnostics,
     wait_for_kv_event_diagnostics,
 )
+from tests.utils.router_nvext import require_router_worker_id
 
 if TYPE_CHECKING:
     from tests.conftest import NatsServer
@@ -549,6 +552,135 @@ def _test_router_two_routers(
             kv_router.__exit__(None, None, None)
 
 
+def _test_frontend_kv_routing(
+    *,
+    frontend_port: int,
+    system_ports: list[int],
+    namespace: str,
+    model_name: str,
+    block_size: int,
+) -> None:
+    """Verify engine events drive HTTP routing to two independently warmed workers."""
+    assert len(system_ports) == 2
+    url = f"http://localhost:{frontend_port}/v1/chat/completions"
+    prompts = [
+        "Amber rabbits explore quiet meadows. " * 96,
+        "Violet submarines navigate distant oceans. " * 96,
+    ]
+
+    async def run_test() -> None:
+        """Warm both workers and verify event-driven routing converges."""
+        with managed_runtime() as runtime:
+            worker_ids = sorted(
+                await poll_for_worker_instances(
+                    runtime.endpoint(f"{namespace}.backend.generate"), 2
+                )
+            )
+            assert len(worker_ids) == 2, worker_ids
+
+            async with aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=30)
+            ) as session:
+
+                async def send(
+                    prompt: str,
+                    *,
+                    query_only: bool = False,
+                    worker_id: int | None = None,
+                ) -> tuple[int, float | None]:
+                    """Send one request and return its selected worker and KV hit rate."""
+                    payload = {
+                        "model": model_name,
+                        "messages": [{"role": "user", "content": prompt}],
+                        "max_tokens": 8,
+                        "temperature": 0,
+                        "stream": True,
+                        "nvext": {
+                            "extra_fields": ["worker_id", "timing"],
+                            "annotations": ["query_instance_id:"] if query_only else [],
+                        },
+                    }
+                    headers = (
+                        {
+                            "x-dynamo-worker-instance-id": str(worker_id),
+                            "x-dynamo-dp-rank": "0",
+                        }
+                        if worker_id is not None
+                        else None
+                    )
+                    nvext, generated = await send_router_chat_request(
+                        session, url, payload, headers
+                    )
+                    selected = require_router_worker_id({"nvext": nvext})
+                    selected_id = selected["decode_worker_id"]
+                    assert selected_id in worker_ids, selected
+                    assert selected["prefill_worker_id"] == selected_id, selected
+                    assert selected["decode_dp_rank"] == 0, selected
+                    hit_rate = nvext.get("timing", {}).get("kv_hit_rate")
+                    if query_only:
+                        assert not generated, nvext
+                        assert len(nvext.get("token_ids", [])) >= block_size * 4, nvext
+                    else:
+                        assert generated, "Request completed without generating text"
+                        assert isinstance(hit_rate, (int, float)), nvext
+                        assert 0 <= hit_rate <= 1, nvext
+                    return selected_id, hit_rate
+
+                baselines = {
+                    port: await get_stored_kv_event_counts(session, port)
+                    for port in system_ports
+                }
+                for prompt in prompts:
+                    await send(prompt, query_only=True)
+                for prompt, worker_id in zip(prompts, worker_ids):
+                    selected, _ = await send(prompt, worker_id=worker_id)
+                    assert selected == worker_id, (selected, worker_id)
+
+                deadline = time.monotonic() + 60
+                observed = []
+                counts = {}
+                while time.monotonic() < deadline:
+                    observed = [
+                        await send(prompt, query_only=True) for prompt in prompts
+                    ]
+                    counts = {
+                        port: await get_stored_kv_event_counts(session, port)
+                        for port in system_ports
+                    }
+                    if all(
+                        selected == expected
+                        for (selected, _), expected in zip(observed, worker_ids)
+                    ) and all(
+                        all(
+                            current > baseline
+                            for current, baseline in zip(counts[port], baselines[port])
+                        )
+                        for port in system_ports
+                    ):
+                        break
+                    await asyncio.sleep(0.1)
+                else:
+                    raise AssertionError(
+                        f"KV events did not converge: expected workers={worker_ids}, "
+                        f"routing={observed}, Stored counters={counts}, baselines={baselines}"
+                    )
+
+                for prompt_index in (0, 0, 1, 0, 1, 1):
+                    selected, hit_rate = await send(prompts[prompt_index])
+                    assert selected == worker_ids[prompt_index], (
+                        prompt_index,
+                        selected,
+                        worker_ids,
+                    )
+                    assert hit_rate is not None and hit_rate >= 0.5, (
+                        prompt_index,
+                        selected,
+                        hit_rate,
+                    )
+
+    asyncio.run(run_test())
+
+
 def _test_session_affinity(
     engine_workers,
     block_size: int,

tests/router/helper.py

@@ -15,7 +15,9 @@
 import aiohttp
 
 from dynamo.llm import KvRouter
+from dynamo.prometheus_names import kv_publisher, name_prefix
 from dynamo.runtime import DistributedRuntime
+from tests.utils.prometheus import sum_metric_samples
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +44,50 @@ def parse_sse_json_chunks(body: str) -> list[dict[str, Any]]:
     return chunks
 
 
+async def send_router_chat_request(
+    session: aiohttp.ClientSession,
+    url: str,
+    payload: dict[str, Any],
+    headers: dict[str, str] | None = None,
+) -> tuple[dict[str, Any], bool]:
+    """Return merged routing metadata and whether a streaming request generated text."""
+    async with session.post(url, json=payload, headers=headers) as response:
+        body = await response.text()
+        assert response.status == 200, body
+
+    assert "data: [DONE]" in body, f"Incomplete SSE response: {body}"
+    nvext: dict[str, Any] = {}
+    generated = False
+    for chunk in parse_sse_json_chunks(body):
+        assert "error" not in chunk, chunk
+        nvext.update(chunk.get("nvext") or {})
+        for choice in chunk.get("choices", []):
+            delta = choice.get("delta") or {}
+            generated |= any(
+                delta.get(field)
+                for field in ("content", "reasoning_content", "reasoning")
+            )
+    return nvext, generated
+
+
+async def get_stored_kv_event_counts(
+    session: aiohttp.ClientSession, system_port: int
+) -> tuple[float, float]:
+    """Read one worker's received and accepted Stored-event counters."""
+    async with session.get(f"http://localhost:{system_port}/metrics") as response:
+        response.raise_for_status()
+        metrics = await response.text()
+    metric_name = f"{name_prefix.COMPONENT}_{kv_publisher.ZMQ_EVENTS_TOTAL}"
+    return (
+        sum_metric_samples(
+            metrics, metric_name, {"stage": "received", "event_type": "stored"}
+        ),
+        sum_metric_samples(
+            metrics, metric_name, {"stage": "accepted", "event_type": "stored"}
+        ),
+    )
+
+
 def generate_random_suffix() -> str:
     """Generate a 10-character random alphabetic suffix for namespace isolation."""
     return "".join(random.choices(string.ascii_lowercase, k=10))  # noqa: S311