From 5d6a0abff2e3e1db90233ba6040aeac99ff0e1e1 Mon Sep 17 00:00:00 2001 From: chaog992 <338238273+chaog992@users.noreply.github.com> Date: Fri, 16 Oct 2026 08:08:45 +0800 Subject: [PATCH 1/3] fix(lanes): a pipelined stream adopted between rounds starts in verify A stream adopted between rounds holds neither a mode nor a queued step: detach drops a verify or exit stream's mode, and a stream another driver opened never had one. pipelinedRound read the missing mode as pipe and unwrapped a step that was never queued; it now starts such a stream as verify, from its pending token. --- zig/src/core/lanes/engine.zig | 3 +- zig/src/core/lanes/engine_test.zig | 95 +++++++++++++++++++++++++++++- 2 files changed, 96 insertions(+), 2 deletions(-) diff --git a/zig/src/core/lanes/engine.zig b/zig/src/core/lanes/engine.zig index cf0a97ea4a..e0bc43a1de 100644 --- a/zig/src/core/lanes/engine.zig +++ b/zig/src/core/lanes/engine.zig @@ -470,7 +470,8 @@ pub const Engine = struct { /// One token, the next forward queued first; a copied continuation ahead switches to verify windows. fn pipelinedRound(e: *Engine, s: *Stream) !Result { const before = s.mode; - const mode = s.mode orelse .pipe; + // an adopted stream holds neither a mode nor a queued step: it starts where a drain step leaves one + const mode: sm.Mode = s.mode orelse if (s.inflight == null) .verify else .pipe; var result: Result = .{ .rows = 1, .keep = 1, .got = &.{} }; if (mode == .verify or mode == .exit) { const proposal = if (mode == .verify and e.cfg.family_width >= 2) try win.copyProposal(e, s) else &.{}; diff --git a/zig/src/core/lanes/engine_test.zig b/zig/src/core/lanes/engine_test.zig index 2100cc3762..bb32189fa3 100644 --- a/zig/src/core/lanes/engine_test.zig +++ b/zig/src/core/lanes/engine_test.zig @@ -88,9 +88,14 @@ const Classes = struct { }; fn model() !Config { + return modelWith(true); +} + +/// The test family drafting with its head (`mtp`) or by copies alone, when drafted streams step pipelined. +fn modelWith(mtp: bool) !Config { var costs: [16]@import("config.zig").Cost = undefined; for (&costs, 1..) |*c, w| c.* = .{ .width = @intCast(w), .ms = 5.0 + 0.8 * @as(f64, @floatFromInt(w)) }; - return Config.init(gpa, .{ .exact_width = 16, .gpu_tokens = true, .mtp = true, .speculate = true, .speculate_early = false, .drafts = 4, .window_costs = &costs, .mtp_step_ms = 0.5, .hidden_rows = true, .batch_rows = 32, .max_streams = 8, .draft_streams = true }, 16, 15); + return Config.init(gpa, .{ .exact_width = 16, .gpu_tokens = true, .mtp = mtp, .speculate = true, .speculate_early = false, .drafts = 4, .window_costs = &costs, .mtp_step_ms = 0.5, .hidden_rows = true, .batch_rows = 32, .max_streams = 8, .draft_streams = true }, 16, 15); } /// Every case's emitted tokens, the cases admitted together and stepped until done. @@ -440,6 +445,94 @@ test "structured output: where the grammar spells its own text, its drafts fill } } +/// Each token the fake target draws one at a time after `prompt`, as `got` must hold them. +fn expectDecode(prompt: []const u32, got: []const u32) !void { + var history: std.ArrayList(u32) = .empty; + defer history.deinit(gpa); + try history.appendSlice(gpa, prompt); + for (got) |t| { + try std.testing.expectEqual(fake.next(history.items, null, history.items.len), t); + try history.append(gpa, t); + } +} + +test "a lone stream handed to another driver between rounds and back replies as if kept, pipelined or windowed" { + // a plain stream left in verify by a shared round, a windowed drafted one, a copy-drafted one left there by a drain + const Set = struct { cases: []const Case, mtp: bool }; + const sets = [_]Set{ + .{ .cases = &.{ .{ .prompt = &p1, .drafts = false }, .{ .prompt = &p2, .max_new = 6 } }, .mtp = true }, + .{ .cases = &.{.{ .prompt = &p2 }}, .mtp = true }, + .{ .cases = &.{.{ .prompt = &p2 }}, .mtp = false }, + }; + for (sets) |set| { + var cfg = try modelWith(set.mtp); + defer cfg.deinit(gpa); + var target: fake.Fake = .{ .gpa = gpa }; + defer target.deinit(); + var clock: fake.FixedClock = .{}; + var engine = Engine.init(gpa, &cfg, target.backend(), clock.clock()); + defer engine.deinit(); + var proposers: [2]SuffixLookup = undefined; + var streams: [2]sm.Stream = undefined; + for (set.cases, proposers[0..set.cases.len], streams[0..set.cases.len]) |c, *p, *s| { + p.* = try SuffixLookup.init(gpa, .{ .min_match = 4 }); + s.* = try sm.Stream.init(gpa, .{ + .id = "s", + .prompt = c.prompt, + .max_new = c.max_new, + .eos = c.eos, + .drafts = c.drafts, + .proposer = p.proposer(), + }); + } + defer for (proposers[0..set.cases.len], streams[0..set.cases.len]) |*p, *s| { + s.deinit(gpa); + p.deinit(); + }; + for (streams[0..set.cases.len]) |*s| try engine.addStream(s); + var handed: usize = 0; + while (engine.activeCount() > 0) { + try engine.step(); + if (engine.live.items.len != 1) continue; + const s = engine.live.items[0]; + if (!engine.detach(s)) continue; + handed += 1; + try engine.adopt(s); + } + try std.testing.expect(handed > 0); + for (set.cases, streams[0..set.cases.len]) |c, *s| try expectDecode(c.prompt, s.emitted()); + } +} + +test "a stream another driver opened and hands to the lane core replies as its one-token decode" { + for ([_]bool{ true, false }) |mtp| { + var cfg = try modelWith(mtp); + defer cfg.deinit(gpa); + var target: fake.Fake = .{ .gpa = gpa }; + defer target.deinit(); + var clock: fake.FixedClock = .{}; + var engine = Engine.init(gpa, &cfg, target.backend(), clock.clock()); + defer engine.deinit(); + var p = try SuffixLookup.init(gpa, .{ .min_match = 4 }); + defer p.deinit(); + var s = try sm.Stream.init(gpa, .{ + .id = "s", + .prompt = &p2, + .max_new = 40, + .eos = &.{96}, + .proposer = p.proposer(), + }); + defer s.deinit(gpa); + // as a lone driver hands a stream over after its first token (lane_host_lone_test.zig's Lone) + _ = try target.backend().opening(gpa, &s); + s.pending = s.context.items[s.context.items.len - 1]; + s.cache_len = s.prompt_len; + try engine.adopt(&s); + while (engine.activeCount() > 0) try engine.step(); + try expectDecode(&p2, s.emitted()); + } +} + test "a family that keeps copies to shared rounds proposes none for a lone stream" { for ([_]bool{ true, false }) |lone_copies| { var cfg = try Config.init(gpa, .{ .exact_width = 16, .mtp = true, .speculate = true, .drafts = 4, .lone_copies = lone_copies }, 16, 15); From b0b630f2617b432ddc9f833d843c44850c5bb851 Mon Sep 17 00:00:00 2001 From: chaog992 <338238273+chaog992@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:21:23 +0800 Subject: [PATCH 2/3] test(lanes): the adoption tests check each whole reply and its finish reason expectDecode compared only the tokens a reply held with the fake target's decode, so a reply that ended early passed. It now builds the decode to the case's max_new or first eos token and compares the whole reply and the stream's reason; like expectRows, it takes the case. --- zig/src/core/lanes/engine_test.zig | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/zig/src/core/lanes/engine_test.zig b/zig/src/core/lanes/engine_test.zig index bb32189fa3..d5415bf5c8 100644 --- a/zig/src/core/lanes/engine_test.zig +++ b/zig/src/core/lanes/engine_test.zig @@ -445,15 +445,22 @@ test "structured output: where the grammar spells its own text, its drafts fill } } -/// Each token the fake target draws one at a time after `prompt`, as `got` must hold them. -fn expectDecode(prompt: []const u32, got: []const u32) !void { +/// The fake target's one-token decode after `c.prompt`, to `c.max_new` tokens or an eos one: `s`'s reply and reason. +fn expectDecode(c: Case, s: *const sm.Stream) !void { var history: std.ArrayList(u32) = .empty; defer history.deinit(gpa); - try history.appendSlice(gpa, prompt); - for (got) |t| { - try std.testing.expectEqual(fake.next(history.items, null, history.items.len), t); + try history.appendSlice(gpa, c.prompt); + var reason: sm.Reason = .length; + while (history.items.len - c.prompt.len < c.max_new) { + const t = fake.next(history.items, c.sampling, history.items.len); try history.append(gpa, t); + if (std.mem.indexOfScalar(u32, c.eos, t) != null) { + reason = .stop; + break; + } } + try std.testing.expectEqualSlices(u32, history.items[c.prompt.len..], s.emitted()); + try std.testing.expectEqual(reason, s.reason); } test "a lone stream handed to another driver between rounds and back replies as if kept, pipelined or windowed" { @@ -500,11 +507,12 @@ test "a lone stream handed to another driver between rounds and back replies as try engine.adopt(s); } try std.testing.expect(handed > 0); - for (set.cases, streams[0..set.cases.len]) |c, *s| try expectDecode(c.prompt, s.emitted()); + for (set.cases, streams[0..set.cases.len]) |c, *s| try expectDecode(c, s); } } test "a stream another driver opened and hands to the lane core replies as its one-token decode" { + const c: Case = .{ .prompt = &p2 }; for ([_]bool{ true, false }) |mtp| { var cfg = try modelWith(mtp); defer cfg.deinit(gpa); @@ -517,9 +525,9 @@ test "a stream another driver opened and hands to the lane core replies as its o defer p.deinit(); var s = try sm.Stream.init(gpa, .{ .id = "s", - .prompt = &p2, - .max_new = 40, - .eos = &.{96}, + .prompt = c.prompt, + .max_new = c.max_new, + .eos = c.eos, .proposer = p.proposer(), }); defer s.deinit(gpa); @@ -529,7 +537,7 @@ test "a stream another driver opened and hands to the lane core replies as its o s.cache_len = s.prompt_len; try engine.adopt(&s); while (engine.activeCount() > 0) try engine.step(); - try expectDecode(&p2, s.emitted()); + try expectDecode(c, &s); } } From 721c29b6b0f08e302e44b0079d04a1d854a8e767 Mon Sep 17 00:00:00 2001 From: chaog992 <338238273+chaog992@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:50:02 +0800 Subject: [PATCH 3/3] test(lanes): the hand-back test runs a round on what an adopted round leaves The hand-back test detached the lone stream again at every chance, so when an adopted round left no step queued, no round ran on what it left: after a copy window that keeps more than one token, the stream still holds neither a mode nor a queued step and starts in verify again. It now skips the hand-back in the step right after one, and the copy-drafted set runs 60 tokens, so that round comes before its reply ends. --- zig/src/core/lanes/engine_test.zig | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/zig/src/core/lanes/engine_test.zig b/zig/src/core/lanes/engine_test.zig index d5415bf5c8..570e8b03ea 100644 --- a/zig/src/core/lanes/engine_test.zig +++ b/zig/src/core/lanes/engine_test.zig @@ -469,7 +469,7 @@ test "a lone stream handed to another driver between rounds and back replies as const sets = [_]Set{ .{ .cases = &.{ .{ .prompt = &p1, .drafts = false }, .{ .prompt = &p2, .max_new = 6 } }, .mtp = true }, .{ .cases = &.{.{ .prompt = &p2 }}, .mtp = true }, - .{ .cases = &.{.{ .prompt = &p2 }}, .mtp = false }, + .{ .cases = &.{.{ .prompt = &p2, .max_new = 60 }}, .mtp = false }, }; for (sets) |set| { var cfg = try modelWith(set.mtp); @@ -498,13 +498,19 @@ test "a lone stream handed to another driver between rounds and back replies as }; for (streams[0..set.cases.len]) |*s| try engine.addStream(s); var handed: usize = 0; + var settle = false; // a round runs on what the adopted round leaves before the next hand-back while (engine.activeCount() > 0) { try engine.step(); if (engine.live.items.len != 1) continue; + if (settle) { + settle = false; + continue; + } const s = engine.live.items[0]; if (!engine.detach(s)) continue; handed += 1; try engine.adopt(s); + settle = true; } try std.testing.expect(handed > 0); for (set.cases, streams[0..set.cases.len]) |c, *s| try expectDecode(c, s);