From 82a337769f227d4b904d0d0592e64589f0fb509f Mon Sep 17 00:00:00 2001 From: cheese-cakee Date: Wed, 5 Aug 2026 02:12:41 +0530 Subject: [PATCH 1/4] test(ds4): preserve reference-exact verification --- server/CMakeLists.txt | 10 + server/docs/DS4.md | 15 + server/docs/ENVIRONMENT.md | 1 + .../scripts/test_ds4_exact_verify_parity.py | 357 ++++++++++++++++++ server/src/deepseek4/deepseek4_backend.cpp | 23 ++ .../src/deepseek4/deepseek4_dspark_spec.cpp | 5 +- server/tests/test_ds4_exact_verify_parity.py | 160 ++++++++ 7 files changed, 569 insertions(+), 2 deletions(-) create mode 100644 server/scripts/test_ds4_exact_verify_parity.py create mode 100644 server/tests/test_ds4_exact_verify_parity.py diff --git a/server/CMakeLists.txt b/server/CMakeLists.txt index c295aed46..9ea3d5afd 100644 --- a/server/CMakeLists.txt +++ b/server/CMakeLists.txt @@ -1185,6 +1185,16 @@ if(DFLASH27B_TESTS) endif() # ─── Unit tests (no GPU, no model files) ──────────────────────────── + find_package(Python3 COMPONENTS Interpreter QUIET) + if(Python3_Interpreter_FOUND) + add_test( + NAME ds4_exact_verify_parser + COMMAND "${Python3_EXECUTABLE}" -m unittest discover + -s "${CMAKE_CURRENT_SOURCE_DIR}/tests" + -p test_ds4_exact_verify_parity.py + -v) + endif() + # The production CUDA and ROCm images share this entrypoint. Keep native # server defaults intact unless an operator explicitly supplies an env # override; otherwise container launches can silently disable features diff --git a/server/docs/DS4.md b/server/docs/DS4.md index 6ce6cdb98..131d10f7f 100644 --- a/server/docs/DS4.md +++ b/server/docs/DS4.md @@ -316,6 +316,21 @@ verification with full rollback snapshots for byte-identity checks. Neither fused verification nor the separate `--ds4-expert-top-k 4` approximation should be presented as byte-identical AR. +For a model-backed reference-exact gate, run +`server/scripts/test_ds4_exact_verify_parity.py`. It starts matched greedy AR +and reference-exact servers, compares every generated token ID, and fails if +the reference run did not execute speculative work with at least one rejected +draft prefix and full-snapshot rollback. The gate requires SHA-256 identities +for the server binary, target GGUF, DSpark GGUF, and prompt file. +Use the same `hip:0` device, exact prefill, q=4, prompt, context, seed, and token +limit for both arms. Pass a new `--log-dir` for each run; the gate retains both +server logs and a JSON manifest and refuses to overwrite existing evidence. +`DFLASH_DS4_EXACT_VERIFY_TRACE` is a default-off diagnostic used by that gate +to emit the complete generated-token trace; it does not make the fused +throughput profile exact. The two arms share a sanitized DS4 execution-policy +environment. Only the reference arm enables DSpark, selects its draft, and +enables reference-exact verification. + DSpark can verify against in-process heterogeneous expert placement. The drafter remains local to its selected GPU backend; a failed draft load is reported and falls back to normal autoregressive decode. The target cache and diff --git a/server/docs/ENVIRONMENT.md b/server/docs/ENVIRONMENT.md index a877c3812..e1f577735 100644 --- a/server/docs/ENVIRONMENT.md +++ b/server/docs/ENVIRONMENT.md @@ -98,6 +98,7 @@ consolidation of this list into CLI flags is tracked as follow-up work. - `DFLASH_DS4_DRAFT_BACKEND` - deepseek4_backend.cpp - `DFLASH_DS4_DRAFT_GPU` - deepseek4_backend.cpp - `DFLASH_DS4_DSPARK_DEBUG` - deepseek4_graph.cpp +- `DFLASH_DS4_EXACT_VERIFY_TRACE` - deepseek4_backend.cpp - `DFLASH_DS4_FUSED_VERIFY` - deepseek4_dspark_spec.cpp, deepseek4_loader.cpp - `DFLASH_DS4_HOTNESS_CSV` - deepseek4_backend.cpp - `DFLASH_DS4_MOE_TP` - deepseek4_backend.cpp diff --git a/server/scripts/test_ds4_exact_verify_parity.py b/server/scripts/test_ds4_exact_verify_parity.py new file mode 100644 index 000000000..401b09feb --- /dev/null +++ b/server/scripts/test_ds4_exact_verify_parity.py @@ -0,0 +1,357 @@ +#!/usr/bin/env python3 +"""Model-backed token parity gate for DS4 reference-exact verification.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import time +import urllib.request +from pathlib import Path +from typing import NamedTuple + +TRACE_RE = re.compile( + r"^\[ds4-exact-verify-trace\] reference_exact=([01]) speculation=([01]) " + r"n=(\d+) ids=\[([0-9 ]*)\]$", + re.MULTILINE, +) +SPEC_SUMMARY_RE = re.compile( + r"^\[ds4-spec\] gen=(\d+) steps=(\d+) matched=(\d+) offered=(\d+) " + r".*\bfull_snap=([01])$", + re.MULTILINE, +) + +SANITIZED_POLICY_PREFIXES = ( + "DFLASH_DS4_", + "DFLASH_EXPERT_", + "DFLASH_MOE_", +) +SANITIZED_POLICY_NAMES = { + "DFLASH_MMQ_SUB_BATCH", + "GGML_BATCH_PEER_COPIES", + "GGML_CUDA_BATCH_PEER_COPIES", + "LUCE_MMVQ_MAX_NCOLS", +} + + +class TokenTrace(NamedTuple): + reference_exact: bool + speculation: bool + tokens: tuple[int, ...] + + +class SpeculationSummary(NamedTuple): + generated: int + steps: int + matched: int + offered: int + full_snapshot: bool + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def require_sha256(path: Path, expected: str) -> None: + actual = sha256_file(path) + if actual.lower() != expected.lower(): + raise RuntimeError(f"SHA-256 mismatch for {path}: expected {expected}, got {actual}") + + +def parse_token_trace( + log: str, + *, + expected_reference_exact: bool, + expected_speculation: bool, +) -> TokenTrace: + matches = list(TRACE_RE.finditer(log)) + if len(matches) != 1: + raise RuntimeError(f"expected exactly one token trace, found {len(matches)}") + match = matches[0] + declared_count = int(match.group(3)) + raw_ids = match.group(4) + tokens = tuple(int(token) for token in raw_ids.split()) if raw_ids else () + if declared_count != len(tokens): + raise RuntimeError(f"token trace declared {declared_count} IDs but contained {len(tokens)}") + if not tokens: + raise RuntimeError("token trace is empty") + trace = TokenTrace( + reference_exact=match.group(1) == "1", + speculation=match.group(2) == "1", + tokens=tokens, + ) + if trace.reference_exact != expected_reference_exact: + raise RuntimeError( + "reference-exact mode mismatch: " + f"expected {int(expected_reference_exact)}, got {int(trace.reference_exact)}" + ) + if trace.speculation != expected_speculation: + raise RuntimeError( + "speculation mode mismatch: " + f"expected {int(expected_speculation)}, got {int(trace.speculation)}" + ) + return trace + + +def require_speculation_work(log: str) -> SpeculationSummary: + matches = list(SPEC_SUMMARY_RE.finditer(log)) + if len(matches) != 1: + raise RuntimeError(f"expected exactly one DS4 speculation summary, found {len(matches)}") + match = matches[0] + summary = SpeculationSummary( + generated=int(match.group(1)), + steps=int(match.group(2)), + matched=int(match.group(3)), + offered=int(match.group(4)), + full_snapshot=match.group(5) == "1", + ) + if summary.generated <= 0 or summary.steps <= 0 or summary.offered <= 0: + raise RuntimeError( + "DS4 speculation did no work: " + f"generated={summary.generated}, steps={summary.steps}, " + f"offered={summary.offered}" + ) + if summary.matched >= summary.offered: + raise RuntimeError( + "reference-exact run did not exercise rejection rollback: " + f"matched={summary.matched}, offered={summary.offered}" + ) + if not summary.full_snapshot: + raise RuntimeError("reference-exact run did not enable full rollback snapshots") + if "[ds4-spec] reference-exact verifier:" not in log: + raise RuntimeError("reference-exact verifier activation banner is missing") + return summary + + +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 case_environment(args: argparse.Namespace, reference_exact: bool) -> dict[str, str]: + env = os.environ.copy() + for name in tuple(env): + if name in SANITIZED_POLICY_NAMES or name.startswith(SANITIZED_POLICY_PREFIXES): + env.pop(name) + env.update( + { + "DFLASH_DS4_ADAPTIVE_WIDTH": "0", + "DFLASH_DS4_EXACT_VERIFY_TRACE": "1", + "DFLASH_DS4_SPEC_Q": str(args.spec_q), + "LUCE_MMVQ_MAX_NCOLS": str(args.mmvq_max_ncols), + } + ) + if reference_exact: + env.update( + { + "DFLASH_DS4_DRAFT": str(args.draft), + "DFLASH_DS4_SPEC": "1", + "DFLASH_DS4_SPEC_REFERENCE_EXACT": "1", + } + ) + return env + + +def run_case( + args: argparse.Namespace, + *, + reference_exact: bool, + prompt: str, + log_path: Path, +) -> TokenTrace: + command = [ + str(args.server_bin), + str(args.target), + "--host", + "127.0.0.1", + "--port", + str(args.port), + "--max-ctx", + str(args.max_ctx), + "--chunk", + str(args.prefill_chunk), + "--target-device", + args.target_device, + "--ds4-fused-decode", + "--ds4-prefill", + "exact", + ] + with log_path.open("wb") as log: + proc = subprocess.Popen( + command, + env=case_environment(args, reference_exact), + stdout=log, + stderr=subprocess.STDOUT, + ) + try: + wait_ready(args.port, proc, args.startup_timeout) + body = json.dumps( + { + "model": "dflash", + "messages": [{"role": "user", "content": prompt}], + "temperature": 0, + "seed": args.seed, + "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) + + log_text = log_path.read_text(errors="replace") + trace = parse_token_trace( + log_text, + expected_reference_exact=reference_exact, + expected_speculation=reference_exact, + ) + if reference_exact: + require_speculation_work(log_text) + return trace + + +def build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Compare DS4 reference-exact speculative decode with greedy AR", + ) + parser.add_argument("--server-bin", required=True, type=Path) + parser.add_argument("--server-sha256", required=True) + parser.add_argument("--target", required=True, type=Path) + parser.add_argument("--target-sha256", required=True) + parser.add_argument("--draft", required=True, type=Path) + parser.add_argument("--draft-sha256", required=True) + parser.add_argument("--prompt-file", required=True, type=Path) + parser.add_argument("--prompt-sha256", required=True) + parser.add_argument("--log-dir", required=True, type=Path) + parser.add_argument("--target-device", default="hip:0") + parser.add_argument("--port", type=int, default=18084) + parser.add_argument("--max-ctx", type=int, default=4096) + parser.add_argument("--prefill-chunk", type=int, default=512) + parser.add_argument("--max-tokens", type=int, default=64) + parser.add_argument("--seed", type=int, default=1234) + parser.add_argument("--spec-q", type=int, default=4) + parser.add_argument("--mmvq-max-ncols", type=int, default=4) + parser.add_argument("--startup-timeout", type=float, default=180) + parser.add_argument("--request-timeout", type=float, default=600) + return parser + + +def main() -> int: + parser = build_arg_parser() + args = parser.parse_args() + for path in (args.server_bin, args.target, args.draft, args.prompt_file): + if not path.is_file(): + parser.error(f"file not found: {path}") + if args.max_tokens < 2: + parser.error("--max-tokens must be at least 2 so speculation can run") + if not 2 <= args.spec_q <= 4: + parser.error("--spec-q must be between 2 and 4") + if args.mmvq_max_ncols < args.spec_q: + parser.error("--mmvq-max-ncols must be at least --spec-q") + + for path, expected in ( + (args.server_bin, args.server_sha256), + (args.target, args.target_sha256), + (args.draft, args.draft_sha256), + (args.prompt_file, args.prompt_sha256), + ): + require_sha256(path, expected) + prompt = args.prompt_file.read_text(encoding="utf-8") + if not prompt.strip(): + parser.error("--prompt-file must not be empty") + + args.log_dir.mkdir(parents=True, exist_ok=True) + evidence_paths = [ + args.log_dir / "ar.log", + args.log_dir / "reference-exact.log", + args.log_dir / "manifest.json", + ] + existing = [str(path) for path in evidence_paths if path.exists()] + if existing: + parser.error(f"refusing to overwrite evidence files: {', '.join(existing)}") + ar = run_case( + args, + reference_exact=False, + prompt=prompt, + log_path=evidence_paths[0], + ) + exact = run_case( + args, + reference_exact=True, + prompt=prompt, + log_path=evidence_paths[1], + ) + manifest = { + "server": {"path": str(args.server_bin), "sha256": args.server_sha256.lower()}, + "target": {"path": str(args.target), "sha256": args.target_sha256.lower()}, + "draft": {"path": str(args.draft), "sha256": args.draft_sha256.lower()}, + "prompt": {"path": str(args.prompt_file), "sha256": args.prompt_sha256.lower()}, + "target_device": args.target_device, + "max_ctx": args.max_ctx, + "prefill_chunk": args.prefill_chunk, + "max_tokens": args.max_tokens, + "seed": args.seed, + "spec_q": args.spec_q, + "mmvq_max_ncols": args.mmvq_max_ncols, + "ar_tokens": list(ar.tokens), + "reference_exact_tokens": list(exact.tokens), + } + evidence_paths[2].write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") + + if ar.tokens != exact.tokens: + limit = min(len(ar.tokens), len(exact.tokens)) + first = next( + (index for index in range(limit) if ar.tokens[index] != exact.tokens[index]), + limit, + ) + print( + f"FAIL: first token mismatch at {first}: " + f"ar={ar.tokens[first : first + 4]} exact={exact.tokens[first : first + 4]}" + ) + return 1 + print( + f"PASS: {len(ar.tokens)} generated token IDs are identical; " + "reference-exact speculation executed at least one step" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/src/deepseek4/deepseek4_backend.cpp b/server/src/deepseek4/deepseek4_backend.cpp index e9220d6e2..3ff4ba9ec 100644 --- a/server/src/deepseek4/deepseek4_backend.cpp +++ b/server/src/deepseek4/deepseek4_backend.cpp @@ -1769,6 +1769,19 @@ GenerateResult DeepSeek4Backend::generate_from_state( result.decode_s = elapsed_s(t1); result.accept_rate = accept_rate; result.spec_decode_ran = spec_ran; + if (env_flag_enabled("DFLASH_DS4_EXACT_VERIFY_TRACE")) { + const bool reference_exact = + env_flag_enabled("DFLASH_DS4_SPEC_REFERENCE_EXACT"); + std::fprintf(stderr, + "[ds4-exact-verify-trace] reference_exact=%d speculation=%d " + "n=%zu ids=[", + (int) reference_exact, (int) result.spec_decode_ran, + 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"); + } std::fprintf(stderr, "[deepseek4] DSpark decode: %zu tok in %.3fs (%.1f tok/s) accept_rate=%.2f\n", result.tokens.size(), result.decode_s, result.decode_s > 0 ? result.tokens.size() / result.decode_s : 0.0, accept_rate); @@ -1787,6 +1800,16 @@ GenerateResult DeepSeek4Backend::generate_from_state( result.succeed(); result.tokens = std::move(gen_tokens); + if (env_flag_enabled("DFLASH_DS4_EXACT_VERIFY_TRACE")) { + std::fprintf(stderr, + "[ds4-exact-verify-trace] reference_exact=0 speculation=0 " + "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.budget_forced_close = forced_close; maybe_save_routing_stats(); diff --git a/server/src/deepseek4/deepseek4_dspark_spec.cpp b/server/src/deepseek4/deepseek4_dspark_spec.cpp index 21596aa6c..0c4fc4c27 100644 --- a/server/src/deepseek4/deepseek4_dspark_spec.cpp +++ b/server/src/deepseek4/deepseek4_dspark_spec.cpp @@ -1097,9 +1097,10 @@ bool run_deepseek4_dspark_spec_decode( : 0.0f; } std::fprintf(stderr, - "[ds4-spec] gen=%d steps=%ld mean_accept=%.2f/%.2f " + "[ds4-spec] gen=%d steps=%ld matched=%ld offered=%ld " + "mean_accept=%.2f/%.2f " "q_cap=%d full_snap=%d\n", - n_generated, steps, + n_generated, steps, accept_sum, offered_sum, steps ? (double) accept_sum / steps : 0.0, steps ? (double) offered_sum / steps : 0.0, q_cap, (int) full_snap); diff --git a/server/tests/test_ds4_exact_verify_parity.py b/server/tests/test_ds4_exact_verify_parity.py new file mode 100644 index 000000000..7d7fbbef0 --- /dev/null +++ b/server/tests/test_ds4_exact_verify_parity.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +"""Parser regressions for the DS4 reference-exact model gate.""" + +import importlib.util +import os +import unittest +from argparse import Namespace +from pathlib import Path +from unittest import mock + +SERVER_DIR = Path(__file__).resolve().parents[1] +SCRIPT = SERVER_DIR / "scripts" / "test_ds4_exact_verify_parity.py" +SPEC = importlib.util.spec_from_file_location("test_ds4_exact_verify_parity_script", SCRIPT) +MODULE = importlib.util.module_from_spec(SPEC) +assert SPEC.loader is not None +SPEC.loader.exec_module(MODULE) + + +class TokenTraceParserTest(unittest.TestCase): + def test_parses_exact_trace(self): + trace = MODULE.parse_token_trace( + "[ds4-exact-verify-trace] reference_exact=1 speculation=1 n=3 ids=[8 13 21]\n", + expected_reference_exact=True, + expected_speculation=True, + ) + self.assertEqual(trace.tokens, (8, 13, 21)) + + def test_rejects_declared_count_mismatch(self): + with self.assertRaisesRegex(RuntimeError, "declared 3 IDs but contained 2"): + MODULE.parse_token_trace( + "[ds4-exact-verify-trace] reference_exact=1 speculation=1 n=3 ids=[8 13]\n", + expected_reference_exact=True, + expected_speculation=True, + ) + + def test_rejects_empty_trace(self): + with self.assertRaisesRegex(RuntimeError, "token trace is empty"): + MODULE.parse_token_trace( + "[ds4-exact-verify-trace] reference_exact=0 speculation=0 n=0 ids=[]\n", + expected_reference_exact=False, + expected_speculation=False, + ) + + def test_rejects_non_speculative_exact_run(self): + with self.assertRaisesRegex(RuntimeError, "speculation mode mismatch"): + MODULE.parse_token_trace( + "[ds4-exact-verify-trace] reference_exact=1 speculation=0 n=2 ids=[3 5]\n", + expected_reference_exact=True, + expected_speculation=True, + ) + + def test_requires_positive_speculation_summary_and_exact_banner(self): + summary = MODULE.require_speculation_work( + "[ds4-spec] reference-exact verifier: sequential target replay " + "with full rollback snapshots\n" + "[ds4-spec] gen=9 steps=4 matched=5 offered=12 " + "mean_accept=1.25/3.00 q_cap=4 full_snap=1\n" + ) + self.assertEqual(summary.matched, 5) + self.assertEqual(summary.offered, 12) + self.assertTrue(summary.full_snapshot) + + def test_rejects_zero_speculation_steps(self): + with self.assertRaisesRegex(RuntimeError, "did no work"): + MODULE.require_speculation_work( + "[ds4-spec] reference-exact verifier: sequential target replay " + "with full rollback snapshots\n" + "[ds4-spec] gen=0 steps=0 matched=0 offered=0 " + "mean_accept=0/0 q_cap=4 full_snap=1\n" + ) + + def test_rejects_all_accepted_run_without_rollback(self): + with self.assertRaisesRegex(RuntimeError, "did not exercise rejection rollback"): + MODULE.require_speculation_work( + "[ds4-spec] reference-exact verifier: sequential target replay " + "with full rollback snapshots\n" + "[ds4-spec] gen=12 steps=4 matched=12 offered=12 " + "mean_accept=3.00/3.00 q_cap=4 full_snap=1\n" + ) + + def test_rejects_summary_without_exact_counters(self): + with self.assertRaisesRegex(RuntimeError, "found 0"): + MODULE.require_speculation_work( + "[ds4-spec] reference-exact verifier: sequential target replay " + "with full rollback snapshots\n" + "[ds4-spec] gen=9 steps=4 mean_accept=1.25/3.00 q_cap=4 full_snap=1\n" + ) + + def test_rejects_run_without_full_snapshots(self): + with self.assertRaisesRegex(RuntimeError, "did not enable full rollback snapshots"): + MODULE.require_speculation_work( + "[ds4-spec] reference-exact verifier: sequential target replay " + "with full rollback snapshots\n" + "[ds4-spec] gen=9 steps=4 matched=5 offered=12 " + "mean_accept=1.25/3.00 q_cap=4 full_snap=0\n" + ) + + +class CaseEnvironmentTest(unittest.TestCase): + def setUp(self): + self.args = Namespace( + draft=Path("draft.gguf"), + spec_q=4, + mmvq_max_ncols=4, + ) + + def test_arms_differ_only_by_reference_exact_activation(self): + inherited = { + "PATH": os.environ.get("PATH", ""), + "DFLASH_DS4_FUSED_VERIFY": "1", + "DFLASH_DS4_DRAFT_GPU": "7", + "DFLASH_EXPERT_BUDGET_MB": "1234", + "DFLASH_MOE_TP_BACKEND": "cuda", + "DFLASH_MMQ_SUB_BATCH": "1", + "GGML_BATCH_PEER_COPIES": "1", + "LUCE_MMVQ_MAX_NCOLS": "99", + } + with mock.patch.dict(os.environ, inherited, clear=True): + ar = MODULE.case_environment(self.args, False) + exact = MODULE.case_environment(self.args, True) + + intentional = { + "DFLASH_DS4_DRAFT": "draft.gguf", + "DFLASH_DS4_SPEC": "1", + "DFLASH_DS4_SPEC_REFERENCE_EXACT": "1", + } + self.assertEqual({key: exact[key] for key in intentional}, intentional) + self.assertEqual( + {key: value for key, value in exact.items() if key not in intentional}, + ar, + ) + + def test_common_policy_is_fixed_and_inherited_policy_is_removed(self): + inherited = { + "DFLASH_DS4_ADAPTIVE_WIDTH": "1", + "DFLASH_DS4_FUSED_VERIFY": "1", + "DFLASH_EXPERT_BUDGET_MB": "1234", + "DFLASH_MOE_TP_BACKEND": "cuda", + "DFLASH_MMQ_SUB_BATCH": "1", + "GGML_CUDA_BATCH_PEER_COPIES": "1", + "LUCE_MMVQ_MAX_NCOLS": "99", + } + with mock.patch.dict(os.environ, inherited, clear=True): + ar = MODULE.case_environment(self.args, False) + + self.assertEqual(ar["DFLASH_DS4_ADAPTIVE_WIDTH"], "0") + self.assertEqual(ar["DFLASH_DS4_SPEC_Q"], "4") + self.assertEqual(ar["LUCE_MMVQ_MAX_NCOLS"], "4") + for name in ( + "DFLASH_DS4_FUSED_VERIFY", + "DFLASH_EXPERT_BUDGET_MB", + "DFLASH_MOE_TP_BACKEND", + "DFLASH_MMQ_SUB_BATCH", + "GGML_CUDA_BATCH_PEER_COPIES", + ): + self.assertNotIn(name, ar) + + +if __name__ == "__main__": + unittest.main() From 549d21e5c941c37b7a864af41bec1834d292670f Mon Sep 17 00:00:00 2001 From: cheese-cakee Date: Wed, 5 Aug 2026 03:34:38 +0530 Subject: [PATCH 2/4] fix(ds4): harden exact verification evidence --- server/docs/DS4.md | 6 +- .../scripts/test_ds4_exact_verify_parity.py | 125 ++++++++++++------ server/src/deepseek4/deepseek4_backend.cpp | 44 +++--- server/tests/test_ds4_exact_verify_parity.py | 76 +++++++++++ 4 files changed, 189 insertions(+), 62 deletions(-) diff --git a/server/docs/DS4.md b/server/docs/DS4.md index 131d10f7f..c7797420a 100644 --- a/server/docs/DS4.md +++ b/server/docs/DS4.md @@ -323,8 +323,10 @@ the reference run did not execute speculative work with at least one rejected draft prefix and full-snapshot rollback. The gate requires SHA-256 identities for the server binary, target GGUF, DSpark GGUF, and prompt file. Use the same `hip:0` device, exact prefill, q=4, prompt, context, seed, and token -limit for both arms. Pass a new `--log-dir` for each run; the gate retains both -server logs and a JSON manifest and refuses to overwrite existing evidence. +limit for both arms. A passing run retains both server logs and a JSON manifest +and refuses to overwrite that final evidence. Failed attempts keep their logs +under a unique `failed-*` subdirectory without occupying the final filenames, +so the same `--log-dir` can be rerun after correcting the failure. `DFLASH_DS4_EXACT_VERIFY_TRACE` is a default-off diagnostic used by that gate to emit the complete generated-token trace; it does not make the fused throughput profile exact. The two arms share a sanitized DS4 execution-policy diff --git a/server/scripts/test_ds4_exact_verify_parity.py b/server/scripts/test_ds4_exact_verify_parity.py index 401b09feb..b7f097951 100644 --- a/server/scripts/test_ds4_exact_verify_parity.py +++ b/server/scripts/test_ds4_exact_verify_parity.py @@ -9,6 +9,8 @@ import os import re import subprocess +import sys +import tempfile import time import urllib.request from pathlib import Path @@ -131,6 +133,49 @@ def require_speculation_work(log: str) -> SpeculationSummary: return summary +def token_mismatch_message(ar_tokens: tuple[int, ...], exact_tokens: tuple[int, ...]) -> str | None: + common_length = min(len(ar_tokens), len(exact_tokens)) + first_mismatch = next( + (index for index in range(common_length) if ar_tokens[index] != exact_tokens[index]), + None, + ) + if first_mismatch is not None: + return ( + f"first token mismatch at {first_mismatch}: " + f"ar={ar_tokens[first_mismatch : first_mismatch + 4]} " + f"exact={exact_tokens[first_mismatch : first_mismatch + 4]}" + ) + if len(ar_tokens) != len(exact_tokens): + return ( + f"token trace length mismatch after common prefix of {common_length}: " + f"ar={len(ar_tokens)} exact={len(exact_tokens)}" + ) + return None + + +def retain_failed_attempt(attempt_dir: Path) -> Path: + failed_dir = attempt_dir.with_name(attempt_dir.name.replace("attempt-", "failed-", 1)) + attempt_dir.replace(failed_dir) + return failed_dir + + +def promote_evidence(staged_paths: list[Path], final_paths: list[Path]) -> None: + promoted: list[Path] = [] + try: + for staged, final in zip(staged_paths, final_paths, strict=True): + os.link(staged, final) + promoted.append(final) + except Exception: + for final in reversed(promoted): + final.unlink(missing_ok=True) + raise + for staged in staged_paths: + try: + staged.unlink() + except OSError: + pass + + 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" @@ -306,46 +351,52 @@ def main() -> int: existing = [str(path) for path in evidence_paths if path.exists()] if existing: parser.error(f"refusing to overwrite evidence files: {', '.join(existing)}") - ar = run_case( - args, - reference_exact=False, - prompt=prompt, - log_path=evidence_paths[0], - ) - exact = run_case( - args, - reference_exact=True, - prompt=prompt, - log_path=evidence_paths[1], - ) - manifest = { - "server": {"path": str(args.server_bin), "sha256": args.server_sha256.lower()}, - "target": {"path": str(args.target), "sha256": args.target_sha256.lower()}, - "draft": {"path": str(args.draft), "sha256": args.draft_sha256.lower()}, - "prompt": {"path": str(args.prompt_file), "sha256": args.prompt_sha256.lower()}, - "target_device": args.target_device, - "max_ctx": args.max_ctx, - "prefill_chunk": args.prefill_chunk, - "max_tokens": args.max_tokens, - "seed": args.seed, - "spec_q": args.spec_q, - "mmvq_max_ncols": args.mmvq_max_ncols, - "ar_tokens": list(ar.tokens), - "reference_exact_tokens": list(exact.tokens), - } - evidence_paths[2].write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") - - if ar.tokens != exact.tokens: - limit = min(len(ar.tokens), len(exact.tokens)) - first = next( - (index for index in range(limit) if ar.tokens[index] != exact.tokens[index]), - limit, + attempt_dir = Path(tempfile.mkdtemp(prefix="attempt-", dir=args.log_dir)) + staged_paths = [attempt_dir / path.name for path in evidence_paths] + try: + ar = run_case( + args, + reference_exact=False, + prompt=prompt, + log_path=staged_paths[0], ) - print( - f"FAIL: first token mismatch at {first}: " - f"ar={ar.tokens[first : first + 4]} exact={exact.tokens[first : first + 4]}" + exact = run_case( + args, + reference_exact=True, + prompt=prompt, + log_path=staged_paths[1], ) + manifest = { + "server": {"path": str(args.server_bin), "sha256": args.server_sha256.lower()}, + "target": {"path": str(args.target), "sha256": args.target_sha256.lower()}, + "draft": {"path": str(args.draft), "sha256": args.draft_sha256.lower()}, + "prompt": {"path": str(args.prompt_file), "sha256": args.prompt_sha256.lower()}, + "target_device": args.target_device, + "max_ctx": args.max_ctx, + "prefill_chunk": args.prefill_chunk, + "max_tokens": args.max_tokens, + "seed": args.seed, + "spec_q": args.spec_q, + "mmvq_max_ncols": args.mmvq_max_ncols, + "ar_tokens": list(ar.tokens), + "reference_exact_tokens": list(exact.tokens), + } + staged_paths[2].write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") + + mismatch = token_mismatch_message(ar.tokens, exact.tokens) + if mismatch is not None: + failed_dir = retain_failed_attempt(attempt_dir) + print(f"FAIL: {mismatch}; diagnostics retained in {failed_dir}") + return 1 + promote_evidence(staged_paths, evidence_paths) + except Exception as error: + failed_dir = retain_failed_attempt(attempt_dir) + print(f"FAIL: {error}; diagnostics retained in {failed_dir}", file=sys.stderr) return 1 + try: + attempt_dir.rmdir() + except OSError: + pass print( f"PASS: {len(ar.tokens)} generated token IDs are identical; " "reference-exact speculation executed at least one step" diff --git a/server/src/deepseek4/deepseek4_backend.cpp b/server/src/deepseek4/deepseek4_backend.cpp index 3ff4ba9ec..b5e6d0375 100644 --- a/server/src/deepseek4/deepseek4_backend.cpp +++ b/server/src/deepseek4/deepseek4_backend.cpp @@ -42,6 +42,25 @@ static bool env_flag_enabled(const char * name) { return value && value[0] && std::strcmp(value, "0") != 0; } +static void emit_exact_verify_trace( + bool speculation, + const std::vector & tokens) { + if (!env_flag_enabled("DFLASH_DS4_EXACT_VERIFY_TRACE")) { + return; + } + + const bool reference_exact = + env_flag_enabled("DFLASH_DS4_SPEC_REFERENCE_EXACT"); + std::fprintf(stderr, + "[ds4-exact-verify-trace] reference_exact=%d speculation=%d " + "n=%zu ids=[", + (int) reference_exact, (int) speculation, tokens.size()); + for (size_t i = 0; i < tokens.size(); ++i) { + std::fprintf(stderr, "%s%d", i == 0 ? "" : " ", tokens[i]); + } + std::fprintf(stderr, "]\n"); +} + static void configure_gfx1151_dspark_mmvq_default(int gpu) { #if defined(DFLASH27B_BACKEND_HIP) || defined(GGML_USE_HIP) if (!env_flag_enabled("DFLASH_DS4_SPEC") || @@ -1769,19 +1788,7 @@ GenerateResult DeepSeek4Backend::generate_from_state( result.decode_s = elapsed_s(t1); result.accept_rate = accept_rate; result.spec_decode_ran = spec_ran; - if (env_flag_enabled("DFLASH_DS4_EXACT_VERIFY_TRACE")) { - const bool reference_exact = - env_flag_enabled("DFLASH_DS4_SPEC_REFERENCE_EXACT"); - std::fprintf(stderr, - "[ds4-exact-verify-trace] reference_exact=%d speculation=%d " - "n=%zu ids=[", - (int) reference_exact, (int) result.spec_decode_ran, - 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"); - } + emit_exact_verify_trace(result.spec_decode_ran, result.tokens); std::fprintf(stderr, "[deepseek4] DSpark decode: %zu tok in %.3fs (%.1f tok/s) accept_rate=%.2f\n", result.tokens.size(), result.decode_s, result.decode_s > 0 ? result.tokens.size() / result.decode_s : 0.0, accept_rate); @@ -1800,16 +1807,7 @@ GenerateResult DeepSeek4Backend::generate_from_state( result.succeed(); result.tokens = std::move(gen_tokens); - if (env_flag_enabled("DFLASH_DS4_EXACT_VERIFY_TRACE")) { - std::fprintf(stderr, - "[ds4-exact-verify-trace] reference_exact=0 speculation=0 " - "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"); - } + emit_exact_verify_trace(false, result.tokens); result.decode_s = elapsed_s(t1); result.budget_forced_close = forced_close; maybe_save_routing_stats(); diff --git a/server/tests/test_ds4_exact_verify_parity.py b/server/tests/test_ds4_exact_verify_parity.py index 7d7fbbef0..6c43abcf0 100644 --- a/server/tests/test_ds4_exact_verify_parity.py +++ b/server/tests/test_ds4_exact_verify_parity.py @@ -3,6 +3,7 @@ import importlib.util import os +import tempfile import unittest from argparse import Namespace from pathlib import Path @@ -95,6 +96,81 @@ def test_rejects_run_without_full_snapshots(self): "mean_accept=1.25/3.00 q_cap=4 full_snap=0\n" ) + def test_reports_positional_token_mismatch(self): + self.assertEqual( + MODULE.token_mismatch_message((3, 5, 8), (3, 7, 8)), + "first token mismatch at 1: ar=(5, 8) exact=(7, 8)", + ) + + def test_reports_length_mismatch_after_common_prefix(self): + self.assertEqual( + MODULE.token_mismatch_message((3, 5), (3, 5, 8)), + "token trace length mismatch after common prefix of 2: ar=2 exact=3", + ) + + +class EvidenceLifecycleTest(unittest.TestCase): + def test_failed_attempt_does_not_occupy_final_evidence_paths(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) + attempt_dir = log_dir / "attempt-test" + attempt_dir.mkdir() + (attempt_dir / "ar.log").write_text("diagnostic", encoding="utf-8") + + failed_dir = MODULE.retain_failed_attempt(attempt_dir) + + self.assertEqual(failed_dir.name, "failed-test") + self.assertEqual((failed_dir / "ar.log").read_text(encoding="utf-8"), "diagnostic") + self.assertFalse((log_dir / "ar.log").exists()) + + def test_promotes_complete_evidence_set(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) + attempt_dir = log_dir / "attempt-test" + attempt_dir.mkdir() + staged = [attempt_dir / name for name in ("ar.log", "exact.log", "manifest.json")] + final = [log_dir / path.name for path in staged] + for index, path in enumerate(staged): + path.write_text(str(index), encoding="utf-8") + + MODULE.promote_evidence(staged, final) + + self.assertEqual([path.read_text(encoding="utf-8") for path in final], ["0", "1", "2"]) + + def test_preexisting_final_rolls_back_only_new_links(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) + attempt_dir = log_dir / "attempt-test" + attempt_dir.mkdir() + staged = [attempt_dir / name for name in ("ar.log", "exact.log")] + final = [log_dir / path.name for path in staged] + for path in staged: + path.write_text("diagnostic", encoding="utf-8") + final[1].write_text("existing evidence", encoding="utf-8") + + with self.assertRaises(FileExistsError): + MODULE.promote_evidence(staged, final) + + self.assertTrue(all(path.exists() for path in staged)) + self.assertFalse(final[0].exists()) + self.assertEqual(final[1].read_text(encoding="utf-8"), "existing evidence") + + def test_denied_staged_cleanup_does_not_fail_completed_promotion(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) + attempt_dir = log_dir / "attempt-test" + attempt_dir.mkdir() + staged = [attempt_dir / name for name in ("ar.log", "exact.log")] + final = [log_dir / path.name for path in staged] + for path in staged: + path.write_text("evidence", encoding="utf-8") + + with mock.patch.object(MODULE.Path, "unlink", side_effect=PermissionError): + MODULE.promote_evidence(staged, final) + + self.assertTrue(all(path.exists() for path in final)) + self.assertTrue(all(path.exists() for path in staged)) + class CaseEnvironmentTest(unittest.TestCase): def setUp(self): From f471eda358634538f7bc489543f62b54f0b0b001 Mon Sep 17 00:00:00 2001 From: Cheese Cake Date: Wed, 5 Aug 2026 03:43:30 +0530 Subject: [PATCH 3/4] Update server/scripts/test_ds4_exact_verify_parity.py Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- server/scripts/test_ds4_exact_verify_parity.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/server/scripts/test_ds4_exact_verify_parity.py b/server/scripts/test_ds4_exact_verify_parity.py index b7f097951..6bd862816 100644 --- a/server/scripts/test_ds4_exact_verify_parity.py +++ b/server/scripts/test_ds4_exact_verify_parity.py @@ -394,9 +394,10 @@ def main() -> int: print(f"FAIL: {error}; diagnostics retained in {failed_dir}", file=sys.stderr) return 1 try: - attempt_dir.rmdir() - except OSError: - pass +try: + attempt_dir.rmdir() +except OSError as error: + print(f"warning: could not remove staging dir {attempt_dir}: {error}", file=sys.stderr) print( f"PASS: {len(ar.tokens)} generated token IDs are identical; " "reference-exact speculation executed at least one step" From 626e51384e3282e597ada97ed08e7db5cb5e531e Mon Sep 17 00:00:00 2001 From: Cheese Cake Date: Wed, 5 Aug 2026 03:49:52 +0530 Subject: [PATCH 4/4] Update server/scripts/test_ds4_exact_verify_parity.py Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- server/scripts/test_ds4_exact_verify_parity.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/server/scripts/test_ds4_exact_verify_parity.py b/server/scripts/test_ds4_exact_verify_parity.py index 6bd862816..a0fd2753f 100644 --- a/server/scripts/test_ds4_exact_verify_parity.py +++ b/server/scripts/test_ds4_exact_verify_parity.py @@ -394,10 +394,9 @@ def main() -> int: print(f"FAIL: {error}; diagnostics retained in {failed_dir}", file=sys.stderr) return 1 try: -try: - attempt_dir.rmdir() -except OSError as error: - print(f"warning: could not remove staging dir {attempt_dir}: {error}", file=sys.stderr) + attempt_dir.rmdir() + except OSError as error: + print(f"warning: could not remove staging dir {attempt_dir}: {error}", file=sys.stderr) print( f"PASS: {len(ar.tokens)} generated token IDs are identical; " "reference-exact speculation executed at least one step"