GLM-5.3-Flash on CUDA keeps conversation states on disk, every rank's part (--learn-conversations)

docs/api.md

@@ -602,7 +602,9 @@ store for each model and build. That store takes the place of `--learn`'s store
 with the flag neither reads nor writes the prefix files a plain `--learn` server keeps, and they stay on disk until the
 cap below needs their room. After a restart, a crash or an eviction from RAM, the next turn reads the latest state back
 instead of reading the conversation again from its first token. Qwen3.8-27B, Nemotron, Flash Next and GLM-5.3-Flash on
-one Mac and Nemotron on CUDA keep them; other engines refuse the flag at startup.
+one Mac, and Nemotron, the Qwen3.5 hybrids, Flash Next on one rank and GLM-5.3-Flash on CUDA keep them; other engines
+refuse the flag at startup. GLM-5.3-Flash on CUDA across ranks writes every rank's part to rank 0's disk, so a state is
+read back on every rank or on none; it takes neither `--spill-gib` nor `--checkpoint-slots`.
 
 The engine stages an owned copy between rounds, and an I/O thread writes and fsyncs it. Staging has a 1 GiB byte
 permit and at most eight jobs; when either is full, the save is skipped. The copy waits for its GPU producers before

zig/src/core/rows_pool_select_test.zig

@@ -86,3 +86,115 @@ test "restore selection: pooled buffers trim only when supported and never steal
         s.release(borrowed);
     }
 }
+
+test "restore selection: a pool without trim reads into the spare storage an eviction leaves, as a keep would" {
+    var tmp = t.tmpDir(.{});
+    defer tmp.cleanup();
+    const dir = try tmp.dir.realPathFileAlloc(t.io, ".", a);
+    defer a.free(dir);
+    var prompt: [64]u32 = undefined;
+    for (&prompt, 0..) |*token, i| token.* = @intCast(100 + i);
+    const rules: pc.Rules = .{ .min_prompt = 0, .min_gap = 1, .lookahead = 1 };
+    {
+        var f: Fake = .{ .gpa = a, .at = 20, .sum = fresh(prompt[0..20]) };
+        var table = f.pooled().vtable.*;
+        table.rows = &Image.api;
+        const family: pc.Snapshots = .{ .ptr = &f, .vtable = &table };
+        const r = try rows.Rows.open(a, family, options(dir));
+        defer r.deinit();
+        var s = pc.Store.init(a, family, rules, 1000);
+        defer s.deinit();
+        s.disk = r;
+        try t.expect(s.keep(&prompt, 20, null, &.{}, &.{}));
+        while (rows.write.step(r)) {}
+    }
+    var f: Fake = .{ .gpa = a, .at = 5, .sum = fresh(prompt[0..5]) };
+    var table = f.pooled().vtable.*;
+    table.rows = &Image.api;
+    table.trim = null;
+    const family: pc.Snapshots = .{ .ptr = &f, .vtable = &table };
+    const r = try rows.Rows.open(a, family, options(dir));
+    defer r.deinit();
+    var s = pc.Store.init(a, family, rules, 240);
+    defer s.deinit();
+    try t.expect(s.keep(&prompt, 5, null, &.{}, &.{}));
+    const borrowed = s.entries.items[0];
+    borrowed.borrowers = 1;
+    prompt[0] = 999;
+    f.at = 30; // another conversation's state: its storage, once evicted, can take the read (120 >= 110)
+    f.sum = fresh(prompt[0..30]);
+    try t.expect(s.keep(&prompt, 30, null, &.{}, &.{}));
+    prompt[0] = 100;
+    s.disk = r;
+    try t.expectEqual(.disk, rows.read.resumable(&s, &prompt, &.{}, &.{}).from);
+    const restored = rows.read.recall(&s, &prompt, 64, &.{}, &.{}, borrowed).?;
+    try t.expectEqual(@as(u32, 20), restored.at);
+    try t.expectEqual(@as(u64, 1), r.counts.reads);
+    try t.expectEqual(@as(u64, 1), s.counts.evicted);
+    try t.expect(s.held + s.spare() <= s.budget);
+    try family.vtable.restore(family.ptr, null, restored.saved);
+    try t.expectEqual(fresh(&prompt), f.pass(&s, &prompt, .{ .from = restored.at }));
+    s.release(borrowed);
+}
+
+/// A disk read into new storage, whatever spare storage the pool holds.
+const Fresh = struct {
+    const api: rows.Image = .{ .parts = Image.parts, .gather = Image.gather, .blank = blank, .scatter = Image.scatter };
+    fn blank(ptr: *anyopaque, at: u32) !pc.Saved {
+        const f: *Fake = @ptrCast(@alignCast(ptr));
+        const previous = f.at;
+        f.at = at;
+        defer f.at = previous;
+        return f.snapshots().vtable.save(ptr, null, at);
+    }
+};
+
+test "restore selection: a pool that also trims reads a disk state only within its budget, whatever it reuses" {
+    var tmp = t.tmpDir(.{});
+    defer tmp.cleanup();
+    const dir = try tmp.dir.realPathFileAlloc(t.io, ".", a);
+    defer a.free(dir);
+    var prompt: [64]u32 = undefined;
+    for (&prompt, 0..) |*token, i| token.* = @intCast(100 + i);
+    const rules: pc.Rules = .{ .min_prompt = 0, .min_gap = 1, .lookahead = 1 };
+    {
+        var f: Fake = .{ .gpa = a, .at = 20, .sum = fresh(prompt[0..20]) };
+        var table = f.pooled().vtable.*;
+        table.rows = &Image.api;
+        const family: pc.Snapshots = .{ .ptr = &f, .vtable = &table };
+        const r = try rows.Rows.open(a, family, options(dir));
+        defer r.deinit();
+        var s = pc.Store.init(a, family, rules, 1000);
+        defer s.deinit();
+        s.disk = r;
+        try t.expect(s.keep(&prompt, 20, null, &.{}, &.{}));
+        while (rows.write.step(r)) {}
+    }
+    var f: Fake = .{ .gpa = a, .at = 5, .sum = fresh(prompt[0..5]) };
+    var table = f.pooled().vtable.*;
+    table.rows = &Fresh.api;
+    try t.expect(table.reuses != null and table.trim != null);
+    const family: pc.Snapshots = .{ .ptr = &f, .vtable = &table };
+    const r = try rows.Rows.open(a, family, options(dir));
+    defer r.deinit();
+    var s = pc.Store.init(a, family, rules, 240);
+    defer s.deinit();
+    try t.expect(s.keep(&prompt, 5, null, &.{}, &.{}));
+    const borrowed = s.entries.items[0];
+    borrowed.borrowers = 1;
+    prompt[0] = 999;
+    f.at = 30; // another conversation's state: once evicted, its spare storage is one the family says takes the read
+    f.sum = fresh(prompt[0..30]);
+    try t.expect(s.keep(&prompt, 30, null, &.{}, &.{}));
+    prompt[0] = 100;
+    s.disk = r;
+    try t.expectEqual(.disk, rows.read.resumable(&s, &prompt, &.{}, &.{}).from);
+    const restored = rows.read.recall(&s, &prompt, 64, &.{}, &.{}, borrowed).?;
+    try t.expectEqual(@as(u32, 20), restored.at);
+    try t.expectEqual(@as(u64, 1), r.counts.reads);
+    try t.expectEqual(@as(u64, 1), s.counts.evicted);
+    try t.expect(s.held + s.spare() <= s.budget); // the spare storage was trimmed for the new storage, not left over it
+    try family.vtable.restore(family.ptr, null, restored.saved);
+    try t.expectEqual(fresh(&prompt), f.pass(&s, &prompt, .{ .from = restored.at }));
+    s.release(borrowed);
+}

zig/src/core/rows_read.zig

@@ -91,9 +91,10 @@ pub fn recall(s: *pc.Store, prompt: []const u32, born: u32, starts: []const u32,
     const name = r.states.items[chosen].name;
     const bytes = s.family.vtable.bytes(s.family.ptr, r.states.items[chosen].at);
     if (bytes > s.budget or s.most == 0) return have;
-    if (!ready(s, bytes)) present(r, chosen) catch return have; // nothing goes to make room for a state whose file is gone
+    const at = r.states.items[chosen].at;
+    if (!ready(s, at, bytes)) present(r, chosen) catch return have; // nothing goes to make room for a state whose file is gone
     if (r.spill != .off) r.touch(chosen); // the newest: the spills that make room for it push older states off first
-    while (!ready(s, bytes)) {
+    while (!ready(s, at, bytes)) {
         if (s.family.vtable.trim) |trim| {
             const spare = s.spare();
             trim(s.family.ptr, s.budget -| s.held -| bytes);
@@ -128,8 +129,11 @@ pub fn recall(s: *pc.Store, prompt: []const u32, born: u32, starts: []const u32,
     return e;
 }
 
-fn ready(s: *const pc.Store, bytes: u64) bool {
+fn ready(s: *const pc.Store, at: u32, bytes: u64) bool {
     if (s.held > s.budget or s.entries.items.len >= s.most) return false;
+    const f = s.family.vtable;
+    // a pool without trim reads into spare storage that takes the state, as a keep would
+    if (f.trim == null) if (f.reuses) |reuses| if (reuses(s.family.ptr, at)) return true;
     const room = s.budget - s.held;
     const spare = s.spare();
     return spare <= room and bytes <= room - spare;

zig/src/core/rows_select.zig

@@ -39,12 +39,15 @@ fn capacity(s: *const pc.Store, have: ?*pc.Entry) ?u64 {
     var protected: usize = 0;
     const f = s.family.vtable;
     for (s.entries.items) |e| {
-        if (e == have or e.borrowers > 0) protected += 1 else if (f.spare == null or f.trim != null) {
+        if (e == have or e.borrowers > 0) protected += 1 else if (f.spare == null or f.trim != null or f.reuses != null) {
             reclaimable +|= e.bytes;
         }
     }
+    if (protected >= s.most) return null;
+    // a pool that reuses spare storage without trimming it (GLM's snapshot pages): a read takes spare or evicted pages
+    if (f.reuses != null and f.trim == null) return s.spare() +| reclaimable;
     const fixed = (s.held -| reclaimable) +| if (f.trim == null) s.spare() else 0;
-    if (protected >= s.most or fixed > s.budget) return null;
+    if (fixed > s.budget) return null;
     return s.budget - fixed;
 }
 

zig/src/core/rows_test.zig

@@ -247,16 +247,18 @@ fn gridRequest(a: Allocator, s: *pc.Store, f: *Fake, p: []const u32, history: u3
     return .{ .from = plan.from, .sum = f.sum };
 }
 
-test "a grid family's requests name no chunk starts: its states go to disk on the grid and resume after a restart" {
+/// A grid family's states through one writer (`sync`: the test's thread, else the I/O worker), read after a restart.
+fn gridRoundTrip(sync: bool, tag: []const u8) !void {
     const gpa = std.testing.allocator;
     var arena: std.heap.ArenaAllocator = .init(gpa);
     defer arena.deinit();
     const a = arena.allocator();
     var buf: [128]u8 = undefined;
-    const dir = root(&buf, "grid");
+    const dir = root(&buf, tag);
     defer rmTree(dir);
     const grid: pc.Rules = .{ .lookahead = 1, .min_prompt = 0, .min_gap = 1, .planned = true, .grid = 512 };
     var o = options(dir);
+    o.sync_writes = sync;
     o.planned = true;
     o.grid = 512;
     const p = try prompt(a, 0, 5, 1300); // the engine cuts its chunks on the grid: 512 and 1024
@@ -268,7 +270,7 @@ test "a grid family's requests name no chunk starts: its states go to disk on th
         defer s.deinit();
         s.disk = r;
         _ = try gridRequest(a, &s, &f, p, 1200);
-        drain(r);
+        _ = rows.write.drain(r, rows.write.nowNs() + 3 * std.time.ns_per_s); // the worker publishes the states
         try std.testing.expect(r.states.items.len > 0);
         for (r.states.items) |st| {
             try std.testing.expectEqual(@as(u32, 0), st.at % 512);
@@ -289,6 +291,14 @@ test "a grid family's requests name no chunk starts: its states go to disk on th
     try std.testing.expectEqual(@as(u64, 1), r.counts.reads);
 }
 
+test "a grid family's requests name no chunk starts: its states go to disk on the grid and resume after a restart" {
+    try gridRoundTrip(true, "grid");
+}
+
+test "a grid family's chunk starts reach the I/O worker's writes as well, the writer GLM's store uses" {
+    try gridRoundTrip(false, "grid-worker");
+}
+
 test "an entry evicted before its write leaves the queue; once written, an eviction costs nothing" {
     const gpa = std.testing.allocator;
     var arena: std.heap.ArenaAllocator = .init(gpa);

zig/src/families/glm/cuda.zig

@@ -37,6 +37,8 @@ pub const follow = @import("cuda_follow.zig");
 pub const lanes = @import("cuda_lanes.zig");
 pub const native = @import("cuda_native.zig");
 pub const snapshots = @import("cuda_snapshots.zig");
+pub const disk_image = @import("cuda_disk_image.zig");
+pub const disk_ranks = @import("cuda_disk_ranks.zig");
 
 test {
     _ = @import("cuda_tile.zig");
@@ -47,6 +49,7 @@ test {
     _ = @import("cuda_images_interleave_test.zig");
     _ = @import("cuda_pages_test.zig");
     _ = @import("cuda_ranks_test.zig");
+    _ = @import("cuda_disk_image.zig");
     _ = experts;
     _ = exl3;
     _ = vision;