diff --git a/README.md b/README.md index 8c06621..048643e 100644 --- a/README.md +++ b/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 diff --git a/package.json b/package.json index 4955a73..e35d375 100644 --- a/package.json +++ b/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", diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 74403f2..3137a77 100644 --- a/src/core/mesh-store.ts +++ b/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 { // 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(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 { // 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); } } diff --git a/src/core/wire-protocol.ts b/src/core/wire-protocol.ts index 9d660af..d869b3b 100644 --- a/src/core/wire-protocol.ts +++ b/src/core/wire-protocol.ts @@ -33,6 +33,13 @@ export interface SerialisedState { rooms: Record; messages: Record; dms: Record; + /** + * 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; } // --------------------------------------------------------------------------- diff --git a/src/test/downtime-replay.integration.test.ts b/src/test/downtime-replay.integration.test.ts new file mode 100644 index 0000000..4675876 --- /dev/null +++ b/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, +): Promise { + 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 { + 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); +}); diff --git a/src/test/downtime-replay.test.ts b/src/test/downtime-replay.test.ts new file mode 100644 index 0000000..e29521f --- /dev/null +++ b/src/test/downtime-replay.test.ts @@ -0,0 +1,125 @@ +/** + * Unit tests for downtime delivery replay (#28): events that accumulated in peers' delivery queues while a target's process was down fire onDelivery when the target applies its first snapshot, without transport involvement. + */ + +import * as assert from "node:assert/strict"; +import { test } from "node:test"; +import { MeshStore } from "../core/mesh-store.js"; +import type { SerialisedState } from "../core/wire-protocol.js"; +import type { DeliveryEvent } from "../core/types.js"; + +/** A wire-accurate snapshot: production always applies parsed (cloned) state. */ +function snapshotOf(store: MeshStore): SerialisedState { + return structuredClone(store.serialise()); +} + +/** A local-only store: no transport start, so no ports and no flake. */ +function makeStore(): MeshStore { + return new MeshStore(); +} + +void test("events queued while the target was down replay on its first snapshot", async () => { + const sender = makeStore(); + const author = await sender.registerAgent({ + name: "author", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + // The target exists on the mesh (known to the sender) but its process is "down": modelled by only ever syncing snapshots into a future store. + const target = makeStore(); + const targetAgent = await target.registerAgent({ + name: "target", + harness: "claude-code", + cwd: "/tmp/t", + pid: process.pid, + visibility: "visible", + tags: [], + }); + // Make the sender aware of the target, then take the target's store away. + sender.applyStateSync(target.serialise()); + await sender.createRoom({ + name: "room", + type: "public", + owner: author.id, + description: "x", + }); + await sender.joinRoom("room", targetAgent.id); + + // While the target is down, a room message is sent to it. + await sender.sendRoomMessage("room", author.id, "while you were away"); + const pending = snapshotOf(sender).deliveryQueues[targetAgent.id]; + assert.ok(pending !== undefined && pending.length > 0); + + // The target returns (fresh process, same agent id) and receives the sender's snapshot: the pending event must fire onDelivery. + const returned = makeStore(); + const deliveries: DeliveryEvent[] = []; + returned.onDelivery = (_id, ev) => { + deliveries.push(ev); + }; + returned.peerId = targetAgent.id; + returned.applyStateSync(snapshotOf(sender)); + assert.equal( + deliveries.some( + (ev) => + ev.type === "room_message" && + ev.message.content === "while you were away", + ), + true, + ); + + // A second snapshot of the same state does not duplicate any push (the + // join notification replays too, exactly once). + returned.applyStateSync(snapshotOf(sender)); + assert.equal( + deliveries.filter( + (ev) => + ev.type === "room_message" && + ev.message.content === "while you were away", + ).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 () => { + const sender = makeStore(); + const author = await sender.registerAgent({ + name: "author", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const target = makeStore(); + const targetAgent = await target.registerAgent({ + name: "target", + harness: "claude-code", + cwd: "/tmp/t", + pid: process.pid, + visibility: "visible", + tags: [], + }); + sender.applyStateSync(target.serialise()); + await sender.createRoom({ + name: "room", + type: "public", + owner: author.id, + description: "x", + }); + await sender.joinRoom("room", targetAgent.id); + + for (let i = 0; i < 120; i++) { + await sender.sendRoomMessage("room", author.id, `msg-${String(i)}`); + } + const queued = snapshotOf(sender).deliveryQueues[targetAgent.id] ?? []; + assert.equal(queued.length, 100); + assert.equal( + queued[0]?.type === "room_message" && + queued[0].message.content === "msg-20", + true, + ); +});