diff --git a/CHANGELOG.md b/CHANGELOG.md
index 604325c54e..e00644fc2d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,8 @@ Each release's page on GitHub has its notes and measurements. See [the 1.0.0 rel
 
 ## Unreleased
 
+- Conversation states saved to disk: a save that fits the staging permit (1 GiB and eight jobs) on its own now waits for a write in flight to finish instead of being skipped, and a block still being written is charged and copied once, however many saves name it (#1548, for #1538). A save larger than the permit is still skipped. So is a waiting save whose state leaves RAM, unless `--spill-gib` copies it then; that copy, like any made at an eviction, is skipped when the permit is full. Thanks to @wojo.
+
 - Qwen3.8-27B on CUDA: ModelOpt checkpoints (NVFP4 and FP8), such as `nvidia/Qwen3.8-27B-NVFP4`, now serve on cards from sm_80 to sm_89, such as a Jetson AGX Orin, where decode stopped at the first projection with `MissingLanePart`. The engine reserves the projections' partial sums for windows under 256 rows (about 21 MB for the 27B, and only on those cards) and reduces them without clusters, as Nemotron's ModelOpt path does; a card older than sm_80 is refused at load with the reason. It has not run on an sm_80 to sm_89 card yet. On sm_90 and newer, including the GB10, nothing changes (#1470, for #1452). Thanks to @javierbits and @valacroix.
 
 - More CUDA kernels take their `ldmatrix` loads and f16 `mma` from one home each: the prompt attention and Qwen 3.5 tensor-core attention take `qmm_frag`'s transposed load, the EXL3 grouped-expert, dense prompt and prompt kernels take its `ldmatrix4` and `ldmatrix2`, and the grouped EXL3 experts take `exl3_decode.cuh`'s f16 `mma16816`. All 1,935 kernel functions in the 14 CUDA images this reaches compile to the same code as before at sm_80, sm_86, sm_89, sm_120 and sm_121 (#1591). Thanks to @jschmied.
diff --git a/docs/api.md b/docs/api.md
index 419e3c8d54..681efe4d3c 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -604,9 +604,12 @@ cap below needs their room. After a restart, a crash or an eviction from RAM, th
 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.
 
-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
-RAM can be reused. A state becomes available for reuse only after all its files have been written and synced.
+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. A save that finds either full but would fit alone waits in the queue until a write finishes; one
+that cannot fit even then is skipped. So is a waiting save whose state leaves RAM, unless `--spill-gib` copies it then
+(below); that copy, like any made at an eviction, is skipped when the permit is full. A block still being written is
+charged and copied once, however many saves name it. The copy waits for its GPU producers before RAM can be reused. A
+state becomes available for reuse only after all its files have been written and synced.
 A planned stop (SIGTERM or SIGINT) drains writes for up to 20 seconds, then cancels the remainder; a blocked syscall
 can finish later using only its private temporary files and staged bytes. A killed server loses unfinished saves.
 Rows a stored state already holds for the same tokens are written once, so ten
@@ -656,11 +659,11 @@ publication.
 On CUDA, Nemotron, the Qwen3.5 hybrids and Flash Next take `--spill-gib` and `--checkpoint-slots` with the same store
 and cap as a Mac. At an eviction the engine thread stages a copy of the state into host memory, waiting for the GPU
 work that produced it, then releases its RAM entry and goes on; the I/O thread writes and fsyncs the copy while the
-engine decodes. Staging holds at most 1 GiB and eight jobs: a state whose new bytes (rows not already on disk) do not
-fit what is free is not saved, so one past 1 GiB never is, and a later request for it prefills. When the engine closes,
-at a planned stop (SIGTERM or SIGINT) or an idle unload, it saves the states still in RAM, newest first, within the same
-disk cap and 20-second stop budget. While other streams run, a queued request whose state is on disk has it read ahead
-on an I/O thread, as described above.
+engine decodes. Staging holds at most 1 GiB and eight jobs: a state whose new bytes (rows neither on disk nor being
+written) do not fit what is free is not saved, so one past 1 GiB never is, and a later request for it prefills. When the
+engine closes, at a planned stop (SIGTERM or SIGINT) or an idle unload, it saves the states still in RAM, newest first,
+within the same disk cap and 20-second stop budget. While other streams run, a queued request whose state is on disk has
+it read ahead on an I/O thread, as described above.
 
 Three routes let a client such as the Mac app ask about a conversation. Each takes the body of a chat completion
 (`messages`, `tools`, `chat_template_kwargs`), renders it as that request would be rendered, and answers on the engine's
diff --git a/zig/src/core/rows_async_fixture.zig b/zig/src/core/rows_async_fixture.zig
index ae97eae302..a202a0187c 100644
--- a/zig/src/core/rows_async_fixture.zig
+++ b/zig/src/core/rows_async_fixture.zig
@@ -49,6 +49,15 @@ pub const Fixture = struct {
             rows.write.io.nap();
         }
     }
+
+    /// The engine's idle turns until `states` are on disk: a queued save is staged whenever `ready` says so.
+    pub fn idle(x: *Fixture, states: usize) !void {
+        const end = rows.write.nowNs() + 3 * std.time.ns_per_s;
+        while (x.r.states.items.len < states) {
+            if (rows.write.nowNs() >= end) return error.WriterTimedOut;
+            if (rows.write.ready(x.r)) _ = rows.write.step(x.r) else rows.write.io.nap();
+        }
+    }
 };
 
 pub fn prompt(salt: u32) [1100]u32 {
@@ -88,7 +97,7 @@ pub const Gate = struct {
     entered: std.atomic.Value(bool) = .init(false),
     release: std.atomic.Value(bool) = .init(false),
     left: std.atomic.Value(bool) = .init(false),
-    mode: enum { good, full, short, sync_failure } = .good,
+    mode: enum { good, full, short, sync_failure, first_fails } = .good,
     calls: usize = 0,
     sync_entered: std.atomic.Value(bool) = .init(false),
     sync_release: std.atomic.Value(bool) = .init(true),
@@ -103,7 +112,7 @@ pub const Gate = struct {
         while (!g.release.load(.acquire)) rows.write.io.nap();
         defer g.left.store(true, .release);
         g.calls += 1;
-        if (g.mode == .full) return error.NoSpaceLeft;
+        if (g.mode == .full or (g.mode == .first_fails and g.calls == 1)) return error.NoSpaceLeft;
         if (g.mode == .short and g.calls > 1) return 0;
         const n = platform.c.write(fd, bytes.ptr, if (g.mode == .short) @min(bytes.len, 7) else bytes.len);
         if (n < 0) return error.RowsWrite;
diff --git a/zig/src/core/rows_async_test.zig b/zig/src/core/rows_async_test.zig
index dc4f2b7e5c..cf3d26992b 100644
--- a/zig/src/core/rows_async_test.zig
+++ b/zig/src/core/rows_async_test.zig
@@ -347,7 +347,7 @@ test "spill I/O: killing a child inside a partial write leaves no published stat
     try std.testing.expectEqual(@as(usize, 0), fixture.files(reopened.dir, ".part"));
 }
 
-test "spill I/O: a full byte permit skips another save without waiting for the first job" {
+test "spill I/O: a full byte permit holds another save that fits it alone until the first job is written" {
     const x = try Fixture.open("full-permit", .finish, false, 1 << 20);
     defer x.close();
     var hold: Hold = .{};
@@ -362,13 +362,18 @@ test "spill I/O: a full byte permit skips another save without waiting for the f
     try x.keep(&second, 600);
     try std.testing.expectEqual(@as(usize, 1), x.r.worker.?.pending());
     try std.testing.expectEqual(used, x.r.worker.?.bytes());
-    try std.testing.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    try std.testing.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(u64, 0), x.r.failures.count);
+    try std.testing.expect(rows.write.pending(x.r) and !rows.write.ready(x.r) and !rows.write.step(x.r));
     hold.release.store(true, .release);
-    try x.settle();
-    try std.testing.expectEqual(@as(usize, 1), x.r.states.items.len);
+    try x.idle(2);
+    try std.testing.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(u64, 2), x.r.counts.writes);
 }
 
-test "spill I/O: the job permit bounds metadata when many small saves queue behind a blocked write" {
+test "spill I/O: the job permit holds a ninth small save behind eight blocked writes until one is written" {
     const x = try Fixture.open("job-permit", .finish, false, 1 << 30);
     defer x.close();
     var hold: Hold = .{};
@@ -380,10 +385,11 @@ test "spill I/O: the job permit bounds metadata when many small saves queue behi
         try x.keep(p, 600);
     }
     try std.testing.expectEqual(@as(usize, 8), x.r.worker.?.pending());
-    try std.testing.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try std.testing.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try std.testing.expectEqual(@as(u64, 0), x.r.counts.skipped);
     hold.release.store(true, .release);
-    try x.settle();
-    try std.testing.expectEqual(@as(usize, 8), x.r.states.items.len);
+    try x.idle(9);
+    try std.testing.expectEqual(@as(usize, 0), x.r.queue.items.len);
 }
 
 test "spill I/O: shutdown keeps a completed job's own filesystem error" {
diff --git a/zig/src/core/rows_fake.zig b/zig/src/core/rows_fake.zig
index 199e65fa90..49eb559e8e 100644
--- a/zig/src/core/rows_fake.zig
+++ b/zig/src/core/rows_fake.zig
@@ -33,6 +33,7 @@ pub const Fake = struct {
     slots: ?usize = null, // a family whose states each take one of so many slots (Nemotron CUDA's): past them, none
     gathers: ?usize = null, // copies out of kept states that succeed before every later one fails (null: all succeed)
     scatters: ?Fault = null, // copies into new states that succeed before every later one fails (null: all succeed)
+    gathered: usize = 0, // copies out of kept states made so far
 
     pub fn snapshots(f: *Fake) pc.Snapshots {
         return .{ .ptr = f, .vtable = &.{ .bytes = bytesFn, .save = saveFn, .restore = restoreFn, .drop = dropFn, .rows = &.{ .parts = partsFn, .gather = gatherFn, .blank = blankFn, .scatter = scatterFn } } };
@@ -134,6 +135,7 @@ pub const Fake = struct {
     }
     fn gatherFn(ptr: *anyopaque, saved: pc.Saved, ranges: []const image.Range, out: []u8) anyerror!void {
         const f = of(ptr);
+        f.gathered += 1;
         if (f.gathers) |*n| if (n.* == 0) return error.GatherFailed else {
             n.* -= 1;
         };
diff --git a/zig/src/core/rows_io.zig b/zig/src/core/rows_io.zig
index 24c1997e9a..17db37d925 100644
--- a/zig/src/core/rows_io.zig
+++ b/zig/src/core/rows_io.zig
@@ -36,6 +36,11 @@ pub const Job = struct {
     taken: bool = false,
     failure: ?anyerror = null,
 
+    /// The block files it writes: every file but the last, which is the state's own.
+    pub fn blocks(j: *const Job) []File {
+        return j.files[0..j.files.len -| 1];
+    }
+
     pub fn destroy(j: *Job) void {
         for (j.files) |*f| if (f.writer) |*w| w.abort();
         j.arena.deinit();
@@ -43,6 +48,9 @@ pub const Job = struct {
     }
 };
 
+/// The stems of the blocks jobs in flight are writing.
+pub const Promised = std.AutoHashMapUnmanaged(chain.Stem, void);
+
 pub const Worker = struct {
     dir: [:0]u8,
     io: file.IO,
@@ -78,6 +86,15 @@ pub const Worker = struct {
         return false;
     }
 
+    /// Whether the jobs in flight alone keep `amount` out: it fits an empty permit, and one is staged to be taken.
+    pub fn busy(w: *Worker, amount: u64) bool {
+        w.lock();
+        defer w.mutex.unlock();
+        if (w.stopped.load(.acquire) or amount > w.limit) return false;
+        for (w.slots) |s| if (s != null) return true;
+        return false;
+    }
+
     pub fn submit(w: *Worker, j: *Job) void {
         w.lock();
         defer w.mutex.unlock();
@@ -90,16 +107,32 @@ pub const Worker = struct {
         unreachable;
     }
 
-    /// Completed jobs stay owned here until the engine takes them, retaining their permits meanwhile.
+    /// Completed jobs stay owned here until the engine takes them, oldest first, retaining their permits meanwhile.
     pub fn take(w: *Worker) ?*Job {
         w.lock();
         defer w.mutex.unlock();
-        for (&w.slots) |*s| if (s.*) |j| if (j.done.load(.acquire)) {
-            s.* = null;
-            w.used -= j.staged;
-            return j;
+        const s = w.oldest(finished) orelse return null;
+        const j = s.*.?;
+        s.* = null;
+        w.used -= j.staged;
+        return j;
+    }
+
+    /// The slot of the earliest staged job `ok` accepts: ids rise as jobs are staged, whichever slot each took.
+    fn oldest(w: *Worker, comptime ok: fn (*const Job) bool) ?*?*Job {
+        var out: ?*?*Job = null;
+        for (&w.slots) |*s| if (s.*) |j| if (ok(j) and (out == null or j.id < out.?.*.?.id)) {
+            out = s;
         };
-        return null;
+        return out;
+    }
+
+    fn finished(j: *const Job) bool {
+        return j.done.load(.acquire);
+    }
+
+    fn untaken(j: *const Job) bool {
+        return !j.taken;
     }
 
     pub fn names(w: *Worker, name: chain.Name) bool {
@@ -112,9 +145,28 @@ pub const Worker = struct {
     pub fn wanted(w: *Worker, stem: chain.Stem) bool {
         w.lock();
         defer w.mutex.unlock();
-        for (w.slots) |s| if (s) |j| for (j.wants) |want| {
-            if (std.mem.eql(u8, want.name[0..16], &stem)) return true;
+        for (w.slots) |s| if (s) |j| if (lists(j, stem)) return true;
+        return false;
+    }
+
+    /// Replaces `set` with the blocks jobs that can still publish write; a job that skipped a block writes none of it.
+    pub fn promised(w: *Worker, gpa: std.mem.Allocator, set: *Promised) !void {
+        set.clearRetainingCapacity();
+        w.lock();
+        defer w.mutex.unlock();
+        const generation = w.generation.load(.acquire);
+        for (w.slots) |s| if (s) |j| if (j.generation == generation and !lost(j)) for (j.blocks()) |f| {
+            try set.put(gpa, chain.stem(f.trailer.name), {});
         };
+    }
+
+    /// Whether the job finished with an error (the worker sets `failure` before it publishes `done`).
+    fn lost(j: *const Job) bool {
+        return j.done.load(.acquire) and j.failure != null;
+    }
+
+    fn lists(j: *const Job, stem: chain.Stem) bool {
+        for (j.wants) |want| if (std.mem.eql(u8, want.name[0..16], &stem)) return true;
         return false;
     }
 
@@ -167,11 +219,9 @@ pub const Worker = struct {
     fn next(w: *Worker) ?*Job {
         w.lock();
         defer w.mutex.unlock();
-        for (w.slots) |s| if (s) |j| if (!j.taken) {
-            j.taken = true;
-            return j;
-        };
-        return null;
+        const j = (w.oldest(untaken) orelse return null).*.?;
+        j.taken = true;
+        return j;
     }
 
     fn run(w: *Worker) void {
@@ -274,3 +324,59 @@ fn compare(fd: c_int, bytes: []const u8, offset: *u64) !void {
         offset.* += got;
     }
 }
+
+const testing = std.testing;
+
+fn named(b: u8) chain.Name {
+    return @splat(b);
+}
+
+fn fileOf(b: u8, kind: file.Kind) File {
+    return .{ .trailer = .{ .kind = @backingInt(kind), .payload = 0, .payload_hash = 0, .name = named(b) }, .payload = &.{} };
+}
+
+/// A job as the worker's slots see it: an id, the blocks its state names and the files it writes (the last is its own).
+fn job(id: u64, wants: []Want, files: []File) Job {
+    return .{ .arena = undefined, .id = id, .generation = 1, .name = undefined, .at = 0, .shared = false, .tokens = &.{}, .spans = &.{}, .starts = &.{}, .wants = wants, .files = files, .bytes = 0, .staged = 1 };
+}
+
+fn threadless() Worker {
+    return .{ .dir = undefined, .io = .{}, .limit = 1 << 20, .thread = undefined };
+}
+
+test "rows io: jobs are run and taken in the order they were staged, whatever slot each holds" {
+    var w = threadless();
+    var jobs = [_]Job{ job(1, &.{}, &.{}), job(2, &.{}, &.{}), job(3, &.{}, &.{}), job(4, &.{}, &.{}) };
+    for (jobs[0..3]) |*j| w.submit(j);
+    try testing.expectEqual(&jobs[0], w.next().?);
+    jobs[0].done.store(true, .release);
+    try testing.expectEqual(&jobs[0], w.take().?);
+    w.submit(&jobs[3]); // the slot job 1 left: the lowest, though the newest
+    for (jobs[1..]) |*j| try testing.expectEqual(j, w.next().?);
+    try testing.expectEqual(@as(?*Job, null), w.next());
+    for (jobs[1..]) |*j| j.done.store(true, .release);
+    for (jobs[1..]) |*j| try testing.expectEqual(j, w.take().?);
+    try testing.expectEqual(@as(?*Job, null), w.take());
+}
+
+test "rows io: only the blocks a job that can still publish writes are promised" {
+    var w = threadless();
+    var wants = [_]Want{ .{ .name = named(1), .window = 0, .k = 0, .bytes = 1 }, .{ .name = named(2), .window = 0, .k = 1, .bytes = 1 } };
+    var writes = [_]File{ fileOf(1, .block), fileOf(2, .block), fileOf(8, .state) };
+    var skips = [_]File{ fileOf(3, .block), fileOf(9, .state) }; // it names blocks 1 and 2 and leaves them to the first job
+    var jobs = [_]Job{ job(1, &wants, &writes), job(2, &wants, &skips) };
+    for (&jobs) |*j| w.submit(j);
+    var set: Promised = .empty;
+    defer set.deinit(testing.allocator);
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 3), set.count());
+    for ([_]u8{ 1, 2, 3 }) |b| try testing.expect(set.contains(chain.stem(named(b))));
+    jobs[0].failure = error.NoSpaceLeft;
+    jobs[0].done.store(true, .release);
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 1), set.count());
+    try testing.expect(set.contains(chain.stem(named(3))));
+    w.fence();
+    try w.promised(testing.allocator, &set);
+    try testing.expectEqual(@as(u32, 0), set.count());
+}
diff --git a/zig/src/core/rows_stage_test.zig b/zig/src/core/rows_stage_test.zig
new file mode 100644
index 0000000000..5485058b70
--- /dev/null
+++ b/zig/src/core/rows_stage_test.zig
@@ -0,0 +1,308 @@
+//! Staging: a block still being written is charged and gathered once, and a save that fits the permit waits for it.
+const std = @import("std");
+const rows = @import("rows_store.zig");
+const fixture = @import("rows_async_fixture.zig");
+const Fixture = fixture.Fixture;
+const Hold = fixture.Gate;
+const t = std.testing;
+
+/// A prompt that shares its first 513 tokens with `fixture.prompt(17)`, so their first two blocks are the same files.
+fn sibling(salt: u32) [1100]u32 {
+    var p = fixture.prompt(salt);
+    const base = fixture.prompt(17);
+    @memcpy(p[0..513], base[0..513]);
+    return p;
+}
+
+fn pair() struct { first: [1100]u32, second: [1100]u32 } {
+    return .{ .first = fixture.prompt(17), .second = sibling(18) };
+}
+
+test "rows stage: a block still being written is charged and gathered once" {
+    const x = try Fixture.open("stage-share", .finish, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const p = pair();
+    try x.keep(&p.first, 900);
+    try Hold.wait(&hold.entered);
+    try x.keep(&p.second, 900);
+    try t.expectEqual(@as(usize, 2), x.r.worker.?.pending());
+    const charged = x.r.worker.?.reserved();
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 2), x.r.states.items.len);
+    try t.expectEqual(x.r.held(), charged);
+    try t.expectEqual(x.r.blocks.count() + 2, x.f.gathered);
+    const saved = try rows.read.read(x.r, 1);
+    defer x.f.snapshots().vtable.drop(&x.f, saved);
+    try x.f.snapshots().vtable.restore(&x.f, null, saved);
+    try t.expect(x.f.restored_ok);
+}
+
+test "rows stage: a state whose earlier job failed to write its block is not published" {
+    const x = try Fixture.open("stage-lost", .finish, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{ .mode = .first_fails };
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const p = pair();
+    try x.keep(&p.first, 900);
+    try Hold.wait(&hold.entered);
+    try x.keep(&p.second, 900);
+    try t.expectEqual(@as(usize, 2), x.r.worker.?.pending());
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 0), x.r.states.items.len);
+    try t.expectEqual(@as(usize, 0), fixture.files(x.r.dir, ".st"));
+    try t.expectEqual(@as(usize, 0), fixture.files(x.r.dir, ".blk"));
+    try t.expectEqual(@as(usize, 0), fixture.files(x.r.dir, ".part"));
+    try t.expectEqual(@as(u64, 2), x.r.failures.count);
+    try t.expectEqual(error.NoSpaceLeft, x.r.failures.records[0].?.err);
+    try t.expectEqual(error.MissingBlock, x.r.failures.records[1].?.err);
+    try t.expectEqual(@as(u64, 2), x.r.counts.failed);
+    try t.expectEqual(@as(u64, 0), x.r.counts.writes);
+    try t.expectEqual(@as(u64, 0), x.r.admission.reserved);
+}
+
+test "rows stage: a block its writer failed to write is written by the next save, though a job that skipped it is in flight" {
+    const x = try Fixture.open("stage-chain", .finish, false, 1 << 20);
+    defer x.close();
+    var failing: Hold = .{ .mode = .first_fails };
+    defer failing.release.store(true, .release);
+    var held: Hold = .{};
+    defer held.release.store(true, .release);
+    x.r.file_io = failing.io();
+    const first = fixture.prompt(17);
+    const second = sibling(18);
+    const third = sibling(19);
+    try x.keep(&first, 900);
+    try Hold.wait(&failing.entered);
+    try x.keep(&second, 900);
+    x.r.worker.?.io = held.io(); // what the worker writes after the first job waits on `held`
+    failing.release.store(true, .release);
+    try Hold.wait(&held.entered);
+    rows.write.reap(x.r);
+    x.r.admission.retry_at = 0; // the back-off the first failure set has run out
+    try t.expectEqual(@as(usize, 1), x.r.worker.?.pending());
+    try x.keep(&third, 900);
+    try t.expectEqual(@as(usize, 2), x.r.worker.?.pending());
+    held.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 1), x.r.states.items.len);
+    try t.expectEqual(error.NoSpaceLeft, x.r.failures.records[0].?.err);
+    try t.expectEqual(error.MissingBlock, x.r.failures.records[1].?.err);
+    try t.expectEqual(@as(u64, 2), x.r.counts.failed);
+    try t.expectEqual(@as(usize, 0), fixture.files(x.r.dir, ".part"));
+    const saved = try rows.read.read(x.r, 0);
+    defer x.f.snapshots().vtable.drop(&x.f, saved);
+    try x.f.snapshots().vtable.restore(&x.f, null, saved);
+    try t.expect(x.f.restored_ok);
+}
+
+test "rows stage: a job a fence cancelled promises no block, so the next save writes its own" {
+    const x = try Fixture.open("stage-fence", .finish, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const p = pair();
+    try x.keep(&p.first, 900);
+    try Hold.wait(&hold.entered);
+    rows.write.fence(x.r);
+    try x.keep(&p.second, 900);
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 1), x.r.states.items.len);
+    try t.expectEqual(@as(u64, 1), x.r.counts.failed);
+    try t.expectEqual(error.Cancelled, x.r.failures.records[0].?.err);
+    const saved = try rows.read.read(x.r, 0);
+    defer x.f.snapshots().vtable.drop(&x.f, saved);
+    try x.f.snapshots().vtable.restore(&x.f, null, saved);
+    try t.expect(x.f.restored_ok);
+}
+
+test "rows stage: a save larger than the permit alone is skipped, not held" {
+    const x = try Fixture.open("stage-big", .finish, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(22);
+    try x.keep(&first, 600);
+    try Hold.wait(&hold.entered);
+    x.r.worker.?.limit = x.r.worker.?.bytes();
+    const second = fixture.prompt(23);
+    try x.keep(&second, 1000);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try t.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try t.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    try t.expectEqual(@as(usize, 1), x.r.worker.?.pending());
+    try t.expect(!rows.write.ready(x.r));
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 1), x.r.states.items.len);
+}
+
+test "rows stage: an eviction cannot wait, so a full permit skips the copy" {
+    const x = try Fixture.open("stage-evict", .evicted, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(24);
+    try x.keep(&first, 600);
+    x.s.remove(0);
+    try Hold.wait(&hold.entered);
+    x.r.worker.?.limit = x.r.worker.?.bytes();
+    const second = fixture.prompt(25);
+    try x.keep(&second, 600);
+    x.s.remove(0);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try t.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try t.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 1), x.r.states.items.len);
+}
+
+test "rows stage: a newer save that fits beside the jobs in flight is staged ahead of one that waits" {
+    const x = try Fixture.open("stage-ahead", .finish, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(26);
+    try x.keep(&first, 600);
+    try Hold.wait(&hold.entered);
+    const used = x.r.worker.?.bytes();
+    x.r.worker.?.limit = 2 * used - 1; // room beside the first job for less than another like it
+    const second = fixture.prompt(27);
+    try x.keep(&second, 600);
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try t.expect(!rows.write.ready(x.r));
+    const third = fixture.prompt(28);
+    try x.keep(&third, 300);
+    try t.expectEqual(@as(usize, 2), x.r.worker.?.pending());
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try t.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    hold.release.store(true, .release);
+    try x.idle(3);
+    try t.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    try t.expectEqual(@as(usize, 0), x.r.queue.items.len);
+}
+
+test "rows stage: a close writes the saves that waited for room" {
+    for ([_]rows.Spill{ .off, .finish }) |spill| {
+        const x = try Fixture.open("stage-close", spill, false, 1 << 20);
+        defer x.close();
+        var hold: Hold = .{};
+        defer hold.release.store(true, .release);
+        x.r.file_io = hold.io();
+        const first = fixture.prompt(29);
+        try x.keep(&first, 600);
+        try Hold.wait(&hold.entered);
+        x.r.worker.?.limit = x.r.worker.?.bytes();
+        const second = fixture.prompt(30);
+        try x.keep(&second, 600);
+        try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+        hold.release.store(true, .release);
+        try t.expectEqual(@as(usize, 0), rows.write.drain(x.r, rows.write.nowNs() + 3 * std.time.ns_per_s));
+        try t.expectEqual(@as(usize, 2), x.r.states.items.len);
+        try t.expectEqual(@as(u64, 0), x.r.counts.skipped);
+    }
+}
+
+test "rows stage: a save that waited for room and lost its entry is counted as skipped" {
+    const x = try Fixture.open("stage-lost-entry", .off, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(31);
+    try x.keep(&first, 600);
+    try Hold.wait(&hold.entered);
+    x.r.worker.?.limit = x.r.worker.?.bytes();
+    const second = fixture.prompt(32);
+    try x.keep(&second, 600);
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    for (x.s.entries.items, 0..) |e, i| if (e == x.r.waiting.?) {
+        x.s.remove(i);
+        break;
+    };
+    try t.expect(x.r.waiting == null);
+    try t.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try t.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expectEqual(@as(usize, 1), x.r.states.items.len);
+}
+
+test "rows stage: losing the older of two waiting saves counts it as skipped" {
+    const x = try Fixture.open("stage-two-waiters", .off, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(41);
+    try x.keep(&first, 600);
+    try Hold.wait(&hold.entered);
+    x.r.worker.?.limit = x.r.worker.?.bytes();
+    const second = fixture.prompt(42);
+    try x.keep(&second, 600);
+    const older = x.r.waiting.?;
+    const third = fixture.prompt(43);
+    try x.keep(&third, 600);
+    try t.expectEqual(@as(usize, 2), x.r.queue.items.len);
+    const newer = x.r.waiting.?;
+    try t.expect(newer != older);
+    for (x.s.entries.items, 0..) |e, i| if (e == older) {
+        x.s.remove(i);
+        break;
+    };
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try t.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    try t.expect(x.r.waiting == newer);
+    try t.expect(x.r.queue.items[0].entry == newer and x.r.queue.items[0].waited);
+    hold.release.store(true, .release);
+    try x.idle(2);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+}
+
+test "rows stage: a save that waited and loses its entry after a reap still counts, under an id no job holds" {
+    const x = try Fixture.open("stage-lost-after-reap", .off, false, 1 << 20);
+    defer x.close();
+    var hold: Hold = .{};
+    defer hold.release.store(true, .release);
+    x.r.file_io = hold.io();
+    const first = fixture.prompt(51);
+    try x.keep(&first, 600);
+    try Hold.wait(&hold.entered);
+    const limit = x.r.worker.?.limit;
+    x.r.worker.?.limit = x.r.worker.?.bytes();
+    const second = fixture.prompt(52);
+    try x.keep(&second, 600);
+    const lost = x.r.waiting.?;
+    hold.release.store(true, .release);
+    try x.settle();
+    try t.expect(x.r.waiting == null);
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    x.r.worker.?.limit = limit;
+    const third = fixture.prompt(53);
+    try x.keep(&third, 600);
+    const staged_as = x.r.next_job;
+    try t.expectEqual(@as(usize, 1), x.r.queue.items.len);
+    for (x.s.entries.items, 0..) |e, i| if (e == lost) {
+        x.s.remove(i);
+        break;
+    };
+    try t.expectEqual(@as(usize, 0), x.r.queue.items.len);
+    try t.expectEqual(@as(u64, 1), x.r.counts.skipped);
+    try t.expectEqual(error.StagingFull, x.r.failures.records[0].?.err);
+    try t.expect(x.r.failures.records[0].?.id != staged_as);
+    try x.settle();
+    try t.expectEqual(@as(usize, 2), x.r.states.items.len);
+}
diff --git a/zig/src/core/rows_store.zig b/zig/src/core/rows_store.zig
index 6642b998f2..382ea4cb52 100644
--- a/zig/src/core/rows_store.zig
+++ b/zig/src/core/rows_store.zig
@@ -112,6 +112,7 @@ pub const Rows = struct {
     stopped: bool = false,
     flush_clock: ?u64 = null, // a close protects states touched after this clock from older queued saves
     next_job: u64 = 0,
+    waiting: ?*pc.Entry = null, // the newest save refused room; while it is next, step waits for a reap
     failures: write.Failures = .{},
     fetch: ?*@import("rows_prefetch.zig").Prefetch = null, // states read ahead for queued requests (a lane host's)
 
@@ -583,10 +584,12 @@ test {
     _ = image;
     _ = chain;
     _ = file;
+    _ = write.io;
     _ = @import("rows_test.zig");
     _ = @import("rows_spill_test.zig");
     _ = @import("rows_async_test.zig");
     _ = @import("rows_close_test.zig");
+    _ = @import("rows_stage_test.zig");
     _ = @import("rows_select_test.zig");
     _ = @import("rows_recall_select_test.zig");
     _ = @import("rows_pool_select_test.zig");
diff --git a/zig/src/core/rows_write.zig b/zig/src/core/rows_write.zig
index b445bc5072..0a924ae9f3 100644
--- a/zig/src/core/rows_write.zig
+++ b/zig/src/core/rows_write.zig
@@ -31,8 +31,16 @@ pub fn queued(r: *Rows, e: *pc.Entry, starts: []const u32) void {
     sync.queued(r, e, starts);
 }
 
-/// Staged work owns no entry pointer; eviction only removes an entry still waiting to be staged.
+/// Eviction removes only a save still queued (staged work owns no entry); one that waited for room counts as skipped.
 pub fn dropped(r: *Rows, e: *pc.Entry) void {
+    if (r.waiting == e) r.waiting = null;
+    for (r.queue.items) |q| if (q.entry == e) {
+        if (q.waited and !r.stopped) {
+            r.next_job +%= 1; // the id a save staged now would take, so no job in flight shares it
+            skip(r, chain.state(r.keying, e.tokens, e.decode_spans, q.starts, e.at), e.at, error.StagingFull);
+        }
+        break;
+    };
     sync.dropped(r, e);
 }
 
@@ -43,7 +51,13 @@ pub fn pending(r: *const Rows) bool {
 
 pub fn ready(r: *const Rows) bool {
     if (r.sync_writes) return sync.pending(r);
-    return (if (r.worker) |w| w.ready() else false) or (!r.stopped and r.spill != .evicted and r.queue.items.len > 0);
+    return (if (r.worker) |w| w.ready() else false) or (!r.stopped and r.spill != .evicted and r.queue.items.len > 0 and !blocked(r));
+}
+
+/// Whether the newest queued save was refused room by the jobs in flight and none has been taken since.
+fn blocked(r: *const Rows) bool {
+    const e = r.waiting orelse return false;
+    return r.queue.items.len > 0 and r.queue.items[r.queue.items.len - 1].entry == e;
 }
 
 /// A spill copies the entry before RAM releases it, without waiting for filesystem work or a byte permit.
@@ -52,7 +66,8 @@ pub fn evicted(r: *Rows, e: *pc.Entry) void {
     if (r.spill == .off or r.stopped) return;
     reap(r);
     for (r.queue.items, 0..) |q, i| if (q.entry == e) {
-        _ = stage(r, r.queue.orderedRemove(i));
+        // The entry leaves RAM now, so a full permit skips it rather than hold it.
+        _ = stage(r, r.queue.orderedRemove(i), false);
         return;
     };
 }
@@ -62,14 +77,13 @@ pub fn wanted(r: *const Rows, n: chain.Stem) bool {
     return if (r.worker) |w| w.wanted(n) else false;
 }
 
-/// Reconcile finished jobs and stage at most one queued entry; an outstanding write alone does not spin the engine.
+/// Reconcile finished jobs and stage at most one queued entry; a save waiting for room is work only a reap can free.
 pub fn step(r: *Rows) bool {
     if (r.sync_writes) return sync.step(r);
     reap(r);
-    if (r.stopped or (r.spill == .evicted and r.flush_clock == null)) return false;
+    if (r.stopped or blocked(r) or (r.spill == .evicted and r.flush_clock == null)) return false;
     const q = r.queue.pop() orelse return false;
-    _ = stage(r, q);
-    return true;
+    return stage(r, q, true);
 }
 
 pub fn abort(r: *Rows, j: *Job) void {
@@ -145,6 +159,7 @@ pub fn reap(r: *Rows) void {
     const w = r.worker orelse return;
     while (w.take()) |j| {
         defer j.destroy();
+        r.waiting = null;
         const err: ?anyerror = j.failure orelse if (j.generation != w.generation.load(.acquire)) error.Cancelled else null;
         if (err) |e| {
             failed(r, j, e);
@@ -154,7 +169,8 @@ pub fn reap(r: *Rows) void {
         }
         publish(r, j) catch |e| {
             failed(r, j, e);
-            r.admission.finish(r.dir, j.bytes, false);
+            // The job that owed the block already backed admission off.
+            r.admission.finish(r.dir, j.bytes, e == error.MissingBlock);
             r.collect();
             continue;
         };
@@ -164,6 +180,7 @@ pub fn reap(r: *Rows) void {
 
 fn publish(r: *Rows, j: *io.Job) !void {
     if (r.find(j.name)) |_| return;
+    if (!complete(r, j)) return error.MissingBlock;
     var s: store.State = .{ .name = j.name, .at = j.at, .tokens = &.{}, .spans = &.{}, .starts = &.{}, .own = j.files[j.files.len - 1].bytes, .rows = 0, .used = 0, .shared = j.shared };
     s.tokens = try r.gpa.dupe(u32, j.tokens);
     errdefer r.gpa.free(s.tokens);
@@ -190,51 +207,79 @@ fn publish(r: *Rows, j: *io.Job) !void {
     if (!@import("builtin").is_test) std.log.info("imprint: {d} tokens on disk: {d} new blocks and its own {d} MiB ({d} MiB this state, {d} MiB of rows it shares) in {d:.0} ms", .{ j.at, j.files.len - 1, s.own >> 20, j.bytes >> 20, s.rows >> 20, @as(f64, @floatFromInt(nowNs() - j.began)) / 1e6 });
 }
 
-fn stage(r: *Rows, q: Queued) bool {
+/// Whether every block the job's state names is on disk or among the files it publishes now.
+fn complete(r: *const Rows, j: *const io.Job) bool {
+    blocks: for (j.wants) |want| {
+        const n = chain.stem(want.name);
+        if (r.blocks.contains(n)) continue;
+        for (j.blocks()) |f| if (std.mem.eql(u8, &chain.stem(f.trailer.name), &n)) continue :blocks;
+        return false;
+    }
+    return true;
+}
+
+/// Whether `q` left the queue, staged or skipped; false when it waits for room and is queued again (`may_wait`).
+fn stage(r: *Rows, q: Queued, may_wait: bool) bool {
     const at = q.entry.at;
     r.next_job +%= 1;
     const name = chain.state(r.keying, q.entry.tokens, q.entry.decode_spans, q.starts, at);
     const j = sync.prepare(r, q) catch |err| {
-        r.counts.skipped += 1;
-        r.failures.add(.{ .id = r.next_job, .generation = if (r.worker) |w| w.generation.load(.acquire) else 0, .name = name, .at = at, .err = err });
-        pc.note("imprint: saving {d} tokens skipped ({s})", .{ at, @errorName(err) });
-        return false;
-    } orelse return false;
+        skip(r, name, at, err);
+        return true;
+    } orelse return true;
+    var queued_again = false;
     defer {
-        r.gpa.free(j.starts);
+        if (!queued_again) r.gpa.free(j.starts);
         r.gpa.free(j.parts);
         r.gpa.free(j.wants);
         r.gpa.free(j.own);
         r.gpa.destroy(j);
     }
-    stageJob(r, j) catch |err| {
-        r.counts.skipped += 1;
-        r.failures.add(.{ .id = r.next_job, .generation = if (r.worker) |w| w.generation.load(.acquire) else 0, .name = j.name, .at = at, .err = err });
+    stageJob(r, j, may_wait) catch |err| {
+        if (err == error.StagingBusy) {
+            queued_again = true;
+            r.waiting = q.entry;
+            r.queue.appendAssumeCapacity(.{ .entry = q.entry, .starts = q.starts, .waited = true });
+            return false;
+        }
+        skip(r, j.name, at, err);
         if (err == error.GatherFailed or err == error.SnapshotGpuFailure) r.counts.failed += 1;
-        pc.note("imprint: saving {d} tokens skipped ({s})", .{ at, @errorName(err) });
-        return false;
+        return true;
     };
     return true;
 }
 
-fn stageJob(r: *Rows, prepared: *const Job) !void {
+/// A save given up: counted, recorded under the job id it would have had, and noted.
+fn skip(r: *Rows, name: chain.Name, at: u32, err: anyerror) void {
+    r.counts.skipped += 1;
+    r.failures.add(.{ .id = r.next_job, .generation = if (r.worker) |w| w.generation.load(.acquire) else 0, .name = name, .at = at, .err = err });
+    pc.note("imprint: saving {d} tokens skipped ({s})", .{ at, @errorName(err) });
+}
+
+fn stageJob(r: *Rows, prepared: *const Job, may_wait: bool) !void {
     if (r.worker == null) r.worker = try io.Worker.open(r.dir, r.staging_bytes, r.file_io);
     const w = r.worker.?;
     if (w.names(prepared.name)) return;
     const e = prepared.entry;
     const metadata = 4 * (3 + e.tokens.len + 2 * e.decode_spans.len + prepared.starts.len);
-    var bytes: u64 = 0;
+    var promised: io.Promised = .empty;
+    defer promised.deinit(r.gpa);
     var count: usize = 0;
+    try w.promised(r.gpa, &promised);
+    var bytes = newBytes(r, &promised, prepared, metadata, &count);
     var charge: u64 = 0;
     while (true) {
-        bytes = newBytes(r, prepared, metadata, &count);
         charge = bytes + @sizeOf(io.Job) + count * @sizeOf(io.File) + prepared.wants.len * @sizeOf(Want) + 4 * @sizeOf([]const u8);
-        if (!w.fits(charge)) return error.StagingFull;
+        if (!w.fits(charge)) return if (may_wait and w.busy(charge)) error.StagingBusy else error.StagingFull;
         const reserved = w.reserved();
         if (!r.budget(bytes + reserved)) return error.DailyBudgetSpent;
         if (!r.room(bytes + reserved)) return error.OverCap;
-        var after_count: usize = 0;
-        if (newBytes(r, prepared, metadata, &after_count) == bytes) break;
+        try w.promised(r.gpa, &promised); // a fence inside room frees the blocks cancelled jobs promised
+        var again_count: usize = 0;
+        const again = newBytes(r, &promised, prepared, metadata, &again_count);
+        if (again == bytes) break;
+        bytes = again;
+        count = again_count;
     }
     if (r.admission.reserve(r.dir, bytes) != .ready) return error.UnderFreeDiskFloor;
     errdefer r.admission.finish(r.dir, bytes, true);
@@ -245,7 +290,7 @@ fn stageJob(r: *Rows, prepared: *const Job) !void {
     const a = arena.allocator();
     const files = try a.alloc(io.File, count);
     var built: usize = 0;
-    for (prepared.wants) |want| if (!r.blocks.contains(chain.stem(want.name))) {
+    for (prepared.wants) |want| if (lacks(r, &promised, want)) {
         var ranges: std.ArrayList(image.Range) = .empty;
         defer ranges.deinit(r.gpa);
         try image.blockRanges(prepared.parts, r.keying.block, want.window, want.k, &ranges, r.gpa);
@@ -267,10 +312,16 @@ fn stageJob(r: *Rows, prepared: *const Job) !void {
     w.submit(j);
 }
 
-fn newBytes(r: *const Rows, j: *const Job, metadata: usize, count: *usize) u64 {
+/// Whether the state's block is for this job to write: the disk lacks it and no job in flight is writing it.
+fn lacks(r: *const Rows, promised: *const io.Promised, want: Want) bool {
+    const n = chain.stem(want.name);
+    return !r.blocks.contains(n) and !promised.contains(n);
+}
+
+fn newBytes(r: *const Rows, promised: *const io.Promised, j: *const Job, metadata: usize, count: *usize) u64 {
     var bytes = image.total(j.own) + metadata + @sizeOf(file.Trailer);
     count.* = 1;
-    for (j.wants) |want| if (!r.blocks.contains(chain.stem(want.name))) {
+    for (j.wants) |want| if (lacks(r, promised, want)) {
         bytes += want.bytes;
         count.* += 1;
     };
diff --git a/zig/src/core/rows_write_sync.zig b/zig/src/core/rows_write_sync.zig
index 536bdd0a17..f6c4b396a8 100644
--- a/zig/src/core/rows_write_sync.zig
+++ b/zig/src/core/rows_write_sync.zig
@@ -17,7 +17,11 @@ const piece: u64 = 16 << 20;
 pub const Want = struct { name: Name, window: u32, k: u32, bytes: u64 };
 
 /// A kept entry waiting to be written, with the chunk starts below it (a planned family's state depends on them).
-pub const Queued = struct { entry: *pc.Entry, starts: []u32 };
+pub const Queued = struct {
+    entry: *pc.Entry,
+    starts: []u32,
+    waited: bool = false, // the jobs in flight once left it no room, so losing its entry counts as a skipped save
+};
 
 /// A kept state being written: the blocks the disk lacks, then its own file a piece at a time.
 pub const Job = struct {