fix(mesh-store): replay pending deliveries to an agent returning from downtime

README.md

@@ -34,7 +34,7 @@ graph LR
     B_Bridge -- "channel notification" --> B_LLM
 ```
 
-All state is held in memory and synchronised between peers. Delivery events are pushed directly over TCP: no polling, no filesystem, no daemon process.
+All state is held in memory and synchronised between peers. Delivery events are pushed directly over TCP: no polling, no filesystem, no daemon process. Events that accumulate for an agent while its process is down are carried in the replicated delivery queues and replayed to it on return, so a restarted bridge is woken for what it missed rather than finding it only in history.
 
 ### Coordinator pattern
 

package.json

@@ -52,7 +52,7 @@
     "generate:server-json": "tsx scripts/sync-release-metadata.ts",
     "publish:mcp-registry": "tsx scripts/publish-mcp-registry.ts",
     "lint": "eslint .",
-    "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/ws-broadcast-window.integration.test.js dist/test/tls-transport.integration.test.js dist/test/state-sync-convergence.test.js",
+    "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/ws-broadcast-window.integration.test.js dist/test/tls-transport.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js",
     "test:visibility": "node --test dist/test/visibility.integration.test.js",
     "test:delivery": "node dist/test/delivery-receipt.runner.js",
     "test:all": "pnpm test && pnpm test:delivery",

src/core/mesh-store.ts

@@ -52,6 +52,13 @@ import type { ListenerInfo, ListenerPolicy } from "./transport.js";
 const DEFAULT_COORDINATOR_PORT = 19876;
 const COORDINATOR_HOST = "127.0.0.1";
 
+/**
+ * Bound on pending delivery events held per target agent. Events beyond the
+ * bound drop oldest-first: a long-offline agent's queue cannot grow without
+ * limit in memory or in synced snapshots (#28).
+ */
+const MAX_QUEUED_DELIVERIES_PER_AGENT = 100;
+
 /**
  * Merge an incoming append-only message history into the local one: add
  * entries the local list does not have and union read receipts on the ones
@@ -125,6 +132,7 @@ export class MeshStore implements CommsStore {
       rooms: Object.fromEntries(this.rooms),
       messages: Object.fromEntries(this.messages),
       dms: Object.fromEntries(this.dms),
+      deliveryQueues: Object.fromEntries(this.deliveryQueues),
     };
   }
 
@@ -285,12 +293,7 @@ export class MeshStore implements CommsStore {
   ): Promise<void> {
     // If we have state and the peer doesn't, send state sync
     if (this.agents.size > 0) {
-      const state: SerialisedState = {
-        agents: Object.fromEntries(this.agents),
-        rooms: Object.fromEntries(this.rooms),
-        messages: Object.fromEntries(this.messages),
-        dms: Object.fromEntries(this.dms),
-      };
+      const state: SerialisedState = this.serialise();
       await this.transport.send(handle, {
         method: "state_sync",
         state,
@@ -361,6 +364,20 @@ export class MeshStore implements CommsStore {
         }
         mergeMessageHistories(existing, dmMsgs);
       }
+      for (const [agentId, events] of Object.entries(state.deliveryQueues)) {
+        const seen = new Set(
+          (this.deliveryQueues.get(agentId) ?? []).map((e) =>
+            JSON.stringify(e),
+          ),
+        );
+        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);
+        }
+      }
     }
   }
 
@@ -423,9 +440,7 @@ export class MeshStore implements CommsStore {
       name: request.name,
       fingerprint: request.fingerprint,
     };
-    const arr = this.deliveryQueues.get(this.peerId) ?? [];
-    arr.push(event);
-    this.deliveryQueues.set(this.peerId, arr);
+    this.queueDelivery(this.peerId, event);
     if (this.onDelivery) {
       void this.onDelivery(this.peerId, event);
     }
@@ -538,6 +553,16 @@ export class MeshStore implements CommsStore {
     };
   }
 
+  /** Append to a target agent's delivery queue, bounded oldest-first (#28). */
+  private queueDelivery(agentId: string, event: DeliveryEvent): void {
+    const arr = this.deliveryQueues.get(agentId) ?? [];
+    arr.push(event);
+    if (arr.length > MAX_QUEUED_DELIVERIES_PER_AGENT) {
+      arr.splice(0, arr.length - MAX_QUEUED_DELIVERIES_PER_AGENT);
+    }
+    this.deliveryQueues.set(agentId, arr);
+  }
+
   /** Bump an entity's sync revision; call before broadcasting a local mutation. */
   private bump<T extends { version: number }>(entity: T): T {
     entity.version += 1;
@@ -614,9 +639,7 @@ export class MeshStore implements CommsStore {
         break;
       }
       case "delivery": {
-        const arr = this.deliveryQueues.get(patch.agentId) ?? [];
-        arr.push(patch.event);
-        this.deliveryQueues.set(patch.agentId, arr);
+        this.queueDelivery(patch.agentId, patch.event);
         if (patch.agentId === this.peerId && this.onDelivery) {
           // Deduplicate against local deliveries
           const eventKey = JSON.stringify(patch.event);
@@ -672,9 +695,7 @@ export class MeshStore implements CommsStore {
     event: DeliveryEvent,
   ): Promise<void> {
     // Local delivery
-    const arr = this.deliveryQueues.get(agentId) ?? [];
-    arr.push(event);
-    this.deliveryQueues.set(agentId, arr);
+    this.queueDelivery(agentId, event);
 
     // Auto-emit delivered status for messages
     if (event.type === "room_message") {
@@ -688,37 +709,61 @@ export class MeshStore implements CommsStore {
       await this.emitDeliveryStatus(event.message.id, agentId, "delivered");
     }
 
-    if (agentId === this.peerId && this.onDelivery) {
-      // Deduplicate: skip if this exact event was already delivered locally.
-      // The mesh can echo delivery patches through multiple peer paths,
-      // causing applyPatch to fire onDelivery for the same event.
-      const eventKey = JSON.stringify(event);
-      if (this.localDeliveryKeys.has(eventKey)) return;
-      this.localDeliveryKeys.add(eventKey);
-      // Prevent unbounded growth — evict oldest when cap reached
-      if (this.localDeliveryKeys.size > 50) {
-        const oldest = this.localDeliveryKeys.values().next().value;
-        if (oldest !== undefined) this.localDeliveryKeys.delete(oldest);
-      }
-      void this.onDelivery(agentId, event);
-      // Auto-mark read — scheduled as a macrotask to yield to the event
-      // loop (see matching comment in applyPatch for rationale).
-      const timer = setTimeout(() => {
-        if (this.isShutDown) return;
-        if (event.type === "room_message") {
-          void this.markRead(event.message.id, agentId, event.message.room);
-        } else if (event.type === "dm") {
-          void this.markRead(event.message.id, agentId);
-        }
-      }, 0);
-      if (!this.isShutDown) this.pendingMarkReadTimers.push(timer);
-    }
+    this.fireLocalDelivery(agentId, event);
 
     // Remote delivery
     const patch: MeshStatePatch = { type: "delivery", agentId, event };
     await this.broadcastPatch(patch);
   }
 
+  /**
+   * Fire onDelivery for an event targeting this peer's own agent, deduped
+   * against events already delivered in this process. Used both for live
+   * deliveries and for replays of events that accumulated while this
+   * process was down (#28): the dedup set is per-process, so a replayed
+   * event this process never saw fires, and one it already handled does
+   * not.
+   */
+  private fireLocalDelivery(agentId: string, event: DeliveryEvent): void {
+    if (agentId !== this.peerId || !this.onDelivery) return;
+    // A room message or DM this agent has already read was already pushed
+    // and consumed: read receipts mutate the event between snapshots, so a
+    // plain structural key would miss and re-fire on the next sync (#28).
+    if (
+      (event.type === "room_message" || event.type === "dm") &&
+      event.message.readBy.includes(agentId)
+    ) {
+      return;
+    }
+    const eventKey = JSON.stringify(event);
+    if (this.localDeliveryKeys.has(eventKey)) return;
+    this.localDeliveryKeys.add(eventKey);
+    // Prevent unbounded growth — evict oldest when cap reached
+    if (this.localDeliveryKeys.size > 50) {
+      const oldest = this.localDeliveryKeys.values().next().value;
+      if (oldest !== undefined) this.localDeliveryKeys.delete(oldest);
+    }
+    void this.onDelivery(agentId, event);
+    // Delivered to this process, so no longer pending for it. Peers that
+    // never fired the event keep their copies, which is what a restart
+    // replays from (#28). Key-based match: replayed events are JSON clones.
+    const queued = this.deliveryQueues.get(agentId);
+    if (queued !== undefined) {
+      const idx = queued.findIndex((e) => JSON.stringify(e) === eventKey);
+      if (idx !== -1) queued.splice(idx, 1);
+    }
+    // Auto-mark read — scheduled as a macrotask to yield to the event loop.
+    const timer = setTimeout(() => {
+      if (this.isShutDown) return;
+      if (event.type === "room_message") {
+        void this.markRead(event.message.id, agentId, event.message.room);
+      } else if (event.type === "dm") {
+        void this.markRead(event.message.id, agentId);
+      }
+    }, 0);
+    if (!this.isShutDown) this.pendingMarkReadTimers.push(timer);
+  }
+
   private async deliverToRoom(
     roomId: string,
     event: DeliveryEvent,
@@ -1423,6 +1468,7 @@ export class MeshStore implements CommsStore {
       this.agents.delete(id);
       this.peerInfo.delete(id);
       this.identityCache.delete(id);
+      this.deliveryQueues.delete(id);
     }
   }
 

src/core/wire-protocol.ts

@@ -33,6 +33,13 @@ export interface SerialisedState {
   rooms: Record<string, Room>;
   messages: Record<string, RoomMessage[]>;
   dms: Record<string, DmMessage[]>;
+  /**
+   * Pending delivery events per target agent, replicated on every peer. A
+   * returning peer replays its own queue from the first snapshot it
+   * receives, so events pushed while its process was down still fire
+   * onDelivery (#28).
+   */
+  deliveryQueues: Record<string, DeliveryEvent[]>;
 }
 
 // ---------------------------------------------------------------------------

src/test/downtime-replay.integration.test.ts

@@ -0,0 +1,149 @@
+/**
+ * Integration test for issue #28: a bridge restarted with a persisted identity must be push-delivered the events that accumulated while its process was down, not just find them in synced history.
+ *
+ * Peer B goes away; A sends a room message while B is down; B restarts in the same identity slot and must fire onDelivery for the missed message.
+ */
+
+import * as assert from "node:assert/strict";
+import * as fs from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { MeshStore } from "../core/mesh-store.js";
+import { TlsTransport } from "../core/tls-transport.js";
+import { generateIdentity } from "../core/identity.js";
+import {
+  loadOrCreateIdentity,
+  releaseIdentityLock,
+  type IdentitySlot,
+} from "../core/identity-store.js";
+import type { DeliveryEvent } from "../core/types.js";
+import type { PeerIdentity } from "../core/identity.js";
+
+const TEST_PORT = 19896;
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+interface Peer {
+  store: MeshStore;
+  deliveries: DeliveryEvent[];
+}
+
+/** A peer wired like a real bridge: TLS transport, fingerprint peer ID. */
+function makePeer(identity: PeerIdentity): Peer {
+  const store = new MeshStore(TEST_PORT);
+  store.peerId = identity.fingerprint;
+  store.setTransport(new TlsTransport(store.events, identity));
+  const deliveries: DeliveryEvent[] = [];
+  return { store, deliveries };
+}
+
+/** Poll until the predicate holds, or fail with the message. */
+async function waitFor(
+  what: string,
+  check: () => Promise<boolean>,
+): Promise<void> {
+  for (let i = 0; i < 20; i++) {
+    if (await check()) return;
+    await sleep(100);
+  }
+  assert.ok(false, `timed out waiting for ${what}`);
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(tmpdir(), "agent-comms-downtime-"));
+  const slot: IdentitySlot = { harness: "pi", cwd: "/tmp/project", dir };
+
+  // A is a normal ephemeral bridge that stays up throughout.
+  const a = makePeer(generateIdentity());
+  await a.store.init();
+  await a.store.registerAgent({
+    name: "peer-a",
+    harness: "claude-code",
+    cwd: "/tmp/a",
+    pid: process.pid,
+    visibility: "visible",
+    tags: [],
+  });
+  await a.store.createRoom({
+    name: "downtime",
+    type: "public",
+    owner: a.store.peerId,
+    description: "issue 28 acceptance",
+  });
+  await sleep(200);
+
+  // B joins with a persisted identity and becomes a room member.
+  const identityB = loadOrCreateIdentity(slot);
+  const b1 = makePeer(identityB);
+  await b1.store.init();
+  // Registration must wait for the TLS data connections to establish (#23).
+  await sleep(300);
+  await b1.store.registerAgent({
+    name: "peer-b",
+    harness: "pi",
+    cwd: "/tmp/project",
+    pid: process.pid,
+    visibility: "visible",
+    tags: [],
+  });
+  await waitFor("the room to reach the joiner", async () => {
+    const rooms = await b1.store.listRooms(b1.store.peerId);
+    return rooms.some((room) => room.id === "downtime");
+  });
+  await b1.store.joinRoom("downtime", b1.store.peerId);
+  await waitFor("membership to reach the sender", async () => {
+    const room = await a.store.getRoom("downtime");
+    return room?.members.includes(b1.store.peerId) === true;
+  });
+
+  // B goes down.
+  await b1.store.shutdown();
+  await sleep(200);
+
+  // A sends a room message while B is down.
+  await a.store.sendRoomMessage(
+    "downtime",
+    a.store.peerId,
+    "while you were down",
+  );
+  await sleep(200);
+
+  // B restarts in the same slot: same identity, same agent ID.
+  const identityB2 = loadOrCreateIdentity(slot);
+  assert.equal(identityB2.fingerprint, identityB.fingerprint);
+  const b2 = makePeer(identityB2);
+  b2.store.onDelivery = (_id, ev) => {
+    b2.deliveries.push(ev);
+  };
+  await b2.store.init();
+  await b2.store.registerAgent({
+    name: "peer-b",
+    harness: "pi",
+    cwd: "/tmp/project",
+    pid: process.pid,
+    visibility: "visible",
+    tags: [],
+  });
+
+  // The message sent during downtime must push to the restarted bridge.
+  await waitFor(
+    "the downtime message to push to the restarted bridge",
+    async () =>
+      b2.deliveries.some(
+        (ev) =>
+          ev.type === "room_message" &&
+          ev.message.content === "while you were down",
+      ),
+  );
+
+  await b2.store.shutdown();
+  await a.store.shutdown();
+  releaseIdentityLock(slot);
+  console.log("✓ downtime messages push-deliver on restart");
+}
+
+main().catch((err: unknown) => {
+  console.error("Test failed:", err);
+  process.exitCode = 1;
+  // The sequence above keeps mesh handles open when it fails partway; exit explicitly so a failure cannot hang the runner.
+  process.exit(1);
+});