lanes: a copy window may be priced by its stream's own copies (Model.own_copy_rates)

zig/src/core/lanes/config.zig

@@ -14,6 +14,7 @@ pub const default_prior = [_]f64{ 0.85, 0.75, 0.7, 0.65, 0.6, 0.55, 0.5, 0.5, 0.
 pub const Model = struct {
     root_copies: bool = false, // verified context copies precede head drafts and skip their proposal pass
     copy_drafts: bool = true, // global serving switch, independent of shared verification capacity
+    own_copy_rates: bool = false, // a copy window is priced by its stream's own copies, not copy_rate's prior
 
     exact_width: u32 = 1, // the widest window that reproduces one-token steps bit for bit
     first_copy_rows: u32 = 0, // a lone stream's first copy window (0: the exact width)
@@ -52,6 +53,7 @@ pub const Config = struct {
     root_copies: bool = false,
     family_width: u32,
     max_copy: u32,
+    own_copy_rates: bool = false, // Model.own_copy_rates
     first_copy: u32,
     base_width: u32,
     family_mtp: bool,
@@ -134,6 +136,7 @@ pub const Config = struct {
             .lone_cost_tile = m.lone_cost_tile,
             .family_width = width,
             .max_copy = max_copy,
+            .own_copy_rates = m.own_copy_rates,
             .first_copy = first_copy,
             .base_width = @min(width, first_copy + 1),
             .family_mtp = m.mtp and width >= 2 and m.speculate,

zig/src/core/lanes/copied_rate_test.zig

@@ -0,0 +1,56 @@
+//! Model.own_copy_rates: a shared round's copy windows priced by each stream's own copies, else by copy_rate.
+const std = @import("std");
+const Config = @import("config.zig").Config;
+const Cost = @import("config.zig").Cost;
+const Engine = @import("engine.zig").Engine;
+const Stream = @import("stream.zig").Stream;
+const win = @import("windows.zig");
+const fake = @import("fake.zig");
+
+const gpa = std.testing.allocator;
+const copied = [_]u32{ 5, 6, 7, 8, 9, 10, 11 };
+
+/// The copy rows allocation grants each of four streams' seven-row copy windows in one shared round.
+fn granted(e: *Engine, streams: []Stream, out: []usize) !void {
+    _ = e.arena.reset(.retain_capacity);
+    var plans: [4]win.Plan = undefined;
+    for (plans[0..streams.len], streams) |*p, *s| p.* = .{ .stream = s, .position = s.context.items.len, .kind = .copy, .tokens = &copied };
+    try win.allocate(e, plans[0..streams.len]);
+    for (plans[0..streams.len], out) |p, *o| o.* = p.count();
+}
+
+test "own_copy_rates: copies that keep breaking are cut, by source; without it every copy is priced at copy_rate" {
+    // plain rows 10 ms, 0.6 ms a row past four: seven copy rows a stream pay at the prior, not landing 1 a copy
+    var costs: [32]Cost = undefined;
+    for (&costs, 1..) |*c, r| c.* = .{ .width = @intCast(r), .ms = 10 + 0.6 * @as(f64, @floatFromInt(@max(r, 4) - 4)) };
+    for ([_]bool{ false, true }) |own| {
+        var cfg = try Config.init(gpa, .{ .exact_width = 16, .hidden_rows = true, .window_costs = &costs, .batch_rows = 32, .own_copy_rates = own }, 16, 15);
+        defer cfg.deinit(gpa);
+        var target: fake.Fake = .{ .gpa = gpa };
+        defer target.deinit();
+        var clock: fake.FixedClock = .{};
+        var e = Engine.init(gpa, &cfg, target.backend(), clock.clock());
+        defer e.deinit();
+        var streams: [4]Stream = undefined;
+        var made: usize = 0;
+        defer for (streams[0..made]) |*s| s.deinit(gpa);
+        for (&streams) |*s| {
+            s.* = try Stream.init(gpa, .{ .id = "copied", .prompt = &.{ 1, 2 }, .max_new = 64 });
+            made += 1;
+        }
+        var rows: [4]usize = undefined;
+        try granted(&e, &streams, &rows);
+        try std.testing.expectEqualSlices(usize, &.{ 7, 7, 7, 7 }, &rows);
+        for (0..8) |_| win.observeCopied(&e, &streams[0], false, 1, 7); // stream 0's context copies land one row of seven
+        try granted(&e, &streams, &rows);
+        if (own) {
+            try std.testing.expect(rows[0] <= 1);
+            try std.testing.expectEqualSlices(usize, &.{ 7, 7, 7 }, rows[1..]);
+            try std.testing.expect(win.copiedRate(&e, &streams[0], false) < 0.6);
+            try std.testing.expectApproxEqAbs(cfg.copy_rate, win.copiedRate(&e, &streams[0], true), 1e-12); // Recall's keep theirs
+        } else {
+            try std.testing.expectEqualSlices(usize, &.{ 7, 7, 7, 7 }, &rows);
+            try std.testing.expectEqual(@as(f64, -1), streams[0].copied_landed[0]); // no history kept
+        }
+    }
+}

zig/src/core/lanes/engine.zig

@@ -344,9 +344,7 @@ pub const Engine = struct {
                 for (path[1..]) |r| own_kept += @intFromBool(r < own_end);
                 const own: usize = own_end - 1;
                 if (p.kind == .copy) {
-                    if (p.recalled) {
-                        if (s.recall) |r| r.observe(@intCast(own), @intCast(own_kept), s.rounds);
-                    } else if (s.proposer) |pr| pr.observe(@intCast(own), @intCast(own_kept));
+                    win.observeCopied(e, s, p.recalled, own_kept, own);
                     if (e.alone) {
                         const width = s.copy_width orelse e.cfg.first_copy;
                         s.copy_width = if (own_kept == own) @min(e.cfg.max_copy, 2 * width + 1) else e.cfg.first_copy;

zig/src/core/lanes/lanes.zig

@@ -65,6 +65,7 @@ test {
     _ = @import("fill_test.zig");
     _ = @import("fill_trim_test.zig");
     _ = @import("fair_turns_test.zig");
+    _ = @import("copied_rate_test.zig");
 }
 
 pub const learned_disk = @import("learned_disk.zig");

zig/src/core/lanes/stream.zig

@@ -200,6 +200,8 @@ pub const Stream = struct {
     prices_seeded: bool = false, // its own prices started from the engine's kept calibration
     copy_landed: [2]f64 = .{ -1, -1 }, // a lone context copy's rows that landed, decayed, by short and long match (-1: unprimed)
     copy_broke: [2]f64 = .{ 0, 0 }, // and its chains that broke before their end: their ratio is a copied row's landing rate
+    copied_landed: [2]f64 = .{ -1, -1 }, // a copy window's rows that landed, decayed, by source: context, Recall (-1: unprimed)
+    copied_broke: [2]f64 = .{ 0, 0 }, // and its copies that broke before their end
     drafted: u64 = 0,
     accepted: u64 = 0,
     branch_rows: u64 = 0, // suffix-match branch rows verified (lanes/fill.zig)

zig/src/core/lanes/windows.zig

@@ -248,7 +248,8 @@ pub fn allocate(e: *Engine, plans: []Plan) !void {
             pr.* = &.{};
         } else if (p.kind == .copy) {
             fx.* = 1;
-            pr.* = try alloc.chainProbabilities(a, &.{e.cfg.copy_rate}, n);
+            const rate = if (e.cfg.own_copy_rates) copiedRate(e, p.stream, p.recalled) else e.cfg.copy_rate;
+            pr.* = try alloc.chainProbabilities(a, &.{rate}, n);
         } else {
             fx.* = 1;
             const chances = try planChances(e, p, n);
@@ -362,6 +363,31 @@ pub fn observeCopy(e: *const Engine, s: *Stream, trusted: bool, kept: u64, offer
     }
 }
 
+/// A copy window's history weighs this many rows of cfg.copy_rate's prior before the stream's own copies teach it.
+const copied_prior = 4;
+
+/// A copy window's row landing rate for this stream and source: landed rows over those plus breaks, decayed a copy.
+pub fn copiedRate(e: *const Engine, s: *Stream, recalled: bool) f64 {
+    const i = @intFromBool(recalled);
+    if (s.copied_landed[i] < 0) {
+        s.copied_landed[i] = copied_prior * e.cfg.copy_rate;
+        s.copied_broke[i] = copied_prior * (1 - e.cfg.copy_rate);
+    }
+    return s.copied_landed[i] / (s.copied_landed[i] + s.copied_broke[i]);
+}
+
+/// A verified copy window of `offered` rows, `kept` landing: its source's counters, then its own rate's history.
+pub fn observeCopied(e: *const Engine, s: *Stream, recalled: bool, kept: usize, offered: usize) void {
+    if (recalled) {
+        if (s.recall) |r| r.observe(@intCast(offered), @intCast(kept), s.rounds);
+    } else if (s.proposer) |pr| pr.observe(@intCast(offered), @intCast(kept));
+    if (!e.cfg.own_copy_rates) return;
+    _ = copiedRate(e, s, recalled);
+    const i = @intFromBool(recalled);
+    s.copied_landed[i] = copy_decay * s.copied_landed[i] + @as(f64, @floatFromInt(kept));
+    s.copied_broke[i] = copy_decay * s.copied_broke[i] + @as(f64, @floatFromInt(@intFromBool(kept < offered)));
+}
+
 /// A context copy's appended rows: each its parent's chance times the stream's copy rate; the drafts keep their own.
 fn planChances(e: *Engine, p: Plan, n: usize) ![]const f64 {
     if (!p.context_copy or p.branch_from == 0 or p.parents == null or n < p.branch_from) return nodeChances(e, p.stream, n);