From 02d556dbcfb93b48e45ed9a5811b4d0a4adc672f Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:46:10 +0100 Subject: [PATCH 1/7] feat(core): fan out room.send to every member over directed wire sends sendRoomMessage and sendDm now deliver via real, wire-authenticated directed room.send requests to each member (deliverRoomSendToMember), replacing the legacy broadcastPatch full-state replication and deliverLocallyAndBroadcast loop for message delivery specifically. Every send carries the sender's own persisted room:member token. handleRoomSend now branches on the room-path's own shape: a DM path stores a DmMessage and fires a dm event, matching sendDm's own dm-path keying, since a DM is just a room-path variant riding the same verb. It also reads room.send's own "reply" message-ref back into replyTo and its "streaming-behavior" extension field into StreamingBehavior, both previously only supported by the legacy path. A send to a member who isn't currently reachable is queued (pendingRoomSends) rather than thrown or dropped, and retried the moment that member's connection is (re)established (flushPendingRoomSends, hooked into handlePeerConnected) -- the fan-out's own substitute for the full-state-sync's automatic eventual consistency, since a direct request to a disconnected peer fails immediately with no protocol-level retry of its own. joinRoom's own admission gate no longer treats "this room is already known locally" as proof of membership for this store's own identity: the legacy full-state-sync replicates a room's metadata to every connected peer well before that peer is ever admitted, so a room already present in this.rooms said nothing about whether this store actually held a valid room:member token for it. The gate is now that token's presence, so a self-join always goes through real wire-level admission when this store has never actually been granted one. --- src/core/mesh-store.ts | 203 ++++++++++++++++++++++++++++++++++++----- 1 file changed, 178 insertions(+), 25 deletions(-) diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 878b3d7..26b4185 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -32,6 +32,7 @@ import { FederationManager } from "./federation.js"; import type { FedLink } from "./federation.js"; import { getCertificateFingerprint } from "./identity.js"; import { + bytesFromHex, bytesToHex, deviceIdFromHex, deviceIdToHex, @@ -48,6 +49,10 @@ import { roomJoinOkSchema, roomSendSchema, } from "wire-mesh-core/generated/protocol"; +import type { + CapabilityToken, + MessageRef, +} from "wire-mesh-core/generated/protocol"; import type { RoomVerbHandler } from "./room-router.js"; import { ROOM_MEMBER_CAPABILITY, @@ -63,6 +68,7 @@ import type { TransportEvents, } from "./transport.js"; import type { CommsStore } from "./comms-store.js"; +import { StreamingBehavior } from "./types.js"; import type { AgentIdentity, AgentStatus, @@ -74,7 +80,6 @@ import type { Room, RoomMessage, RoomType, - StreamingBehavior, Visibility, } from "./types.js"; import type { ListenerInfo, ListenerPolicy } from "./transport.js"; @@ -148,6 +153,11 @@ export class MeshStore implements CommsStore { private dms = new Map(); private deliveryQueues = new Map(); private identityCache = new Map(); + /** Directed room.send requests that failed because their target member wasn't reachable at send time, held for retry when that member's own connection is (re)established -- the wire-authenticated fan-out's substitute for the legacy full-state-sync's own automatic eventual consistency, since a direct request to a disconnected peer fails immediately with no protocol-level retry of its own. Keyed by member device-id hex, bounded oldest-first per member with the same cap ordinary delivery queues use. */ + private pendingRoomSends = new Map< + string, + { roomPath: string; params: Record }[] + >(); private transport: MeshTransport | undefined; private storeIdentity: MeshStoreIdentity | undefined; @@ -395,6 +405,7 @@ export class MeshStore implements CommsStore { state, }); } + await this.flushPendingRoomSends(handle.id); } /** @@ -1255,8 +1266,26 @@ export class MeshStore implements CommsStore { }; } + /** Reads the "reply" message-ref out of a room.send's own params (if any) and returns the hex id it names -- room.send's replyTo carries a single parent message, so the first reply-relation ref is the one that matters; any further refs are a future relation this handler doesn't yet act on. */ + private static replyToFromRefs( + refs: readonly MessageRef[] | undefined, + ): string | undefined { + const reply = refs?.find((ref) => ref.relation === "reply"); + return reply === undefined ? undefined : bytesToHex(reply.id); + } + + /** Reads room.send's own "streaming-behavior" extension field (open params tail, not a named schema field) and validates it against the same StreamingBehavior contract every other delivery path already enforces -- an unrecognised or malformed value is dropped rather than rejecting the whole send, matching core/room's own obligation to ignore what it doesn't understand instead of failing closed on an extension field. */ + private static streamingBehaviorFromParams( + params: Readonly>, + ): StreamingBehavior | undefined { + const raw = params["streaming-behavior"]; + if (raw === undefined) return undefined; + const result = StreamingBehavior.safeParse(raw); + return result.success ? result.data : undefined; + } + /** - * Receiving side of a directed room.send (P3.5): verifies the presented token against all six of core/room's own obligations, then delivers the message locally exactly once -- the manage-response this returns IS the delivery receipt, so there is no separate "delivered" event to emit the way the legacy broadcastPatch path needed one. + * Receiving side of a directed room.send (P3.5): verifies the presented token against all six of core/room's own obligations, then delivers the message locally exactly once -- the manage-response this returns IS the delivery receipt, so there is no separate "delivered" event to emit the way the legacy broadcastPatch path needed one. Branches on the room-path's own shape: an owner-named path stores a RoomMessage in this room's own history and fires a room_message event; a DM path stores a DmMessage keyed by the same dm-path sendDm already uses and fires a dm event -- both ride the identical room:member-gated verb, since a DM is just a room-path variant, not a separate verb. */ private async handleRoomSend( request: IncomingManageRequest, @@ -1286,20 +1315,44 @@ export class MeshStore implements CommsStore { return { result: "error", code: "malformed_params" }; } const params = parsedParams.data; + const replyTo = MeshStore.replyToFromRefs(params.refs); + const streamingBehavior = MeshStore.streamingBehaviorFromParams(params); + const id = bytesToHex(params["message-id"]); + const timestamp = new Date(params["sent-at"]).toISOString(); + + const parsedPath = parseRoomPath(roomPath); + let event: DeliveryEvent; + if (parsedPath.kind === "dm") { + const message: DmMessage = { + id, + from: handle.id, + to: this.peerId, + content: params.text, + timestamp, + readBy: [handle.id], + ...(streamingBehavior !== undefined && { streamingBehavior }), + }; + const history = this.dms.get(roomPath) ?? []; + history.push(message); + this.dms.set(roomPath, history); + event = { type: "dm", message }; + } else { + const message: RoomMessage = { + id, + from: handle.id, + room: roomPath, + content: params.text, + timestamp, + readBy: [handle.id], + ...(replyTo !== undefined && { replyTo }), + ...(streamingBehavior !== undefined && { streamingBehavior }), + }; + const history = this.messages.get(roomPath) ?? []; + history.push(message); + this.messages.set(roomPath, history); + event = { type: "room_message", message }; + } - const message: RoomMessage = { - id: bytesToHex(params["message-id"]), - from: handle.id, - room: roomPath, - content: params.text, - timestamp: new Date(params["sent-at"]).toISOString(), - readBy: [handle.id], - }; - const history = this.messages.get(roomPath) ?? []; - history.push(message); - this.messages.set(roomPath, history); - - const event: DeliveryEvent = { type: "room_message", message }; this.queueDelivery(this.peerId, event); this.fireLocalDelivery(this.peerId, event); @@ -1344,6 +1397,58 @@ export class MeshStore implements CommsStore { } } + /** Records a room.send that couldn't reach memberId right now, for a later flushPendingRoomSends to retry once that member reconnects. Bounded oldest-first with the same cap ordinary delivery queues use, so an indefinitely-offline member cannot grow this without limit. */ + private queuePendingRoomSend( + memberId: string, + roomPath: string, + params: Record, + ): void { + const queue = this.pendingRoomSends.get(memberId) ?? []; + queue.push({ roomPath, params }); + if (queue.length > MAX_QUEUED_DELIVERIES_PER_AGENT) { + queue.splice(0, queue.length - MAX_QUEUED_DELIVERIES_PER_AGENT); + } + this.pendingRoomSends.set(memberId, queue); + } + + /** + * Sends one directed room.send to a single member, queuing it for retry instead of throwing when the member isn't currently reachable -- the fan-out's own per-recipient primitive, distinct from sendRoomMessageDirected's deliberate throw-on-failure contract for a caller sending to one specific, known recipient. Silently drops a send this store no longer holds a token for (no longer a member of the room) rather than queuing something that will only fail again on retry. + */ + private async deliverRoomSendToMember( + memberId: string, + roomPath: string, + token: CapabilityToken, + params: Record, + ): Promise { + const outcome = await this.requireTransport().sendRoomRequest( + memberId, + { verb: ROOM_MEMBER_CAPABILITY, params }, + { kind: "room", path: roomPath }, + token, + ); + if (outcome.result !== "ok") { + this.queuePendingRoomSend(memberId, roomPath, params); + } + } + + /** Retries every room.send queued for memberId since it was last reachable, dropping (not re-queuing) any whose room this store no longer holds a token for. Called once a connection to memberId is (re)established -- handlePeerConnected fires for both a fresh introduction and a reconnection after downtime, exactly the two cases a queued send needs to be retried on. */ + private async flushPendingRoomSends(memberId: string): Promise { + const queue = this.pendingRoomSends.get(memberId); + if (queue === undefined || queue.length === 0) return; + this.pendingRoomSends.delete(memberId); + const { slot } = this.requireIdentity(); + for (const pending of queue) { + const token = loadRoomTokens(slot)[pending.roomPath]; + if (token === undefined) continue; + await this.deliverRoomSendToMember( + memberId, + pending.roomPath, + token, + pending.params, + ); + } + } + /** * Owner-side admission for an incoming room.join request, against either a named room this store owns or a DM path this store is a participant of. Named-room admission always needs a human decision; DM admission needs one only for the party being contacted first -- the reply half of the two-round consent flow (section 6) auto-approves, since a reply on a path this node itself opened is not unsolicited contact. */ @@ -1608,7 +1713,16 @@ export class MeshStore implements CommsStore { saveRoomToken(slot, dmPath, parsedOutcome.data["granted-token"]); } + /** + * Joins a room. For this store's own identity, "already known locally" is not the right gate for skipping real admission: the legacy full-state-sync replicates a room's metadata to every mesh-connected peer the moment it's created, well before that peer has ever been admitted, so a room already present in this.rooms says nothing about whether this store actually holds a valid room:member token for it. The real gate is that token's presence -- absent, this always goes through joinRemoteRoom's real wire-level admission regardless of what this.rooms already knows, so a peer that merely heard about a room never mistakes hearing about it for having joined it. Joining on behalf of a DIFFERENT agentId (this store's own convergence/admin bookkeeping, exercised directly by state-sync-convergence.test.ts) is untouched -- that's a pure local CRDT mutation with no admission concept at all. + */ async joinRoom(roomId: string, agentId: string): Promise { + if (agentId === this.peerId) { + const { slot } = this.requireIdentity(); + if (loadRoomTokens(slot)[roomId] === undefined) { + return this.joinRemoteRoom(roomId, agentId); + } + } const room = this.rooms.get(roomId); if (!room) return this.joinRemoteRoom(roomId, agentId); @@ -1819,6 +1933,9 @@ export class MeshStore implements CommsStore { // CommsStore — Messages // ----------------------------------------------------------------------- + /** + * Sends a room message via a real, wire-authenticated room.send fan-out (P3.5): one directed request per member, each carrying this sender's own persisted room:member token, rather than the legacy broadcastPatch's full-state replication. A member unreachable right now is queued for retry (see deliverRoomSendToMember/flushPendingRoomSends) instead of blocking or failing the whole send -- delivery to any one recipient is independent of every other. + */ async sendRoomMessage( roomId: string, from: string, @@ -1832,34 +1949,50 @@ export class MeshStore implements CommsStore { if (!room.members.includes(from)) throw new CommsError(`Not a member of ${roomId}`, "NOT_MEMBER"); - const id = `${String(Date.now())}-${nanoid(6)}`; + const { slot, clock } = this.requireIdentity(); + const token = loadRoomTokens(slot)[roomId]; + if (token === undefined) { + throw new CommsError(`No room:member token for ${roomId}`, "NOT_MEMBER"); + } + + const messageId = randomId(); + const id = bytesToHex(messageId); const message: RoomMessage = { id, from, room: roomId, content, timestamp: new Date().toISOString(), - replyTo, readBy: [from], + ...(replyTo !== undefined && { replyTo }), ...(streamingBehavior !== undefined && { streamingBehavior }), }; const arr = this.messages.get(roomId) ?? []; arr.push(message); this.messages.set(roomId, arr); - await this.broadcastPatch({ type: "message_add", roomId, message }); // Forward to federated links if the room is federated if (room.federated) { await this.federation.forwardRoomMessage(roomId, message); } + const params: Record = { + verb: "room.send", + "message-id": messageId, + "sent-at": clock.now(), + text: content, + ...(replyTo !== undefined && { + refs: [{ id: bytesFromHex(replyTo), relation: "reply" }], + }), + ...(streamingBehavior !== undefined && { + "streaming-behavior": streamingBehavior, + }), + }; + for (const memberId of room.members) { if (memberId !== from) { - await this.deliverLocallyAndBroadcast(memberId, { - type: "room_message", - message, - }); + await this.deliverRoomSendToMember(memberId, roomId, token, params); } } @@ -1880,6 +2013,9 @@ export class MeshStore implements CommsStore { // CommsStore — DMs // ----------------------------------------------------------------------- + /** + * Sends a DM via the same wire-authenticated room.send fan-out sendRoomMessage uses (P3.5): a DM is just a dm-shaped room path with exactly one other member, so it rides the identical mechanism rather than a separate one. Self-DM is the one exception -- a purely local scratchpad note that never leaves the process, so it needs no token and no wire round trip at all. + */ async sendDm( from: string, to: string, @@ -1894,7 +2030,8 @@ export class MeshStore implements CommsStore { throw new CommsError(`Cannot DM agent ${to}`, "AGENT_NOT_FOUND"); } - const id = `${String(Date.now())}-${nanoid(6)}`; + const messageId = randomId(); + const id = bytesToHex(messageId); const message: DmMessage = { id, from, @@ -1910,9 +2047,25 @@ export class MeshStore implements CommsStore { const arr = this.dms.get(key) ?? []; arr.push(message); this.dms.set(key, arr); - await this.broadcastPatch({ type: "dm_add", key, message }); - await this.deliverLocallyAndBroadcast(to, { type: "dm", message }); + if (to !== from) { + const { slot, clock } = this.requireIdentity(); + const token = loadRoomTokens(slot)[key]; + if (token === undefined) { + throw new CommsError(`No room:member token for ${key}`, "NOT_MEMBER"); + } + const params: Record = { + verb: "room.send", + "message-id": messageId, + "sent-at": clock.now(), + text: content, + ...(streamingBehavior !== undefined && { + "streaming-behavior": streamingBehavior, + }), + }; + await this.deliverRoomSendToMember(to, key, token, params); + } + return message; } From 3062b65bcb2fe8adee4f973757893e6dfd5cd171 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:46:21 +0100 Subject: [PATCH 2/7] test(core): cover the directed room.send retry queue Three unit tests against a fake MeshTransport: a send to an unreachable member is queued and retried once handlePeerConnected fires for it; the queue is bounded oldest-first at the same cap ordinary delivery queues use; and a flush drops (rather than re-queuing) a send whose room this store no longer holds a token for. --- src/test/room-send-retry-queue.test.ts | 182 +++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 src/test/room-send-retry-queue.test.ts diff --git a/src/test/room-send-retry-queue.test.ts b/src/test/room-send-retry-queue.test.ts new file mode 100644 index 0000000..571a984 --- /dev/null +++ b/src/test/room-send-retry-queue.test.ts @@ -0,0 +1,182 @@ +/** + * Unit tests for the directed room.send fan-out's own retry queue (P3.5): a send that fails because its target member isn't currently reachable is queued rather than thrown or dropped, and retried the moment that member's connection is (re)established -- handlePeerConnected fires for exactly that event, so these tests trigger it directly via store.events.onPeerConnected rather than driving a real second peer (see room-send-retry.integration.test.ts for the real, two-peer version of this same acceptance behaviour). + */ + +import * as assert from "node:assert/strict"; +import { test } from "node:test"; +import type { ManageOutcome } from "wire-mesh-core/domain/mesh-session"; +import { MeshStore } from "../core/mesh-store.js"; +import type { MeshTransport } from "../core/transport.js"; +import { deleteRoomToken } from "../core/identity-store.js"; +import { waitFor, wireTestTransport } from "./test-transport.js"; + +const MEMBER_ID = "b".repeat(64); +const QUEUE_CAP = 100; + +/** A MeshTransport whose sendRoomRequest always fails until told otherwise -- flippable mid-test to simulate the member becoming reachable, and recording every attempt made against it (including retries) so a test can assert exactly which sends were retried. */ +function fakeTransport(): { + transport: MeshTransport; + attempts: unknown[]; + setConnected: (connected: boolean) => void; +} { + let connected = false; + const attempts: unknown[] = []; + const transport: MeshTransport = { + dataPort: 0, + isCoordinator: false, + hasCoordinatorConnection: false, + startDataServer: async () => {}, + connectToCoordinator: async () => {}, + becomeCoordinator: async () => {}, + connectToPeer: async () => {}, + send: async () => {}, + acceptConnection: async () => {}, + rejectConnection: async () => {}, + connectToRemote: async () => {}, + broadcast: async () => {}, + sendRoomRequest: async (_memberId, command): Promise => { + attempts.push(command.params); + if (!connected) { + return { result: "error", code: "not_connected" }; + } + return { result: "ok" }; + }, + addListener: async () => "id", + removeListener: async () => {}, + listListeners: () => [], + shutdown: async () => {}, + unref: () => {}, + }; + return { + transport, + attempts, + setConnected: (value: boolean) => { + connected = value; + }, + }; +} + +void test("a room.send to an unreachable member is queued and retried once it reconnects", async () => { + const store = new MeshStore(); + await wireTestTransport(store); + const owner = await store.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const room = await store.createRoom({ + name: "general", + type: "public", + owner: owner.id, + description: "", + }); + + await store.joinRoom(room.id, MEMBER_ID); + const { transport, attempts, setConnected } = fakeTransport(); + store.setTransport(transport); + + await store.sendRoomMessage(room.id, owner.id, "hello"); + assert.equal(attempts.length, 1, "the first, failed attempt was made"); + + setConnected(true); + store.events.onPeerConnected( + { id: MEMBER_ID }, + { id: MEMBER_ID, port: 0, startedAt: new Date().toISOString() }, + ); + + await waitFor( + () => attempts.length === 2, + "the queued send to retry once the member reconnects", + ); +}); + +void test("the retry queue is bounded oldest-first per member", async () => { + const store = new MeshStore(); + await wireTestTransport(store); + const owner = await store.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const room = await store.createRoom({ + name: "general", + type: "public", + owner: owner.id, + description: "", + }); + + await store.joinRoom(room.id, MEMBER_ID); + const { transport, attempts, setConnected } = fakeTransport(); + store.setTransport(transport); + + const overflow = 20; + for (let i = 0; i < QUEUE_CAP + overflow; i++) { + await store.sendRoomMessage(room.id, owner.id, `msg-${String(i)}`); + } + assert.equal(attempts.length, QUEUE_CAP + overflow); + + attempts.length = 0; + setConnected(true); + store.events.onPeerConnected( + { id: MEMBER_ID }, + { id: MEMBER_ID, port: 0, startedAt: new Date().toISOString() }, + ); + + await waitFor( + () => attempts.length === QUEUE_CAP, + "exactly the cap's worth of retries to fire", + ); + + function textOf(params: unknown): string | undefined { + if (typeof params !== "object" || params === null) return undefined; + if (!("text" in params)) return undefined; + return typeof params.text === "string" ? params.text : undefined; + } + assert.equal(textOf(attempts[0]), `msg-${String(overflow)}`); + assert.equal( + textOf(attempts[attempts.length - 1]), + `msg-${String(QUEUE_CAP + overflow - 1)}`, + ); +}); + +void test("a flush drops (not re-queues) a send whose room this store no longer holds a token for", async () => { + const store = new MeshStore(); + const slot = await wireTestTransport(store); + const owner = await store.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const room = await store.createRoom({ + name: "general", + type: "public", + owner: owner.id, + description: "", + }); + + await store.joinRoom(room.id, MEMBER_ID); + const { transport, attempts } = fakeTransport(); + store.setTransport(transport); + + await store.sendRoomMessage(room.id, owner.id, "hello"); + assert.equal(attempts.length, 1); + + deleteRoomToken(slot, room.id); + + store.events.onPeerConnected( + { id: MEMBER_ID }, + { id: MEMBER_ID, port: 0, startedAt: new Date().toISOString() }, + ); + // No token to present -- give the flush a moment to run, then confirm it made no further attempt. + await new Promise((resolve) => setTimeout(resolve, 50)); + assert.equal(attempts.length, 1); +}); From 4b70f61e2f992a5861b87418898a38570231eca8 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:46:29 +0100 Subject: [PATCH 3/7] test: replace downtime-replay.integration.test.ts with the retry-queue version Same acceptance scenario as issue #28 (a room message sent while its recipient is offline must still push-deliver once the recipient reconnects), re-verified against the directed fan-out's own retry mechanism instead of the legacy deliveryQueues/applyStateSync machinery the fan-out no longer uses for message delivery. --- ...ts => room-send-retry.integration.test.ts} | 41 ++++++++++--------- 1 file changed, 21 insertions(+), 20 deletions(-) rename src/test/{downtime-replay.integration.test.ts => room-send-retry.integration.test.ts} (70%) diff --git a/src/test/downtime-replay.integration.test.ts b/src/test/room-send-retry.integration.test.ts similarity index 70% rename from src/test/downtime-replay.integration.test.ts rename to src/test/room-send-retry.integration.test.ts index 08841c6..38b7b28 100644 --- a/src/test/downtime-replay.integration.test.ts +++ b/src/test/room-send-retry.integration.test.ts @@ -1,7 +1,5 @@ /** - * 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. + * Integration test for issue #28's own acceptance behaviour, re-verified against P3.5's directed fan-out: a room message sent while its recipient is offline must still reach the recipient once it reconnects. Unlike the legacy full-state-sync this replaces, a directed room.send to a disconnected member fails immediately rather than eventually converging, so the fan-out queues it for retry (pendingRoomSends) and flushes that queue the moment the member's connection is (re)established (handlePeerConnected). Replaces downtime-replay.integration.test.ts, whose own scenario relied on deliveryQueues/applyStateSync -- machinery this fan-out no longer uses for message delivery. */ import * as assert from "node:assert/strict"; @@ -23,7 +21,7 @@ import type { DeliveryEvent } from "../core/types.js"; import type { PeerIdentity } from "../core/identity.js"; import { ownerNamedRoomPath } from "../core/room-path.js"; -const TEST_PORT = 19896; +const TEST_PORT = 19897; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); interface Peer { @@ -64,12 +62,12 @@ async function waitFor( } async function main(): Promise { - const dir = fs.mkdtempSync(path.join(tmpdir(), "agent-comms-downtime-")); + const dir = fs.mkdtempSync(path.join(tmpdir(), "agent-comms-send-retry-")); const slot: IdentitySlot = { harness: "pi", cwd: "/tmp/project", dir }; const slotA: IdentitySlot = { harness: "claude-code", cwd: "/tmp/a", - dir: fs.mkdtempSync(path.join(tmpdir(), "agent-comms-downtime-a-")), + dir: fs.mkdtempSync(path.join(tmpdir(), "agent-comms-send-retry-a-")), }; // A is a normal ephemeral bridge that stays up throughout. @@ -92,7 +90,7 @@ async function main(): Promise { }); await sleep(200); - // B joins with a persisted identity and becomes a room member. + // B joins with a persisted identity and becomes a room member -- a real admitted join, since the room already replicating to B via legacy full-state-sync says nothing about B holding a room:member token for it. const identityB = loadOrCreateIdentity(slot); const b1 = await makePeer(identityB, slot); await b1.store.init(); @@ -106,25 +104,28 @@ async function main(): Promise { 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 === roomId); - }); - await b1.store.joinRoom(roomId, b1.store.peerId); - await waitFor("membership to reach the sender", async () => { - const room = await a.store.getRoom(roomId); - return room?.members.includes(b1.store.peerId) === true; - }); + const joinPromise = b1.store.joinRoom(roomId, b1.store.peerId); + await waitFor("A to see B's pending join request", async () => + Promise.resolve( + a.store + .listPendingRoomJoins() + .some( + (p) => p.roomPath === roomId && p.requesterId === b1.store.peerId, + ), + ), + ); + a.store.acceptRoomJoin(roomId, b1.store.peerId); + await joinPromise; // B goes down. await b1.store.shutdown(); await sleep(200); - // A sends a room message while B is down. + // A sends a room message while B is down: the directed send to B fails immediately (B isn't connected), so it's queued for retry rather than thrown or silently dropped. await a.store.sendRoomMessage(roomId, a.store.peerId, "while you were down"); await sleep(200); - // B restarts in the same slot: same identity, same agent ID. + // B restarts in the same slot: same identity, same agent ID, same persisted room:member token -- the queued send retries the moment the reconnection to A completes. const identityB2 = loadOrCreateIdentity(slot); assert.equal( deviceIdToHex(Uint8Array.from(identityB2.deviceId)), @@ -144,7 +145,7 @@ async function main(): Promise { tags: [], }); - // The message sent during downtime must push to the restarted bridge. + // The message sent during downtime must push to the restarted bridge, once A's own pendingRoomSends queue flushes against the new connection. await waitFor( "the downtime message to push to the restarted bridge", async () => @@ -158,7 +159,7 @@ async function main(): Promise { await b2.store.shutdown(); await a.store.shutdown(); releaseIdentityLock(slot); - console.log("✓ downtime messages push-deliver on restart"); + console.log("✓ a queued room.send retries and delivers on reconnect"); } main().catch((err: unknown) => { From 9bd1999bb789c2ab1fdcdd3f394fe72a411ca2b5 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:46:38 +0100 Subject: [PATCH 4/7] test: split invite replay out of downtime-replay.test.ts The other two tests in that file covered room_message replay via applyStateSync/deliveryQueues -- machinery the directed fan-out no longer uses for message delivery (see room-send-retry.integration.test.ts for that behaviour's own replacement). Invite delivery itself is untouched by that migration, so its own test keeps the legacy replay path it actually exercises, just renamed to reflect that it's now the only thing left in the file. --- src/test/downtime-replay.test.ts | 189 ------------------------------- src/test/invite-replay.test.ts | 75 ++++++++++++ 2 files changed, 75 insertions(+), 189 deletions(-) delete mode 100644 src/test/downtime-replay.test.ts create mode 100644 src/test/invite-replay.test.ts diff --git a/src/test/downtime-replay.test.ts b/src/test/downtime-replay.test.ts deleted file mode 100644 index 4d52ea4..0000000 --- a/src/test/downtime-replay.test.ts +++ /dev/null @@ -1,189 +0,0 @@ -/** - * 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"; -import { ownerNamedRoomPath } from "../core/room-path.js"; -import { wireTestTransport } from "./test-transport.js"; - -/** A wire-accurate snapshot: production always applies parsed (cloned) state. */ -function snapshotOf(store: MeshStore): SerialisedState { - return structuredClone(store.serialise()); -} - -/** A local-only store: transport is set (registerAgent's own broadcastPatch needs one) but never started, so no ports and no flake -- the stores in these tests never actually connect. */ -async function makeStore(): Promise { - const store = new MeshStore(); - await wireTestTransport(store); - return store; -} - -void test("events queued while the target was down replay on its first snapshot", async () => { - const sender = await 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 = await 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()); - const roomId = ownerNamedRoomPath(author.id, "room"); - await sender.createRoom({ - name: "room", - type: "public", - owner: author.id, - description: "x", - }); - await sender.joinRoom(roomId, targetAgent.id); - - // While the target is down, a room message is sent to it. - await sender.sendRoomMessage(roomId, 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 = await 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, - ); - - // 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( - (ev) => - ev.type === "room_message" && - ev.message.content === "while you were away", - ).length, - 1, - ); -}); - -void test("a queue is bounded oldest-first so downtime cannot grow it without limit", async () => { - const sender = await makeStore(); - const author = await sender.registerAgent({ - name: "author", - harness: "pi", - cwd: "/tmp/p", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const target = await makeStore(); - const targetAgent = await target.registerAgent({ - name: "target", - harness: "claude-code", - cwd: "/tmp/t", - pid: process.pid, - visibility: "visible", - tags: [], - }); - sender.applyStateSync(target.serialise()); - const roomId = ownerNamedRoomPath(author.id, "room"); - await sender.createRoom({ - name: "room", - type: "public", - owner: author.id, - description: "x", - }); - await sender.joinRoom(roomId, targetAgent.id); - - for (let i = 0; i < 120; i++) { - await sender.sendRoomMessage(roomId, 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, - ); -}); - -void test("a pending invite replays until accepted or declined", async () => { - const sender = await makeStore(); - const author = await sender.registerAgent({ - name: "owner", - harness: "pi", - cwd: "/tmp/p", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const target = await makeStore(); - const targetAgent = await target.registerAgent({ - name: "invitee", - harness: "claude-code", - cwd: "/tmp/t", - pid: process.pid, - visibility: "visible", - tags: [], - }); - sender.applyStateSync(target.serialise()); - const roomId = ownerNamedRoomPath(author.id, "private-room"); - await sender.createRoom({ - name: "private-room", - type: "private", - owner: author.id, - description: "x", - }); - await sender.inviteToRoom(roomId, targetAgent.id, author.id); - - const returned = await 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(roomId, targetAgent.id, "not now"); - const declined = await 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); -}); diff --git a/src/test/invite-replay.test.ts b/src/test/invite-replay.test.ts new file mode 100644 index 0000000..0f99cc8 --- /dev/null +++ b/src/test/invite-replay.test.ts @@ -0,0 +1,75 @@ +/** + * Unit test for invite replay (#28): a pending room invite that accumulated in a peer's delivery queue while its target's process was down fires onDelivery when the target applies its first snapshot, without transport involvement, and stops replaying once the invite is consumed (accepted or declined). Split out of downtime-replay.test.ts, whose own other two tests covered room_message replay via applyStateSync/deliveryQueues -- machinery P3.5's directed room.send fan-out no longer uses for message delivery (see room-send-retry.integration.test.ts for that behaviour's own replacement). Invite delivery itself is untouched by that migration: room.invite's own receiving side is still P3.6 work, so this legacy replay path remains the real mechanism for it today. + */ + +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"; +import { ownerNamedRoomPath } from "../core/room-path.js"; +import { wireTestTransport } from "./test-transport.js"; + +/** A wire-accurate snapshot: production always applies parsed (cloned) state. */ +function snapshotOf(store: MeshStore): SerialisedState { + return structuredClone(store.serialise()); +} + +/** A local-only store: transport is set (registerAgent's own broadcastPatch needs one) but never started, so no ports and no flake -- the stores in these tests never actually connect. */ +async function makeStore(): Promise { + const store = new MeshStore(); + await wireTestTransport(store); + return store; +} + +void test("a pending invite replays until accepted or declined", async () => { + const sender = await makeStore(); + const author = await sender.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const target = await makeStore(); + const targetAgent = await target.registerAgent({ + name: "invitee", + harness: "claude-code", + cwd: "/tmp/t", + pid: process.pid, + visibility: "visible", + tags: [], + }); + sender.applyStateSync(target.serialise()); + const roomId = ownerNamedRoomPath(author.id, "private-room"); + await sender.createRoom({ + name: "private-room", + type: "private", + owner: author.id, + description: "x", + }); + await sender.inviteToRoom(roomId, targetAgent.id, author.id); + + const returned = await 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(roomId, targetAgent.id, "not now"); + const declined = await 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); +}); From d4256a3a47527080c8fabed84b534b0f0fa85ac1 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:47:00 +0100 Subject: [PATCH 5/7] test: admit real room joins before sending across integration tests joinRoom's own admission gate now requires this store to actually hold a room:member token before treating a self-join as already satisfied, so every test that joined a room already known locally via legacy full-state-sync (and every DM sent with no prior consent) needs to drive the real admission/consent round trip: poll for the pending request, accept it, and (for the admitting side specifically) wait for its own local membership record to reflect the join before sending, since acceptRoomJoin only resolves the held-open request's own promise and its membership update lands on a later tick. mesh-smoke.runner.ts and delivery-receipt.helper.ts each gained this as an inline poll-and-accept step; mesh-e2e and identity-restart gained it via the same pattern already used elsewhere in this codebase's own integration tests. --- src/test/delivery-receipt.helper.ts | 71 ++++++++++++++++--- src/test/identity-restart.integration.test.ts | 29 +++++++- src/test/mesh-e2e.integration.test.ts | 35 +++++++-- src/test/mesh-smoke.runner.ts | 16 ++++- 4 files changed, 134 insertions(+), 17 deletions(-) diff --git a/src/test/delivery-receipt.helper.ts b/src/test/delivery-receipt.helper.ts index 4e650f8..0832071 100644 --- a/src/test/delivery-receipt.helper.ts +++ b/src/test/delivery-receipt.helper.ts @@ -56,6 +56,59 @@ async function cleanup(...stores: MeshStore[]): Promise { } } +/** Polls until predicate holds or a fixed budget elapses -- these helpers wait on a real, cross-process wire round trip (admission, DM consent), not a fixed sleep. */ +async function pollUntil( + predicate: () => boolean | Promise, + what: string, +): Promise { + const deadline = Date.now() + 5000; + while (!(await predicate())) { + if (Date.now() >= deadline) { + throw new Error(`timed out waiting for ${what}`); + } + await sleep(50); + } +} + +/** + * member's own real, admitted room.join: the room already replicated to member via legacy full-state-sync, but knowing about a room is not the same as holding a room:member token for it -- member's own join still goes through real wire-level admission, held open until owner approves it. Also waits for owner's own local membership record to actually reflect the join before returning: acceptRoomJoin only resolves the held-open request's promise, and its continuation (minting the token, updating the room) runs on a later tick, so a caller sending immediately afterward could otherwise read owner's room.members before that update lands. + */ +async function joinAndAccept( + owner: MeshStore, + member: MeshStore, + roomId: string, +): Promise { + const joinPromise = member.joinRoom(roomId, member.peerId); + await pollUntil( + () => + owner + .listPendingRoomJoins() + .some((p) => p.roomPath === roomId && p.requesterId === member.peerId), + "owner to see member's pending join request", + ); + owner.acceptRoomJoin(roomId, member.peerId); + await joinPromise; + await pollUntil(async () => { + const room = await owner.getRoom(roomId); + return room?.members.includes(member.peerId) ?? false; + }, "owner's own room record to reflect the join"); +} + +/** The two-round DM consent flow (section 6): from's own outbound room.join is what authorises the DM, and to (the party contacted first) still needs a human decision. */ +async function dmConsent(from: MeshStore, to: MeshStore): Promise { + const accessPromise = from.requestDmAccess(to.peerId); + await pollUntil( + () => to.listPendingRoomJoins().some((p) => p.requesterId === from.peerId), + "to see from's pending DM request", + ); + const pending = to + .listPendingRoomJoins() + .find((p) => p.requesterId === from.peerId); + if (pending === undefined) throw new Error("pending DM request vanished"); + to.acceptRoomJoin(pending.roomPath, from.peerId); + await accessPromise; +} + // --------------------------------------------------------------------------- // Test implementations // --------------------------------------------------------------------------- @@ -103,8 +156,7 @@ async function testPushRoom(): Promise { description: "Push delivery test", }); await sleep(200); - await b.joinRoom(room.id, b.peerId); - await sleep(200); + await joinAndAccept(a, b, room.id); deliveriesB.length = 0; await a.sendRoomMessage(room.id, a.peerId, "Hello push!"); @@ -161,6 +213,8 @@ async function testPushDm(): Promise { }); await sleep(300); + await dmConsent(a, b); + deliveriesB.length = 0; await a.sendDm(a.peerId, b.peerId, "Direct push!"); await sleep(300); @@ -213,8 +267,7 @@ async function testDrainRoom(): Promise { description: "Drain delivery test", }); await sleep(200); - await b.joinRoom(room.id, b.peerId); - await sleep(200); + await joinAndAccept(a, b, room.id); await a.sendRoomMessage(room.id, a.peerId, "Hello drain!"); await sleep(300); @@ -269,6 +322,7 @@ async function testDrainDm(): Promise { }); await sleep(300); + await dmConsent(a, b); await a.sendDm(a.peerId, b.peerId, "Direct drain!"); await sleep(300); @@ -327,8 +381,7 @@ async function testReadReceiptPush(): Promise { description: "Read receipt push test", }); await sleep(200); - await b.joinRoom(room.id, b.peerId); - await sleep(200); + await joinAndAccept(a, b, room.id); deliveriesA.length = 0; await a.sendRoomMessage(room.id, a.peerId, "Read me"); @@ -384,8 +437,7 @@ async function testReadReceiptDrain(): Promise { description: "Read receipt drain test", }); await sleep(200); - await b.joinRoom(room.id, b.peerId); - await sleep(200); + await joinAndAccept(a, b, room.id); deliveriesA.length = 0; await a.sendRoomMessage(room.id, a.peerId, "Drain then read"); @@ -447,8 +499,7 @@ async function testReadbyArray(): Promise { description: "readBy test", }); await sleep(200); - await b.joinRoom(room.id, b.peerId); - await sleep(200); + await joinAndAccept(a, b, room.id); const msg = await a.sendRoomMessage(room.id, a.peerId, "Check readBy"); await sleep(800); diff --git a/src/test/identity-restart.integration.test.ts b/src/test/identity-restart.integration.test.ts index 0101326..3a83d48 100644 --- a/src/test/identity-restart.integration.test.ts +++ b/src/test/identity-restart.integration.test.ts @@ -108,7 +108,17 @@ async function main(): Promise { description: "issue 14 acceptance", }); await sleep(200); - await b.store.joinRoom(roomId, b.store.peerId); + // The room already replicated to B via legacy full-state-sync, but knowing about a room is not the same as holding a room:member token for it -- B's own join still goes through real wire-level admission, held open until A approves it. + const joinPromise = b.store.joinRoom(roomId, b.store.peerId); + await waitFor("A to see B's pending join request", async () => + Promise.resolve( + a1.store + .listPendingRoomJoins() + .some((p) => p.roomPath === roomId && p.requesterId === b.store.peerId), + ), + ); + a1.store.acceptRoomJoin(roomId, b.store.peerId); + await joinPromise; await sleep(200); const agentIdA = a1.store.peerId; @@ -152,7 +162,22 @@ async function main(): Promise { ), ); - // DMs targeted at the persisted ID must deliver too. + // DMs targeted at the persisted ID must deliver too -- the two-round consent flow first. + const dmAccessPromise = b.store.requestDmAccess(agentIdA); + await waitFor("restarted A to see B's pending DM request", async () => + Promise.resolve( + a2.store + .listPendingRoomJoins() + .some((p) => p.requesterId === b.store.peerId), + ), + ); + const pendingDm = a2.store + .listPendingRoomJoins() + .find((p) => p.requesterId === b.store.peerId); + assert.ok(pendingDm); + a2.store.acceptRoomJoin(pendingDm.roomPath, b.store.peerId); + await dmAccessPromise; + await b.store.sendDm(b.store.peerId, agentIdA, "dm after restart"); await waitFor("the DM to push to restarted A", async () => a2.deliveries.some( diff --git a/src/test/mesh-e2e.integration.test.ts b/src/test/mesh-e2e.integration.test.ts index 6306c3e..95dde93 100644 --- a/src/test/mesh-e2e.integration.test.ts +++ b/src/test/mesh-e2e.integration.test.ts @@ -10,7 +10,7 @@ import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; import type { DeliveryEvent } from "../core/types.js"; import * as assert from "node:assert/strict"; -import { wireTestTransport } from "./test-transport.js"; +import { waitFor, wireTestTransport } from "./test-transport.js"; const E2E_PORT = 19878; @@ -87,9 +87,20 @@ async function main(): Promise { console.log(` B sees ${String(roomsB.length)} room(s)`); assert.ok(roomsB.length >= 1, "B should see the room"); - // --- Test: B joins room --- + // --- Test: B joins room --- The room already replicated to B via legacy full-state-sync, but knowing about a room is not the same as holding a room:member token for it -- B's own join still goes through real wire-level admission, held open until A approves it. console.log("Test: B joins room..."); - await b.store.joinRoom(room.id, b.store.peerId); + const joinPromise = b.store.joinRoom(room.id, b.store.peerId); + await waitFor( + () => + a.store + .listPendingRoomJoins() + .some( + (p) => p.roomPath === room.id && p.requesterId === b.store.peerId, + ), + "A to see B's pending join request", + ); + a.store.acceptRoomJoin(room.id, b.store.peerId); + await joinPromise; await sleep(200); @@ -116,7 +127,23 @@ async function main(): Promise { assert.strictEqual(roomMsg.type, "room_message"); assert.strictEqual(roomMsg.message.content, "Hello from A!"); - // --- Test: DM from A to B --- + // --- Test: DM from A to B --- The two-round DM consent flow (section 6): A's own outbound room.join is what authorises the DM, and B (the party contacted first) still needs a human decision. + console.log("Test: A requests DM access from B..."); + const dmAccessPromise = a.store.requestDmAccess(b.store.peerId); + await waitFor( + () => + b.store + .listPendingRoomJoins() + .some((p) => p.requesterId === a.store.peerId), + "B to see A's pending DM request", + ); + const pendingDm = b.store + .listPendingRoomJoins() + .find((p) => p.requesterId === a.store.peerId); + assert.ok(pendingDm); + b.store.acceptRoomJoin(pendingDm.roomPath, a.store.peerId); + await dmAccessPromise; + console.log("Test: DM from A to B..."); b.deliveries.length = 0; await a.store.sendDm(a.store.peerId, b.store.peerId, "Hey B!"); diff --git a/src/test/mesh-smoke.runner.ts b/src/test/mesh-smoke.runner.ts index 0f3c3a8..b4e7fef 100644 --- a/src/test/mesh-smoke.runner.ts +++ b/src/test/mesh-smoke.runner.ts @@ -211,7 +211,21 @@ async function main(): Promise { ` description: "Smoke test room",`, `});`, `log({ type: "room_created", data: { id: room.id } });`, - `await new Promise(r => setTimeout(r, 1500));`, + // B's own join is a real, admitted wire-level room.join -- A (the room's owner) must accept it before B can ever hold a working room:member token, so this polls for and accepts the request rather than assuming legacy replication alone made B a member. acceptRoomJoin only resolves the held-open request's own promise; its continuation (minting the token, updating this room's own membership) runs on a later tick, so this also waits for that membership update to actually land before sending -- otherwise sendRoomMessage's own fan-out reads room.members before admitRoomJoin has added B to it. + `for (let i = 0; i < 20; i++) {`, + ` const pending = store.listPendingRoomJoins();`, + ` if (pending.some(p => p.roomPath === room.id)) {`, + ` store.acceptRoomJoin(room.id, pending.find(p => p.roomPath === room.id).requesterId);`, + ` log({ type: "accepted_join", data: { room: room.id } });`, + ` break;`, + ` }`, + ` await new Promise(r => setTimeout(r, 100));`, + `}`, + `for (let i = 0; i < 20; i++) {`, + ` const current = await store.getRoom(room.id);`, + ` if (current && current.members.length >= 2) break;`, + ` await new Promise(r => setTimeout(r, 100));`, + `}`, `await store.sendRoomMessage(room.id, agent.id, "Hello from A!");`, `log({ type: "sent", data: { room: room.id } });`, ].join("\n "), From 4fa249707001a2a1c99415a9fc75a3f0635d1163 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:47:09 +0100 Subject: [PATCH 6/7] test: give the self-join CRDT-merge test a token before joinRoom registerAgent's own id is always this.peerId, so joiner in "room membership changes converge and stale member lists are rejected" is B's own peer identity -- joinRoom's admission gate now sees this as a real self-join and, absent a token, would route it through a genuine (and here, transport-free, therefore impossible) wire-level join instead of the CRDT membership-merge logic this test actually targets. Minting and persisting a token for the room on B's own identity first restores that intent without changing any assertion. --- src/test/state-sync-convergence.test.ts | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/src/test/state-sync-convergence.test.ts b/src/test/state-sync-convergence.test.ts index 5ee7eb1..6b31c72 100644 --- a/src/test/state-sync-convergence.test.ts +++ b/src/test/state-sync-convergence.test.ts @@ -6,9 +6,14 @@ import * as assert from "node:assert/strict"; import { test } from "node:test"; +import { deviceIdFromHex } from "wire-mesh-core/domain/device-id"; +import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; +import { mintCapabilityToken } from "wire-mesh-core/domain/tokens"; import { MeshStore } from "../core/mesh-store.js"; import type { SerialisedState } from "../core/wire-protocol.js"; import { ownerNamedRoomPath } from "../core/room-path.js"; +import { loadOrCreateIdentity, saveRoomToken } from "../core/identity-store.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; import { wireTestTransport } from "./test-transport.js"; /** A local-only store: transport is set (registerAgent's own broadcastPatch needs one) but never started, so no ports and no flake -- the stores in these tests never actually connect. */ @@ -73,7 +78,8 @@ void test("a current holder rejects a stale snapshot instead of regressing", asy void test("room membership changes converge and stale member lists are rejected", async () => { const a = await makeStore(); - const b = await makeStore(); + const b = new MeshStore(); + const bSlot = await wireTestTransport(b); const owner = await a.registerAgent({ name: "owner", harness: "pi", @@ -103,6 +109,22 @@ void test("room membership changes converge and stale member lists are rejected" b.applyStateSync(snapshotOf(a)); assert.equal((await a.getRoom(roomId))?.members.includes(joiner.id), false); + // joiner.id is B's own peerId (registerAgent's own id is always this.peerId), so this is a real self-join as far as joinRoom's own admission gate is concerned -- give B a token for the room first so the gate sees an already-admitted member and exercises the CRDT membership-merge logic this test actually targets, not a real (and here, transport-free, therefore impossible) wire-level join. + const bClock = createSystemClock(); + const bIdentity = await toIdentityPort(loadOrCreateIdentity(bSlot)); + const grantVerdict = await mintCapabilityToken({ + identity: bIdentity, + clock: bClock, + tokenId: new Uint8Array([1]), + bearer: deviceIdFromHex(joiner.id), + capability: "room:member", + scope: { kind: "room", path: roomId }, + expires: bClock.now() + 60_000, + delegationsRemaining: 0, + }); + assert.ok(grantVerdict.ok, "expected the fixture grant to mint successfully"); + if (grantVerdict.ok) saveRoomToken(bSlot, roomId, grantVerdict.token); + await b.joinRoom(roomId, joiner.id); a.applyStateSync(snapshotOf(b)); assert.equal((await a.getRoom(roomId))?.members.includes(joiner.id), true); From b553704ff0daf1b68ccced039e853d3a203dadb3 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 12 Sep 2026 10:53:22 +0100 Subject: [PATCH 7/7] fix(deps): bump wire-mesh-core to 1.1.0 for bytesFromHex mesh-store.ts's own room.send fan-out needs bytesFromHex (added upstream in ExaDev/wire-mesh#90) to encode a reply's referenced message-id back into the wire's message-ref bytes; 1.0.3 predates that export. --- package.json | 2 +- pnpm-lock.yaml | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/package.json b/package.json index 927db33..ddbcac5 100644 --- a/package.json +++ b/package.json @@ -97,7 +97,7 @@ "cbor2": "2.3.0", "preact": "10.29.7", "typebox": "1.3.6", - "wire-mesh-core": "1.0.3", + "wire-mesh-core": "1.1.0", "ws": "8.21.1", "zod": "4.4.3" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7190d36..583771f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -36,8 +36,8 @@ importers: specifier: 1.3.6 version: 1.3.6 wire-mesh-core: - specifier: 1.0.3 - version: 1.0.3 + specifier: 1.1.0 + version: 1.1.0 ws: specifier: 8.21.1 version: 8.21.1 @@ -3235,8 +3235,8 @@ packages: engines: {node: '>= 8'} hasBin: true - wire-mesh-core@1.0.3: - resolution: {integrity: sha512-jauVnnL5c9EJynIvvztXa0V4eRXZbMz2a6zGKCTh1ZHB9srwTRg03rE49FltC+2MXtOT//dpOIZFTshEAHwKhg==} + wire-mesh-core@1.1.0: + resolution: {integrity: sha512-IibIoBv72Mzo4pWutmB7PM76saraaTdEqm0vM+RqXw6Hp57Ap06iVttZATP0NbEf61E2sCWbbG/+Fsv9MIXjdA==} word-wrap@1.2.5: resolution: {integrity: sha512-BN22B5eaMMI9UMtjrGd5g5eCYPpCPDUy0FJXbYsaT5zYxjFOckS53SQDE3pWkVoWpHXVb3BrYcEN4Twa55B5cA==} @@ -6780,7 +6780,7 @@ snapshots: dependencies: isexe: 2.0.0 - wire-mesh-core@1.0.3: + wire-mesh-core@1.1.0: dependencies: cbor2: 2.3.0 cddl.js: 1.0.1