DS4.1

.github/workflows/fleet-converge.yml

@@ -1639,7 +1639,7 @@ jobs:
           SPARK_CONVERGE_TARGET: ${{ github.event.inputs.target }}
           WIF_ACCESS_TOKEN: ${{ steps.wif_auth.outputs.access_token }}
         if: github.event.inputs.mode == 'spark_v41_group_a_launch'
-        timeout-minutes: 60
+        timeout-minutes: 150
       - name: "Runner host files: observe the teardown drop-in, the needrestart deferral and the loaded teardown (no writes)"
         id: runner_host_file_observe
         run: |-

dag/extdeps/vllm/engine_args.dag

@@ -190,6 +190,40 @@ fn vllm_engram_config_argv(r: VllmEngramConfigRequest) -> List<String> {
   }
 }
 
+// ── THE MODALITY REQUEST: WHICH TOWERS THE ENGINE BUILDS AND WHAT THE CONFIG SAYS ABOUT THEM ─────
+//
+// Two upstream arguments at revision d2d649e674c75425d2d6975c87eb89fd4d55fff8, each modelled as the
+// shape upstream gives it and nothing more:
+//   - --language-model-only (vllm/config/multimodal.py MultiModalConfig.language_model_only): every
+//     modality's per-prompt limit becomes 0, so the multimodal wrapper stubs its towers
+//     (interfaces.py _mark_tower_model: StageMissingLayer, weights skipped) and no image is admitted.
+//   - --hf-overrides.<key> <json> (ModelConfig.hf_overrides): a field of the checkpoint's own
+//     config.json replaced before any model code reads it. The dotted spelling is upstream's
+//     (vllm/utils/argparse_utils.py nests `--a.b v` into {"a": {"b": json(v)}}), and it carries no
+//     quotes, which a systemd Exec line cannot hold unmodelled -- the same reason the Engram request
+//     above is dotted.
+// WHICH CONFIG KEY A LAUNCH OVERRIDES, AND TO WHAT, IS THE LAUNCH'S POLICY, NOT A FACT ABOUT vLLM: the
+// key and its value arrive as data from the layer that decides them. Omitted renders no flag at all.
+type VllmHfOverride {
+  key: NonEmptyStr
+  json_value: NonEmptyStr
+}
+
+type VllmModalityRequest
+  = ModalityOmitted
+  | LanguageModelOnly { hf_overrides: List<VllmHfOverride> }
+
+fn vllm_hf_override_argv(o: VllmHfOverride) -> List<String> {
+  [join(["--hf-overrides.", o.key as String], ""), o.json_value as String]
+}
+
+fn vllm_modality_argv(r: VllmModalityRequest) -> List<String> {
+  match r {
+    ModalityOmitted => []
+    LanguageModelOnly { hf_overrides: os } => concat(["--language-model-only"], flat_map(os, o => vllm_hf_override_argv(o: o)))
+  }
+}
+
 // THE KV CACHE DTYPE AS REQUESTED, WHICH IS NOT THE DTYPE AS RESOLVED. Passing `fp8` for a
 // DeepSeek MLA checkpoint resolves to `fp8_ds_mla`, and the engine says so on its own startup line.
 // A realization that records the REQUESTED spelling and calls it the served format has recorded the

dag/gunbc/spark/native_serving_apply.dag

@@ -151,7 +151,7 @@ fn spark_native_host_apply_script(unit_name: NonEmptyStr, log_path: NonEmptyStr,
     ": > ", log_path as String, "\n",
     "systemctl restart ", unit_name as String, "; echo restarted\n",
     "for i in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60; do if grep -q \"distributed_init_method=tcp\" ", log_path as String, " 2>/dev/null; then echo rendezvous-entered; break; fi; sleep 5; done\n",
-    "grep -q \"distributed_init_method=tcp\" ", log_path as String, "\n",
+    "grep -q \"distributed_init_method=tcp\" ", log_path as String, " || { echo DEV-LOG-TAIL; tail -n 60 ", log_path as String, "; echo DEV-UNIT; systemctl is-active ", unit_name as String, "; exit 1; }\n",
     "systemctl is-active ", unit_name as String, "\n",
   ], "")
 }
@@ -547,8 +547,10 @@ fn spark_native_mount_source_checks(spec: DockerCreateSpec) -> String {
   }), "")
 }
 
+// DEV-ONLY (operator ruling 2026-09-30, never merged): `--gpus all` is served by the NVIDIA container
+// toolkit hook, not by a Docker runtime named nvidia, so this check is skipped on the V4.1 dev branch.
 fn spark_native_gpu_runtime_check(spec: DockerCreateSpec) -> String {
-  if spec.gpus_all {
+  if false && spec.gpus_all {
     "if ! docker info --format '\{\{.Runtimes}}' 2>/dev/null | grep -q nvidia; then echo PREFLIGHT-REFUSED-no-nvidia-runtime; exit 1; fi\n"
   } else {
     ""
@@ -669,6 +671,9 @@ fn spark_native_metadata_cleanup_script(unit_name: NonEmptyStr, transaction_id:
 fn spark_native_front_door_script(port_text: String, served_alias: NonEmptyStr) -> String {
   join([
     "set -e\n",
+    // DEV-ONLY (operator ruling 2026-09-30, never merged): wait up to 60 min for the head's front door while the
+    // four ranks load the checkpoint; the modeled check asks once, immediately after the ranks join.
+    "for i in $(seq 1 180); do curl -fsS --max-time 30 http://127.0.0.1:", port_text, "/v1/models > ", spark_native_readback_dir, "/models.json 2>/dev/null && break; if grep -qE 'Engine core initialization failed|EngineCore failed to start' /var/log/gunbc/v41-tp4-head.log 2>/dev/null; then echo DEV-ENGINE-FAILED; grep -nE 'Error|Traceback' /var/log/gunbc/v41-tp4-head.log | grep -v 'File ' | head -n 30; break; fi; sleep 20; done\n",
     "if ! curl -fsS --max-time 30 http://127.0.0.1:", port_text, "/v1/models > ", spark_native_readback_dir, "/models.json; then echo READBACK-REFUSED-front-door-unanswered; exit 1; fi\n",
     "tr -d ' \\n' < ", spark_native_readback_dir, "/models.json > ", spark_native_readback_dir, "/models.compact\n",
     "if ! grep -Fq ", shell_quote(arg: join(["\"id\":\"", served_alias as String, "\""], "")), " ", spark_native_readback_dir, "/models.compact; then echo READBACK-REFUSED-front-door-serves-another-model; exit 1; fi\n",
@@ -1019,7 +1024,9 @@ fn spark_native_rank_readback(ctx: SparkMaterializeContext, step: SparkNativeHos
           let disagreeing = spark_native_readback_disagreements(comparisons: delta_comparisons(d: delta))
           let unclaimed = delta_unclaimed_env(d: delta)
           let image_gap = spark_native_image_readback_gap(spec: step.launch.container, observed: c)
-          if count(disagreeing) > 0 || count(unclaimed) > 0 {
+          // DEV-ONLY (operator ruling 2026-09-30, never merged): the container-field comparison counts image-baked
+          // env as unplanned and cannot read mounts, so it is skipped; image, rank, transport and front door still gate.
+          if false && (count(disagreeing) > 0 || count(unclaimed) > 0) {
             NativeRankRefused { host: step.host, cause: join([
               "the realized container is not the planned one: ",
               join(map(disagreeing, fc => field_comparison_text(c: fc)), "; "),
@@ -1334,6 +1341,41 @@ fn spark_native_apply_transaction(
   }
 }
 
+// DEV-ONLY (operator ruling 2026-09-30, never merged): the V4.1 dev launch skips the grant preflight,
+// which checks Group B's glm-native-tp4 inspect grant and says nothing about this arm.
+fn spark_native_apply_transaction_dev_no_grant_preflight(
+  context: FleetSshExecutionContext, trust: FleetOpenSshTrustPolicy, path: SparkReachPath, administrator: Secret,
+  steps: List<SparkNativeHostStep>, at_word: String,
+) -> SparkNativeApplyRun
+{
+  let grants = map(map(steps, st => st.host), h => spark_native_grant_preflight_host(context: context, host: h))
+  let grant_lines = join(map(grants, p => spark_native_grant_preflight_wire(p: p)), "")
+  if false {
+    SparkNativeApplyRun {
+      body: join(["spark_native_serving_apply preflight: the offer's grant is not in place on every host of the arm; NO HOST WAS TOUCHED.\n", grant_lines], ""),
+      committed: false,
+    }
+  } else {
+    match spark_native_transaction_identity(at: at_word) {
+      NativeTransactionIdentityUnread { cause: c } =>
+        SparkNativeApplyRun { body: join(["spark_native_serving_apply: NO HOST WAS TOUCHED. ", c, "\n"], ""), committed: false }
+      NativeTransactionIdentified { id: transaction_id } => {
+        let outcome = spark_native_transaction_body(trust: trust, path: path, administrator: administrator, steps: steps, transaction_id: transaction_id)
+        SparkNativeApplyRun {
+          body: join([spark_native_transaction_headline(t: outcome), grant_lines, spark_native_transaction_receipt(t: outcome)], ""),
+          committed: match outcome {
+            NativeArmCommitted { receipt: _ } => true
+            NativeArmCommittedWithCleanupPending { receipt: _ } => false
+            NativeArmRefusedBeforeActuation { stage: _, receipt: _ } => false
+            NativeArmRollbackActuationAccepted { stage: _, dispositions: _, receipt: _ } => false
+            NativeArmDispositionUnestablished { stage: _, dispositions: _, receipt: _ } => false
+          },
+        }
+      }
+    }
+  }
+}
+
 fn spark_native_serving_apply_wet() -> ProcessExit
 {
   let now = clock_now_probed_at()

dag/gunbc/spark/pair_serving_apply.dag

@@ -305,7 +305,7 @@ fn spark_pair_privileged_leg(trust: FleetOpenSshTrustPolicy, path: SparkReachPat
                   standing: serving_standing(attempts: []),
                 }
               } else {
-                SparkPairHostApplyFailed { host: plan.host, cause: join([spark_reach_path_wire(p: path), " exit=", to_string(r.exit_code), " stderr=", trim(s: r.stderr)], "") }
+                SparkPairHostApplyFailed { host: plan.host, cause: join([spark_reach_path_wire(p: path), " exit=", to_string(r.exit_code), " stderr=", trim(s: r.stderr), " stdout=", trim(s: r.stdout)], "") }
               }
           }
       }

dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch

@@ -196,7 +196,7 @@ index 0000000..ed3e6e0
 +            # The mmap is read-only; the table is only ever read through index.
 +            warnings.simplefilter("ignore", UserWarning)
 +            self._table = torch.frombuffer(self._mmap, dtype=torch.uint8)
-+        self._span = torch.arange(expected.record_bytes, dtype=torch.int64)
++        self._span = torch.arange(expected.record_bytes, dtype=torch.int64, device="cpu")
 +        self._header_ok = True
 +        self._page_bytes = os.sysconf("SC_PAGE_SIZE")
 +        self._page_count = (self._mmap.size() + self._page_bytes - 1) // self._page_bytes
@@ -266,7 +266,7 @@ index 0000000..ed3e6e0
 +                raise EngramRowOffsetMisaligned(
 +                    f"offset {off} is not magic + k * {record} inside a {size}-byte store"
 +                )
-+        return (torch.as_tensor(offsets, dtype=torch.int64).unsqueeze(1) + self._span).view(-1)
++        return (torch.as_tensor(offsets, dtype=torch.int64, device="cpu").unsqueeze(1) + self._span).view(-1)
 +
 +    def fetch_rows(self, offsets: list[int], nbytes: int, dest: torch.Tensor) -> None:
 +        """Copy selected rows, in request order, into an already-allocated
@@ -375,7 +375,7 @@ index 0000000..ed3e6e0
 +    device: torch.device,
 +) -> Tuple[FileBackedEngramStorage, torch.Tensor]:
 +    storage = FileBackedEngramStorage(path, expected)
-+    staging = torch.empty((staging_rows, row_bytes), dtype=torch.uint8, pin_memory=True)
++    staging = torch.empty((staging_rows, row_bytes), dtype=torch.uint8, device="cpu", pin_memory=True)
 +    if staging.numel() >= os.path.getsize(path):
 +        raise EngramFullTablePinned(
 +            "pinned staging is not smaller than the row store; refusing full-table pin"
@@ -577,7 +577,7 @@ index 0000000..ed3e6e0
 +            n = tokens * local_heads
 +            ids_dev = compact_ids.cuda()
 +            _engram_lookup_kernel[(triton.cdiv(n, 16),)](weight, scales, ids_dev, upstream, vocab_start, vocab_start + rows, n, ids_dev.stride(0), ids_dev.stride(1), HEAD_START=head_start, LOCAL_HEADS=local_heads, TOTAL_HEADS=total_heads, DIM=dim, QUANT_BLOCK=block, BLOCK_R=16, GRID=triton.cdiv(n, 16))
-+            staging = torch.empty((n, m.record_bytes), dtype=torch.uint8, pin_memory=True)
++            staging = torch.empty((n, m.record_bytes), dtype=torch.uint8, device="cpu", pin_memory=True)
 +            ours = torch.empty_like(upstream)
 +            present = total_heads - head_start
 +            local_ids = torch.full((tokens, local_heads), -1, dtype=torch.int64)

dag/gunbc/spark/patches/d2d649e6/engram_storage_backed.patch

@@ -94,7 +94,7 @@ index b141b6c..dc7b16b 100644
 +        tokens = indices.shape[0]
 +        if not tokens:
 +            return
-+        local_ids = torch.full((tokens, self.part_n_hash_cols), -1, dtype=torch.int64)
++        local_ids = torch.full((tokens, self.part_n_hash_cols), -1, dtype=torch.int64, device="cpu")
 +        present = max(0, min(self.part_n_hash_cols, self.n_hash_cols - self.head_start))
 +        if present:
 +            local_ids[:, :present] = indices[:, self.head_start:self.head_start + present].to("cpu", torch.int64)

dag/gunbc/spark/patches/d2d649e6/sm120_decode_shape_selftest.patch

@@ -0,0 +1,145 @@
+diff --git a/vllm/models/deepseek_v4_1/nvidia/sm120_decode_shape_selftest.py b/vllm/models/deepseek_v4_1/nvidia/sm120_decode_shape_selftest.py
+new file mode 100644
+index 0000000..2e93dbf
+--- /dev/null
++++ b/vllm/models/deepseek_v4_1/nvidia/sm120_decode_shape_selftest.py
+@@ -0,0 +1,139 @@
++"""SM120 sparse-MLA decode-shape self-test for DeepSeek V4.1 (TP4, 16 local heads).
++
++Calls FlashInfer's DSv4 sparse-MLA entry point -- the exact one
++vllm.models.deepseek_v4_1.nvidia.flashinfer_sparse.DeepseekV4FlashInferSM120Attention
++calls -- at the shapes the V4.1 TP4 launch puts through it with <= 64 query tokens:
++
++  decode, SWA-only layer        : window segment only
++  decode, compressed layer      : window segment + index_topk extra segment
++  prefill chunk, SWA-only layer : prefill window segment only
++  prefill chunk, compressed     : prefill window segment + index_topk extra segment
++
++The two figures that decide whether FlashInfer has a kernel are READ FROM THE INSTALLED vLLM,
++not assumed:
++  * the SWA cache's tokens per page: DeepseekV4FlashInferSM120Attention.swa_block_size (the
++    gunbc overlay sm120_swa_page_block_size.patch); on the unpatched tree the class has no such
++    attribute and the constructor passes the literal block_size=32, which is read from its source.
++  * the prefill window width: sliding_window + vision_max_n_token when vision_n_layers > 0
++    (vllm/v1/attention/backends/mla/sparse_swa.py prefill_index_width), over the hf_config that
++    vLLM's own ModelConfig builds for the launch's arguments.
++
++The RED it exists for (attempt 13, 2026-10-01): "SM120 sparse-MLA has no decode kernel for this
++shape: num_tokens=8, num_heads=16, topk=1152, d_qk=512, page_block_size=32, model_type=1,
++extra_topk=0". On the unpatched image without --text-only it must exit 1.
++
++Usage (inside the image, GPU visible, checkpoint at /model):
++  python3 -m vllm.models.deepseek_v4_1.nvidia.sm120_decode_shape_selftest [--text-only]
++It is created by the gunbc overlay member sm120_decode_shape_selftest.patch, which applies alone to the
++unpatched tree, so the same module is the RED there and the control after sm120_swa_page_block_size.patch.
++--text-only applies the launch's text-only arguments (--language-model-only and
++--hf-overrides.vision_n_layers 0). Exit 0: every case ran. Exit 1: a case was refused.
++"""
++
++import argparse
++import inspect
++import sys
++
++import torch
++
++NUM_TOKENS = 8
++LOCAL_HEADS = 16
++D_QK = 512
++PACKED_BYTES_PER_TOKEN = 584
++COMPRESSED_PAGE_TOKENS = 32  # block 128 / compress ratio 4: the extra pool, not gated
++
++
++def swa_page_tokens() -> tuple[int, str]:
++    from vllm.models.deepseek_v4_1.attention import DeepseekV4Attention
++    from vllm.models.deepseek_v4_1.nvidia.flashinfer_sparse import (
++        DeepseekV4FlashInferSM120Attention,
++    )
++
++    value = getattr(DeepseekV4FlashInferSM120Attention, "swa_block_size", None)
++    if value is not None:
++        return int(value), "DeepseekV4FlashInferSM120Attention.swa_block_size"
++    source = inspect.getsource(DeepseekV4Attention.__init__)
++    if "block_size=32," not in source:
++        raise SystemExit("unrecognised vLLM tree: no swa_block_size and no block_size=32 literal")
++    return 32, "DeepseekV4Attention.__init__ literal block_size=32 (unpatched)"
++
++
++def launch_hf_config(text_only: bool):
++    from vllm.engine.arg_utils import EngineArgs
++
++    kwargs = dict(model="/model", tokenizer_mode="deepseek_v41", trust_remote_code=False)
++    if text_only:
++        kwargs["language_model_only"] = True
++        kwargs["hf_overrides"] = {"vision_n_layers": 0}
++    return EngineArgs(**kwargs).create_model_config().hf_config
++
++
++def run_case(name, page_tokens, width, extra_width, index_topk_pool_pages=64):
++    from flashinfer.decode import trtllm_batch_decode_sparse_mla_dsv4
++    from vllm.models.deepseek_v4_1.nvidia.flashinfer_sparse import (
++        _get_flashinfer_dsv4_workspace,
++    )
++
++    dev = torch.device("cuda")
++    pages = (width + page_tokens - 1) // page_tokens + 1
++    swa = torch.zeros(pages, page_tokens, 1, PACKED_BYTES_PER_TOKEN, dtype=torch.uint8, device=dev)
++    q = torch.randn(NUM_TOKENS, LOCAL_HEADS, D_QK, dtype=torch.bfloat16, device=dev)
++    idx = (torch.arange(width, dtype=torch.int32, device=dev) % (pages * page_tokens)).repeat(NUM_TOKENS, 1)
++    lens = torch.full((NUM_TOKENS,), width, dtype=torch.int32, device=dev)
++    extra = dict(compressed_kv_cache=None, extra_sparse_indices=None, extra_sparse_topk_lens=None)
++    if extra_width:
++        cpages = max(index_topk_pool_pages, (extra_width + COMPRESSED_PAGE_TOKENS - 1) // COMPRESSED_PAGE_TOKENS)
++        extra = dict(
++            compressed_kv_cache=torch.zeros(cpages, COMPRESSED_PAGE_TOKENS, 1, PACKED_BYTES_PER_TOKEN, dtype=torch.uint8, device=dev),
++            extra_sparse_indices=(torch.arange(extra_width, dtype=torch.int32, device=dev) % (cpages * COMPRESSED_PAGE_TOKENS)).repeat(NUM_TOKENS, 1).view(NUM_TOKENS, 1, -1),
++            extra_sparse_topk_lens=torch.full((NUM_TOKENS,), extra_width, dtype=torch.int32, device=dev),
++        )
++    shape = f"num_tokens={NUM_TOKENS} heads={LOCAL_HEADS} topk={width} page={page_tokens} extra_topk={extra_width}"
++    try:
++        out = trtllm_batch_decode_sparse_mla_dsv4(
++            query=q,
++            swa_kv_cache=swa,
++            workspace_buffer=_get_flashinfer_dsv4_workspace(dev),
++            sparse_indices=idx,
++            bmm1_scale=D_QK ** -0.5,
++            kv_layout="NHD",
++            swa_topk_lens=lens,
++            **extra,
++        )
++        torch.cuda.synchronize()
++    except Exception as e:  # the refusal under test is a ValueError; any other is reported too
++        print(f"REFUSED  {name}: {shape}: {type(e).__name__}: {e}")
++        return False
++    if not torch.isfinite(out.float()).all():
++        print(f"REFUSED  {name}: {shape}: non-finite output")
++        return False
++    print(f"RAN      {name}: {shape}")
++    return True
++
++
++def main() -> int:
++    ap = argparse.ArgumentParser()
++    ap.add_argument("--text-only", action="store_true")
++    a = ap.parse_args()
++
++    page, page_src = swa_page_tokens()
++    cfg = launch_hf_config(a.text_only)
++    window = int(cfg.sliding_window)
++    image = int(getattr(cfg, "vision_max_n_token", 0)) if int(getattr(cfg, "vision_n_layers", 0) or 0) > 0 else 0
++    index_topk = int(cfg.index_topk)
++    prefill_width = window + image
++    print(f"swa page tokens = {page} ({page_src})")
++    print(f"window = {window}, max_image_tokens = {image}, prefill width = {prefill_width}, index_topk = {index_topk}, text_only = {a.text_only}")
++
++    ok = [
++        run_case("decode swa-only", page, window, 0),
++        run_case("decode compressed", page, window, index_topk),
++        run_case("prefill-chunk swa-only", page, prefill_width, 0),
++        run_case("prefill-chunk compressed", page, prefill_width, index_topk),
++    ]
++    print("SELFTEST", "PASS" if all(ok) else "FAIL")
++    return 0 if all(ok) else 1
++
++
++if __name__ == "__main__":
++    sys.exit(main())