fix(ds4): preserve exact speculative verification

server/docs/DS4.md

@@ -183,13 +183,13 @@ export DFLASH_DS4_SPEC_Q=4
   --ds4-expert-top-k 4
 ```
 
-`DFLASH_DS4_FUSED_VERIFY=1` is the opt-in throughput profile. Its persistent
-whole-model GPU graph uses stable padded reduction shapes, so near-tied greedy
-logits can select a different token than the normal causal verifier even at
-temperature 0. Leave it unset when comparing against the normal verifier, or
-set `DFLASH_DS4_SEQ_VERIFY=1` for the slower token-at-a-time verification
-diagnostic. Neither fused verification nor the separate
-`--ds4-expert-top-k 4` approximation should be presented as byte-identical AR.
+`DFLASH_DS4_FUSED_VERIFY=1` requests the fused throughput profile, but falls
+back to the normal verifier because the persistent whole-model GPU graph can
+select a different greedy token at near-tied logits. The approximate graph is
+available only for explicit research runs by also setting
+`DFLASH_DS4_ALLOW_APPROX_FUSED_VERIFY=1`; it must not be presented as
+byte-identical. `DFLASH_DS4_SEQ_VERIFY=1` remains the slower token-at-a-time
+diagnostic. The separate `--ds4-expert-top-k 4` policy is also approximate.
 
 DSpark currently requires monolithic target placement. On HIP,
 `--ds4-fused-decode` selects that placement; if the target falls back to hybrid
@@ -216,6 +216,8 @@ confidence of the proposed prefix. It adds the projection to the same fused
 Markov graph and reads its scores in the existing token-id synchronization; no
 additional host round trip is introduced. Artifacts without a compatible
 confidence head transparently retain the existing acceptance-EWMA policy.
+Set `DFLASH_DS4_ADAPTIVE_WIDTH=0` to hold the verify width fixed for parity
+tests; the environment value takes precedence over the legacy `/tmp` control.
 
 On the gfx1151 validation host, confidence-adaptive width retained 10/10
 GSM+Math accuracy and measured 31.94 tok/s weighted, within 0.6% of fixed q=4

server/scripts/test_ds4_fused_verify_parity.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Model-backed token parity regression for the public DS4 fused-verify flag."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import tempfile
+import time
+import urllib.request
+from pathlib import Path
+
+
+TOKEN_RE = re.compile(r"\[ds4-parity-tokens\] n=\d+ ids=\[([0-9 ]*)\]")
+
+
+def wait_ready(port: int, proc: subprocess.Popen[bytes], timeout: float) -> None:
+    deadline = time.monotonic() + timeout
+    url = f"http://127.0.0.1:{port}/health"
+    while time.monotonic() < deadline:
+        if proc.poll() is not None:
+            raise RuntimeError(f"server exited before readiness: {proc.returncode}")
+        try:
+            with urllib.request.urlopen(url, timeout=2) as response:
+                if response.status == 200:
+                    return
+        except OSError:
+            time.sleep(1)
+    raise TimeoutError(f"server did not become ready within {timeout:.0f}s")
+
+
+def stop_server(proc: subprocess.Popen[bytes]) -> None:
+    if proc.poll() is not None:
+        return
+    proc.terminate()
+    try:
+        proc.wait(timeout=30)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=10)
+
+
+def run_case(args: argparse.Namespace, fused_requested: bool, log_path: Path) -> list[int]:
+    env = os.environ.copy()
+    for name in (
+        "DFLASH_DS4_FUSED_VERIFY",
+        "DFLASH_DS4_ALLOW_APPROX_FUSED_VERIFY",
+        "DFLASH_DS4_SEQ_VERIFY",
+        "DFLASH_DS4_DSPARK_DEBUG",
+    ):
+        env.pop(name, None)
+    env.update({
+        "DFLASH_DS4_SPEC": "1",
+        "DFLASH_DS4_DRAFT": str(args.draft),
+        "DFLASH_DS4_SPEC_Q": "4",
+        "DFLASH_DS4_ADAPTIVE_WIDTH": "0",
+        "DFLASH_DS4_PARITY_TRACE": "1",
+        "LUCE_MMVQ_MAX_NCOLS": "4",
+    })
+    if fused_requested:
+        env["DFLASH_DS4_FUSED_VERIFY"] = "1"
+
+    cmd = [
+        str(args.server_bin), str(args.target),
+        "--host", "127.0.0.1", "--port", str(args.port),
+        "--max-ctx", str(args.max_ctx), "--chunk", "512",
+        "--target-device", "hip:0", "--ds4-fused-decode",
+    ]
+    with log_path.open("wb") as log:
+        proc = subprocess.Popen(cmd, env=env, stdout=log, stderr=subprocess.STDOUT)
+        try:
+            wait_ready(args.port, proc, args.startup_timeout)
+            body = json.dumps({
+                "model": "dflash",
+                "messages": [{"role": "user", "content": args.prompt}],
+                "temperature": 0,
+                "seed": 1234,
+                "max_tokens": args.max_tokens,
+                "stream": False,
+            }).encode()
+            request = urllib.request.Request(
+                f"http://127.0.0.1:{args.port}/v1/chat/completions",
+                data=body,
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(request, timeout=args.request_timeout) as response:
+                if response.status != 200:
+                    raise RuntimeError(f"generation returned HTTP {response.status}")
+                json.load(response)
+        finally:
+            stop_server(proc)
+
+    matches = TOKEN_RE.findall(log_path.read_text(errors="replace"))
+    if not matches:
+        raise RuntimeError(f"token trace missing from {log_path}")
+    return [int(token) for token in matches[-1].split()]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--server-bin", required=True, type=Path)
+    parser.add_argument("--target", required=True, type=Path)
+    parser.add_argument("--draft", required=True, type=Path)
+    parser.add_argument("--port", type=int, default=18084)
+    parser.add_argument("--max-ctx", type=int, default=4096)
+    parser.add_argument("--max-tokens", type=int, default=32)
+    parser.add_argument("--startup-timeout", type=float, default=180)
+    parser.add_argument("--request-timeout", type=float, default=600)
+    parser.add_argument(
+        "--prompt",
+        default="Explain why a bicycle stays upright while moving.",
+    )
+    args = parser.parse_args()
+    for path in (args.server_bin, args.target, args.draft):
+        if not path.is_file():
+            parser.error(f"file not found: {path}")
+
+    with tempfile.TemporaryDirectory(prefix="ds4-fused-parity-") as tmp:
+        tmp_path = Path(tmp)
+        normal = run_case(args, False, tmp_path / "normal.log")
+        fused_flag = run_case(args, True, tmp_path / "fused-flag.log")
+
+    if normal != fused_flag:
+        limit = min(len(normal), len(fused_flag))
+        first = next((i for i in range(limit) if normal[i] != fused_flag[i]), limit)
+        print(f"FAIL: first token mismatch at {first}: normal={normal[first:first+4]} "
+              f"fused_flag={fused_flag[first:first+4]}")
+        return 1
+    print(f"PASS: {len(normal)} generated token IDs are identical")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server/src/deepseek4/deepseek4_backend.cpp

@@ -1040,6 +1040,13 @@ GenerateResult DeepSeek4Backend::generate_impl(const GenerateRequest & req,
         }
         result.succeed();
         result.tokens = std::move(gen);
+        if (env_flag_enabled("DFLASH_DS4_PARITY_TRACE")) {
+            std::fprintf(stderr, "[ds4-parity-tokens] n=%zu ids=[", result.tokens.size());
+            for (size_t i = 0; i < result.tokens.size(); ++i) {
+                std::fprintf(stderr, "%s%d", i == 0 ? "" : " ", result.tokens[i]);
+            }
+            std::fprintf(stderr, "]\n");
+        }
         result.decode_s = elapsed_s(t1);
         result.accept_rate = accept_rate;
         result.spec_decode_ran = spec_ran;

server/src/deepseek4/deepseek4_dspark_spec.cpp

@@ -423,13 +423,17 @@ bool run_deepseek4_dspark_spec_decode(
 
     const bool debug = spec_env_flag("DFLASH_DS4_DSPARK_DEBUG");
     const bool timing = spec_env_flag("DFLASH_DS4_TIMING");
+    const bool parity_trace = spec_env_flag("DFLASH_DS4_PARITY_TRACE");
     const bool full_snap = spec_env_flag("DFLASH_DS4_FULL_SNAP");
     const bool seq_verify_mode = spec_env_flag("DFLASH_DS4_SEQ_VERIFY");
     // Laguna-style adaptive verify width: EWMA of accepted candidates, width =
     // ewma + 2 (avg_commit << block means the wide tail is usually wasted).
     // /tmp/ds4_awidth: 1 = on, 0 = off (default on).
     bool adaptive_width = true;
-    if (std::FILE * f = std::fopen("/tmp/ds4_awidth", "r")) {
+    const char * adaptive_env = std::getenv("DFLASH_DS4_ADAPTIVE_WIDTH");
+    if (adaptive_env && *adaptive_env) {
+        adaptive_width = *adaptive_env != '0';
+    } else if (std::FILE * f = std::fopen("/tmp/ds4_awidth", "r")) {
         int v = 1;
         if (std::fscanf(f, "%d", &v) == 1) adaptive_width = (v != 0);
         std::fclose(f);
@@ -477,6 +481,7 @@ bool run_deepseek4_dspark_spec_decode(
 
     DeepSeek4DFlashTarget target(target_w, target_cache, backend, snap_backend,
                                  drafter.capture_layer_ids, drafter.mask_token_id);
+    target.set_keep_logits(parity_trace);
     DraftWeights dw = make_dspark_shim(drafter);
     DeepSeek4SpecRollback rollback;
     DeepSeek4StepTelemetry tel{};
@@ -677,6 +682,31 @@ bool run_deepseek4_dspark_spec_decode(
         }
         tm_verify += spec_ms_since(t0);
 
+        if (parity_trace) {
+            std::vector<float> trace_logits;
+            if (target.read_verify_logits(q, trace_logits)) {
+                for (int row = 0; row < q; ++row) {
+                    const float * logits = trace_logits.data() + (size_t) row * target_w.n_vocab;
+                    int best = 0;
+                    int second = -1;
+                    for (int tok = 1; tok < target_w.n_vocab; ++tok) {
+                        if (logits[tok] > logits[best]) {
+                            second = best;
+                            best = tok;
+                        } else if (second < 0 || logits[tok] > logits[second]) {
+                            second = tok;
+                        }
+                    }
+                    const float second_logit = second >= 0 ? logits[second] : logits[best];
+                    std::fprintf(stderr,
+                        "[ds4-parity-logits] step=%ld pos=%d row=%d best=%d val=%.9g "
+                        "second=%d val2=%.9g margin=%.9g\n",
+                        steps, pos, row, best, logits[best], second, second_logit,
+                        logits[best] - second_logit);
+                }
+            }
+        }
+
         // Accept the longest matching prefix. accept counts the seed (slot 0)
         // plus each candidate the target agrees with.
         int accept = 1;

server/src/deepseek4/deepseek4_fused_verify.inc

@@ -14,7 +14,15 @@ static void ds4_fv_set(ggml_tensor * t, const void * data, size_t nbytes) {
 static bool ds4_fused_verify_enabled() {
     static int enabled = -1;
     if (enabled < 0) {
-        enabled = ds4_env_flag("DFLASH_DS4_FUSED_VERIFY") ? 1 : 0;
+        const bool requested = ds4_env_flag("DFLASH_DS4_FUSED_VERIFY");
+        const bool allow_approx = ds4_env_flag("DFLASH_DS4_ALLOW_APPROX_FUSED_VERIFY");
+        enabled = requested && allow_approx ? 1 : 0;
+        if (requested && !allow_approx) {
+            std::fprintf(stderr,
+                "[ds4-fused-verify] exact token parity is not guaranteed; "
+                "using the normal verifier. Set DFLASH_DS4_ALLOW_APPROX_FUSED_VERIFY=1 "
+                "only for explicit approximate experiments.\n");
+        }
     }
     return enabled == 1;
 }

server/src/deepseek4/deepseek4_graph.cpp

@@ -6529,7 +6529,8 @@ bool deepseek4_step_layer_range(
             ggml_context * ctx = ggml_init(params);
             if (!ctx) return false;
 
-            const bool last_only = n_tokens > 1;
+            const bool need_all_logits = verify_hooks && verify_hooks->all_logits_out;
+            const bool last_only = n_tokens > 1 && !need_all_logits;
             const int output_tokens = last_only ? 1 : n_tokens;
             ggml_tensor * inp = ggml_new_tensor_2d(
                 ctx, GGML_TYPE_F32, n_embd, output_tokens);