diff --git a/src/main.zig b/src/main.zig index c77ae41a..726e5d07 100644 --- a/src/main.zig +++ b/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") }; diff --git a/src/presence.zig b/src/presence.zig index 13450d72..8b98b9fe 100644 --- a/src/presence.zig +++ b/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) diff --git a/src/presence_ack.zig b/src/presence_ack.zig new file mode 100644 index 00000000..c409888a --- /dev/null +++ b/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); +} diff --git a/src/shared_tree_batch.zig b/src/shared_tree_batch.zig index b9391b3c..363ef88d 100644 --- a/src/shared_tree_batch.zig +++ b/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