qwen27: serve the 27B pair, refuse what a pair cannot keep, and die with the peer

build.zig

@@ -153,7 +153,7 @@ pub fn build(b: *std.Build) void {
         const gemma4_host_mods = metalEngineModules(b, b.graph.host, .Debug, draft_ids);
         gemma4_host_mods.sources.addImport("nemotron_packed_metallib", embedded(b, b.addWriteFiles().add("gemma4-compile-empty.metallib", ""), "nemotron_packed.metallib"));
         @import("zig/build/gemma4_host.zig").compileHost(b, gemma4_host_mods.metal, gemma4_host_mods.engine, gemma4_host_mods.lanes, gemma4_host_mods.sources);
-        @import("zig/build/qwen27_host.zig").compileHost(b, gemma4_host_mods.metal, gemma4_host_mods.engine, gemma4_host_mods.lanes);
+        test_step.dependOn(@import("zig/build/qwen27_host.zig").compileHost(b, gemma4_host_mods.metal, gemma4_host_mods.engine, gemma4_host_mods.lanes));
     }
     @import("zig/build/qwen27_host.zig").tests(b);
     const xgrammar = grammar_build.option(b);

docs/kernel-contract.md

@@ -20,15 +20,16 @@ The test of a kernel fills a `contract.spec.Spec` and hands the harness a few ho
   and the magnitude of the terms each output sums, and can replay the sums in another accumulator.
 - Its served envelope: the shapes the engine calls it at, and for each dimension its tile, the step a valid size
   moves by and the least valid size.
-- Its accumulator (`f32` for every kernel here) and the depth of its longest chain of roundings at a shape, plus any
+- Its accumulator (`f32` where a kernel sums) and the depth of its longest chain of roundings at a shape, plus any
   fp32 operations after the sums (`ops`) and intermediate roundings to the output type (`mids`).
 - Whether it may write in place. If it replaces another kernel, the kernel whose bits it must equal.
 - Its NaN rule: `propagate` (as float64 arithmetic on its formula does), or a stated rule the test checks itself
   (top-k, argmax, sampling and routing kernels state theirs).
 - For gate 5, its launch in `zig/src/core/contract/registry.zig`: threads a group and threadgroup bytes (the source's,
   and any its encoder sets) on Metal, or symbol, image, threads a block, dynamic shared bytes and cluster width on CUDA,
   with the limits its compiles recorded. HIP launches are in `zig/src/core/contract/hip_records.zig`: symbol, code
-  object, threads a workgroup and dynamic LDS bytes.
+  object, threads a workgroup and dynamic LDS bytes. The seam's records are in `metal_pair_records.zig`, which
+  `registry.zig` concatenates.
 
 ## The gates
 
@@ -99,7 +100,8 @@ The program exits 1 on any failed gate, any control that stayed green, or any li
    the inputs for a shape and a gate's fill (usually through an input family), `run` launches once and waits, `count`
    and `reference` give the outputs and their float64 values, `simulate` replays the sums for gate 4.
 2. Add it to the runner's list, with its served shapes and tiles.
-3. Add its launches to `registry.zig`, and fill the records from `--offline` (Metal) or `kernel-contract-limits`
+3. Add its launches to `registry.zig`, or to a records file it concatenates as the seam's are, and fill them from
+   `--offline` (Metal) or `kernel-contract-limits`
    (CUDA: nvcc 13.0's in `cuda`, and each later release's set where that release compiles it differently); a HIP
    kernel's go to `hip_records.zig`, filled from `hip-kernel-contract-limits`.
 4. Post the `SUMMARY` lines from each backend you ran in the pull request, and say which chips you could not run.
@@ -126,11 +128,17 @@ float64 output class, without an inherited exception.
 
 ## Coverage
 
-Metal: softmax, RMS norm, bf16 top-k, the row projection (3, 4, 6 and 8 bits), affine_mm (4 and 8 bits), simd_qmm
-(4 bits) and simd_qmm_bits (3, 6 and 8 bits), and the lane projection (6-bit groups of 32 on every chip; 4- and 8-bit
+Metal: softmax, RMS norm, bf16 top-k, the row projection (3, 4, 6 and 8 bits), affine_mm (4 and 8 bits), simd_qmm (4
+bits) and simd_qmm_bits (3, 6 and 8 bits), and the lane projection (6-bit groups of 32 on every chip; 4- and 8-bit
 formats on the M5's tensor units, where its reg kernel must equal the cooperative one bit for bit), and tile_qmm's
-prompt tiles and their decode twin (4, 6 and 8 bits; the twin bit-equal to the tiles). Gate 5 also holds the rest of
-tile_qmm's pipelines and GLM-5.3-Flash's hc_expand_10.
+prompt tiles and their decode twin (4, 6 and 8 bits; the twin bit-equal to the tiles), and the row-split TP seam's
+five kernels (`pair_rows.metal`, launched by `pair_rows.zig`): `pair_add_rms` on the fp32 stream, and the bf16 frame's
+`pair_stage`, `pair_gather`, `pair_argmax` and `pair_pick`. `pair_stage` reads only its own rank's half of a frame and
+copies it into the slot the relay sends, and `pair_gather` writes both halves of every row, its own half natively and
+the peer's from the relayed words, with no arithmetic between them: no exchange rounds a sum across ranks, and
+`pair_argmax` and `pair_pick` state their own NaN and tie rules. Gate 5 also holds the rest of tile_qmm's pipelines
+and GLM-5.3-Flash's hc_expand_10.
+
 CUDA: the RMS norm rows (16-byte form and the fp32-out tree), the bf16 residual's norm tails (MLX's rounded rows and
 the once-rounded ones, each checked on its residual and its output, with four controls each dropping a rounding point), Nemotron's router logits (and their 128-row tiles,
 bit-equal to them) and top-k, the latent top-k's decode form (bit-equal to `tf_super_topk`), the latent combine and

zig/build/qwen27_host.zig

@@ -42,9 +42,14 @@ pub fn tests(b: *std.Build) void {
     b.step("test-qwen27-equality-host", "Shared lanes drafted, resumed and concurrent equality on the fake target").dependOn(&equal.step);
 }
 
-pub fn compileHost(b: *std.Build, metal: *std.Build.Module, engine: *std.Build.Module, lanes: *std.Build.Module) void {
+pub fn compileHost(b: *std.Build, metal: *std.Build.Module, engine: *std.Build.Module, lanes: *std.Build.Module) *std.Build.Step {
     const api = b.createModule(.{ .root_source_file = b.path("zig/src/core/engine_api.zig"), .target = b.graph.host, .link_libc = true, .imports = &.{.{ .name = "lanes", .module = lanes }} });
     const host = b.createModule(.{ .root_source_file = b.path("zig/src/native/qwen27_served.zig"), .target = b.graph.host, .link_libc = true, .imports = &.{ .{ .name = "metal", .module = metal }, .{ .name = "tensorfold", .module = engine }, .{ .name = "engine_api", .module = api } } });
+    // pair refusals are decided before a model loads: the host's own tests run on CPU alone
+    const served = b.step("test-qwen27-served-host", "Rank 1's named refusal and the pair's --vision refusal, no shaders or GPU");
+    const served_run = b.addRunArtifact(b.addTest(.{ .root_module = host }));
+    served_run.has_side_effects = true; // a rerun must re-execute the tests, not replay a cached pass
+    served.dependOn(&served_run.step);
     const boundary = b.addTest(.{ .root_module = b.createModule(.{ .root_source_file = b.path("zig/qwen27_bank_boundary_host.zig"), .target = b.graph.host, .optimize = .debug, .link_libc = true, .imports = &.{.{ .name = "tensorfold", .module = engine }} }), .filters = &.{"qwen27 bank boundary:"} });
     b.step("test-qwen27-bank-boundary-host", "Runner pending-keep guards with no device or shader execution").dependOn(&b.addRunArtifact(boundary).step);
     const object = b.addObject(.{ .name = "qwen27-speed-compile", .root_module = b.createModule(.{
@@ -56,6 +61,7 @@ pub fn compileHost(b: *std.Build, metal: *std.Build.Module, engine: *std.Build.M
     }) });
     _ = object.getEmittedBin();
     b.step("check-qwen27-speed-host", "Compile the actual Qwen27 startup and encoding without shaders").dependOn(&object.step);
+    return served;
 }
 
 fn tryNames(b: *std.Build, names: *std.ArrayList([]const u8), text: []const u8) void {

zig/cost_rule_cases.zig

@@ -16,7 +16,9 @@ test "Nemotron initial and drift passes share interleaved head sampling; the 27B
     try std.testing.expect(std.mem.indexOf(u8, body, "if (b.head != null) &head[0] else null") != null);
     try std.testing.expect(std.mem.indexOf(u8, body, "drifted(&head, &r.head)") != null);
     try std.testing.expect(std.mem.indexOf(u8, body, "fastest(heads") == null);
+    const warm = @embedFile("src/native/qwen27_warm.zig"); // the window timing moved here from the lane backend
+    try std.testing.expect(std.mem.indexOf(u8, warm, "cost_rule.fastest(timer, i)") != null);
+    try std.testing.expect(std.mem.indexOf(u8, warm, "cost_rule.measure(") == null);
     const qwen = @embedFile("src/native/qwen27_lanes.zig");
-    try std.testing.expect(std.mem.indexOf(u8, qwen, "cost_rule.fastest(timer, i)") != null);
     try std.testing.expect(std.mem.indexOf(u8, qwen, "cost_rule.measure(") == null);
 }

zig/kernels/metal/core/pair_rows.metal

@@ -0,0 +1,150 @@
+// Two Macs' tensor-parallel exchanges: the host relays parts through shared events; these kernels only do arithmetic.
+#include <metal_stdlib>
+using namespace metal;
+
+// Sync page words (shared with pair_rows.zig): FAILED = the host gave up on a seam; OBSERVED = this side's refusals.
+constant constexpr uint FAILED = 16, OBSERVED = 20;
+
+// True once the host gave up: FAILED is set before the wait releases, so a later kernel poisons its output rows.
+inline bool pair_link_broken(device atomic_uint* sync, uint t) {
+  if (atomic_load_explicit(&sync[FAILED], memory_order_relaxed) == 0u) return false;
+  if (t == 0) atomic_fetch_add_explicit(&sync[OBSERVED], 1u, memory_order_relaxed);
+  return true;
+}
+
+inline float pair_lanes(float v) {
+  for (ushort d = 16; d > 0; d >>= 1) v += simd_shuffle_xor(v, d);
+  return v;
+}
+
+// Kernels' argument block, byte-identical to pair_rows.zig's PairRows: four u32 then four f32; no payload element.
+struct PairRows {
+  uint width, stride, rank, x;
+  float eps, pad0, pad1, pad2;
+};
+
+// bf16 bits as the exact fp32 value they name (argmax reads a bf16 logits stream this way).
+inline float pair_bf16(uint bits) {
+  return as_type<float>(bits << 16);
+}
+
+// Row r: h += rank 0's part + rank 1's part, then y = w * (h / rms(h)); payload is fp32 (streamBytes).
+kernel void pair_add_rms(device atomic_uint* sync [[buffer(0)]], constant PairRows& a [[buffer(1)]],
+                         const device float* mine [[buffer(2)]], device atomic_uint* peer [[buffer(3)]],
+                         device float* h [[buffer(4)]], const device bfloat* w [[buffer(5)]],
+                         device float* y [[buffer(6)]], uint r [[threadgroup_position_in_grid]],
+                         uint t [[thread_index_in_threadgroup]], uint sg [[simdgroup_index_in_threadgroup]],
+                         uint lane [[thread_index_in_simdgroup]]) {
+  threadgroup float part[8];
+  if (pair_link_broken(sync, t)) {
+    for (uint i = t; i < a.width; i += 256) y[ulong(r) * a.stride + i] = NAN;
+    return;
+  }
+  device float* x = h + ulong(r) * a.stride;
+  float s = 0.0f;
+  for (uint i = t; i < a.width; i += 256) {
+    const ulong at = ulong(r) * a.width + i;
+    const float other = as_type<float>(atomic_load_explicit(&peer[at], memory_order_relaxed));
+    const float sum = a.rank == 0 ? mine[at] + other : other + mine[at];
+    const float v = x[i] + sum;
+    x[i] = v;
+    s = fma(v, v, s);
+  }
+  s = pair_lanes(s);
+  if (lane == 0) part[sg] = s;
+  threadgroup_barrier(mem_flags::mem_threadgroup);
+  float var = 0.0f;
+  for (uint g = 0; g < 8; ++g) var += part[g];
+  const float rs = precise::rsqrt(var / float(a.width) + a.eps);
+  for (uint i = t; i < a.width; i += 256) y[ulong(r) * a.stride + i] = float(w[i]) * (x[i] * rs);
+}
+
+// Row r: this rank's own half of the frame row into the exchange slot; a bf16 copy, so it carries the frame's bits.
+kernel void pair_stage(device atomic_uint* sync [[buffer(0)]], constant PairRows& a [[buffer(1)]],
+                       const device ushort* src [[buffer(2)]], device ushort* dst [[buffer(3)]],
+                       uint r [[threadgroup_position_in_grid]], uint t [[thread_index_in_threadgroup]]) {
+  const ulong in = ulong(r) * a.stride;
+  const ulong out = ulong(r) * a.width;
+  if (pair_link_broken(sync, t)) {
+    for (uint i = t; i < a.width; i += 256) dst[out + i] = 0x7FC0u; // bf16 NaN: the link stays down
+    return;
+  }
+  for (uint i = t; i < a.width; i += 256) dst[out + i] = src[in + i];
+}
+
+// Row r: this rank's half at its own offset in the frame row and the peer's at the other; a bf16 copy, no arithmetic.
+kernel void pair_gather(device atomic_uint* sync [[buffer(0)]], constant PairRows& a [[buffer(1)]],
+                        const device ushort* mine [[buffer(2)]], device atomic_uint* peer [[buffer(3)]],
+                        device ushort* out [[buffer(4)]], uint r [[threadgroup_position_in_grid]],
+                        uint t [[thread_index_in_threadgroup]]) {
+  const ulong mrow = ulong(r) * a.width;
+  const ulong orow = ulong(r) * a.stride;
+  const ulong own = a.rank == 0 ? orow : orow + a.width;
+  const ulong got = a.rank == 0 ? orow + a.width : orow;
+  if (pair_link_broken(sync, t)) {
+    for (uint i = t; i < a.width; i += 256) {
+      out[own + i] = 0x7FC0u;
+      out[got + i] = 0x7FC0u;
+    }
+    return;
+  }
+  for (uint i = t; i < a.width; i += 256) {
+    // The relay wrote the peer's half: an untracked frame read, so one atomic word (two bf16) at a time.
+    const uint word = atomic_load_explicit(&peer[(mrow + i) >> 1], memory_order_relaxed);
+    out[own + i] = mine[mrow + i];
+    out[got + i] = ushort(word >> ((i & 1) * 16));
+  }
+}
+
+// Row r's argmax over this rank's logits (ids lo..lo+n): (fp32 value, id), lower id on tie; dims.w 0 = fp32, 1 = bf16.
+kernel void pair_argmax(const device float* logits [[buffer(0)]], constant uint4& dims [[buffer(1)]],
+                        device uint* out [[buffer(2)]], uint r [[threadgroup_position_in_grid]],
+                        uint t [[thread_index_in_threadgroup]], uint sg [[simdgroup_index_in_threadgroup]],
+                        uint lane [[thread_index_in_simdgroup]]) {
+  const ulong row = ulong(r) * dims.x;
+  const device float* xf = logits + row;
+  const device ushort* xb = reinterpret_cast<const device ushort*>(logits) + row;
+  const bool wide = dims.w != 0u;
+  float best = -INFINITY;
+  uint at = 0xFFFFFFFFu;
+  for (uint i = t; i < dims.z; i += 1024) {
+    const float v = wide ? pair_bf16(uint(xb[i])) : xf[i];
+    if (v > best) best = v, at = dims.y + i;
+  }
+  for (ushort o = 16; o > 0; o >>= 1) {
+    const float ob = simd_shuffle_xor(best, o);
+    const uint oa = simd_shuffle_xor(at, o);
+    if (ob > best || (ob == best && oa < at)) best = ob, at = oa;
+  }
+  threadgroup float vb[32];
+  threadgroup uint va[32];
+  if (lane == 0) vb[sg] = best, va[sg] = at;
+  threadgroup_barrier(mem_flags::mem_threadgroup);
+  if (sg != 0) return;
+  best = vb[lane];
+  at = va[lane];
+  for (ushort o = 16; o > 0; o >>= 1) {
+    const float ob = simd_shuffle_xor(best, o);
+    const uint oa = simd_shuffle_xor(at, o);
+    if (ob > best || (ob == best && oa < at)) best = ob, at = oa;
+  }
+  if (lane == 0) out[2 * r] = as_type<uint>(best), out[2 * r + 1] = at;
+}
+
+// Each row's pick over both Macs' argmaxes: the larger value, the lower id on a tie (one Mac's argmax).
+kernel void pair_pick(device atomic_uint* sync [[buffer(0)]], constant uint2& rx [[buffer(1)]],
+                      const device uint* mine [[buffer(2)]], device atomic_uint* peer [[buffer(3)]],
+                      device uint* picks [[buffer(4)]], uint t [[thread_index_in_threadgroup]]) {
+  if (pair_link_broken(sync, t)) {
+    if (t == 0) for (uint r = 0; r < rx.x; ++r) picks[r] = 0xFFFFFFFFu;  // no such token: visibly wrong, never a token
+    return;
+  }
+  if (t != 0) return;
+  for (uint r = 0; r < rx.x; ++r) {
+    const float vm = as_type<float>(mine[2 * r]);
+    const uint am = mine[2 * r + 1];
+    const float vp = as_type<float>(atomic_load_explicit(&peer[2 * r], memory_order_relaxed));
+    const uint ap = atomic_load_explicit(&peer[2 * r + 1], memory_order_relaxed);
+    picks[r] = (vp > vm || (vp == vm && ap < am)) ? ap : am;
+  }
+}

zig/kernels/metal/qwen27/attention_prompt.metal

@@ -4,7 +4,8 @@
 using namespace metal;
 using namespace mpp::tensor_ops;
 constant constexpr int PG = 6, PD = 256, PH = 128, PCK = 512, PTK = 32;
-struct Q27PromptArgs { int start, rows, chunks, rp, key_stride, value_stride; float scale2; };
+// kv_heads is the rank's, not the model's: a split attention writes local node rows.
+struct Q27PromptArgs { int start, rows, chunks, rp, key_stride, value_stride; float scale2; int kv_heads; };
 typedef vec<float, 8> q27f8;
 typedef vec<bfloat, 8> q27b8;
 typedef vec<half, 8> q27h8;
@@ -166,5 +167,6 @@ kernel void q27_attn_prompt_merge(device const float* PO [[buffer(0)]], device c
     m = nm;
   }
   const int node = r / PG, head = hk * PG + r % PG;
-  for (int i = 0; i < DP; i++) OUT[((int64_t)node * (PG * 4) + head) * PD + lane * DP + i] = static_cast<bfloat>(o[i] / l);
+  const int64_t at = ((int64_t)node * (PG * args.kv_heads) + head) * PD + lane * DP;
+  for (int i = 0; i < DP; i++) OUT[at + i] = static_cast<bfloat>(o[i] / l);
 }

zig/kernels/metal/sources.zig

@@ -34,6 +34,8 @@ pub const core_exlr8_dense = @embedFile("core/exlr8_dense.metal");
 pub const core_exlr8_prompt = @embedFile("core/exlr8_prompt.metal");
 pub const core_exlr8_dense_prompt = @embedFile("core/exlr8_dense_prompt.metal");
 pub const core_pair = @embedFile("core/pair.metal");
+/// The row-split seam's kernels (see `pair_rows.zig`): `pair.metal` above stays as it is for glm_dsa.
+pub const core_pair_rows = @embedFile("core/pair_rows.metal");
 
 /// GLM-5.3 (glm_moe_dsa): MLA with rope over a bf16 latent cache, the DSA indexer and its top-k, fp32 glue.
 pub const glm_dsa_mla = @embedFile("glm_dsa/mla.metal");

zig/src/core/contract.zig

@@ -19,6 +19,7 @@ test {
     _ = @import("contract/vit_records.zig");
     _ = @import("contract/simd_pairs_records.zig");
     _ = @import("contract/glm_rows_records.zig");
+    _ = @import("contract/metal_pair_records.zig");
     _ = spec;
     _ = gates;
     _ = rows;

zig/src/core/contract/metal_pair_records.zig

@@ -0,0 +1,26 @@
+//! The pair seam's five launches (core/pair_rows.metal) with the M1/M2 limits the offline compile measures.
+const std = @import("std");
+const limits = @import("limits.zig");
+const registry = @import("registry.zig");
+
+pub const records = [_]limits.MetalLaunch{
+    .{ .name = "metal pair add_rms pair_add_rms", .threads = 256, .tg_memory = 32, .m1 = 1024, .m2 = 1024 },
+    .{ .name = "metal pair stage pair_stage", .threads = 256, .tg_memory = 0, .m1 = 1024, .m2 = 1024 },
+    .{ .name = "metal pair gather pair_gather", .threads = 256, .tg_memory = 0, .m1 = 1024, .m2 = 1024 },
+    .{ .name = "metal pair argmax pair_argmax", .threads = 1024, .tg_memory = 256, .m1 = 1024, .m2 = 1024 },
+    .{ .name = "metal pair pick pair_pick", .threads = 32, .tg_memory = 0, .m1 = 1024, .m2 = 1024 },
+};
+
+test "the pair seam's five launches are in the live and offline collector and fit every family" {
+    for (records) |record| {
+        var found: usize = 0;
+        for (registry.all_metal) |active| if (std.mem.eql(u8, active.name, record.name)) {
+            try std.testing.expectEqualDeep(record, active);
+            found += 1;
+        };
+        try std.testing.expectEqual(@as(usize, 1), found);
+        try std.testing.expectEqual(@as(usize, 0), try registry.checkMetal(std.testing.allocator, &.{record}, true));
+        // A zero here would be a launch no M1/M2 compiler was asked about.
+        try std.testing.expect(record.m1 > 0 and record.m2 > 0);
+    }
+}

zig/src/core/contract/registry.zig

@@ -570,7 +570,7 @@ test {
     _ = paging;
 }
 /// The Metal collector keeps measured slots and the restored vision declarations.
-pub const all_metal = metal ++ @import("delta_slots_records.zig").records ++ @import("flashnext_pending.zig").records ++ @import("simd_pairs_records.zig").records ++ @import("glm_rows_records.zig").records ++ @import("weight_only_records.zig").metal ++ @import("hc_activation_records.zig").metal ++ @import("qwen4_records.zig").metal;
+pub const all_metal = metal ++ @import("delta_slots_records.zig").records ++ @import("flashnext_pending.zig").records ++ @import("simd_pairs_records.zig").records ++ @import("metal_pair_records.zig").records ++ @import("glm_rows_records.zig").records ++ @import("weight_only_records.zig").metal ++ @import("hc_activation_records.zig").metal ++ @import("qwen4_records.zig").metal;
 const learning = @import("registry_learning.zig");
 pub const gradient_pending = learning.gradient_pending;
 pub const qwen_learning_pending = learning.qwen_learning_pending;