Route reconstructed pages to C++ CUDA placement

deploy/deepseek_v4/tp4_profile.json

@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "4998b24f4f504aeeb9bf92769ec720e282f546e6726d89fdfd06c4efa8d17c10"
+    "source_sha256": "f7c0565521fddeff7085e4cc08043cb8d1e2bde33abc67f83b8608a162d05b88"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",

deploy/glm52_35bpw/profile.json

@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "4998b24f4f504aeeb9bf92769ec720e282f546e6726d89fdfd06c4efa8d17c10"
+    "source_sha256": "f7c0565521fddeff7085e4cc08043cb8d1e2bde33abc67f83b8608a162d05b88"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",

sparkcache/spark_context_cache_connector.py

@@ -168,7 +168,10 @@ def _load_native_components() -> SimpleNamespace:
         ArenaMode=placement.ArenaMode,
         RecordKind=placement.RecordKind,
         execute_native_restore=restore.execute_native_restore,
-        execute_native_hybrid_placement=(hybrid_restore.execute_native_hybrid_restore),
+        execute_native_hybrid_restore=hybrid_restore.execute_native_hybrid_restore,
+        execute_native_hybrid_placement=(
+            hybrid_restore.execute_native_hybrid_placement
+        ),
         bind_page_reference=binding.bind_page_reference,
         hybrid_page_cuda_capability=binding.CAP_HYBRID_PAGE_CUDA,
     )
@@ -683,7 +686,8 @@ def __init__(
         self._native_adapter: Any = None
         self._native_adapters: list[Any] = []
         self._native_execute_restore: Any = None
-        self._native_execute_hybrid: Any = None
+        self._native_execute_hybrid_restore: Any = None
+        self._native_execute_hybrid_placement: Any = None
         self._native_required_record_mask = 0
         # Scheduler state.
         self._need_load: dict[str, tuple[str, int]] = {}
@@ -2067,9 +2071,12 @@ def _configure_native_hybrid_restore(self) -> None:
                 )
                 adapters.append(adapter)
                 adapter.configure_pages(layout, self._layer_tensors)
-            execute = components.execute_native_hybrid_placement
-            if not callable(execute):
-                raise TypeError("native hybrid restore orchestrator is not callable")
+            execute_restore = components.execute_native_hybrid_restore
+            execute_placement = components.execute_native_hybrid_placement
+            if not callable(execute_restore):
+                raise TypeError("native hybrid direct-restore orchestrator is not callable")
+            if not callable(execute_placement):
+                raise TypeError("native hybrid page-placement orchestrator is not callable")
         except Exception as error:
             for adapter in adapters:
                 with contextlib.suppress(Exception):
@@ -2079,7 +2086,8 @@ def _configure_native_hybrid_restore(self) -> None:
             ) from error
         self._native_adapters = adapters
         self._native_adapter = adapters[0]
-        self._native_execute_hybrid = execute
+        self._native_execute_hybrid_restore = execute_restore
+        self._native_execute_hybrid_placement = execute_placement
         self.counters["native_hybrid_configured"] = 1
         logger.info(
             "spark-context-cache: native hybrid restore configured"
@@ -3220,14 +3228,16 @@ def _load_hybrid_pages(
         if len(groups) != len(layout.groups):
             raise HybridCodecError("request block tables disagree with page groups")
         if self._native_restore_enabled and lookup.root_kind != "page_delta":
-            if not self._native_adapters or not callable(self._native_execute_hybrid):
+            if not self._native_adapters or not callable(
+                self._native_execute_hybrid_restore
+            ):
                 raise RuntimeError(
                     "native hybrid restore selected without a configured adapter"
                 )
             if not 0 <= native_lane < len(self._native_adapters):
                 raise RuntimeError("native hybrid restore lane is unavailable")
             try:
-                result = self._native_execute_hybrid(
+                result = self._native_execute_hybrid_restore(
                     adapter=self._native_adapters[native_lane],
                     request_id=plan.request_id,
                     lookup=lookup,
@@ -3316,14 +3326,16 @@ def _load_hybrid_pages(
                 time.perf_counter_ns() - reassembly_started,
             )
         if self._native_restore_enabled:
-            if not self._native_adapters or not callable(self._native_execute_hybrid):
+            if not self._native_adapters or not callable(
+                self._native_execute_hybrid_placement
+            ):
                 raise RuntimeError(
                     "native hybrid restore selected without a configured adapter"
                 )
             if not 0 <= native_lane < len(self._native_adapters):
                 raise RuntimeError("native hybrid restore lane is unavailable")
             try:
-                result = self._native_execute_hybrid(
+                result = self._native_execute_hybrid_placement(
                     adapter=self._native_adapters[native_lane],
                     request_id=plan.request_id,
                     encoded_pages=encoded_pages,

sparkcache/test_spark_context_cache_connector.py

@@ -9,6 +9,7 @@
 
 import dataclasses
 import hashlib
+import inspect
 import json
 import os
 import struct
@@ -153,6 +154,19 @@ def _get_connector_metadata(self):
 )
 
 
+class NativeHybridDispatchTests(unittest.TestCase):
+    def test_loader_exposes_distinct_direct_and_materialized_page_paths(self) -> None:
+        components = connector_module._load_native_components()
+        direct = components.execute_native_hybrid_restore
+        materialized = components.execute_native_hybrid_placement
+
+        self.assertIsNot(direct, materialized)
+        self.assertIn("lookup", inspect.signature(direct).parameters)
+        self.assertNotIn("encoded_pages", inspect.signature(direct).parameters)
+        self.assertIn("encoded_pages", inspect.signature(materialized).parameters)
+        self.assertNotIn("lookup", inspect.signature(materialized).parameters)
+
+
 class CodecTests(unittest.TestCase):
     def test_owned_positions_interleave_one(self) -> None:
         self.assertEqual(codec.owned_positions(8, 4, 0), (0, 4))
@@ -558,7 +572,7 @@ def native_placement(**kwargs):
 
             connector._native_restore_enabled = True
             connector._native_adapters = [object()]
-            connector._native_execute_hybrid = native_placement
+            connector._native_execute_hybrid_placement = native_placement
             self.assertTrue(
                 connector._load_one(
                     _ReqPlan(