feat(router): add decode-to-prefill KV reuse with retained GPU transport

.github/filters.yaml

@@ -495,6 +495,7 @@ frontend:
 
 rust:
   - '.github/workflows/pre-merge.yml'
+  - 'scripts/ci/**'
   - '**/*.rs'
   - '**/Cargo.toml'
   - '**/Cargo.lock'

.github/workflows/pre-merge.yml

@@ -470,7 +470,19 @@ jobs:
     # Have an explicit step to build tests first to separate time spent on build vs execution.
     - name: Compile Tests
       working-directory: ${{ matrix.dir }}
-      run: cargo test --locked --no-run
+      run: |
+        if [ "${{ matrix.dir }}" = "." ]; then
+          set -o pipefail
+          set +e
+          cargo test --locked --no-run --message-format=json-render-diagnostics \
+            | tee "$RUNNER_TEMP/completed-rust-test-artifacts.jsonl"
+          compile_status=("${PIPESTATUS[@]}")
+          if [ "${compile_status[0]}" -ne 0 ]; then
+            exit "${compile_status[0]}"
+          fi
+          exit "${compile_status[1]}"
+        fi
+        cargo test --locked --no-run
     - name: Run Doc Tests
       working-directory: ${{ matrix.dir }}
       run: cargo doc --no-deps && cargo test --locked --doc
@@ -527,6 +539,19 @@ jobs:
       if: matrix.dir == 'lib/bindings/python'
       working-directory: ${{ matrix.dir }}
       run: cargo test --locked --features ais-forward-pass -- --nocapture
+    - name: Reclaim completed Rust test executables
+      if: matrix.dir == '.'
+      working-directory: ${{ matrix.dir }}
+      run: |
+        set -euo pipefail
+        python3 -m unittest discover -s scripts/ci -p test_reclaim_completed_test_executables.py
+        reclaim_metadata="$RUNNER_TEMP/reclaim-cargo-metadata.json"
+        cargo metadata --locked --no-deps --format-version 1 > "$reclaim_metadata"
+        workspace_root="$(python3 -c 'import json, sys; from pathlib import Path; print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["workspace_root"])' "$reclaim_metadata")"
+        target_dir="$(python3 -c 'import json, sys; from pathlib import Path; print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["target_directory"])' "$reclaim_metadata")"
+        python3 scripts/ci/reclaim_completed_test_executables.py \
+          --cargo-messages "$RUNNER_TEMP/completed-rust-test-artifacts.jsonl" \
+          --workspace "$workspace_root" --target-dir "$target_dir"
     - name: Compile feature-gated benchmark entrypoints
       if: matrix.dir == '.'
       working-directory: ${{ matrix.dir }}

components/src/dynamo/common/configuration/groups/kv_router_args.py

@@ -54,6 +54,14 @@
     "enable_session_prefix_index",
     "shared_cache_type",
     "router_predicted_ttl_secs",
+    "enable_decode_to_prefill_reuse",
+    "enable_decode_to_prefill_gpu_reuse",
+    "transfer_source_lookup_budget_ms",
+    "minimum_transfer_prefix_blocks",
+    "minimum_estimated_ttft_saving_ms",
+    "transfer_source_estimated_block_us",
+    "transfer_source_gpu_estimated_block_us",
+    "transfer_source_prefill_token_us",
 )
 
 # Authoritative field list — used by kv_router_kwargs() to extract values.
@@ -213,6 +221,14 @@ class KvRouterConfigBase(ConfigBase):
     conditional_disagg_prefill_busy_threshold: Optional[float] = None
     conditional_disagg_decode_busy_threshold: Optional[float] = None
     router_predicted_ttl_secs: Optional[float] = None
+    enable_decode_to_prefill_reuse: bool = False
+    enable_decode_to_prefill_gpu_reuse: bool = False
+    transfer_source_lookup_budget_ms: int = 5
+    minimum_transfer_prefix_blocks: int = 1
+    minimum_estimated_ttft_saving_ms: float = 0.0
+    transfer_source_estimated_block_us: Optional[float] = None
+    transfer_source_gpu_estimated_block_us: float | None = None
+    transfer_source_prefill_token_us: Optional[float] = None
     load_aware: bool = False
 
     @cached_property
@@ -603,6 +619,78 @@ def add_arguments(self, parser) -> None:
             ),
             arg_type=str,
         )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--router-decode-to-prefill-reuse",
+            env_var="DYN_ROUTER_DECODE_TO_PREFILL_REUSE",
+            dest="enable_decode_to_prefill_reuse",
+            default=False,
+            help=(
+                "Experimental decode-to-prefill KV reuse. Requires a certified "
+                "router_hint_transfer_domain, local KV events, and calibrated "
+                "transfer-source cost bounds."
+            ),
+        )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--router-decode-to-prefill-gpu-reuse",
+            env_var="DYN_ROUTER_DECODE_TO_PREFILL_GPU_REUSE",
+            dest="enable_decode_to_prefill_gpu_reuse",
+            default=False,
+            help=(
+                "Experimental direct decode-GPU to prefill-GPU KV reuse. "
+                "Requires GPU-only pin/transport endpoints, certified layout, "
+                "ready gpu_direct KV events, and separately calibrated GPU costs. "
+                "Does not enable the CPU offload transfer path."
+            ),
+        )
+        for name, arg_type, default, help_text in (
+            (
+                "transfer_source_lookup_budget_ms",
+                int,
+                5,
+                "Maximum donor lookup time in milliseconds.",
+            ),
+            (
+                "minimum_transfer_prefix_blocks",
+                int,
+                1,
+                "Minimum extra reusable blocks beyond the existing prefix.",
+            ),
+            (
+                "minimum_estimated_ttft_saving_ms",
+                float,
+                0.0,
+                "Minimum estimated TTFT saving in milliseconds.",
+            ),
+            (
+                "transfer_source_estimated_block_us",
+                float,
+                None,
+                "Measured upper bound per missing block, including acquisition, copy, onboarding, donor contention, and incremental P-to-D cost.",
+            ),
+            (
+                "transfer_source_gpu_estimated_block_us",
+                float,
+                None,
+                "Separately measured direct GPU acquisition/transfer cost per missing block in microseconds; CPU copy/onboarding costs do not qualify it.",
+            ),
+            (
+                "transfer_source_prefill_token_us",
+                float,
+                None,
+                "Measured conservative prefill time saved per reused token in microseconds.",
+            ),
+        ):
+            add_argument(
+                g,
+                flag_name="--router-" + name.replace("_", "-"),
+                env_var="DYN_ROUTER_" + name.upper(),
+                dest=name,
+                arg_type=arg_type,
+                default=default,
+                help=help_text,
+            )
         add_negatable_bool_argument(
             g,
             flag_name="--router-conditional-disagg",

components/src/dynamo/common/constants.py

@@ -7,10 +7,15 @@
 
 # TODO: Rename these constants and wire values with the matching KVCC names.
 KV_HINT_TRANSFER_CAPABILITY_KEY = "router_hint"
+KV_HINT_TRANSFER_DOMAIN_RUNTIME_KEY = "router_hint_transfer_domain"
 KV_HINT_TRANSFER_WORKER_TYPE_RUNTIME_KEY = "router_hint_worker_type"
 KV_HINT_TRANSFER_SOURCE_CONTROL_ENDPOINTS_RUNTIME_KEY = (
     "router_hint_source_control_endpoints"
 )
+KV_HINT_GPU_DIRECT_CAPABILITY_KEY = "router_hint_gpu_direct"
+KV_HINT_GPU_SOURCE_CONTROL_ENDPOINTS_RUNTIME_KEY = (
+    "router_hint_gpu_source_control_endpoints"
+)
 
 
 class DisaggregationMode(Enum):