Route undrained-ack reservation refusal around the capacity retry

The reserve_write_cp undrained-ack refusal returned a bare
HiCacheWriteFailure, which _reserve_write_cp_indices_no_collective
interpreted as a host-capacity failure: with free host space the
retry path tripped the predicted-no-deficit RuntimeError (crashing
the scheduler in exactly the scenario the gate exists to handle
gracefully), and with a deficit it triggered pointless evictions
before refusing again.

Give HiCacheWriteFailure an explicit reason (default host_capacity
keeps all existing constructors/semantics); the wrapper returns
non-capacity failures immediately as skip-this-round, and a
wrapper-level test pins that undrained_ack reaches neither the retry
admission nor host eviction.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

python/sglang/srt/managers/cache_controller.py

@@ -321,6 +321,11 @@ class HiCacheWriteResult:
 class HiCacheWriteFailure:
     required_host_slots: int
     metadata: object = None
+    # Why the reservation failed. "host_capacity" failures are retryable via
+    # deterministic host eviction; "undrained_ack" means the node's previous
+    # backup ack has not been drained yet — callers must skip this backup
+    # round instead of entering the capacity retry/eviction path.
+    reason: str = "host_capacity"
 
 
 @dataclass
@@ -1039,7 +1044,7 @@ class HiCacheController:
                 "has an undrained write ack; skipping duplicate backup",
                 node_id,
             )
-            return HiCacheWriteFailure(required_host_slots=0)
+            return HiCacheWriteFailure(required_host_slots=0, reason="undrained_ack")
 
         page_size = self.page_size
         layout = self.cp_shared_kv_layout

python/sglang/srt/mem_cache/hiradix_cache.py

@@ -2212,6 +2212,12 @@ class HiRadixCache(RadixCache):
         )
         if not isinstance(result, HiCacheWriteFailure):
             return result
+        if getattr(result, "reason", "host_capacity") != "host_capacity":
+            # Not a capacity failure (e.g. undrained_ack: the node's previous
+            # backup ack is still queued). Eviction cannot help, and entering
+            # the retry path would trip the predicted-no-deficit fail-fast
+            # below. Skip this backup round.
+            return result
 
         retry_admission = self._cp_build_write_admission(
             device_indices, node_id=node_id, phase="retry_after_local_failure"

test/registered/unit/mem_cache/test_cp_hicache_metadata.py

@@ -2224,6 +2224,41 @@ class TestHiRadixCacheCPBackup(CustomTestCase):
             HiCacheController.node_has_undrained_write_ack(controller, 9999)
         )
 
+    def test_reserve_wrapper_skips_capacity_retry_on_undrained_ack(self):
+        """An undrained_ack refusal must return through
+        _reserve_write_cp_indices_no_collective as a skip-this-round failure:
+        entering the capacity retry would either trip the
+        predicted-no-deficit RuntimeError (free host) or trigger pointless
+        evictions (full host)."""
+        from sglang.srt.managers.cache_controller import HiCacheWriteFailure
+
+        cache = HiRadixCache.__new__(HiRadixCache)
+        refusal = HiCacheWriteFailure(required_host_slots=0, reason="undrained_ack")
+        cache.cache_controller = types.SimpleNamespace(
+            reserve_write_cp=lambda device_indices, node_id: refusal,
+        )
+
+        def _admission(device_indices, node_id, phase):
+            if phase != "initial":
+                raise AssertionError(
+                    f"undrained_ack must not reach admission phase {phase!r}"
+                )
+            return types.SimpleNamespace(
+                deficit_by_owner=[0],
+                eviction_plan=types.SimpleNamespace(remaining_deficit=[0]),
+            )
+
+        cache._cp_build_write_admission = _admission
+        cache._evict_cp_host_for_write_admission = lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("undrained_ack must not trigger host eviction")
+        )
+
+        result = cache._reserve_write_cp_indices_no_collective(
+            torch.arange(4, dtype=torch.int64), 4614
+        )
+        self.assertIs(result, refusal)
+        self.assertEqual(result.reason, "undrained_ack")
+
     def test_rollback_pending_backup_scrubs_acks_and_cancels_state(self):
         """The write_backup exception rollback must not orphan an ack or a
         half-submitted layer-write state: an orphaned ack re-opens the