From 1aa5179e429b398643cad5c84ee3a481ddbfd170 Mon Sep 17 00:00:00 2001
From: Joseph Mearman <joseph@mearman.co.uk>
Date: Wed, 9 Sep 2026 11:52:46 +0100
Subject: [PATCH] fix(mesh-store): replay only delivery events that carry
 consumption evidence

Replay fired for every event merged from a snapshot, but transient
notifications (room_members, member_joined, connection_request,
delivery status) have no consumption evidence: replaying them could
only ever duplicate-notify, bounded by a structural-key LRU whose
eviction window admitted occasional re-fires. Replay now fires only
for events whose consumption is observable: room messages and DMs
(readBy carries the reader) and room invites (the invited list carries
membership of a pending invite). Transient notifications still merge
into the queue, so drain bridges see them, and the state they describe
arrives through the synced room and agent records.

This closes the replay re-fire hole exactly rather than bounding it,
superseding the documented boundary on #28.
---
 src/core/mesh-store.ts           | 21 +++++++++--
 src/test/downtime-replay.test.ts | 63 ++++++++++++++++++++++++++++++--
 2 files changed, 78 insertions(+), 6 deletions(-)

diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts
index 3137a77..a158c65 100644
--- a/src/core/mesh-store.ts
+++ b/src/core/mesh-store.ts
@@ -373,9 +373,24 @@ export class MeshStore implements CommsStore {
         for (const event of events) {
           if (seen.has(JSON.stringify(event))) continue;
           this.queueDelivery(agentId, event);
-          // A returning peer replays its own pending queue: events pushed
-          // while its process was down fire onDelivery now (#28).
-          this.fireLocalDelivery(agentId, event);
+          // Replay fires only for events with consumption evidence (#28):
+          // messages are consumed by reading (readBy), invites by acceptance
+          // or decline (no longer in the invited list). Transient
+          // notifications (member_joined, room_members, connection_request,
+          // delivery status) carry no consumption evidence, so replaying
+          // them could only ever duplicate-notify; the state they describe
+          // arrives via the synced room and agent records instead. They
+          // still merge into the queue, so drain bridges see them.
+          if (event.type === "room_message" || event.type === "dm") {
+            this.fireLocalDelivery(agentId, event);
+          } else if (event.type === "room_invite") {
+            const stillInvited = this.rooms
+              .get(event.room)
+              ?.invited.includes(agentId);
+            if (stillInvited === true) {
+              this.fireLocalDelivery(agentId, event);
+            }
+          }
         }
       }
     }
diff --git a/src/test/downtime-replay.test.ts b/src/test/downtime-replay.test.ts
index e29521f..939135f 100644
--- a/src/test/downtime-replay.test.ts
+++ b/src/test/downtime-replay.test.ts
@@ -70,8 +70,15 @@ void test("events queued while the target was down replay on its first snapshot"
     true,
   );
 
-  // A second snapshot of the same state does not duplicate any push (the
-  // join notification replays too, exactly once).
+  // Transient notifications carry no consumption evidence, so they merge into the queue (drain bridges still see them) but never replay-fire.
+  assert.equal(deliveries.filter((ev) => ev.type === "room_members").length, 0);
+  const queue = snapshotOf(returned).deliveryQueues[targetAgent.id] ?? [];
+  assert.equal(
+    queue.some((ev) => ev.type === "room_members"),
+    true,
+  );
+
+  // A second snapshot of the same state does not duplicate the push.
   returned.applyStateSync(snapshotOf(sender));
   assert.equal(
     deliveries.filter(
@@ -81,7 +88,6 @@ void test("events queued while the target was down replay on its first snapshot"
     ).length,
     1,
   );
-  assert.equal(deliveries.filter((ev) => ev.type === "room_members").length, 1);
 });
 
 void test("a queue is bounded oldest-first so downtime cannot grow it without limit", async () => {
@@ -123,3 +129,54 @@ void test("a queue is bounded oldest-first so downtime cannot grow it without li
     true,
   );
 });
+
+void test("a pending invite replays until accepted or declined", async () => {
+  const sender = makeStore();
+  const author = await sender.registerAgent({
+    name: "owner",
+    harness: "pi",
+    cwd: "/tmp/p",
+    pid: process.pid,
+    visibility: "visible",
+    tags: [],
+  });
+  const target = makeStore();
+  const targetAgent = await target.registerAgent({
+    name: "invitee",
+    harness: "claude-code",
+    cwd: "/tmp/t",
+    pid: process.pid,
+    visibility: "visible",
+    tags: [],
+  });
+  sender.applyStateSync(target.serialise());
+  await sender.createRoom({
+    name: "private-room",
+    type: "private",
+    owner: author.id,
+    description: "x",
+  });
+  await sender.inviteToRoom("private-room", targetAgent.id, author.id);
+
+  const returned = makeStore();
+  const deliveries: DeliveryEvent[] = [];
+  returned.onDelivery = (_id, ev) => {
+    deliveries.push(ev);
+  };
+  returned.peerId = targetAgent.id;
+
+  // Still on the invited list: the invite replays.
+  returned.applyStateSync(snapshotOf(sender));
+  assert.equal(deliveries.filter((ev) => ev.type === "room_invite").length, 1);
+
+  // Declined (no longer invited): the same snapshot no longer replays it.
+  await sender.declineInvite("private-room", targetAgent.id, "not now");
+  const declined = makeStore();
+  const deliveries2: DeliveryEvent[] = [];
+  declined.onDelivery = (_id, ev) => {
+    deliveries2.push(ev);
+  };
+  declined.peerId = targetAgent.id;
+  declined.applyStateSync(snapshotOf(sender));
+  assert.equal(deliveries2.filter((ev) => ev.type === "room_invite").length, 0);
+});