test(router): add GMS placement E2E coverage

Signed-off-by: Maksim Khadkevich <mkhadkevich@nvidia.com>

Author: hutm

docs/components/frontend/configuration.md

@@ -43,6 +43,7 @@ The Rust HTTP server also reads these environment variables (not exposed as CLI
 | `--router-assume-kv-reuse` / `--no-router-assume-kv-reuse` | `DYN_ROUTER_ASSUME_KV_REUSE` | `true` | Assume KV cache reuse when tracking active blocks |
 | `--router-track-output-blocks` / `--no-router-track-output-blocks` | `DYN_ROUTER_TRACK_OUTPUT_BLOCKS` | `false` | Track output blocks with fractional decay during generation |
 | `--router-track-prefill-tokens` / `--no-router-track-prefill-tokens` | `DYN_ROUTER_TRACK_PREFILL_TOKENS` | `true` | Track prompt-side prefill load in worker load accounting |
+| `--router-gms-decode-transfer` / `--no-router-gms-decode-transfer` | `DYN_ROUTER_GMS_DECODE_TRANSFER` | `false` | Experimental GMS decode-to-decode transfer orchestration for GMS-capable aggregated decode workers |
 | `--router-prefill-load-model` | `DYN_ROUTER_PREFILL_LOAD_MODEL` | `none` | Prompt-side load model: `none` for static load, `aic` for oldest-prefill decay using an AIC prediction |
 | `--router-event-threads` | `DYN_ROUTER_EVENT_THREADS` | `4` | KV indexer worker threads. >1 enables the concurrent radix tree, including with `--no-router-kv-events` |
 | `--router-queue-threshold` | `DYN_ROUTER_QUEUE_THRESHOLD` | `16.0` | Queue threshold fraction of prefill capacity. Priority hints only affect requests waiting in this queue |
@@ -73,6 +74,7 @@ For MoE models, AIC requires `aic_tp_size * aic_attention_dp_size == aic_moe_tp_
 | CLI Argument | Env Var | Default | Description |
 |-------------|---------|---------|-------------|
 | `--migration-limit` | `DYN_MIGRATION_LIMIT` | `0` | Max request migrations per worker disconnect. 0 = disabled |
+| — | `DYN_MIGRATION_RETRY_CONCURRENCY` | `0` | Optional process-wide cap for concurrent migrated retry streams. 0 = unlimited |
 | `--active-decode-blocks-threshold` | `DYN_ACTIVE_DECODE_BLOCKS_THRESHOLD` | `1.0` | KV cache utilization fraction (0.0–1.0) for busy detection. Pass `None` to disable |
 | `--active-prefill-tokens-threshold` | `DYN_ACTIVE_PREFILL_TOKENS_THRESHOLD` | `10000000` | Absolute token count for prefill busy detection. Pass `None` to disable |
 | `--active-prefill-tokens-threshold-frac` | `DYN_ACTIVE_PREFILL_TOKENS_THRESHOLD_FRAC` | `64.0` | Fraction of `max_num_batched_tokens` for prefill busy detection. OR logic with absolute threshold. Pass `None` to disable |

docs/components/router/README.md

@@ -30,6 +30,7 @@ For Kubernetes, set `DYN_ROUTER_MODE=kv` on the Frontend service. For event-driv
 | `--router-queue-threshold` | `16.0` | Backpressure queue threshold; priority hints only reorder requests while this queue is non-empty |
 | `--router-queue-policy` | `fcfs` | Queue scheduling policy: `fcfs` (tail TTFT), `wspt` (avg TTFT), or `lcfs` (comparison-only reverse ordering) |
 | `--no-router-track-prefill-tokens` | disabled | Ignore prompt-side prefill tokens in router load accounting; useful for decode-only routing paths |
+| `--router-gms-decode-transfer` | disabled | Experimental GMS decode-to-decode transfer orchestration for GMS-capable aggregated decode workers |
 
 ### Standalone Router
 

docs/components/router/router-configuration.md

@@ -14,6 +14,7 @@ This page collects the main router flags for frontend-embedded and standalone de
 - `--load-aware`: Preset for load-aware KV routing without cache-reuse signals. On the frontend, it implies `--router-mode kv`. It sets `overlap_score_credit=0`, disables KV events, durable KV events, and KV reuse assumptions, enables active-block and prefill-token load tracking, disables remote/shared cache indexers, and preserves `--router-prefill-load-scale`.
 - `--router-temperature`: Controls worker selection randomness through softmax sampling of normalized router cost logits. A value of 0 (default) ensures deterministic selection of the lowest-cost worker, while higher values introduce more randomness.
 - `--router-track-prefill-tokens`: Enables prompt-side load accounting in the worker cost model. This should stay enabled if you want queue thresholds, `active_prefill_tokens`, and AIC prefill load decay to reflect prompt work.
+- `--router-gms-decode-transfer`: **Experimental, disabled by default.** Enables GMS decode-to-decode KV transfer orchestration for GMS-capable aggregated decode workers. Vanilla Dynamo routing is unchanged unless this flag is explicitly enabled.
 - `--router-prefill-load-model`: Selects the router's prompt-side load model. `none` keeps the existing static prompt load accounting. `aic` predicts one expected prefill duration per admitted request and lazily decays only the oldest active prefill request on each worker.
 - `--router-queue-threshold`: Queue threshold fraction for prefill token capacity (default: 16.0). The router holds incoming requests in a priority queue while all eligible workers exceed `threshold * max_num_batched_tokens`, releasing them when capacity frees up. This defers dispatch rather than rejecting work, so routing decisions use the freshest load metrics at the moment a request is actually sent to a worker. `nvext.agent_hints.priority` only affects ordering while requests are pending in this queue. Must be greater than or equal to 0; use `0.0` for maximum queueing sensitivity. Set to `None` to disable queueing. See the SGLang note under [Tuning Guidelines](#tuning-guidelines) for caveats around how `max_num_batched_tokens` is populated on that backend, and see [Priority Scheduling](../../agents/priority-scheduling.md) for how router priority differs from backend engine priority.
 - `--router-queue-policy`: Scheduling policy for the router queue (default: `fcfs`).

docs/components/router/router-guide.md

@@ -41,6 +41,7 @@ Backend workers register themselves using the `register_model` API. For accurate
 | `--router-kv-overlap-score-credit <float>` | `1.0` | Credit multiplier for device-local prefix overlap, from 0.0 to 1.0 |
 | `--router-prefill-load-scale <float>` | `1.0` | Scale adjusted prompt-side prefill load before adding decode blocks |
 | `--router-track-prefill-tokens` / `--no-router-track-prefill-tokens` | `--router-track-prefill-tokens` | Include prompt-side load in active worker load accounting |
+| `--router-gms-decode-transfer` / `--no-router-gms-decode-transfer` | `--no-router-gms-decode-transfer` | Experimental GMS decode-to-decode transfer orchestration for GMS-capable aggregated decode workers |
 | `--router-prefill-load-model <none\|aic>` | `none` | Prompt-side load model; see [Routing Concepts](router-concepts.md#active-load-modeling) and [Configuration and Tuning](router-configuration.md#aic-prefill-load-model) |
 | `--router-queue-threshold <float>` | `16.0` | Queue threshold fraction; priority hints only reorder requests while this queue is non-empty |
 | `--router-queue-policy <str>` | `fcfs` | Scheduling policy for the queue: `fcfs` (tail TTFT), `wspt` (avg TTFT), or `lcfs` (comparison-only reverse ordering) |

tests/conftest.py

@@ -407,6 +407,10 @@ def pytest_runtestloop(session: pytest.Session) -> bool | None:
 @pytest.fixture()
 def set_ucx_tls_no_mm():
     """Set UCX env defaults for all tests."""
+    if os.environ.get("DYNAMO_TEST_PRESERVE_UCX_ENV") == "1":
+        yield
+        return
+
     mp = pytest.MonkeyPatch()
     # CI note:
     # - Affected test: tests/fault_tolerance/cancellation/test_vllm.py::test_request_cancellation_vllm_decode_cancel
@@ -454,6 +458,10 @@ def download_models(model_list=None, ignore_weights=False):
 
     failures = []
     for model_id in model_list:
+        if Path(str(model_id)).exists():
+            logging.info("Using local model path; skipping pre-download: %s", model_id)
+            continue
+
         logging.info(
             f"Pre-downloading {'model (no weights)' if ignore_weights else 'model'}: {model_id}"
         )

tests/router/common.py

@@ -1969,6 +1969,8 @@ def _test_router_decisions_disagg(
     durable_kv_events: bool = False,
     router_aic_config: Optional[dict[str, Any]] = None,
     enable_bootstrap: bool = False,
+    strict_timing: bool = True,
+    progressive_request_count: int = 4,
 ):
     """Validate KV cache prefix reuse in disaggregated prefill-decode setup via HTTP frontend.
 
@@ -1992,6 +1994,8 @@ def _test_router_decisions_disagg(
         store_backend: Storage backend to use ("etcd" or "file"). Defaults to "etcd".
         durable_kv_events: If True, use durable KV events (JetStream). Defaults to False.
         router_aic_config: Optional AIC router perf-model config for frontend KV routing.
+        strict_timing: If False, allow backend responses that omit optional timing fields.
+        progressive_request_count: Number of overlapping-prefix requests to send.
 
     Raises:
         AssertionError: If prefill_worker_ids differ across requests (prefix reuse failure)
@@ -2030,16 +2034,19 @@ def _test_router_decisions_disagg(
             )
         )
 
+        if progressive_request_count < 2:
+            raise ValueError("progressive_request_count must be at least 2")
+
         async def send_progressive_requests():
-            """Send 4 progressive requests with overlapping prefixes and collect worker IDs."""
+            """Send progressive requests with overlapping prefixes and collect worker IDs."""
             prefill_worker_ids = []
             decode_worker_ids = []
 
             # Generate base tokens for progressive prefix extension
             base_content = test_payload["messages"][0]["content"]
 
             async with aiohttp.ClientSession() as session:
-                for i in range(4):
+                for i in range(progressive_request_count):
                     # Build progressive content by repeating base content
                     # Each iteration adds more content to extend the prefix
                     progressive_content = " ".join([base_content] * (i + 1))
@@ -2058,7 +2065,7 @@ async def send_progressive_requests():
                     }
 
                     logger.info(
-                        f"Sending request {i + 1}/4 with progressive prefix "
+                        f"Sending request {i + 1}/{progressive_request_count} with progressive prefix "
                         f"(~{len(progressive_content)} chars)"
                     )
 
@@ -2067,10 +2074,13 @@ async def send_progressive_requests():
                             response.status == 200
                         ), f"Request {i + 1} failed with status {response.status}"
 
-                        # Collect all chunks and look for nvext with worker_id and timing
+                        # Collect all chunks and look for nvext with worker_id, timing,
+                        # and actual generated text. A backend can otherwise surface a
+                        # transfer failure as a 200 stream with worker IDs but no tokens.
                         prefill_wid = None
                         decode_wid = None
                         timing_info = None
+                        generated_parts: list[str] = []
 
                         async for line in response.content:
                             if not line:
@@ -2086,6 +2096,12 @@ async def send_progressive_requests():
 
                             try:
                                 data = json.loads(data_str)
+                                for choice in data.get("choices", []):
+                                    delta = choice.get("delta") or {}
+                                    content = delta.get("content")
+                                    if content:
+                                        generated_parts.append(content)
+
                                 # Check for nvext in the response
                                 nvext = data.get("nvext", {})
                                 if nvext:
@@ -2106,9 +2122,15 @@ async def send_progressive_requests():
                             except json.JSONDecodeError:
                                 continue
 
+                        generated_text = "".join(generated_parts)
                         logger.info(
                             f"Request {i + 1}: prefill_worker_id={prefill_wid}, "
-                            f"decode_worker_id={decode_wid}, timing={timing_info}"
+                            f"decode_worker_id={decode_wid}, timing={timing_info}, "
+                            f"generated_chars={len(generated_text)}"
+                        )
+                        assert generated_text.strip(), (
+                            f"Request {i + 1}: Expected generated content in the response stream, "
+                            "but the backend returned no text tokens"
                         )
 
                         if prefill_wid is not None:
@@ -2122,7 +2144,12 @@ async def send_progressive_requests():
                         assert (
                             timing_info is not None
                         ), f"Request {i + 1}: Expected timing info in final chunk, got None"
-                        verify_response_timing(timing_info, disagg=not enable_bootstrap)
+                        verify_response_timing(
+                            timing_info,
+                            disagg=not enable_bootstrap,
+                            require_ttft=strict_timing,
+                            require_kv_transfer_latency=strict_timing,
+                        )
 
                     # Small delay between requests
                     await asyncio.sleep(1)
@@ -2136,8 +2163,8 @@ async def send_progressive_requests():
         logger.info(f"Collected decode_worker_ids: {decode_ids}")
 
         # Verify we got worker IDs from all requests
-        assert len(prefill_ids) == 4, (
-            f"Expected 4 prefill_worker_ids, got {len(prefill_ids)}. "
+        assert len(prefill_ids) == progressive_request_count, (
+            f"Expected {progressive_request_count} prefill_worker_ids, got {len(prefill_ids)}. "
             f"Make sure nvext.extra_fields=['worker_id'] is being processed."
         )
 
@@ -2148,12 +2175,12 @@ async def send_progressive_requests():
         # the second request is routed before the first request's KV "stored" events have been
         # fully ingested. After ingestion, routing stabilizes.
         #
-        # So for TCP we assert that requests 2-4 converge to the same prefill worker; for NATS
+        # So for TCP we assert that requests 2-N converge to the same prefill worker; for NATS
         # request plane we keep the stronger assertion that all 4 match.
         if request_plane == "tcp":
             unique_prefill_ids = set(prefill_ids[1:])
             assert len(unique_prefill_ids) == 1, (
-                f"Expected prefill requests 2-4 to route to the same worker due to prefix reuse, "
+                f"Expected prefill requests 2-{progressive_request_count} to route to the same worker due to prefix reuse, "
                 f"but found {len(unique_prefill_ids)} unique prefill_worker_ids: {unique_prefill_ids}. "
                 f"Full list: {prefill_ids}"
             )

tests/router/configs/trtllm_agg.yaml

@@ -0,0 +1,11 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Extra-engine-args for aggregate TRT-LLM router e2e tests.
+#
+# The TRTLLM attention backend currently takes the FMHA NVRTC path on B200 for
+# TinyLlama generation and can fail when runtime include discovery does not
+# populate cuda.h search paths. VANILLA keeps the test on the PyTorch backend
+# path while still exercising real TRT-LLM workers,
+# router discovery, request forwarding, KV events, and local indexer behavior.
+attn_backend: VANILLA

tests/router/configs/trtllm_disagg_decode.yaml

@@ -4,5 +4,8 @@
 # Extra-engine-args for the decode side of disaggregated TRT-LLM in router
 # e2e tests. Only the transceiver backend is required here; the overlap
 # scheduler can stay on for decode workers.
+attn_backend: VANILLA
+
 cache_transceiver_config:
-  backend: DEFAULT
+  backend: UCX
+  max_tokens_in_buffer: 8192

tests/router/configs/trtllm_disagg_prefill.yaml

@@ -13,6 +13,8 @@
 #     "Context only requests are not supported in pytorch backend when
 #     overlap is enabled with block reuse").
 disable_overlap_scheduler: true
+attn_backend: VANILLA
 
 cache_transceiver_config:
-  backend: DEFAULT
+  backend: UCX
+  max_tokens_in_buffer: 8192