shared-tree checkpoint: one pass covers every live peer and converges

src/main.zig

@@ -560,7 +560,7 @@ test { // ── Unit tests (`zig build test`): pull in tests from imported modu
     _ = .{ @import("beta_feed.zig"), @import("jev_tool.zig"), @import("jev_model_scope.zig"), @import("jev_provider_switch_tests.zig"), @import("jev_effort_state.zig"), @import("gemini_turns.zig"), @import("gemini_turns_tests.zig") };
     _ = .{ @import("mcp_apps.zig"), @import("html_view.zig"), @import("usage_trace.zig"), @import("providers_confirmation_tests.zig"), @import("repl_model_confirmation_tests.zig"), @import("repl_model_pick_tests.zig"), @import("engine_sink_citation_tests.zig"), @import("cite_markup_stream_tests.zig"), @import("oneshot_citation_tests.zig"), @import("acp_permission.zig"), @import("acp_usage.zig"), @import("context_savings.zig") };
     _ = .{ @import("mcp_server_tests.zig"), @import("main_test.zig"), @import("artifact_claim_command.zig"), @import("cli_path_hint.zig"), @import("run_budget.zig") };
-    _ = .{ @import("session_catalog.zig"), @import("session_size_tests.zig"), @import("session_prompt.zig"), @import("session_discovery.zig"), @import("session_workspaces.zig"), @import("commands_resume.zig"), @import("peer_live.zig"), @import("presence_tests.zig"), @import("mcp_rpc_tests.zig") };
+    _ = .{ @import("session_catalog.zig"), @import("session_size_tests.zig"), @import("session_prompt.zig"), @import("session_discovery.zig"), @import("session_workspaces.zig"), @import("commands_resume.zig"), @import("peer_live.zig"), @import("presence_tests.zig"), @import("presence_ack.zig"), @import("mcp_rpc_tests.zig") };
     _ = .{ @import("prompt_astra.zig"), @import("prompt_guidance.zig"), @import("oauth_chatgpt.zig"), @import("chatgpt_reauth.zig"), @import("oauth_anthropic.zig"), @import("anthropic_oauth_wire.zig"), @import("anthropic_login_tests.zig") };
     _ = .{ @import("agent_empty_completion.zig"), @import("agent_length_cutoff.zig"), @import("agent_model_loop.zig"), @import("publication_policy_tests.zig"), @import("jobs_completion_tests.zig"), @import("headless_result.zig"), @import("oneshot.zig"), @import("agent_length.zig"), @import("agent_length_tests.zig") };
     _ = .{ @import("provider_routing_tests.zig"), @import("codegraff_messages_tests.zig"), @import("tool_result_empty_tests.zig"), @import("history_wire.zig"), @import("path_lookup.zig"), @import("agents_prepare.zig"), @import("catalog_price.zig"), @import("readline_paste_number_tests.zig"), @import("test_hooks.zig"), @import("peer_wake_loop.zig") };

src/presence.zig

@@ -328,13 +328,11 @@ pub fn gateCheckIdentity(io: Io, arena: Allocator, identity: []const u8) ?[]cons
     var dir = Io.Dir.cwd().openDir(io, dir_path, .{ .iterate = true }) catch return null;
     defer dir.close(io);
     const peers = listPeers(io, arena, dir);
-    const peer = unackedPeer(peers, identity, proc_identity.selfPid(), g_session, g_acked[0..g_acked_len]) orelse return null;
-    if (g_acked_len < g_acked.len) {
-        g_acked[g_acked_len] = ackKey(peer);
-        g_acked_len += 1;
-    }
+    const acked = @import("presence_ack.zig").ackLive(peers, identity, proc_identity.selfPid(), g_session, &g_acked, &g_acked_len);
+    const peer = acked.first orelse return null;
     const warning = worktree_lease.duplicateOwnerWarning(arena, peer, unixMs(io) - peer.last_seen_ms);
-    return std.fmt.allocPrint(arena, "{s}shared-tree checkpoint: the action was NOT performed. Re-issue the identical call to proceed — this fires once per live peer, across git mutations, file writes, and shell moves alike — coordinate first via the peer_message tool (session \"{s}\"), or keep your edits disjoint from theirs.", .{ warning, peer.session_id }) catch warning;
+    const others = if (acked.fresh > 1) std.fmt.allocPrint(arena, " and {d} other live peer(s)", .{acked.fresh - 1}) catch "" else "";
+    return std.fmt.allocPrint(arena, "{s}shared-tree checkpoint: the action was NOT performed. Re-issue the identical call to proceed — this one checkpoint covers every peer live now; only a peer that starts later fires another — coordinate first via the peer_message tool (session \"{s}\"{s}), or keep your edits disjoint from theirs.", .{ warning, peer.session_id, others }) catch warning;
 }
 
 // The channel's wire format (Message, chanName, postMessage, readNewMessages)

src/presence_ack.zig

@@ -0,0 +1,87 @@
+//! #1622: the shared-tree checkpoint's acknowledgments. It fired once per
+//! live peer, but remembered at most `acked.len` of them and never forgot a
+//! peer that had exited. With peers respawning, the set filled up; every new
+//! peer could then never be acknowledged and every mutating call was refused
+//! for good, each re-issue naming another peer. One checkpoint now covers
+//! every live peer at once, and only live peers keep their acknowledgment.
+
+const std = @import("std");
+const presence = @import("presence.zig");
+const worktree_lease = @import("worktree_lease.zig");
+const Owner = worktree_lease.Owner;
+
+pub const Acked = struct {
+    /// The first newly acknowledged peer, null when every live peer already was.
+    first: ?Owner = null,
+    /// How many live peers were newly acknowledged.
+    fresh: usize = 0,
+};
+
+fn live(peers: presence.Peers, i: usize, my_identity: []const u8, my_pid: i32, my_session: []const u8) bool {
+    if (i >= peers.probes.len) return false;
+    const rec = peers.records[i];
+    if (presence.ownSessionRecord(rec, my_session)) return false;
+    return switch (worktree_lease.ownerVerdict(rec, my_identity, my_pid, peers.probes[i])) {
+        .live_foreign, .live_unverified => true,
+        else => false,
+    };
+}
+
+/// Drop acknowledgments of peers that are no longer live, then acknowledge
+/// every live peer not yet acknowledged. `acked[0..len.*]` is updated in place.
+pub fn ackLive(peers: presence.Peers, my_identity: []const u8, my_pid: i32, my_session: []const u8, acked: []u64, len: *usize) Acked {
+    var kept: usize = 0;
+    for (acked[0..len.*]) |key| {
+        for (peers.records, 0..) |rec, i| {
+            if (presence.ackKey(rec) == key and live(peers, i, my_identity, my_pid, my_session)) {
+                acked[kept] = key;
+                kept += 1;
+                break;
+            }
+        }
+    }
+    len.* = kept;
+    var out: Acked = .{};
+    for (peers.records, 0..) |rec, i| {
+        if (!live(peers, i, my_identity, my_pid, my_session)) continue;
+        const key = presence.ackKey(rec);
+        if (std.mem.indexOfScalar(u64, acked[0..len.*], key) != null) continue;
+        if (out.first == null) out.first = rec;
+        out.fresh += 1;
+        if (len.* < acked.len) {
+            acked[len.*] = key;
+            len.* += 1;
+        }
+    }
+    return out;
+}
+
+test "#1622 one checkpoint covers every live peer, and churn cannot fill the ack set" {
+    const proc_identity = @import("proc_identity.zig");
+    const my_identity = "/repo/.git";
+    var acked: [4]u64 = undefined;
+    var len: usize = 0;
+
+    const a: Owner = .{ .pid = 10, .start_id = 1, .session_id = "s-a", .identity = my_identity };
+    const b: Owner = .{ .pid = 11, .start_id = 2, .session_id = "s-b", .identity = my_identity };
+    const records = [_]Owner{ a, b };
+    const probes = [_]proc_identity.Probe{ .{ .id = 1 }, .{ .id = 2 } };
+    const first = ackLive(.{ .records = &records, .probes = &probes }, my_identity, 1, "s-me", &acked, &len);
+    try std.testing.expectEqualStrings("s-a", first.first.?.session_id);
+    try std.testing.expectEqual(@as(usize, 2), first.fresh);
+    // The identical re-issue proceeds: both peers were acknowledged together.
+    try std.testing.expectEqual(@as(usize, 0), ackLive(.{ .records = &records, .probes = &probes }, my_identity, 1, "s-me", &acked, &len).fresh);
+
+    // Peers keep exiting and respawning: far more than the ack set holds.
+    var pid: i32 = 100;
+    while (pid < 140) : (pid += 1) {
+        const fresh: Owner = .{ .pid = pid, .start_id = @intCast(pid), .session_id = "s-new", .identity = my_identity };
+        const churn = [_]Owner{fresh};
+        const churn_probes = [_]proc_identity.Probe{.{ .id = @intCast(pid) }};
+        const peers: presence.Peers = .{ .records = &churn, .probes = &churn_probes };
+        try std.testing.expectEqual(@as(usize, 1), ackLive(peers, my_identity, 1, "s-me", &acked, &len).fresh);
+        // ...and the re-issue after each checkpoint goes through.
+        try std.testing.expectEqual(@as(usize, 0), ackLive(peers, my_identity, 1, "s-me", &acked, &len).fresh);
+    }
+    try std.testing.expectEqual(@as(usize, 1), len);
+}

src/shared_tree_batch.zig

@@ -1,6 +1,6 @@
 //! #1553: one shared-tree checkpoint holds the whole batch.
 //!
-//! The checkpoint fires once per live peer and acknowledges it on the spot, so
+//! The checkpoint fires once for the peers live now and acknowledges them, so
 //! the re-issued call proceeds. Gating a parallel batch call by call let the
 //! first mutation take the checkpoint and every sibling after it pass the
 //! now-acknowledged gate: one result said "NOT performed" while the other