feat(kv-offload): use external PegaFlow server

Cargo.toml

@@ -159,6 +159,7 @@ sha2 = "0.11"
 tmq = "0.5.0"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"
+tonic = "0.14"
 tokio-util = "0.7"
 toml = "1.1"
 tower = "0.5"

docs/index.md

@@ -117,7 +117,8 @@ Organized by domain (model line / subsystem / playbook / lesson) instead of by l
 | `subsystems/router/kv-aware-routing.md` | Dynamo KV-aware routing on 8×Qwen3-4B (RTX 5090): cache-affinity routing keeps a multi-turn conversation on its home worker, so follow-up-turn TTFT stays flat ~45ms vs round-robin 160–170ms / random 165–180ms (all-turns p50 3.3–3.8× lower). Router prefix overlap 0.72 under KV, 0 under stateless policies; `kv_hit_rate>0` is the gate that the worker↔router block-hash bridge is actually matching. Includes the per-response `prompt_tokens_details.cached_tokens` signal. |
 | `subsystems/runtime/runtime.md` | Runtime complexity is controlled by a shared `openinfer-core` that owns the generation contract and orchestration; per-model crates implement `ModelForward` so prefill/decode and hybrid attention stay hidden from the caller. State (`&mut`) is separated from weights (`&self`) for future bs > 1. |
 | `subsystems/runtime/kv-cache-design.md` | Dynamo 式 logical/physical 分层 KV cache：BlockManager 管 block 生命周期和 admission，PhysicalBackend trait 管 GPU 内存和布局（FullAttention / MLA）。支持 TP / DP。基于 vLLM/Dynamo/pegaflow 调研。 |
-| `subsystems/runtime/pegaflow-offload-integration.md` | 把 `pegaflow-core` 当进程内 Rust 库做 KV 卸载物理后端（HBM→DRAM/SSD/RDMA），补 kvbm 没写的卸载层。**Qwen3-4B full-attn 首发，端到端已在真实 GPU 跑通并验证**（async SAVE+LOAD 接进 executor/scheduler，纯 CPU-hit 与 GPU+CPU 组合 hit 恢复后 logits 与冷算一致）。pegaflow 经 git rev pin（#331+#333）。默认关，server CLI 已接（#316：`--kv-offload`/`--no-prefix-cache`，plain+LoRA）。linear 排除，sparse 暂缓。 |
+| `subsystems/runtime/pegaflow-offload-integration.md` | Historical design and Qwen3 evidence for the retired embedded PegaFlow backend; current external-server ownership is tracked in `external-pegaflow-server.md`. |
+| `subsystems/runtime/external-pegaflow-server.md` | External-only PegaFlow boundary: OpenInfer registers CUDA IPC allocations and issues save/query/load/flush RPCs; the server owns DRAM/SSD/RDMA. Qwen3 and GLM5.2 cross-process gates on 8×H200 are in progress. |
 
 ## subsystems / scheduler
 

docs/subsystems/runtime/external-pegaflow-server.md

@@ -0,0 +1,42 @@
+# External PegaFlow server for KV offload
+
+> **TL;DR:** OpenInfer is a CUDA-IPC/RPC client selected by `--kv-offload-server`; the external PegaFlow process owns the host, SSD, and RDMA tiers. Native registration, terminal Load completion, and Flush are implemented for Qwen3 and GLM5.2, with final-head cross-process replay remaining before deployment.
+>
+> **Last touched:** 2026-07
+
+## Ownership boundary
+
+- External mode is the only PegaFlow integration. OpenInfer no longer embeds a `PegaEngine` or constructs a second pinned-memory pool.
+- OpenInfer owns model KV allocations and exports them through CUDA IPC. PegaFlow imports those allocations, owns the storage hierarchy, and performs GPU↔host transfers.
+- `--kv-offload-server` enables the client. `--kv-offload-namespace` identifies the checkpoint/deployment content domain; vLLM compatibility uses the connector namespace.
+- Server capacity, SSD, RDMA, routing, and topology configuration stay outside OpenInfer.
+
+## Wire contract
+
+- Each registered layer carries one CUDA IPC allocation handle plus view offset, view size, and block stride. This avoids positionally coupling a second stride array to the layer list.
+- The server validates the allocation device and view bounds, opens each allocation once per registration batch, and keeps the mapping alive until unregister.
+- Native registration has an exact capability version. Old and new clients fail before importing memory instead of silently assuming a dense layout.
+- Native `Load` returns only after the GPU transfer completes. The Python connector retains its shared-memory completion path.
+- `Flush` waits for previously submitted saves to become cache-visible and for queued MetaServer registrations to be attempted.
+- External offload uses exportable CUDA allocations; the default offload-disabled path retains the stream-ordered allocator.
+
+## Model layouts
+
+- Qwen3 registers one page-first fused allocation with strided per-layer views.
+- GLM5.2 registers rank-local MLA and index-K arenas by name: 78 MLA plus 21 index-K arenas per EP8 rank.
+- Both clients preserve PegaFlow query leases and load only after a host-tier hit has produced a valid lease.
+
+## Validation
+
+Current PR-head local gates are green:
+
+- PegaFlow server: 23 unit tests; core: 124 passed with one GPU-only test ignored; CUDA 12/13 checks and Python wheel builds pass.
+- OpenInfer KV offload: 9 unit tests; workspace CPU tests, simulated frontend E2E, SM80 CUDA compile, and SM80 CUDA clippy pass.
+
+Cross-process evidence collected during bring-up established the layout and transfer contract:
+
+- Qwen page-first save/load restored three blocks into different HBM block IDs byte-for-byte; an untouched block remained zero.
+- Qwen3 forced CPU-only and combined GPU/CPU prefix restores without a material head-logprob shift.
+- GLM5.2 EP8 registered 99 arenas per rank and restored one 64-token host block with the expected first output tokens.
+
+Those runs predate the final PR heads. Replay the Qwen and GLM5.2 gates after PegaFlow #407 and OpenInfer #729 are merged; do not treat the historical runs as deployment evidence.

docs/subsystems/runtime/pegaflow-offload-integration.md

@@ -1,8 +1,8 @@
 # pegaflow KV 卸载接入 Spec
 
-> **TL;DR**: 把 `pegaflow-core` 当**进程内 Rust 库**做 KV 卸载的物理后端（HBM→DRAM/SSD/RDMA），补上 kvbm 留着没写的卸载层。connector 大脑（决定 load/save 哪些 block）用 kvbm logical/physical 分层思想自建，pegaflow 退为语义无关的 raw block transfer 后端。**路线已调整为 Qwen3-4B full-attn 首发**（原计划 Kimi 首发）：page-first 单 buffer 经 pegaflow `block_stride_bytes`（PR #331）适配。**端到端已在真实 GPU 上跑通并验证**：async SAVE + async LOAD 接进 `Qwen3Executor` + scheduler，`tests/kv_offload_cpu_hit.rs` 覆盖纯 CPU-hit 与 GPU+CPU 组合 hit，恢复后 logits 与冷算一致；连接层 `OffloadEngine` + `tests/cpu_roundtrip.rs` 字节级一致。默认关（builder flag opt-in）；**server CLI 已接**（#316：`--kv-offload` / `--kv-offload-host-gib` / `--no-prefix-cache`，plain 与 `--enable-lora` 两条启动路径都透传）。纯-L2 基准实测 Qwen3-4B mean TTFT 195→40ms（−79%，evict-before-probe → `gpu_hit=0`，全前缀从 host tier 恢复）。**Qwen3.5 linear/SSM state 明确排除**；**DeepSeek sparse 暂缓**。
+> **TL;DR**: 本文保留最初把 `pegaflow-core` 嵌入 OpenInfer 的设计与 Qwen3 验证记录；当前实现已改为 external-only：OpenInfer 通过 `--kv-offload-server` 注册 CUDA IPC allocation 并调用 save/query/load/flush RPC，DRAM/hugepage/SSD/RDMA 全归 PegaFlow server。当前边界和跨进程验证见 [external-pegaflow-server.md](external-pegaflow-server.md)。
 >
-> Last touched: 2026-06
+> Last touched: 2026-07
 
 ## 0. 实现状态（2026-06）
 

openinfer-glm52/src/lib.rs

@@ -43,7 +43,7 @@ use std::{
 use anyhow::{Context as _, Result, bail, ensure};
 use bytesize::ByteSize;
 use openinfer_core::engine::{EngineHandle, KvCapacity, LoadSnapshot};
-use openinfer_kv_offload::{HostConfig, KvArena, OffloadEngine, OffloadHost};
+use openinfer_kv_offload::{KvArena, OffloadEngine, OffloadHost};
 use remote::Glm52RemoteNode;
 use runner::{Glm52RankPlacement, Glm52RankWorker, Glm52Worker};
 
@@ -316,44 +316,25 @@ mod topology_tests {
     }
 }
 
-/// Host-tier KV offload knobs. One `PegaEngine` (one pinned pool) backs all
-/// 8 DP ranks under a single namespace: the MLA latent has no TP sharding
+/// External PegaFlow KV offload for all 8 DP ranks under one namespace. The
+/// MLA latent has no TP sharding
 /// and the non-expert weights are replicated, so any rank's KV for a token
 /// prefix is as good as any other's — the same tolerance as reusing a
 /// rank's own prefix cache (FP reduction order may differ across the batch
 /// shapes that computed it, never the semantics). Any rank restores what
 /// any rank saved.
 #[derive(Clone, Debug)]
 pub struct Glm52KvOffloadOptions {
-    /// Host pinned-memory pool size in bytes, shared by all ranks.
-    pub pinned_pool_bytes: usize,
-    /// Back the pool with hugepages (the box must hold a reservation —
-    /// check `HugePages_Total`).
-    pub use_hugepages: bool,
-    /// `Some` joins the cross-instance P2P mesh: saved block hashes register
-    /// with the MetaServer and missing prefixes are pulled from peer
-    /// instances over RDMA — the P/D disaggregation data plane.
-    pub p2p: Option<Glm52P2pOptions>,
+    /// Out-of-process PegaFlow gRPC endpoint.
+    pub server_addr: String,
+    /// Stable deployment/checkpoint identity for native OpenInfer peers.
+    pub namespace: Option<String>,
     /// `Some` when the P/D prefill peer is vLLM (pegaflow connector): offload
     /// query keys switch from kvbm lineage hashes to vLLM's prefix-cache hash
     /// scheme so this decode node can find the blocks vLLM registered.
-    /// Requires `p2p` (the peer's KV lives in its pegaflow-server's pool,
-    /// even on the same host).
     pub vllm_compat: Option<Glm52VllmCompatOptions>,
 }
 
-/// Cross-instance P2P KV sharing (see `openinfer_kv_offload::P2pConfig`).
-#[derive(Clone, Debug)]
-pub struct Glm52P2pOptions {
-    /// MetaServer gRPC address, e.g. `http://10.0.0.100:50056`.
-    pub metaserver_addr: String,
-    /// This engine's routable `IP:port` (doubles as the embedded transfer
-    /// service's bind address). Must be reachable by every peer.
-    pub advertise_addr: String,
-    /// RDMA NIC device names to register the pinned pool on.
-    pub rdma_nics: Vec<String>,
-}
-
 /// Decode-node settings for a P/D deployment whose prefill node is vLLM with
 /// the pegaflow connector (see `openinfer_kv_offload::VllmBlockHasher` and
 /// `docs/models/glm52/pd-vllm-prefill.md`).
@@ -436,17 +417,6 @@ pub fn launch(model_path: &Path, options: Glm52LaunchOptions) -> Result<EngineHa
         "GLM5.2 --kv-offload requires the EP8 topology (tp8 replicates KV on all ranks; \
          a host-tier restore would land on one)"
     );
-    // The vLLM prefill peer's KV lives in its pegaflow-server's pool (a
-    // separate process even on the same host); without the P2P mesh the
-    // compat keys would query an empty local tier and every request would
-    // wait out the full miss window.
-    ensure!(
-        kv_offload
-            .as_ref()
-            .is_none_or(|kv| kv.vllm_compat.is_none() || kv.p2p.is_some()),
-        "GLM5.2 --kv-pd-vllm-seed requires the KV P2P mesh (--kv-p2p-metaserver-addr, \
-         --kv-p2p-advertise-addr, --kv-p2p-nics)"
-    );
     // The miss window must sit inside the in-flight-fetch ceiling, or the
     // registration phase could never hand over to the fetch phase.
     ensure!(
@@ -728,16 +698,21 @@ fn start_engine(
     // down, and the launch error surfaces only after the ~100 s DeepEP
     // device timeout. The TP8 LL rendezvous rejecting a topology (poison
     // pill, NVLink probe) is a real failure landing exactly in this window.
-    let rank_arenas =
-        match build_rank_models(&loaded.workers, max_model_len, moe_topo, dspark_enabled) {
-            Ok(rank_arenas) => rank_arenas,
-            Err(err) => {
-                for worker in &loaded.workers {
-                    let _ = worker.request_shutdown();
-                }
-                return Err(err);
+    let rank_arenas = match build_rank_models(
+        &loaded.workers,
+        max_model_len,
+        moe_topo,
+        dspark_enabled,
+        kv_offload.is_some(),
+    ) {
+        Ok(rank_arenas) => rank_arenas,
+        Err(err) => {
+            for worker in &loaded.workers {
+                let _ = worker.request_shutdown();
             }
-        };
+            return Err(err);
+        }
+    };
     let vllm_compat = kv_offload
         .as_ref()
         .and_then(|opts| opts.vllm_compat.clone());
@@ -903,11 +878,14 @@ fn build_rank_models(
     max_model_len: usize,
     moe_topo: Glm52MoeTopo,
     dspark_enabled: bool,
+    exportable_kv: bool,
 ) -> Result<Vec<Vec<KvArena>>> {
     let build_started = Instant::now();
     let responses = workers
         .iter()
-        .map(|worker| worker.build_model_async(max_model_len, moe_topo, dspark_enabled))
+        .map(|worker| {
+            worker.build_model_async(max_model_len, moe_topo, dspark_enabled, exportable_kv)
+        })
         .collect::<Result<Vec<_>>>()?;
     let mut rank_arenas = Vec::with_capacity(responses.len());
     for (rank, response) in responses.into_iter().enumerate() {
@@ -970,28 +948,19 @@ fn build_offload_engines(
             .all(|arena| arena.bytes_per_block == mla_page_size * mla_bytes_per_token)),
         "GLM5.2 KV offload ranks disagree on MLA cache layout"
     );
-    let host = OffloadHost::new(HostConfig {
-        pinned_pool_bytes: opts.pinned_pool_bytes,
-        use_hugepages: opts.use_hugepages,
-        runtime_threads: 2,
-        p2p: opts
-            .p2p
-            .as_ref()
-            .map(|p2p| openinfer_kv_offload::P2pConfig {
-                metaserver_addr: p2p.metaserver_addr.clone(),
-                advertise_addr: p2p.advertise_addr.clone(),
-                rdma_nics: p2p.rdma_nics.clone(),
-            }),
-    })
-    .map_err(|err| anyhow::anyhow!("GLM5.2 KV offload host: {err}"))?;
+    let host = OffloadHost::connect(&opts.server_addr, 2)
+        .map_err(|err| anyhow::anyhow!("GLM5.2 KV offload host: {err}"))?;
     // vLLM-compat mode joins the *P side's* content domain: the pegaflow
     // connector derives an 8-hex namespace from vLLM config (and logs it at
     // startup); reproducing that derivation would mean chasing Python repr
     // of vLLM internals, so the operator passes it through explicitly.
     let namespace = match &opts.vllm_compat {
         Some(compat) => compat.namespace.clone(),
         None => format!(
-            "openinfer-glm52-l{GLM52_LAYERS}-p{}-mla{}-idxk{}",
+            "openinfer-glm52-{}-l{GLM52_LAYERS}-p{}-mla{}-idxk{}",
+            opts.namespace
+                .as_deref()
+                .context("GLM5.2 native KV offload requires a checkpoint namespace")?,
             mla_page_size,
             mla_bytes_per_token,
             config::GLM52_INDEX_HEAD_DIM + 4,
@@ -1029,10 +998,9 @@ fn build_offload_engines(
             .filter(|&layer| config::glm52_layer_has_full_indexer(layer))
             .count();
     log::info!(
-        "GLM5.2 KV offload up: {} pinned host pool (hugepages: {}), namespace {namespace}, \
+        "GLM5.2 KV offload up: server={}, namespace {namespace}, \
          {} rank instances x {arenas_per_rank} arenas",
-        ByteSize(opts.pinned_pool_bytes as u64),
-        opts.use_hugepages,
+        opts.server_addr,
         engines.len(),
     );
     Ok(engines)

openinfer-glm52/src/model/mod.rs

@@ -21,6 +21,7 @@ use cudarc::driver::{CudaSlice, CudaStream, DevicePtr as _, PinnedHostSlice};
 use half::bf16;
 use openinfer_core::cuda_graph::CudaGraphDumpSummary;
 use openinfer_core::cuda_graph::CudaGraphState;
+use openinfer_kernels::exportable::alloc_ipc_zeros;
 use openinfer_kernels::ops::{
     GLM52_FLASHMLA_SPARSE_BYTES_PER_TOKEN, GLM52_FLASHMLA_SPARSE_PAGE_SIZE,
     GLM52_FLASHMLA_SPARSE_TOPK, GLM52_GEMV_MMA_SCRATCH_FLOATS_PER_ROW, GLM52_MLA_CACHE_BYTES,
@@ -477,6 +478,7 @@ impl Glm52RankModel {
         moe_topo: crate::Glm52MoeTopo,
         attn_shard: Option<usize>,
         dspark_enabled: bool,
+        exportable_kv: bool,
     ) -> Result<Self> {
         ensure!(
             moe_topo.uses_tensor_replicated_moe() == attn_shard.is_some(),
@@ -523,17 +525,25 @@ impl Glm52RankModel {
                 build::build_decoder_layer(ctx, w, layer, moe_topo, attn_shard)
                     .with_context(|| format!("build GLM5.2 decoder layer {layer}"))?,
             );
+            let mla_len =
+                contract.num_blocks * GLM52_FLASHMLA_SPARSE_PAGE_SIZE * mla_cache_bytes_per_token;
+            let mla_cache = if exportable_kv {
+                alloc_ipc_zeros::<u8>(&ctx.stream, mla_len)
+            } else {
+                ctx.stream.alloc_zeros::<u8>(mla_len)
+            }?;
             caches.push(Glm52LayerCaches {
-                mla_cache: ctx.stream.alloc_zeros::<u8>(
-                    contract.num_blocks
-                        * GLM52_FLASHMLA_SPARSE_PAGE_SIZE
-                        * mla_cache_bytes_per_token,
-                )?,
+                mla_cache,
                 index_k_cache: glm52_layer_has_full_indexer(layer)
                     .then(|| {
-                        ctx.stream
-                            .alloc_zeros::<u8>(index_cache_layout.min_cache_bytes()?)
-                            .map_err(anyhow::Error::from)
+                        let len = index_cache_layout.min_cache_bytes()?;
+                        if exportable_kv {
+                            alloc_ipc_zeros::<u8>(&ctx.stream, len).map_err(anyhow::Error::from)
+                        } else {
+                            ctx.stream
+                                .alloc_zeros::<u8>(len)
+                                .map_err(anyhow::Error::from)
+                        }
                     })
                     .transpose()?,
             });

openinfer-glm52/src/remote.rs

@@ -115,6 +115,7 @@ enum WireRequest {
         max_model_len: usize,
         moe_topo: Glm52MoeTopo,
         dspark_enabled: bool,
+        exportable_kv: bool,
     },
     SetupComm {
         unique_id: Vec<u8>,
@@ -459,6 +460,7 @@ impl Glm52RemoteRankWorker {
         max_model_len: usize,
         moe_topo: Glm52MoeTopo,
         dspark_enabled: bool,
+        exportable_kv: bool,
     ) -> Result<Receiver<Result<Vec<KvArena>>>> {
         let (tx, rx) = bounded(1);
         self.node.shared.submit(
@@ -467,6 +469,7 @@ impl Glm52RemoteRankWorker {
                 max_model_len,
                 moe_topo,
                 dspark_enabled,
+                exportable_kv,
             },
             PendingResp::BuildModel(tx),
         )?;
@@ -809,10 +812,12 @@ fn host_demux_loop(
                 max_model_len,
                 moe_topo,
                 dspark_enabled,
+                exportable_kv,
             } => HostPending::BuildModel(worker.build_model_async(
                 max_model_len,
                 moe_topo,
                 dspark_enabled,
+                exportable_kv,
             )?),
             WireRequest::SetupComm {
                 unique_id,