[KVCache][Engine] fix has_pending_work and move swap/evict to worker layer

## Motivation

BatchRequest.__len__ 混入了 prefetch/swap/evict 任务数量，导致 engine 调度
逻辑（判断是否有待处理工作）出现误判；同时 swap/evict 提交散落在
gpu_model_runner 和 resource_manager 中，职责不清晰。

## Modifications

- engine/request.py: 新增 has_pending_work 属性，__len__ 恢复只计 requests
  数量；has_pending_work 同时感知 prefetch/swap/evict 任务
- engine/common_engine.py: 用 has_pending_work 替换 len(batch_request) > 0
  判断，逻辑更准确
- worker/worker_process.py: 将 submit_swap_tasks 调用移至 worker 层处理，
  处理后清空 metadata 避免重复提交
- worker/gpu_model_runner.py: 移除重复的 submit_swap_tasks 调用
- engine/sched/resource_manager_v1.py: 调整 check_and_add_pending_backup /
  issue_pending_backup / dispatch_pending_prefetches 执行顺序，去掉对
  len(batch_request) 的依赖
- cache_manager/v1/cache_manager.py: 恢复 matched_nodes 按 device/host 分类
  逻辑（之前被误注释）

## Usage or Command

```bash
# 启动服务（单机）
cd baidu/FastDeploy
bash run.sh
```

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: kevincheng2

fastdeploy/cache_manager/v1/cache_manager.py

@@ -516,11 +516,10 @@ def match_prefix(
                 #   Split matched_nodes into device blocks and host blocks
                 if self.enable_host_cache:
                     for node in matched_nodes:
-                        pass
-                        # if node.is_on_device():
-                        #     result.device_nodes.append(node)
-                        # elif node.is_on_host():
-                        #     result.host_nodes.append(node)
+                        if node.is_on_device():
+                            result.device_nodes.append(node)
+                        elif node.is_on_host():
+                            result.host_nodes.append(node)
                 else:
                     result.device_nodes = matched_nodes
 

fastdeploy/engine/common_engine.py

@@ -1112,7 +1112,7 @@ def _fetch_request():
                 batch_request, error_tasks = self.resource_manager.schedule()
 
                 # 3. Send to engine
-                if len(batch_request) > 0:
+                if batch_request.has_pending_work:
                     if self.cfg.scheduler_config.splitwise_role == "decode":
                         for task in batch_request:
                             if task.task_type == RequestType.PREEMPTED:
@@ -1191,7 +1191,7 @@ def _fetch_request():
                             continue
                         self._send_error_response(request_id, failed)
 
-                if len(batch_request) <= 0 and not error_tasks:
+                if not batch_request.has_pending_work and not error_tasks:
                     time.sleep(0.005)
 
             except RuntimeError as e:

fastdeploy/engine/request.py

@@ -697,10 +697,17 @@ def __getitem__(self, index):
         return self.requests[index]
 
     def __len__(self):
-        count = len(self.requests)
-        if self.storage_prefetch_tasks:
-            count += len(self.storage_prefetch_tasks)
-        return count
+        return len(self.requests)
+
+    @property
+    def has_pending_work(self) -> bool:
+        """Whether there is any pending work (inference requests, prefetch/swap/evict tasks)."""
+        return (
+            len(self.requests) > 0
+            or bool(self.storage_prefetch_tasks)
+            or bool(self.cache_swap_metadata)
+            or bool(self.cache_evict_metadata)
+        )
 
     def append(self, batch_request: "BatchRequest"):
         self.requests.extend(batch_request.requests)

fastdeploy/engine/sched/resource_manager_v1.py

@@ -1276,17 +1276,16 @@ def _allocate_decode_and_extend():
             # Issue pending backup tasks to batch_request
             # This handles write_through_selective policy by attaching backup tasks
             # to the batch request, which will be processed by the worker
-            if self.enable_cache_manager_v1 and len(batch_request) > 0:
+            if self.enable_cache_manager_v1:
+                self.cache_manager.check_and_add_pending_backup()
+
                 evict_metadata = self.cache_manager.issue_pending_backup_to_batch_request()
                 if evict_metadata:
                     batch_request.append_evict_metadata([evict_metadata])
 
-            if self.enable_cache_manager_v1:
-                self.cache_manager.check_and_add_pending_backup()
-
-            # Dispatch any pending storage prefetch tasks via batch_request
-            if self.config.cache_config.kvcache_storage_backend:
-                self._dispatch_pending_prefetches(batch_request)
+                # Dispatch any pending storage prefetch tasks via batch_request
+                if self.config.cache_config.kvcache_storage_backend:
+                    self._dispatch_pending_prefetches(batch_request)
 
             return batch_request, error_reqs
 

fastdeploy/worker/gpu_model_runner.py

@@ -804,12 +804,6 @@ def insert_tasks_v1(self, req_dicts: BatchRequest, num_running_requests: int = N
         if self.enable_mm:
             # Sort by idx to ensure attention mask offsets are filled in order during mm prefill
             req_dicts = sorted(req_dicts, key=lambda r: r.idx)
-        if self.enable_cache_manager_v1:
-            # submit_swap_tasks handles:
-            # 1. Waiting for pending evict handlers before submitting new evict
-            # 2. write_back policy: waiting for evict to complete before submitting swap-in
-            # 3. Adding handlers to pending lists appropriately
-            self.cache_controller.submit_swap_tasks(req_dicts.cache_evict_metadata, req_dicts.cache_swap_metadata)
 
         for i in range(req_len):
             request = req_dicts[i]

fastdeploy/worker/worker_process.py

@@ -688,6 +688,14 @@ def event_loop_normal(self) -> None:
                     self._handle_prefetch_tasks(batch_request.storage_prefetch_tasks)
                     batch_request.storage_prefetch_tasks = None
 
+                # Handle swap/evict tasks from batch_request
+                if batch_request.cache_evict_metadata or batch_request.cache_swap_metadata:
+                    self.worker.model_runner.cache_controller.submit_swap_tasks(
+                        batch_request.cache_evict_metadata, batch_request.cache_swap_metadata
+                    )
+                    batch_request.cache_evict_metadata = None
+                    batch_request.cache_swap_metadata = None
+
                 if len(control_reqs) > 0:
                     logger.info(f"Rank: {self.local_rank} received {len(control_reqs)} control request.")
                     for control_req in control_reqs: