feat(kvbm): integrate kvbm-logical into Qwen3-4B KV cache

[xiaguan]

Summary

• Port 11 NVIDIA Dynamo kvbm crates into workspace (Apache-2.0)
• Add pegainfer-kv-cache crate wrapping kvbm-logical's BlockManager with KvCacheManager / RequestKv / KvView
• Replace KvPool/KvState with kvbm-logical lifecycle in Qwen3-4B executor

Architecture change

Before	After
Qwen3Model owns KvPool	Model is pure weights, executor owns KvCacheManager
Per-TP-rank RequestStateStore	Centralized HashMap<RequestId, RequestKv> in executor
Workers manage per-request state	Workers stateless, receive pre-computed KvViews
ensure_capacity/advance inside forward	Forward functions pure compute: &[KvView] + &CudaSlice<bf16> + &KvLayout
Implicit TP page-index consistency	Single BlockManager guarantees identical indices

Block lifecycle

schedule_prefill → prefill_view → batch_prefill (kernel) → apply_prefill
schedule_decode  → decode_view  → batch_decode  (kernel) → apply_decode
drop_request     → RAII reclaims blocks

Test plan

• cargo check --tests --bins clean (0 errors, 0 warnings)
• pegainfer-kv-cache lifecycle tests (10 cases)
• vllm bench serve 200 requests, block pool stable at ~2440, no leak
• E2E greedy regression: cargo test -p pegainfer-qwen3-4b --test e2e
• TPOT bench ±3% vs baseline

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request integrates the Dynamo KV cache management framework into the codebase, introducing several ported crates (such as dynamo-tokens, dynamo-kv-hashing, and dynamo-kv-router), licensing files, and detailed design documentation. The changes establish a logical/physical split KV cache design, a radix tree indexer, and a lower-tier continuation indexer. The review feedback identifies two critical issues: a potential memory leak and stale worker references in anchor_nodes due to a lack of pruning, and a state desynchronization bug in remove_blocks_impl where an early return aborts the removal of subsequent blocks in a batch.

[gemini-code-assist]

The anchor_nodes map is never pruned or cleaned up, leading to unbounded memory growth (memory leak) over time as new anchors are registered. Furthermore, when a worker is removed via RemoveWorker or RemoveWorkerDpRank, the worker is only dropped from nodes present in the thread-local lookup map. Since apply_anchor does not insert the anchor node into lookup, the worker is never removed from the anchor node's full_edge_workers (unless resolve_anchor_lookup happened to be called), leading to stale worker references in the anchor nodes.

Consider implementing a pruning mechanism for anchor_nodes during cleanup (e.g., in sweep_stale_children or run_cleanup_task), removing any anchor nodes that have no active workers and no children. Additionally, ensure that when a worker is removed, they are also cleaned up from the anchor_nodes map.

[gemini-code-assist]

If a batch of block hashes is removed, and one of them is not found in worker_map, remove_blocks_impl immediately returns Err(KvCacheEventError::BlockNotFound) and aborts the loop. This leaves the remaining blocks in the batch registered, leading to state desynchronization between the primary indexer (which continues processing remaining blocks on error) and the lower-tier indexer.

Modify the loop to process all block hashes in the batch, logging a warning or accumulating the error, and only return the error after the loop completes, matching the behavior of the primary indexer.

        let mut kv_cache_err = None;
        let remove_worker_entry = {
            let Some(worker_map) = worker_blocks.get_mut(&worker) else {
                return Err(KvCacheEventError::BlockNotFound);
            };

            for block_hash in block_hashes {
                if let Some(key) = worker_map.remove(block_hash) {
                    self.remove_worker_from_edge(key, worker);
                } else {
                    if kv_cache_err.is_none() {
                        kv_cache_err = Some(KvCacheEventError::BlockNotFound);
                    }
                }
            }

            worker_map.is_empty()
        };

        if remove_worker_entry {
            worker_blocks.remove(&worker);
        }

        kv_cache_err.map_or(Ok(()), Err)