feat(ipc): add DeepSeek4 DSpark shared worker

server/CMakeLists.txt

@@ -260,6 +260,7 @@ add_library(dflash_common STATIC
     src/deepseek4/deepseek4_target_shard_ipc_daemon.cpp
     src/deepseek4/deepseek4_dspark.cpp
     src/deepseek4/deepseek4_dspark_spec.cpp
+    src/deepseek4/deepseek4_dspark_draft_ipc_daemon.cpp
     src/flashprefill_q8.cpp
     src/kv_cache.cpp
     src/kv_quant.cpp

server/docs/DS4.md

@@ -1,6 +1,6 @@
 # DeepSeek V4 Flash — DFlash Integration
 
-This document describes the current DeepSeek V4 Flash implementation in DFlash. Today DeepSeek4 runs through the layer-split path: a local CUDA prefix shard plus either a local follow-on shard or a remote Halo/HIP target shard.
+This document describes the current DeepSeek V4 Flash implementation in DFlash. DeepSeek4 supports the layer-split target path and an opt-in DSpark speculative path. The DSpark proposal blocks can run in a separate HIP process while the complete target remains local.
 
 ## Model Architecture
 
@@ -28,6 +28,7 @@ DeepSeek V4 Flash is a 43-layer MoE model with:
 | Model weights and metadata | `src/deepseek4/deepseek4_internal.h`, `src/deepseek4/deepseek4_loader.cpp` |
 | HC pre/post CUDA kernel | `src/deepseek4/deepseek4_hc_cuda.cu`, `.h` |
 | Remote target-shard daemon | `src/deepseek4/deepseek4_target_shard_ipc_daemon.cpp` |
+| DSpark runtime and remote draft daemon | `src/deepseek4/deepseek4_dspark*.{h,cpp}` |
 | Shared target-shard IPC infrastructure | `src/common/target_shard_ipc.*`, `src/placement/remote_target_shard_config.h` |
 | Backend IPC CLI entry | `src/ipc/backend_ipc_main.cpp` |
 
@@ -75,6 +76,28 @@ For heterogeneous setups, the CUDA-built server can keep the prefix layers on th
 
 This path uses `TargetShardIpcSession`, `deepseek4_target_shard_ipc_daemon.cpp`, and `BackendIpcMode::DeepSeek4TargetShard` rather than the old expert-worker protocol.
 
+### Local target + remote HIP DSpark draft
+
+The speculative IPC mode keeps the complete DeepSeek4 target, KV cache, tied
+embedding, DSpark head, and sampler in the parent. A separate
+`BackendIpcMode::DeepSeek4DSparkDraft` process executes the DSpark proposal
+graph on the selected HIP GPU. The parent currently retains its locally loaded
+draft weights for metadata and fallback; removing that duplicate residency is
+a separate optimization.
+
+Each speculative step transfers:
+
+- target feature captures in token-major
+  `[n_tokens, n_target_layers, n_embd]` layout;
+- the embedded seed/MASK proposal block;
+- the proposal hidden states returned by the HIP worker.
+
+All target-layer captures are uploaded as one feature block and one
+acknowledgement. On POSIX hosts this mode defaults to the existing memfd-backed
+shared payload transport; proposal input and output reuse the same mapping.
+`stream` remains available as an A/B and compatibility fallback. This is shared
+host IPC, not direct `hipIpcMemHandle` peer memory.
+
 ## Shard Boundary State
 
 The shard boundary transfers the **full HC state tensor**, not a separate expert-routing payload:
@@ -102,6 +125,13 @@ The runtime logs the chosen split with a `[deepseek4-split] auto-split:` banner.
 |----------|---------|
 | `DFLASH_DS4_CUDA_LAYERS` | Override the auto-split heuristic and pin the first `N` DeepSeek4 layers to CUDA. The remaining `43 - N` layers run on the Halo shard. |
 | `DFLASH_DS4_TIMING` | Enable DS4 timing logs for the layer-split parent and target-shard daemon. Useful for profiling prefill/decode breakdowns; leave unset for normal runs. |
+| `DFLASH_DS4_SPEC` | Enable the DeepSeek4 DSpark speculative runtime. |
+| `DFLASH_DS4_DRAFT` | DSpark draft GGUF loaded locally for metadata/head use and by the remote worker. |
+| `DFLASH_DS4_DRAFT_IPC_BIN` | Backend IPC daemon executable used for the remote DSpark worker. |
+| `DFLASH_DS4_DRAFT_IPC_GPU` | HIP device index for the remote DSpark proposal blocks. |
+| `DFLASH_DS4_DRAFT_IPC_WORK_DIR` | Scratch directory for the child process. |
+| `DFLASH_DS4_DRAFT_IPC_REQUIRED` | Fail initialization instead of falling back to a local draft when remote startup fails. |
+| `DFLASH_DRAFT_IPC_TRANSPORT` | Payload transport: `auto`, `shared`, or `stream`. DeepSeek4 DSpark defaults to `auto`; other draft modes retain their existing default. |
 
 `DFLASH_DS4_TIMING` enables the existing timing banners:
 

server/src/common/backend_ipc.cpp

@@ -34,6 +34,7 @@ const char * backend_ipc_mode_name(BackendIpcMode mode) {
         case BackendIpcMode::LagunaTargetShard: return "laguna-target-shard";
         case BackendIpcMode::MoeExpertCompute: return "moe-expert-compute";
         case BackendIpcMode::DeepSeek4TargetShard: return "deepseek4-target-shard";
+        case BackendIpcMode::DeepSeek4DSparkDraft: return "deepseek4-dspark-draft";
     }
     return "unknown";
 }
@@ -67,6 +68,10 @@ bool parse_backend_ipc_mode(const std::string & value, BackendIpcMode & out) {
         out = BackendIpcMode::DeepSeek4TargetShard;
         return true;
     }
+    if (value == "deepseek4-dspark-draft") {
+        out = BackendIpcMode::DeepSeek4DSparkDraft;
+        return true;
+    }
     return false;
 }
 

server/src/common/backend_ipc.h

@@ -29,6 +29,7 @@ enum class BackendIpcMode {
     LagunaTargetShard,
     MoeExpertCompute,
     DeepSeek4TargetShard,
+    DeepSeek4DSparkDraft,
 };
 
 const char * backend_ipc_mode_name(BackendIpcMode mode);

server/src/common/dflash_draft_ipc.cpp

@@ -25,14 +25,15 @@ namespace dflash::common {
 
 namespace {
 
-BackendIpcPayloadTransport draft_ipc_transport_from_env() {
+BackendIpcPayloadTransport draft_ipc_transport_from_env(
+        BackendIpcPayloadTransport fallback) {
     const char * raw = std::getenv("DFLASH_DRAFT_IPC_TRANSPORT");
     if (!raw || !*raw) {
-        return BackendIpcPayloadTransport::Stream;
+        return fallback;
     }
-    BackendIpcPayloadTransport transport = BackendIpcPayloadTransport::Stream;
+    BackendIpcPayloadTransport transport = fallback;
     if (!parse_backend_ipc_payload_transport(raw, transport)) {
-        return BackendIpcPayloadTransport::Stream;
+        return fallback;
     }
     return transport;
 }
@@ -47,16 +48,23 @@ bool checked_mul_size(size_t a, size_t b, size_t & out) {
 
 size_t dflash_draft_ipc_required_shared_bytes(int hidden_size,
                                               int block_size,
-                                              int ring_cap) {
-    if (hidden_size <= 0 || block_size <= 0 || ring_cap <= 0) {
+                                              int ring_cap,
+                                              int n_target_layers) {
+    if (hidden_size <= 0 || block_size <= 0 || ring_cap <= 0 ||
+        n_target_layers <= 0) {
         return 0;
     }
-    const size_t max_tokens =
-        (size_t)std::max(block_size, ring_cap);
-    size_t elements = 0;
+    size_t noise_elements = 0;
+    size_t feature_elements = 0;
     size_t bytes = 0;
-    if (!checked_mul_size(max_tokens, (size_t)hidden_size, elements) ||
-        !checked_mul_size(elements, sizeof(float), bytes)) {
+    if (!checked_mul_size((size_t)block_size, (size_t)hidden_size,
+                          noise_elements) ||
+        !checked_mul_size((size_t)ring_cap, (size_t)n_target_layers,
+                          feature_elements) ||
+        !checked_mul_size(feature_elements, (size_t)hidden_size,
+                          feature_elements) ||
+        !checked_mul_size(std::max(noise_elements, feature_elements),
+                          sizeof(float), bytes)) {
         return 0;
     }
     return bytes;
@@ -100,36 +108,110 @@ bool DFlashDraftIpcClient::start(
         const std::string & draft_path,
         int draft_gpu,
         int ring_cap,
-        const std::string & work_dir) {
+        const std::string & work_dir,
+        BackendIpcMode mode) {
 #if defined(_WIN32)
-    (void)bin; (void)draft_path; (void)draft_gpu; (void)ring_cap; (void)work_dir;
+    (void)bin; (void)draft_path; (void)draft_gpu; (void)ring_cap; (void)work_dir; (void)mode;
     std::fprintf(stderr, "DFlash draft IPC is only implemented on POSIX hosts\n");
     return false;
 #else
     close();
-    if (bin.empty() || draft_path.empty() || ring_cap <= 0) return false;
+    if (bin.empty() || draft_path.empty() || ring_cap <= 0 ||
+        (mode != BackendIpcMode::DFlashDraft &&
+         mode != BackendIpcMode::DeepSeek4DSparkDraft)) return false;
     BackendIpcLaunchConfig launch;
     launch.bin = bin;
-    launch.mode = BackendIpcMode::DFlashDraft;
+    launch.mode = mode;
     launch.payload_path = draft_path;
     launch.work_dir = work_dir;
-    launch.payload_transport = draft_ipc_transport_from_env();
+    const BackendIpcPayloadTransport default_transport =
+        mode == BackendIpcMode::DeepSeek4DSparkDraft
+            ? BackendIpcPayloadTransport::Auto
+            : BackendIpcPayloadTransport::Stream;
+    launch.payload_transport = draft_ipc_transport_from_env(default_transport);
+    const int shared_feature_layers =
+        mode == BackendIpcMode::DeepSeek4DSparkDraft ? n_target_layers_ : 1;
     launch.shared_payload_bytes = draft_ipc_shared_bytes_from_env(
-        dflash_draft_ipc_required_shared_bytes(hidden_size_, block_size_, ring_cap));
+        dflash_draft_ipc_required_shared_bytes(
+            hidden_size_, block_size_, ring_cap, shared_feature_layers));
     launch.args.push_back("--ring-cap=" + std::to_string(ring_cap));
     launch.args.push_back("--draft-gpu=" + std::to_string(std::max(0, draft_gpu)));
     if (!process_.start(launch)) {
         std::fprintf(stderr, "draft-ipc backend process start failed\n");
         return false;
     }
     ring_cap_ = ring_cap;
+    mode_ = mode;
     active_ = true;
-    std::printf("[draft-ipc] ready bin=%s gpu=%d ring_cap=%d work_dir=%s\n",
-                bin.c_str(), draft_gpu, ring_cap, process_.work_dir().c_str());
+    std::printf("[draft-ipc] ready mode=%s bin=%s gpu=%d ring_cap=%d work_dir=%s\n",
+                backend_ipc_mode_name(mode), bin.c_str(), draft_gpu, ring_cap,
+                process_.work_dir().c_str());
     return true;
 #endif
 }
 
+bool DFlashDraftIpcClient::send_feature_block(
+        int start_pos,
+        int n_tokens,
+        const float * features,
+        size_t feature_count) {
+#if defined(_WIN32)
+    (void)start_pos; (void)n_tokens; (void)features; (void)feature_count;
+    return false;
+#else
+    FILE * cmd = process_.command_stream();
+    const int stream_fd = process_.stream_fd();
+    const int payload_fd = process_.payload_fd();
+    if (!active_ || mode_ != BackendIpcMode::DeepSeek4DSparkDraft ||
+        !cmd || stream_fd < 0 || start_pos < 0 || n_tokens <= 0 ||
+        n_tokens > ring_cap_) {
+        return false;
+    }
+    const size_t expected = (size_t)n_tokens * (size_t)n_target_layers_ *
+                            (size_t)hidden_size_;
+    if (!features || feature_count != expected) return false;
+    const size_t bytes = feature_count * sizeof(float);
+
+    if (process_.resolved_payload_transport() == BackendIpcPayloadTransport::Shared) {
+        uint64_t seq = 0;
+        if (!process_.write_shared_payload(features, bytes, seq)) {
+            std::fprintf(stderr,
+                         "draft-ipc feature_block shared payload too large bytes=%zu capacity=%zu\n",
+                         bytes, process_.shared_payload_capacity());
+            return false;
+        }
+        std::fprintf(cmd, "feature_block_shared %d %d %zu %" PRIu64 "\n",
+                     start_pos, n_tokens, bytes, seq);
+        std::fflush(cmd);
+        int32_t status = -1;
+        const bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) &&
+                        status == 0;
+        if (!ok) {
+            std::fprintf(stderr,
+                         "draft-ipc feature_block_shared failed status=%d\n",
+                         status);
+        }
+        return ok;
+    }
+
+    if (payload_fd < 0) return false;
+    std::fprintf(cmd, "feature_block_pipe %d %d %zu\n",
+                 start_pos, n_tokens, bytes);
+    std::fflush(cmd);
+    if (!write_exact_fd(payload_fd, features, bytes)) {
+        std::fprintf(stderr, "draft-ipc feature_block payload write failed\n");
+        return false;
+    }
+    int32_t status = -1;
+    const bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) &&
+                    status == 0;
+    if (!ok) {
+        std::fprintf(stderr, "draft-ipc feature_block failed status=%d\n", status);
+    }
+    return ok;
+#endif
+}
+
 bool DFlashDraftIpcClient::send_feature_slice(
         int capture_idx,
         int start_pos,
@@ -216,11 +298,19 @@ bool DFlashDraftIpcClient::propose(
     const int payload_fd = process_.payload_fd();
     if (!active_ || !cmd || stream_fd < 0 || committed < 0 ||
         ctx_len <= 0 || ctx_len > ring_cap_) {
+        std::fprintf(stderr,
+                     "draft-ipc propose rejected active=%d cmd=%p stream_fd=%d committed=%d ctx_len=%d ring_cap=%d\n",
+                     (int)active_, (void *)cmd, stream_fd, committed, ctx_len, ring_cap_);
         return false;
     }
     const size_t noise_expected =
         (size_t)hidden_size_ * block_size_;
-    if (noise_embed.size() != noise_expected) return false;
+    if (noise_embed.size() != noise_expected) {
+        std::fprintf(stderr,
+                     "draft-ipc propose noise size mismatch got=%zu expected=%zu hidden=%d block=%d\n",
+                     noise_embed.size(), noise_expected, hidden_size_, block_size_);
+        return false;
+    }
     const size_t bytes = noise_embed.size() * sizeof(float);
     if (process_.resolved_payload_transport() == BackendIpcPayloadTransport::Shared) {
         uint64_t seq = 0;
@@ -230,15 +320,20 @@ bool DFlashDraftIpcClient::propose(
                          bytes, process_.shared_payload_capacity());
             return false;
         }
-        std::fprintf(cmd, "propose_shared %d %d %zu %" PRIu64 "\n",
+        const bool bidirectional =
+            mode_ == BackendIpcMode::DeepSeek4DSparkDraft;
+        std::fprintf(cmd, bidirectional
+                     ? "propose_shared_bidir %d %d %zu %" PRIu64 "\n"
+                     : "propose_shared %d %d %zu %" PRIu64 "\n",
                      committed, ctx_len, bytes, seq);
         std::fflush(cmd);
         int32_t status = -1;
         bool ok = read_exact_fd(stream_fd, &status, sizeof(status)) && status == 0;
         if (ok) {
             hidden_out.assign(noise_expected, 0.0f);
-            ok = read_exact_fd(stream_fd, hidden_out.data(),
-                               hidden_out.size() * sizeof(float));
+            ok = bidirectional
+                ? process_.read_shared_payload(hidden_out.data(), bytes, seq)
+                : read_exact_fd(stream_fd, hidden_out.data(), bytes);
         }
         if (!ok) {
             std::fprintf(stderr, "draft-ipc propose_shared failed status=%d\n", status);
@@ -348,6 +443,7 @@ void DFlashDraftIpcClient::close() {
     process_.close();
     active_ = false;
     ring_cap_ = 0;
+    mode_ = BackendIpcMode::Invalid;
 }
 
 // ── Remote draft feature copy helper ────────────────────────────────

server/src/common/dflash_draft_ipc.h

@@ -17,6 +17,7 @@
 #include "ggml.h"
 #include "ggml-backend.h"
 
+#include <cstddef>
 #include <cstdint>
 #include <cstdio>
 #include <string>
@@ -41,13 +42,28 @@ class DFlashDraftIpcClient {
                const std::string & draft_path,
                int draft_gpu,
                int ring_cap,
-               const std::string & work_dir);
+               const std::string & work_dir,
+               BackendIpcMode mode = BackendIpcMode::DFlashDraft);
 
     bool send_feature_slice(int capture_idx,
                             int start_pos,
                             int n_tokens,
                             const std::vector<float> & slice);
 
+    // Token-major [n_tokens, n_target_layers, hidden_size] upload. This is the
+    // preferred DSpark path because all capture layers cross IPC in one
+    // command and one acknowledgement.
+    bool send_feature_block(int start_pos,
+                            int n_tokens,
+                            const float * features,
+                            size_t feature_count);
+    bool send_feature_block(int start_pos,
+                            int n_tokens,
+                            const std::vector<float> & features) {
+        return send_feature_block(
+            start_pos, n_tokens, features.data(), features.size());
+    }
+
     bool propose(int committed,
                  int ctx_len,
                  const std::vector<float> & noise_embed,
@@ -71,6 +87,7 @@ class DFlashDraftIpcClient {
     int hidden_size_ = DFLASH27B_TARGET_HIDDEN;
     int block_size_ = DFLASH27B_DRAFT_BLOCK_SIZE;
     int n_target_layers_ = DFLASH27B_DRAFT_N_TARGET_LAYERS;
+    BackendIpcMode mode_ = BackendIpcMode::Invalid;
 };
 
 // ── Remote draft feature copy helper ────────────────────────────────