Prevent duplicate CP HiCache write acks from crashing split drain

CP per-layer write-through can leave duplicated ready ack ids in the
ack queue while pending-split insertion drains write visibility. The
pre-scan observed both duplicates as registered before the first
completion removed radix state, so the second completion raised KeyError
and killed the scheduler.

The write-check path now records completed ack ids until matching queue
entries are drained. Duplicate completed acks are removed with a warning,
while genuinely unknown or unregistered acks still fail fast.

Constraint: CP HiCache pending-split drain may call writing_check outside the normal idle event path.
Rejected: Blind pop(..., None) for all unknown acks | would silently hide truly unattached or corrupt ack state.
Confidence: high
Scope-risk: moderate
Directive: Do not remove the completed-ack short-term memory unless duplicate and stale ack queues are proven impossible across TP MIN synchronization.
Tested: Local py_compile for hiradix_cache.py and test_cp_hicache_metadata.py.
Tested: Remote cjy-glm5-new py_compile and full test_cp_hicache_metadata.py, 121 passed.
Not-tested: Full ETE prefill workload after restarting service.
Co-authored-by: OmX <omx@oh-my-codex.dev>

test/registered/unit/mem_cache/test_cp_hicache_metadata.py

@@ -2174,6 +2174,72 @@ class TestHiRadixCacheCPBackup(CustomTestCase):
         self.assertEqual(cache.ongoing_write_through, {53: attached_node})
         self.assertEqual(dec_locked, [])
 
+    def test_writing_check_ignores_duplicate_ready_ack_for_same_node(self):
+        cache = HiRadixCache.__new__(HiRadixCache)
+        cache._uses_cp_hicache = True
+        cache.tp_world_size = 1
+        cache.enable_storage = False
+        cache.pending_host_backups = {}
+
+        class ReadyEvent:
+            def query(self):
+                return True
+
+            def synchronize(self):
+                pass
+
+        node = TreeNode()
+        node.id = 4614
+        dec_locked = []
+        cache.ongoing_write_through = {4614: node}
+        cache.dec_node_lock_ref = lambda released_node: dec_locked.append(released_node.id)
+        cache.cache_controller = types.SimpleNamespace(
+            ack_write_queue=[
+                HiCacheAck(ReadyEvent(), ReadyEvent(), [4614]),
+                HiCacheAck(ReadyEvent(), ReadyEvent(), [4614]),
+            ],
+        )
+
+        cache.writing_check()
+
+        self.assertEqual(cache.ongoing_write_through, {})
+        self.assertEqual(cache.cache_controller.ack_write_queue, [])
+        self.assertEqual(dec_locked, [4614])
+
+    def test_writing_check_drops_stale_duplicate_ack_before_valid_ack(self):
+        cache = HiRadixCache.__new__(HiRadixCache)
+        cache._uses_cp_hicache = True
+        cache.tp_world_size = 1
+        cache.enable_storage = False
+        cache.pending_host_backups = {}
+        cache._completed_write_ack_ids = {4614}
+
+        class ReadyEvent:
+            def query(self):
+                return True
+
+            def synchronize(self):
+                pass
+
+        node = TreeNode()
+        node.id = 4615
+        dec_locked = []
+        cache.ongoing_write_through = {4615: node}
+        cache.dec_node_lock_ref = lambda released_node: dec_locked.append(released_node.id)
+        cache.cache_controller = types.SimpleNamespace(
+            ack_write_queue=[
+                HiCacheAck(ReadyEvent(), ReadyEvent(), [4614]),
+                HiCacheAck(ReadyEvent(), ReadyEvent(), [4615]),
+            ],
+        )
+
+        cache.writing_check()
+
+        self.assertEqual(cache.ongoing_write_through, {})
+        self.assertEqual(cache.cache_controller.ack_write_queue, [])
+        self.assertEqual(dec_locked, [4615])
+        self.assertEqual(cache._completed_write_ack_ids, set())
+
     def test_write_backup_cp_failfast_on_unplanned_reservation_failure(self):
         cache = HiRadixCache.__new__(HiRadixCache)
         cache._uses_cp_hicache = True