From f7b0e41b5d1a7133d3ca3e0fa6458b2f78a1a401 Mon Sep 17 00:00:00 2001 From: mrsergiopascal <258002037+mrsergiopascal@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:51:24 +0200 Subject: [PATCH] fix(dsv41): align decode-sized replay position views --- tools/dsv41-zig/replay_positions.md | 87 +++++++++++++++ tools/dsv41-zig/repro_replay_positions.py | 123 ++++++++++++++++++++++ tools/dsv41-zig/test_replay_positions.py | 117 ++++++++++++++++++++ zig/src/families/dsv41/prompt.zig | 16 ++- 4 files changed, 338 insertions(+), 5 deletions(-) create mode 100644 tools/dsv41-zig/replay_positions.md create mode 100644 tools/dsv41-zig/repro_replay_positions.py create mode 100644 tools/dsv41-zig/test_replay_positions.py diff --git a/tools/dsv41-zig/replay_positions.md b/tools/dsv41-zig/replay_positions.md new file mode 100644 index 0000000000..28e9bfe87a --- /dev/null +++ b/tools/dsv41-zig/replay_positions.md @@ -0,0 +1,87 @@ +# Decode-sized replay position alignment + +`replayCut` formerly kept `pos0 + first * sizeof(i64)`. With an odd cut and +1–15 surviving rows, that view is 8 modulo 16. Decode-sized split sparse +attention uses POS as its default RBASE/CBASE too; the served aligned AOT +variants cannot accept those pointers. The fix uploads the exact surviving +host positions into the existing aligned `pos0`, in the existing stream order. +Even cuts and chunks larger than `tri.decode_rows` keep the view. No kernel +math, replay geometry, capture metadata or alignment checks change. + +## CPU regression (no model or CUDA required) + +Use Zig 0.17.0, as pinned in `.zig-version`: + +```sh +python3 tools/dsv41-zig/test_replay_positions.py +``` + +The test extracts and compiles the actual `replayCut` function, stubbing only +GPU collaborators. It checks exact positions (including noncontiguous sentinel +values), unchanged host data, alignment, upload count, unchanged even/larger +views, and no-cut/encoder-only chunks. It does not test CUDA execution or stream +ordering. To reproduce the red result without changing the checkout: + +```sh +git show b65c8b92d2403dfdefccf15a3b23ba609a4553e2:zig/src/families/dsv41/prompt.zig > baseline-prompt.zig +python3 tools/dsv41-zig/test_replay_positions.py --source baseline-prompt.zig +``` + +Baseline: two tests fail (`expected 0, found 8`); candidate: all three pass. +Remove or keep the generated baseline file outside your publication tree. + +## Opt-in server reproduction + +`repro_replay_positions.py` uses only synthetic comment padding and a synthetic +`check_code` tool. The URL is a native server base URL without `/v1`. Sizing +calls `/tokenize`; `--run` additionally performs inference and a tool continuation. +It never starts, stops, rebuilds, or reconfigures a server. Use an explicitly +approved test server and preserve each request file between baseline/candidate: + +```sh +python3 tools/dsv41-zig/repro_replay_positions.py --url http://127.0.0.1:8080 \ + --tokens 39039 --request request-39039.json +python3 tools/dsv41-zig/repro_replay_positions.py --url http://127.0.0.1:8080 \ + --tokens 39039 --request request-39039.json --run --output baseline-39039 +# After your separately approved candidate rebuild, send those same saved bytes: +python3 tools/dsv41-zig/repro_replay_positions.py --url http://127.0.0.1:8080 \ + --tokens 39039 --request request-39039.json --run --output candidate-39039 +``` + +Use a new output directory per run. Baseline SSE errors are retained in +`first.sse` before the script exits nonzero. Candidate success requires actual +`response.completed`, exact tool arguments, actual input-token usage matching +the target, and `ALIGNMENT_OK` from a completed tool continuation. HTTP 200 alone +is not success. The restricted chat translation used for sizing is checked +against Responses usage; different templates may require adapting it. + +For the cached-prefix cases, first run the same fixture at 38783 and 122747 +respectively, then 39039 and 123003. Inspect returned cached-token usage: cache +state and scheduling matter, and token residue alone is not a universal trigger. +Check adjacent lengths 39038/39040 and 123002/123004 separately. + +## Existing four-Spark qualification + +These are earlier real GPU results on the recipe's engine base +`59402086381a42c74d53053443224eec2790b4c5` plus its four patches, before/after +this narrow repair, **not** a GPU qualification of the newer PR base. +The recipe is `tonyd2wild/DeepSeek-V4.1-Flash-vLLM-DGX-Spark`, +`runs/2026-10-17-1017-tensorfold` (DeepSeek-V4.1-Flash 1017, TP4, four GB10 Sparks). + +| Rendered tokens | Baseline | Same first request after repair | Cached tokens after repair | +|---:|---|---|---:| +| 10353 | MissingTritonVariant, zero generated tokens | completed + tool continuation | 0 | +| 10363 | MissingTritonVariant, zero generated tokens | completed + tool continuation | 0 | +| 39039 | MissingTritonVariant, zero generated tokens | completed + tool continuation | 36864 | +| 123003 | MissingTritonVariant, zero generated tokens | completed + tool continuation | 120832 | + +All four first requests were byte-identical before/after. The neighbors +39038/39040 and 123002/123004 and prefix warmers also completed with the exact +tool call and continuation. With chunk size 2048 and window 128, the failing +lengths yield cuts at 2033/2043/2047/2043, leaving 15/5/1/5 rows respectively. +The failing sparse-attention launch had ring_size=144 and n_idx=512; +POS/RBASE/CBASE were all 8 modulo 16. Existing H16/H32 captures require alignment. + +Only four-Spark TP4 was GPU-tested. No TP2, other accelerator, benchmark, soak, +private workload replay, broad long-job stability, or token-for-token whole-model +quality claim is made. This PR does not change any running deployment. diff --git a/tools/dsv41-zig/repro_replay_positions.py b/tools/dsv41-zig/repro_replay_positions.py new file mode 100644 index 0000000000..840e0f7e5b --- /dev/null +++ b/tools/dsv41-zig/repro_replay_positions.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Opt-in synthetic Responses reproducer: size once, replay the same JSON before/after a rebuild.""" +import argparse +import copy +import json +from pathlib import Path +import urllib.request + +TOOL = { + "type": "function", "name": "check_code", "description": "Check a synthetic code file.", + "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, + "required": ["path"], "additionalProperties": False}, +} + + +def fixture(model, padding): + return { + "model": model, + "instructions": "You are a coding assistant testing synthetic fixtures. Follow the latest user instruction.", + "input": [ + {"role": "user", "content": "Synthetic repository context. Padding is inert code-comment material:\n#" + + " x" * padding + "\ndef add(a, b): return a + b\n"}, + {"type": "function_call", "call_id": "call_fixture", "name": "check_code", + "arguments": '{"path":"synthetic.py"}'}, + {"type": "function_call_output", "call_id": "call_fixture", "output": '{"ok":true,"synthetic":true}'}, + {"role": "user", "content": "Call check_code with path synthetic.py now. " + "After its result, reply exactly ALIGNMENT_OK."}, + ], + "tools": [TOOL], "stream": True, "temperature": 0, "max_output_tokens": 128, + "chat_template_kwargs": {"enable_thinking": False}, "store": False, + } + + +def chat(body): + # Restricted translation for this fixture, matching responses_translate.zig; verify final usage too. + inp = body["input"] + return { + "messages": [ + {"role": "system", "content": body["instructions"]}, inp[0], + {"role": "assistant", "content": "", "tool_calls": [ + {"id": inp[1]["call_id"], "type": "function", "function": { + "name": inp[1]["name"], "arguments": inp[1]["arguments"]}}]}, + {"role": "tool", "tool_call_id": inp[2]["call_id"], "content": inp[2]["output"]}, inp[3], + ], + "tools": [{"type": "function", "function": {k: v for k, v in TOOL.items() if k != "type"}}], + "chat_template_kwargs": body["chat_template_kwargs"], + } + + +def post(url, path, payload): + req = urllib.request.Request(url.rstrip("/") + path, payload, {"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=900) as response: + return response.read() + + +def completed(raw): + events = [json.loads(line[6:]) for line in raw.decode().splitlines() + if line.startswith("data: ") and line[6:] != "[DONE]"] + failures = [e for e in events if e.get("type") in ("error", "response.failed")] + if failures: + raise RuntimeError(json.dumps(failures)) + responses = [e["response"] for e in events if e.get("type") == "response.completed"] + if not responses or responses[-1]["status"] != "completed": + raise RuntimeError("No completed response in SSE stream") + return responses[-1] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", required=True, help="Native server base URL, without /v1") + parser.add_argument("--model", default="deepseek-v4.1-flash") + parser.add_argument("--tokens", type=int, required=True, help="Rendered input tokens (not text length)") + parser.add_argument("--request", type=Path, required=True, help="Synthetic request JSON to create or replay") + parser.add_argument("--output", type=Path, help="New directory for SSE evidence (required with --run)") + parser.add_argument("--run", action="store_true", help="Send inference and a tool continuation; otherwise only size") + args = parser.parse_args() + if args.tokens < 1 or (args.run and args.output is None): + parser.error("positive --tokens and --output with --run are required") + if not args.request.exists(): + padding = 0 + for _ in range(5): + body = fixture(args.model, padding) + count = json.loads(post(args.url, "/tokenize", json.dumps(chat(body)).encode()))["count"] + if count == args.tokens: + with args.request.open("x") as output: + json.dump(body, output, indent=2) + break + padding += args.tokens - count + if padding < 0: + raise ValueError("Token target is smaller than the fixture") + else: + raise RuntimeError("Could not size fixture exactly") + if not args.run: + print("Request ready; inference was not sent") + return + args.output.mkdir(parents=True, exist_ok=False) + payload = args.request.read_bytes() + body = json.loads(payload) + count = json.loads(post(args.url, "/tokenize", json.dumps(chat(body)).encode()))["count"] + if count != args.tokens: + raise ValueError(f"Saved request renders to {count}, expected {args.tokens}") + raw = post(args.url, "/v1/responses", payload) + (args.output / "first.sse").write_bytes(raw) + response = completed(raw) + if response["usage"]["input_tokens"] != args.tokens: + raise ValueError("Responses usage differs from the tokenizer target") + calls = [item for item in response["output"] if item["type"] == "function_call"] + if len(calls) != 1 or calls[0]["name"] != "check_code" or json.loads(calls[0]["arguments"]) != {"path": "synthetic.py"}: + raise ValueError("Unexpected tool call") + continuation = copy.deepcopy(body) + continuation["input"] += response["output"] + [{"type": "function_call_output", + "call_id": calls[0]["call_id"], "output": '{"ok":true,"synthetic":true}'}] + raw = post(args.url, "/v1/responses", json.dumps(continuation).encode()) + (args.output / "continuation.sse").write_bytes(raw) + result = completed(raw) + text = "".join(part.get("text", "") for item in result["output"] for part in item.get("content", [])) + if text.strip() != "ALIGNMENT_OK": + raise ValueError("Unexpected continuation text") + print(json.dumps({"completed": True, "tool_continuation_completed": True, "usage": response["usage"]})) + + +if __name__ == "__main__": + main() diff --git a/tools/dsv41-zig/test_replay_positions.py b/tools/dsv41-zig/test_replay_positions.py new file mode 100644 index 0000000000..20d690f683 --- /dev/null +++ b/tools/dsv41-zig/test_replay_positions.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""CPU regression/reproducer for replayCut's position view; requires Zig from .zig-version.""" +import argparse +import os +from pathlib import Path +import subprocess +import tempfile + +ROOT = Path(__file__).resolve().parents[2] +PRELUDE = r''' +const std = @import("std"); +const tri = struct { const decode_rows = 16; }; +const Caches = struct {}; +const Shared = struct {}; +const Ops = struct { + fn copyRows(_: Ops, _: u8, _: u64, _: usize, _: u64, _: usize, _: usize, _: usize) !void {} +}; +const Layer = struct { comp_wkv: ?u8 = 1 }; +const Engine = struct { + c: struct { hc: usize = 4, hidden: usize = 8 } = .{}, + w: struct { layers: [1]Layer = .{.{}} } = .{}, + ops: Ops = .{}, s: u8 = 0, +}; +const Chunk = struct { + start: usize, n: usize, pos: u64, pos0: u64, + h: u64 = 0, h_alt: u64 = 0, pre: u64 = 0, pre_f: u64 = 0, +}; +fn attnMixes(_: *const Engine, _: *Chunk, _: usize) !void {} +fn kvSourceUpdate(_: *const Engine, _: *Chunk, _: *const Caches, _: *Shared, _: usize) !void {} +var uploads: usize = 0; +fn upload(_: *const Engine, dst: u64, src: *const anyopaque, bytes: usize) !void { + const to: [*]u8 = @ptrFromInt(dst); + const from: [*]const u8 = @ptrCast(src); + @memcpy(to[0..bytes], from[0..bytes]); + uploads += 1; +} +''' +TESTS = r''' +fn check(origin: usize, remaining: usize, noncontiguous: bool) !void { + const e: Engine = .{}; + const cs: Caches = .{}; + var sh: Shared = .{}; + var host: [2048]i64 = undefined; + var dev: [2048]i64 align(16) = undefined; + for (0..2048) |i| { + host[i] = @intCast(origin + i * @as(usize, if (noncontiguous) 3 else 1)); + dev[i] = host[i]; + } + const expected = host; + const base = @intFromPtr(&dev); + var ch: Chunk = .{ .start = origin, .n = 2048, .pos = base, .pos0 = base }; + const first = 2048 - remaining; + uploads = 0; + try std.testing.expect(try replayCut(&e, &ch, &cs, &sh, 0, origin + first, &host)); + try std.testing.expectEqual(origin + first, ch.start); + try std.testing.expectEqual(remaining, ch.n); + const fixed = first % 2 != 0 and remaining <= 16; + if (remaining <= 16) try std.testing.expectEqual(@as(u64, 0), ch.pos % 16); + try std.testing.expectEqual(@as(usize, if (fixed) 1 else 0), uploads); + try std.testing.expectEqual(if (fixed) base else base + first * 8, ch.pos); + const positions: [*]const i64 = @ptrFromInt(ch.pos); + try std.testing.expectEqualSlices(i64, expected[first..], positions[0..remaining]); + try std.testing.expectEqualSlices(i64, &expected, &host); +} +test "reported lengths and adjacent replay boundaries" { + const lengths = [_]usize{ 10353, 10363, 39039, 123003, 39038, 39040, 123002, 123004 }; + for (lengths) |length| { + const replay = length - 128; + const origin = replay / 2048 * 2048; + try check(origin, 2048 - (replay - origin), false); + } +} +test "all small odd cuts, even cuts, and larger views preserve exact host positions" { + for ([_]usize{ 0, 8192, 36864, 120832 }) |origin| { + for (1..34) |remaining| { + try check(origin, remaining, false); + try check(origin, remaining, true); + } + } +} +test "no cut and encoder-only chunks leave positions alone" { + const e: Engine = .{}; + const cs: Caches = .{}; + var sh: Shared = .{}; + var host: [2048]i64 = undefined; + var dev: [2048]i64 align(16) = undefined; + const base = @intFromPtr(&dev); + var ch: Chunk = .{ .start = 2048, .n = 2048, .pos = base, .pos0 = base }; + uploads = 0; + try std.testing.expect(try replayCut(&e, &ch, &cs, &sh, 0, 0, &host)); + try std.testing.expect(try replayCut(&e, &ch, &cs, &sh, 0, 2048, &host)); + try std.testing.expect(!try replayCut(&e, &ch, &cs, &sh, 0, 4096, &host)); + try std.testing.expectEqual(@as(usize, 0), uploads); + try std.testing.expectEqual(base, ch.pos); + try std.testing.expectEqual(@as(usize, 2048), ch.start); + try std.testing.expectEqual(@as(usize, 2048), ch.n); +} +''' + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--zig", default=os.environ.get("ZIG", "zig")) + parser.add_argument("--source", type=Path, default=ROOT / "zig/src/families/dsv41/prompt.zig") + args = parser.parse_args() + source = args.source.read_text() + # Compile the actual function, not a Python translation; stub only GPU collaborators. + start = source.index("pub fn replayCut(") + end = source.index("\n}\n", start) + 3 + with tempfile.TemporaryDirectory(prefix="replay-positions-") as tmp: + harness = Path(tmp) / "replay_positions.zig" + harness.write_text(PRELUDE + source[start:end] + TESTS) + subprocess.run([args.zig, "test", str(harness)], check=True) + + +if __name__ == "__main__": + main() diff --git a/zig/src/families/dsv41/prompt.zig b/zig/src/families/dsv41/prompt.zig index 4c04130de8..2fb961c169 100644 --- a/zig/src/families/dsv41/prompt.zig +++ b/zig/src/families/dsv41/prompt.zig @@ -170,8 +170,8 @@ pub const Chunk = struct { n: usize = 0, start: usize = 0, ids: u64, // int64 [cap] - pos: u64, // int64 [n]: start .. start + n (after a replay cut a view into pos0, pos[first:]) - pos0: u64, // int64 [cap]: the chunk's positions as begin() wrote them + pos: u64, // int64 [n]: start .. start + n (replay cut: a view, or aligned pos0 for small odd cuts) + pos0: u64, // int64 [cap]: begin() positions, or exact surviving positions after a small odd cut h: u64, // bf16 [cap, hc, D]: the streams h_alt: u64, // their second buffer (hc_pre_pf writes the posted streams there) x: u64, // bf16 [cap, D]: a block's input rows @@ -1360,9 +1360,15 @@ pub fn replayCut(e: *const Engine, ch: *Chunk, cs: *const Caches, sh: *Shared, l std.mem.swap(u64, &ch.pre, &ch.pre_f); ch.start += first; ch.n = n; - // pos[first:]: a view, first * 8 bytes in (its alignment is the one the served kernels were specialized for) - _ = host_pos; - ch.pos += first * 8; + // Decode-sized split attention captures require aligned POS (also its default RBASE/CBASE). + const pos = ch.pos + first * 8; + if (pos % 16 != 0 and n <= tri.decode_rows) { + // Upload begin()'s exact surviving positions after their previous readers on e.s. + try upload(e, ch.pos0, host_pos[first..][0..n].ptr, n * 8); + ch.pos = ch.pos0; + } else { + ch.pos = pos; + } return true; }