feat(core): fan out room.send to every member over directed wire sends - #77
Conversation
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.
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.
…e 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.
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.
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.
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.
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.
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b553704ff0
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| const token = loadRoomTokens(slot)[key]; | ||
| if (token === undefined) { | ||
| throw new CommsError(`No room:member token for ${key}`, "NOT_MEMBER"); |
There was a problem hiding this comment.
For a first-time DM between two peers, the normal agent_comms({ action: "dm" }) path calls sendDm directly, but no public tool action invokes requestDmAccess, so the identity slot cannot contain this DM-path token and every initial DM throws NOT_MEMBER. Initiate the admission flow from the tool/send path or expose a usable action before requiring the token; also avoid adding the failed message to local history before this check.
AGENTS.md reference: AGENTS.md:L203-L207
Useful? React with 👍 / 👎.
| private pendingRoomSends = new Map< | ||
| string, | ||
| { roomPath: string; params: Record<string, unknown> }[] | ||
| >(); |
There was a problem hiding this comment.
When a recipient is offline and the sender bridge restarts before that recipient reconnects, this private in-memory map is discarded and is absent from serialise(), so the queued room message or DM is permanently lost. The previous delivery queue was replicated specifically to survive this sequence; pending directed sends need equivalent replication or another restart-safe replay source.
AGENTS.md reference: AGENTS.md:L37-L37
Useful? React with 👍 / 👎.
| for (const memberId of room.members) { | ||
| if (memberId !== from) { | ||
| await this.deliverLocallyAndBroadcast(memberId, { | ||
| type: "room_message", | ||
| message, | ||
| }); | ||
| await this.deliverRoomSendToMember(memberId, roomId, token, params); |
There was a problem hiding this comment.
After peers are already connected, this fan-out sends each new message only to current members and no longer broadcasts a message_add patch. Consequently, a non-member calling read_room on a public room—and a member who joins after earlier messages were sent—has no local copy of that history, even though CommsTool.readRoom reads only readRoomMessages. Preserve a history-sync/read path in addition to directed delivery.
AGENTS.md reference: AGENTS.md:L232-L234
Useful? React with 👍 / 👎.
| const outcome = await this.requireTransport().sendRoomRequest( | ||
| memberId, | ||
| { verb: ROOM_MEMBER_CAPABILITY, params }, | ||
| { kind: "room", path: roomPath }, | ||
| token, | ||
| ); |
There was a problem hiding this comment.
If a peer disconnects after WireMeshTransport.sendRoomRequest finds its session but before the manage response arrives, session.sendManageRequest rejects rather than returning an error outcome. This uncaught rejection bypasses the retry queue, rejects the whole room send, and prevents later members in the sequential fan-out from being attempted; catch transport rejections here and handle them like not_connected while keeping permanent protocol errors distinct.
Useful? React with 👍 / 👎.
| if (outcome.result !== "ok") { | ||
| this.queuePendingRoomSend(memberId, roomPath, params); | ||
| } |
There was a problem hiding this comment.
On a successful manage response this method simply returns, so room and DM senders no longer receive the delivery_status { status: "delivered" } event previously emitted when the recipient was queued. A later read event is not equivalent for offline/drain consumers, and integrations relying on the documented delivered/read progression can no longer distinguish successful delivery from a pending retry.
AGENTS.md reference: AGENTS.md:L295-L298
Useful? React with 👍 / 👎.
| if (loadRoomTokens(slot)[roomId] === undefined) { | ||
| return this.joinRemoteRoom(roomId, agentId); |
There was a problem hiding this comment.
When this peer already knows the room through state sync, routing it through joinRemoteRoom replaces the existing record with a synthetic version-1 public room and unconditionally resets its message history to []. Joining a private, secret, described, federated, or previously active room therefore loses its synced metadata and history locally; admission should merge the returned membership/token into the existing room instead of using the never-seen-room constructor.
AGENTS.md reference: AGENTS.md:L232-L236
Useful? React with 👍 / 👎.
| const outcome = await this.requireTransport().sendRoomRequest( | ||
| memberId, | ||
| { verb: ROOM_MEMBER_CAPABILITY, params }, | ||
| { kind: "room", path: roomPath }, | ||
| token, | ||
| ); |
There was a problem hiding this comment.
If a peer disconnects after WireMeshTransport.sendRoomRequest finds its session but before the manage response completes, session.sendManageRequest rejects rather than returning an error outcome. This await is not caught, so sendRoomMessage rejects without queuing that recipient and stops its sequential loop before later room members are contacted; catch transport rejections here and queue the failed recipient just like not_connected.
Useful? React with 👍 / 👎.
| if (outcome.result !== "ok") { | ||
| this.queuePendingRoomSend(memberId, roomPath, params); | ||
| } |
There was a problem hiding this comment.
Every protocol error is treated as if the member were merely offline, so an expired or revoked token (unauthorized) or malformed request is silently queued even though reconnecting cannot make the same request succeed. Because the public send already recorded the message and returns success, these messages appear sent but remain in a retry loop until the bounded queue drops them; queue only transient connectivity outcomes and surface or refresh permanent authorization/schema failures.
Useful? React with 👍 / 👎.
| for (const pending of queue) { | ||
| const token = loadRoomTokens(slot)[pending.roomPath]; | ||
| if (token === undefined) continue; | ||
| await this.deliverRoomSendToMember( |
There was a problem hiding this comment.
A queued send is retried solely because the sender still has a token for the path; it never verifies that the room still exists or that memberId remains a member. For example, if the owner sends while a member is offline and then destroys the room, the persisted owner token and pending entry survive, so reconnecting the former member receives a live message for the destroyed room. Drop pending entries when rooms are destroyed or members are removed, or revalidate both conditions during this flush.
Useful? React with 👍 / 👎.
| const history = this.messages.get(roomPath) ?? []; | ||
| history.push(message); | ||
| this.messages.set(roomPath, history); |
There was a problem hiding this comment.
On reconnect, handlePeerConnected awaits a full state sync before flushing pending sends, so the recipient first imports the sender's copy of an offline message into its history and then receives the same message ID through room.send. This unconditional push adds a second copy (and the DM branch does the same), causing readRoomMessages/DM history to return duplicates with divergent readBy state; merge or look up by message ID before appending the directed delivery.
Useful? React with 👍 / 👎.
|
🎉 This PR is included in version 2.10.0 🎉 The release is available on: Your semantic-release bot 📦🚀 |
1 participant
P3.5's own core primitive, building on the token verification and directed room.send delivery #76 landed: sendRoomMessage and sendDm now deliver via real, wire-authenticated directed room.send requests to every room member/DM recipient, replacing the legacy broadcastPatch full-state replication and deliverLocallyAndBroadcast loop for message delivery specifically.
handleRoomSend now branches on the room-path's own shape (owner-named room vs DM), and reads room.send's own "reply" message-ref and "streaming-behavior" extension field, both previously only available on the legacy path.
A send to a member who isn't currently reachable is queued (pendingRoomSends) and retried the moment that member's connection is (re)established (handlePeerConnected), replacing the full-state-sync's automatic eventual consistency with an explicit retry queue -- a direct request to a disconnected peer fails immediately, unlike the old model's implicit convergence.
joinRoom's own admission gate no longer treats "this room is already known locally" as proof of membership for this store's own identity, since the legacy full-state-sync replicates a room's metadata to every connected peer well before that peer is ever actually admitted. The gate is now whether this store holds a real room:member token, so a self-join always goes through genuine wire-level admission when it's never actually been granted one -- this surfaced as a real correctness gap (a production bridge calling join_room for a room its own gossip already knew about would previously never get a working token at all), not just a test artifact.
Every existing test/script that relied on the old "already known locally means already a member" shortcut needed the corresponding real admission or DM-consent round trip added: mesh-e2e, identity-restart, the multi-process smoke test, the delivery-receipt suite, and one state-sync-convergence test whose own self-join (registerAgent's id is always this.peerId) now needs a token to keep exercising the CRDT-merge logic it actually targets rather than a genuine (and there, transport-free, therefore impossible) wire join.
downtime-replay.test.ts and downtime-replay.integration.test.ts are retired: their own room-message-delivery scenarios are replaced by room-send-retry-queue.test.ts (unit, fake transport) and room-send-retry.integration.test.ts (real two-peer, same #28 acceptance scenario re-verified against the new retry mechanism). The one unrelated test in the old file (pending invite replay, untouched by this migration) survives as its own invite-replay.test.ts.
Full local verification: typecheck, lint, build, the full unit/integration suite (143 tests), the multi-process smoke test, and the delivery-receipt suite all green.