fix(dsv41): align decode-sized replay position views

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.

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()

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()

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;
 }