From 06448e5269453b7efcd99b01aa3967eea5e2c7cc Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 9 Sep 2026 11:55:19 +0100 Subject: [PATCH 1/2] feat(mesh-store): add per-agent membership operation maps to rooms Wire-format groundwork for convergent membership merges: four records holding the room revision of each agent's latest join to and leave from the member and invited lists, with members and invited as derived views. --- src/core/types.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/core/types.ts b/src/core/types.ts index bfddf28..d76e24a 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -102,6 +102,18 @@ export const RoomSchema = defineSchema( description: z.string(), members: z.array(z.string()), invited: z.array(z.string()), + /** + * Per-agent membership operations for convergent merges. An agent is a + * member (or invited) iff their latest join's room revision strictly + * exceeds their latest leave's, so a leave at the same revision wins: + * concurrent kicks and joins converge with the kick honoured, while + * joins of different agents never interact. `members` and `invited` are + * derived views, refreshed after every mutation and merge. + */ + memberJoins: z.record(z.string(), z.number()), + memberLeaves: z.record(z.string(), z.number()), + invitedJoins: z.record(z.string(), z.number()), + invitedLeaves: z.record(z.string(), z.number()), federated: z.boolean().optional(), }), ); From af164725f424d4a722fb609043b130e8aca331d5 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 9 Sep 2026 11:58:51 +0100 Subject: [PATCH 2/2] feat(mesh-store): converge room membership as a per-agent element set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Room membership was merged by version-gated record replacement with a union on equal revisions, which protected concurrent joins of different agents but made a leave or kick racing a concurrent join converge with the joiner retained — a kicked agent stayed a member until the next room mutation resolved the tie. Membership is now a last-write-wins element set: each room carries the revision of every agent's latest join to and leave from the member and invited lists, an agent is in a list when their join strictly outranks their leave, and merges take the highest revision per agent per list. Joins of different agents are independent keys and never interact, and an exact join/leave tie resolves to out, so a kick racing a join converges with the kick honoured. The members and invited arrays become derived views refreshed after every mutation and merge, replacing the hand-maintained pushes and filters at each site. Scalar room fields keep the version gate; only membership switched to element merge, so renames and edits still heal through the higher version replacing the record. The FileStore maintains the same operation maps so its records stay shape-compatible. --- src/core/mesh-store.ts | 151 ++++++++++++++++++------ src/core/store.ts | 52 +++++--- src/test/state-sync-convergence.test.ts | 104 ++++++++++++++++ 3 files changed, 250 insertions(+), 57 deletions(-) diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index a158c65..121beb5 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -338,15 +338,7 @@ export class MeshStore implements CommsStore { if (existingVersion !== undefined && room.version < existingVersion) { continue; } - if (existingVersion === room.version) { - const existing = this.rooms.get(id); - if (existing) { - for (const m of existing.members) { - if (!room.members.includes(m)) room.members.push(m); - } - } - } - this.rooms.set(id, room); + this.mergeRoom(room); } for (const [id, msgs] of incoming.messages) { const existing = this.messages.get(id); @@ -578,6 +570,90 @@ export class MeshStore implements CommsStore { this.deliveryQueues.set(agentId, arr); } + /** + * Merge an incoming room record into the local one. Scalar fields follow + * the version gate the caller already applied (an equal or higher version + * reaches here), while membership is always element-merged per agent by + * highest operation revision, so concurrent joins of different agents + * survive and a kick racing a join converges with the kick honoured + * (#27). The members and invited views are re-derived from the merged + * operations. + */ + private mergeRoom(incoming: Room): void { + const existing = this.rooms.get(incoming.id); + if (existing === undefined) { + this.refreshMembership(incoming); + this.rooms.set(incoming.id, incoming); + return; + } + existing.version = Math.max(existing.version, incoming.version); + existing.name = incoming.name; + existing.type = incoming.type; + existing.owner = incoming.owner; + existing.createdAt = incoming.createdAt; + existing.description = incoming.description; + if (incoming.federated !== undefined) + existing.federated = incoming.federated; + existing.memberJoins = MeshStore.mergeMemberOps( + existing.memberJoins, + incoming.memberJoins, + ); + existing.memberLeaves = MeshStore.mergeMemberOps( + existing.memberLeaves, + incoming.memberLeaves, + ); + existing.invitedJoins = MeshStore.mergeMemberOps( + existing.invitedJoins, + incoming.invitedJoins, + ); + existing.invitedLeaves = MeshStore.mergeMemberOps( + existing.invitedLeaves, + incoming.invitedLeaves, + ); + this.refreshMembership(existing); + } + + /** + * Derive the members and invited views from the per-agent operation maps. + * An agent is in the list when their latest join strictly outranks their + * latest leave; equal revisions mean the leave wins, so a kick racing a + * concurrent join converges with the kick honoured (#27). + */ + private refreshMembership(room: Room): void { + room.members = Object.keys(room.memberJoins).filter( + (id) => (room.memberJoins[id] ?? 0) > (room.memberLeaves[id] ?? 0), + ); + room.invited = Object.keys(room.invitedJoins).filter( + (id) => (room.invitedJoins[id] ?? 0) > (room.invitedLeaves[id] ?? 0), + ); + } + + /** Record a membership operation at the room's current revision. */ + private recordMemberOp( + room: Room, + list: "member" | "invited", + op: "join" | "leave", + agentId: string, + ): void { + const joins = list === "member" ? room.memberJoins : room.invitedJoins; + const leaves = list === "member" ? room.memberLeaves : room.invitedLeaves; + const stamp = room.version; + if (op === "join") joins[agentId] = stamp; + else leaves[agentId] = stamp; + } + + /** Merge per-agent operation maps by highest revision per agent. */ + private static mergeMemberOps( + local: Record, + incoming: Record, + ): Record { + const merged: Record = { ...local }; + for (const [id, stamp] of Object.entries(incoming)) { + if (stamp > (merged[id] ?? 0)) merged[id] = stamp; + } + return merged; + } + /** Bump an entity's sync revision; call before broadcasting a local mutation. */ private bump(entity: T): T { entity.version += 1; @@ -625,17 +701,7 @@ export class MeshStore implements CommsStore { // Stale copy from a peer that missed updates (#27). break; } - const merged = patch.room; - if (existing?.version === patch.room.version) { - // Concurrent mutations from the same base: union members so - // simultaneous joins both survive. A strictly higher version - // replaces the record wholesale, which is what heals a leave or - // edit a lagging peer missed. - for (const m of existing.members) { - if (!merged.members.includes(m)) merged.members.push(m); - } - } - this.rooms.set(merged.id, merged); + this.mergeRoom(patch.room); break; } case "room_delete": @@ -1075,6 +1141,10 @@ export class MeshStore implements CommsStore { description: opts.description, members: [opts.owner], invited: [], + memberJoins: { [opts.owner]: 1 }, + memberLeaves: {}, + invitedJoins: {}, + invitedLeaves: {}, federated: opts.federated ?? false, }; @@ -1105,21 +1175,20 @@ export class MeshStore implements CommsStore { if (!room) throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); - if (room.type === "public") { - if (!room.members.includes(agentId)) room.members.push(agentId); - } else { - if ( - !room.invited.includes(agentId) && - room.owner !== agentId && - !room.members.includes(agentId) - ) { + const alreadyMember = room.members.includes(agentId); + if (!alreadyMember && room.type !== "public") { + if (!room.invited.includes(agentId) && room.owner !== agentId) { throw new CommsError(`Not invited to room ${roomId}`, "NOT_INVITED"); } - room.invited = room.invited.filter((id) => id !== agentId); - if (!room.members.includes(agentId)) room.members.push(agentId); } this.bump(room); + this.recordMemberOp(room, "member", "join", agentId); + if (alreadyMember || room.type !== "public") { + // Consuming an invitation (or re-joining) retires the invited entry. + this.recordMemberOp(room, "invited", "leave", agentId); + } + this.refreshMembership(room); this.rooms.set(roomId, room); const agent = this.agents.get(agentId); @@ -1175,8 +1244,9 @@ export class MeshStore implements CommsStore { if (!room) throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); - room.members = room.members.filter((id) => id !== agentId); this.bump(room); + this.recordMemberOp(room, "member", "leave", agentId); + this.refreshMembership(room); this.rooms.set(roomId, room); const agent = this.agents.get(agentId); @@ -1216,10 +1286,11 @@ export class MeshStore implements CommsStore { if (room.owner !== inviterId) throw new CommsError("Only the room owner can invite", "NOT_OWNER"); + this.bump(room); if (!room.invited.includes(targetId) && !room.members.includes(targetId)) { - room.invited.push(targetId); + this.recordMemberOp(room, "invited", "join", targetId); } - this.bump(room); + this.refreshMembership(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); @@ -1249,8 +1320,9 @@ export class MeshStore implements CommsStore { "NOT_INVITED", ); - room.invited = room.invited.filter((id) => id !== agentId); this.bump(room); + this.recordMemberOp(room, "invited", "leave", agentId); + this.refreshMembership(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); @@ -1275,9 +1347,10 @@ export class MeshStore implements CommsStore { if (room.owner !== kickerId) throw new CommsError("Only the room owner can kick", "NOT_OWNER"); - room.members = room.members.filter((id) => id !== targetId); - room.invited = room.invited.filter((id) => id !== targetId); this.bump(room); + this.recordMemberOp(room, "member", "leave", targetId); + this.recordMemberOp(room, "invited", "leave", targetId); + this.refreshMembership(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); } @@ -1633,8 +1706,9 @@ export class MeshStore implements CommsStore { const remoteId = `fed:${agentId}`; if (!room.members.includes(remoteId)) { - room.members.push(remoteId); this.bump(room); + this.recordMemberOp(room, "member", "join", remoteId); + this.refreshMembership(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); } @@ -1659,8 +1733,9 @@ export class MeshStore implements CommsStore { if (!room?.federated) return; const remoteId = `fed:${agentId}`; - room.members = room.members.filter((m) => m !== remoteId); this.bump(room); + this.recordMemberOp(room, "member", "leave", remoteId); + this.refreshMembership(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); diff --git a/src/core/store.ts b/src/core/store.ts index f75ffa2..00df400 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -252,6 +252,10 @@ export class FileStore implements CommsStore { description: opts.description, members: [opts.owner], invited: [], + memberJoins: { [opts.owner]: 1 }, + memberLeaves: {}, + invitedJoins: {}, + invitedLeaves: {}, }; await this.writeJsonFile(this.roomPath(id), room); @@ -292,25 +296,24 @@ export class FileStore implements CommsStore { if (!room) throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); - if (room.type === "public") { - if (!room.members.includes(agentId)) { - room.members.push(agentId); - } - } else { - if ( - !room.invited.includes(agentId) && - room.owner !== agentId && - !room.members.includes(agentId) - ) { - throw new CommsError(`Not invited to room ${roomId}`, "NOT_INVITED"); - } - room.invited = room.invited.filter((id) => id !== agentId); - if (!room.members.includes(agentId)) { - room.members.push(agentId); - } + if ( + room.type !== "public" && + !room.invited.includes(agentId) && + room.owner !== agentId && + !room.members.includes(agentId) + ) { + throw new CommsError(`Not invited to room ${roomId}`, "NOT_INVITED"); } room.version += 1; + room.memberJoins[agentId] = room.version; + room.invitedLeaves[agentId] = room.version; + room.members = Object.keys(room.memberJoins).filter( + (id) => (room.memberJoins[id] ?? 0) > (room.memberLeaves[id] ?? 0), + ); + room.invited = Object.keys(room.invitedJoins).filter( + (id) => (room.invitedJoins[id] ?? 0) > (room.invitedLeaves[id] ?? 0), + ); await this.writeJsonFile(this.roomPath(roomId), room); const agent = await this.getAgent(agentId); @@ -333,8 +336,12 @@ export class FileStore implements CommsStore { if (!room) throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); - room.members = room.members.filter((id) => id !== agentId); room.version += 1; + room.memberLeaves[agentId] = room.version; + room.members = Object.keys(room.memberJoins).filter( + (id) => (room.memberJoins[id] ?? 0) > (room.memberLeaves[id] ?? 0), + ); + await this.writeJsonFile(this.roomPath(roomId), room); const agent = await this.getAgent(agentId); @@ -422,9 +429,16 @@ export class FileStore implements CommsStore { if (room.owner !== kickerId) throw new CommsError("Only the room owner can kick", "NOT_OWNER"); - room.members = room.members.filter((id) => id !== targetId); - room.invited = room.invited.filter((id) => id !== targetId); room.version += 1; + room.memberLeaves[targetId] = room.version; + room.invitedLeaves[targetId] = room.version; + room.members = Object.keys(room.memberJoins).filter( + (id) => (room.memberJoins[id] ?? 0) > (room.memberLeaves[id] ?? 0), + ); + room.invited = Object.keys(room.invitedJoins).filter( + (id) => (room.invitedJoins[id] ?? 0) > (room.invitedLeaves[id] ?? 0), + ); + await this.writeJsonFile(this.roomPath(roomId), room); } diff --git a/src/test/state-sync-convergence.test.ts b/src/test/state-sync-convergence.test.ts index fdfccee..3e9ffc1 100644 --- a/src/test/state-sync-convergence.test.ts +++ b/src/test/state-sync-convergence.test.ts @@ -156,3 +156,107 @@ void test("history sync adds unseen messages and unions read receipts", async () ); assert.equal(merged[0]?.readBy.includes("reader-elsewhere"), true); }); + +void test("a kick racing a concurrent join converges with the kick honoured", async () => { + // Two holders of the same room base: one records X leaving (a kick), the + // other records X joining, both at the same revision because they mutated + // concurrently from the same base. Both directions of the sync must + // converge on X being out — the leave wins an exact tie (#27). + const base = makeStore(); + const owner = await base.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const member = makeStore(); + const x = await member.registerAgent({ + name: "x", + harness: "claude-code", + cwd: "/tmp/x", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await base.createRoom({ + name: "room", + type: "public", + owner: owner.id, + description: "x", + }); + await base.joinRoom("room", x.id); + + const kicker = makeStore(); + const joiner = makeStore(); + kicker.applyStateSync(snapshotOf(base)); + joiner.applyStateSync(snapshotOf(base)); + + // Concurrent mutations from the same base: a kick on one holder, a + // re-join of X recorded on the other. + await kicker.leaveRoom("room", x.id); + await joiner.joinRoom("room", x.id); + assert.equal((await kicker.getRoom("room"))?.members.includes(x.id), false); + assert.equal((await joiner.getRoom("room"))?.members.includes(x.id), true); + + // Exchange both ways: both converge on the kick. + kicker.applyStateSync(snapshotOf(joiner)); + joiner.applyStateSync(snapshotOf(kicker)); + assert.equal((await kicker.getRoom("room"))?.members.includes(x.id), false); + assert.equal((await joiner.getRoom("room"))?.members.includes(x.id), false); +}); + +void test("concurrent joins of different agents both survive the merge", async () => { + // The property the old version-tie union existed to protect: two peers + // each record a different agent joining from the same base, and the + // merged room holds both. + const base = makeStore(); + const owner = await base.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const agents = makeStore(); + const p = await agents.registerAgent({ + name: "p", + harness: "claude-code", + cwd: "/tmp/1", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const q = await agents.registerAgent({ + name: "q", + harness: "codex", + cwd: "/tmp/2", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await base.createRoom({ + name: "room", + type: "public", + owner: owner.id, + description: "x", + }); + base.applyStateSync(snapshotOf(agents)); + + const holderA = makeStore(); + const holderB = makeStore(); + holderA.applyStateSync(snapshotOf(base)); + holderB.applyStateSync(snapshotOf(base)); + await holderA.joinRoom("room", p.id); + await holderB.joinRoom("room", q.id); + + holderA.applyStateSync(snapshotOf(holderB)); + holderB.applyStateSync(snapshotOf(holderA)); + const mergedA = await holderA.getRoom("room"); + const mergedB = await holderB.getRoom("room"); + assert.equal(mergedA?.members.includes(p.id), true); + assert.equal(mergedA?.members.includes(q.id), true); + assert.deepEqual(mergedB?.members, mergedA?.members); +});