feat(pathfinder): select preparation batches in deterministic rotating order

rfc/prepared-endpoint-implementation.md

@@ -58,3 +58,26 @@ matching consumers before the following integration is complete.
 regression tests exercise freshness/publication and the read-only table contract.
 This does not establish mutation coverage, rotating-batch fairness, game
 integration or WASM performance; those are subsequent acceptance checks.
+
+## Follow-up: rotating bounded selection
+
+`PreparationBatch` now provides world-local execution scratch. Each pass receives
+current ECS-derived pending identities, chooses up to the supplied budget in
+stable order after the previous cursor, and wraps. View order and duplicate
+memberships do not change selection. Deleting the cursor entity does not reset
+progress. Newly invalidated entities wait their turn. `finish` advances only at
+the publication barrier after the whole selected batch has been processed;
+rejected stale results still consume their turn. World changes call `reset`.
+
+There is no default budget or authoritative persistent pending list. Memory is
+bounded by the highest requested budget, and allocation failure leaves the batch
+empty without moving the cursor. Discovery currently costs O(candidate count ×
+budget), including duplicate checks. The helper reports inspected candidates;
+it does not claim a wall-time bound. Measure discovery together with snapshots,
+searches and publication before choosing the game's budget or optimizing selection.
+
+Five tests cover rotating fairness, deletion/insertion/wrap, duplicates, zero
+budget, empty views, OOM and every ordering of a small candidate set. There are
+now 81 navigation specs. This helper does not yet discover participants, take
+snapshots, search nodes or publish CMNs. Controller integration remains required;
+only a pending predicate backed by shared freshness may feed this selector.

src/nav/preparation_batch.zig

@@ -0,0 +1,72 @@
+const std = @import("std");
+
+/// Bounded execution scratch, not an authoritative pending queue. Feed the
+/// current ECS-derived pending entity identities on every pass. The owner is
+/// world-local and must reset the cursor on a world change. Full entity IDs
+/// must include generation (or use a stable, generation-qualified ordering).
+pub const PreparationBatch = struct {
+    selected: std.ArrayListUnmanaged(u64) = .empty,
+    cursor: ?u64 = null,
+    budget: usize = 0,
+    inspected: usize = 0,
+
+    pub fn deinit(self: *PreparationBatch, allocator: std.mem.Allocator) void {
+        self.selected.deinit(allocator);
+    }
+
+    /// Reserve before discovery. OOM leaves an empty batch and does not move
+    /// the cursor. There is no truncated allocation fallback.
+    pub fn begin(self: *PreparationBatch, allocator: std.mem.Allocator, budget: usize) !void {
+        self.selected.clearRetainingCapacity();
+        self.budget = 0;
+        self.inspected = 0;
+        try self.selected.ensureTotalCapacityPrecise(allocator, budget);
+        self.budget = budget;
+    }
+
+    fn before(self: *const PreparationBatch, a: u64, b: u64) bool {
+        if (self.cursor) |cursor| {
+            const a_wraps = a <= cursor;
+            const b_wraps = b <= cursor;
+            if (a_wraps != b_wraps) return !a_wraps;
+        }
+        return a < b;
+    }
+
+    /// Feed only pending ordinary entities, excluding current/off-graph CMNs
+    /// and registered MovementNodes. View order and duplicate memberships do
+    /// not change the batch. Work is O(discovered * budget), memory O(highest budget);
+    /// discovery is NOT time-bounded just because searches are count-bounded.
+    pub fn consider(self: *PreparationBatch, entity: u64) void {
+        self.inspected += 1;
+        if (self.budget == 0) return;
+        var index: usize = 0;
+        while (index < self.selected.items.len) : (index += 1) {
+            const existing = self.selected.items[index];
+            if (existing == entity) return;
+            if (self.before(entity, existing)) break;
+        }
+        if (index >= self.budget) return;
+        const old_len = self.selected.items.len;
+        if (old_len < self.budget) self.selected.items.len += 1;
+        var move = self.selected.items.len - 1;
+        while (move > index) : (move -= 1) self.selected.items[move] = self.selected.items[move - 1];
+        self.selected.items[index] = entity;
+    }
+
+    /// Call at the publication barrier after the entire batch was processed.
+    /// Rejected stale results also consume their turn: repeated invalidation
+    /// must not prevent progress. Cancelling before execution leaves the cursor.
+    pub fn finish(self: *PreparationBatch) void {
+        if (self.selected.items.len > 0) self.cursor = self.selected.items[self.selected.items.len - 1];
+        self.selected.clearRetainingCapacity();
+        self.budget = 0;
+    }
+
+    pub fn reset(self: *PreparationBatch) void {
+        self.cursor = null;
+        self.selected.clearRetainingCapacity();
+        self.budget = 0;
+        self.inspected = 0;
+    }
+};

src/nav/root.zig

@@ -87,3 +87,6 @@ pub const Components = struct {
     /// persisted.
     pub const ControllerState = controller.ControllerState;
 };
+
+/// World-local bounded selection scratch for serial endpoint preparation.
+pub const PreparationBatch = @import("preparation_batch.zig").PreparationBatch;

src/pathfinding.zig

@@ -140,3 +140,6 @@ test {
     _ = @import("a_star.zig");
     _ = @import("heuristics.zig");
 }
+
+/// World-local bounded selection scratch for serial endpoint preparation.
+pub const PreparationBatch = @import("nav/preparation_batch.zig").PreparationBatch;

src/root.zig

@@ -115,3 +115,6 @@ pub const algo = pathfinding.algo;
 test {
     _ = pathfinding;
 }
+
+/// World-local bounded selection scratch for serial endpoint preparation.
+pub const PreparationBatch = @import("nav/preparation_batch.zig").PreparationBatch;

tests/nav/preparation_batch_test.zig

@@ -0,0 +1,84 @@
+const std = @import("std");
+const testing = std.testing;
+const Batch = @import("pathfinder").PreparationBatch;
+
+fn select(batch: *Batch, budget: usize, pending: []const u64) ![]const u64 {
+    try batch.begin(testing.allocator, budget);
+    for (pending) |id| batch.consider(id);
+    return batch.selected.items;
+}
+
+test "rotating batches ignore view order and repeated invalidation cannot jump the queue" {
+    var batch = Batch{};
+    defer batch.deinit(testing.allocator);
+    const pending = [_]u64{ 6, 1, 3, 5, 2, 4 };
+    try testing.expectEqualSlices(u64, &.{ 1, 2 }, try select(&batch, 2, &pending));
+    batch.finish();
+    // 1/2 became stale again; they remain pending but cannot jump ahead.
+    try testing.expectEqualSlices(u64, &.{ 3, 4 }, try select(&batch, 2, &pending));
+    batch.finish();
+    try testing.expectEqualSlices(u64, &.{ 5, 6 }, try select(&batch, 2, &pending));
+    batch.finish();
+    try testing.expectEqualSlices(u64, &.{ 1, 2 }, try select(&batch, 2, &pending));
+}
+
+test "deleted cursor, insertion, duplicate views and wrap retain deterministic order" {
+    var batch = Batch{};
+    defer batch.deinit(testing.allocator);
+    batch.cursor = 3;
+    try testing.expectEqualSlices(u64, &.{ 4, 7, 1 }, try select(&batch, 3, &.{ 1, 7, 4, 4, 2, 7 }));
+    try testing.expectEqual(@as(usize, 6), batch.inspected);
+    batch.finish();
+    try testing.expectEqualSlices(u64, &.{ 2, 4, 7 }, try select(&batch, 3, &.{ 7, 4, 2, 1 }));
+    batch.reset();
+    try testing.expectEqualSlices(u64, &.{1}, try select(&batch, 1, &.{ 7, 4, 2, 1 }));
+}
+
+test "zero budget and empty pending view do not advance cursor" {
+    var batch = Batch{};
+    defer batch.deinit(testing.allocator);
+    batch.cursor = 4;
+    try testing.expectEqualSlices(u64, &.{}, try select(&batch, 0, &.{ 1, 5 }));
+    batch.finish();
+    try testing.expectEqual(@as(?u64, 4), batch.cursor);
+    try testing.expectEqualSlices(u64, &.{}, try select(&batch, 4, &.{}));
+    batch.finish();
+    try testing.expectEqual(@as(?u64, 4), batch.cursor);
+}
+
+test "allocation failure cannot publish a partial batch or consume a turn" {
+    var bytes: [0]u8 = .{};
+    var allocator = std.heap.FixedBufferAllocator.init(&bytes);
+    var batch = Batch{ .cursor = 4 };
+    defer batch.deinit(allocator.allocator());
+    try testing.expectError(error.OutOfMemory, batch.begin(allocator.allocator(), 3));
+    batch.consider(5);
+    try testing.expectEqual(@as(usize, 0), batch.selected.items.len);
+    batch.finish();
+    try testing.expectEqual(@as(?u64, 4), batch.cursor);
+}
+
+test "all permutations agree with the rotated ordering and budget" {
+    const values = [_]u64{ 0, 2, 5, 9 };
+    var batch = Batch{};
+    defer batch.deinit(testing.allocator);
+    for (values) |a| for (values) |b| for (values) |c| for (values) |d| {
+        if (a == b or a == c or a == d or b == c or b == d or c == d) continue;
+        for (0..11) |cursor| {
+            for (0..6) |budget| {
+                batch.cursor = cursor;
+                var expected: [4]u64 = undefined;
+                var count: usize = 0;
+                for (values) |v| if (v > cursor) {
+                    expected[count] = v;
+                    count += 1;
+                };
+                for (values) |v| if (v <= cursor) {
+                    expected[count] = v;
+                    count += 1;
+                };
+                try testing.expectEqualSlices(u64, expected[0..@min(budget, 4)], try select(&batch, budget, &.{ a, b, c, d }));
+            }
+        }
+    };
+}

tests/nav/root.zig

@@ -1,6 +1,7 @@
 const std = @import("std");
 
 test {
+    _ = @import("preparation_batch_test.zig");
     _ = @import("prepared_endpoint_test.zig");
     _ = @import("graph_test.zig");
     _ = @import("floyd_warshall_test.zig");