feat(web-console): replicate notice logs through a hub

ts/packages/core/src/domain/mesh-session.ts

@@ -800,6 +800,11 @@ export function createMeshSession(
   addresses: readonly string[] = [],
   /** Fired for every peer-advert entry as it's applied to the directory, regardless of source -- the hook a gossip-expansion consumer (wire-mesh#187, gossip-expansion.ts's own createGossipExpansion) uses to observe newly-gossiped peers without becoming a second, competing consumer of this session's own single-reader events stream (each emitted SessionEvent wakes at most one waiter, so a second for-await loop over events would silently steal events from whichever consumer already reads it). Omit for a caller with no use for it, exactly today's behaviour. */
   onPeerAdvert?: (advert: PeerAdvert) => void,
+  /** Fired once per frame received on this session's connection, with the Connection it arrived on: a new Connection after a reconnect is how a caller notices it has to announce itself again, and a core/data reply is how it receives a log it asked a hub for. Omit for a caller with no use for it, exactly today's behaviour. */
+  onFrame?: (
+    connection: Readonly<Connection>,
+    frame: Frame,
+  ) => void | Promise<void>,
 ): MeshSession {
   const { session } = createSessionCore(
     identity,
@@ -808,6 +813,7 @@ export function createMeshSession(
     async (address) => transport.connect(address),
     onPeerAdvert,
     addresses,
+    onFrame,
   );
   return session;
 }

ts/packages/core/src/domain/notice-board.ts

@@ -175,18 +175,20 @@ export function createNoticeBoard(
   async function readLog(
     peer: DeviceId,
     room: string,
+    dmRootPolicy: "self" | "either-participant",
   ): Promise<NoticeBoardEntry[]> {
     const entries = await readEntries(storage, peer, 0);
     const out: NoticeBoardEntry[] = [];
     for (const entry of entries) {
-      out.push(await readOne(entry, room));
+      out.push(await readOne(entry, room, dmRootPolicy));
     }
     return out;
   }
 
   async function readOne(
     entry: Uint8Array,
     room: string,
+    dmRootPolicy: "self" | "either-participant",
   ): Promise<NoticeBoardEntry> {
     const notice = decodeNotice(entry);
     if (notice === undefined) {
@@ -197,6 +199,7 @@ export function createNoticeBoard(
       clock,
       revocation,
       expectedRoom: room,
+      dmRootPolicy,
     });
     if (!verdict.ok) {
       return { verified: false, reason: verdict.reason };
@@ -246,8 +249,10 @@ export function createNoticeBoard(
   return {
     postEncryptedNotice,
     ingestPeerEntry,
-    readOwnNotices: async (room: string) => readLog(identity.deviceId, room),
+    // A device's own token in a DM is rooted at the other participant, so its own log is read under the either-participant policy; another device's notices must be rooted at this identity.
+    readOwnNotices: async (room: string) =>
+      readLog(identity.deviceId, room, "either-participant"),
     readPeerNotices: async (peer: DeviceId, room: string) =>
-      readLog(peer, room),
+      readLog(peer, room, "self"),
   };
 }

ts/packages/core/src/domain/room-rekey.ts

@@ -44,7 +44,7 @@ export function buildRoomRekeyCommand(
 
 /** Sends a room.rekey for roomPath. targetDevice/token forward directly to MeshSession.sendManageRequest's own identically-named parameters, exactly as sendRoomMessage's own doc comment already describes. */
 export async function sendRoomRekey(
-  session: Readonly<MeshSession>,
+  session: Readonly<Pick<MeshSession, "sendManageRequest">>,
   roomPath: string,
   keyEpoch: number,
   wrappedKey: Uint8Array | readonly Uint8Array[],

ts/packages/core/src/domain/room.ts

@@ -90,6 +90,10 @@ export interface VerifyRoomNoticeOptions {
    * When given, refuses any notice not claiming exactly this room -- "is this the room I actually asked to read", checked against the notice's own self-declared `room` field before any cryptographic work, so a caller scanning a mixed stream of notices can cheaply skip ones for other rooms. Independent of, and layered on top of, this function's own unconditional internal self-consistency check (the embedded token's scope.path MUST equal the notice's own `room` field regardless of whether expectedRoom is given at all) -- a notice can be internally self-consistent yet still be for a room other than the one a caller expected, and this is the option that catches that case. Mirrors verifyCapabilityToken's own optional expectedBearer for the same "also assert it matches what I expected" shape.
    */
   expectedRoom?: RoomPath;
+  /**
+   * How a DM's embedded token must be rooted, passed to verifyRoomToken (see its own `dmRootPolicy`). The default requires the root to be the verifying identity, which is right for a notice another device posted to me. A device reading its own notice holds a token rooted at the other participant, so only "either-participant" can pass there; it is safe because nothing about an own notice rests on the root rule, the device wrote it.
+   */
+  dmRootPolicy?: "self" | "either-participant";
 }
 
 /**
@@ -170,6 +174,9 @@ export async function verifyRoomNotice(
     expectedBearer: claims.poster,
     // Obligation 3 (scope.kind MUST be "room" and scope.path MUST equal this notice's own `room` field): this notice's internal self-consistency, independent of options.expectedRoom above, which instead asserts the room against what the CALLER expected.
     roomPath: claims.room,
+    ...(options.dmRootPolicy !== undefined
+      ? { dmRootPolicy: options.dmRootPolicy }
+      : {}),
   });
   if (!tokenVerdict.ok) {
     return { ok: false, reason: tokenVerdict.reason };

ts/packages/core/test/mesh-session-create.unit.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import type {
   DataHaveFrame,
+  DataRequestFrame,
   GossipFrame,
   HandshakeFrame,
   PeerAdvert,
@@ -642,4 +643,31 @@ describe("createMeshSession", () => {
     expect(result.done).toBe(false);
     await session.close();
   });
+
+  it("hands every received frame to onFrame", async () => {
+    const { transport, connection } = fakeTransport();
+    const onFrame = vi.fn(async (): Promise<void> => Promise.resolve());
+    const session = createMeshSession(
+      transport,
+      testIdentity,
+      testClock,
+      null,
+      [],
+      undefined,
+      onFrame,
+    );
+    await session.connect("ws://node", ["core/data"]);
+
+    const request: DataRequestFrame = {
+      type: "data-request",
+      peer: deviceA,
+      "from-seq": 0,
+    };
+    connection.push(request);
+
+    await vi.waitFor(() => {
+      expect(onFrame).toHaveBeenCalledWith(expect.anything(), request);
+    });
+    await session.close();
+  });
 });

ts/packages/core/test/notice-board.unit.test.ts

@@ -377,3 +377,93 @@ describe("createNoticeBoard", () => {
     expect(encryptedContentType("text/plain")).toBe("text/plain+aes256gcm");
   });
 });
+
+describe("notices in a DM", () => {
+  async function dm(): Promise<{
+    a: IdentityPort;
+    b: IdentityPort;
+    roomPath: string;
+  }> {
+    const first = await generateEs256Identity();
+    const second = await generateEs256Identity();
+    const firstIsLower =
+      deviceIdToHex(first.deviceId) < deviceIdToHex(second.deviceId);
+    const a = firstIsLower ? first : second;
+    const b = firstIsLower ? second : first;
+    return {
+      a,
+      b,
+      roomPath: `${deviceIdToHex(a.deviceId)}+${deviceIdToHex(b.deviceId)}`,
+    };
+  }
+
+  async function boardFor(
+    identity: IdentityPort,
+    roomPath: string,
+    key: Uint8Array,
+  ): Promise<{
+    board: ReturnType<typeof createNoticeBoard>;
+    storage: KeyValueStorage;
+  }> {
+    const keys = memoryKeyStore();
+    await keys.set(roomPath, FIRST_EPOCH, key);
+    const storage = createMemoryStorage();
+    return {
+      board: createNoticeBoard({
+        identity,
+        storage,
+        clock: fixedClock(NOW_MS),
+        revocation: createRevocationView(),
+        roomKeys: keys,
+      }),
+      storage,
+    };
+  }
+
+  it("reads its own notice although the token it holds is rooted at the other participant", async () => {
+    const { a, b, roomPath } = await dm();
+    const { board } = await boardFor(a, roomPath, generateContentKey());
+    await board.postEncryptedNotice({
+      room: roomPath,
+      token: await mintRoomMemberToken(b, a, roomPath),
+      contentType: "text/plain",
+      plaintext: new TextEncoder().encode("mine"),
+    });
+
+    const read = await board.readOwnNotices(roomPath);
+
+    expect(read[0]?.verified).toBe(true);
+    expect(read[0]?.plaintext).toEqual(new TextEncoder().encode("mine"));
+  });
+
+  it("reads the peer's notice when its token was granted by the reader, and refuses one the poster issued itself", async () => {
+    const { a, b, roomPath } = await dm();
+    const key = generateContentKey();
+    const granted = await boardFor(a, roomPath, key);
+    const selfIssued = await boardFor(a, roomPath, key);
+    await granted.board.postEncryptedNotice({
+      room: roomPath,
+      token: await mintRoomMemberToken(b, a, roomPath),
+      contentType: "text/plain",
+      plaintext: new TextEncoder().encode("granted"),
+    });
+    await selfIssued.board.postEncryptedNotice({
+      room: roomPath,
+      token: await mintRoomMemberToken(a, a, roomPath),
+      contentType: "text/plain",
+      plaintext: new TextEncoder().encode("self-issued"),
+    });
+
+    for (const [poster, expected] of [
+      [granted, true],
+      [selfIssued, false],
+    ] as const) {
+      const reader = await boardFor(b, roomPath, key);
+      for (const entry of await readEntries(poster.storage, a.deviceId, 0)) {
+        await reader.board.ingestPeerEntry(a.deviceId, new Uint8Array(entry));
+      }
+      const read = await reader.board.readPeerNotices(a.deviceId, roomPath);
+      expect(read[0]?.verified).toBe(expected);
+    }
+  });
+});

ts/packages/web-console/README.md

@@ -43,7 +43,7 @@ The `webrtc:signal` capability itself carries no target-device field -- a `manag
 
 ## Messaging
 
-Messages travel over `core/room`. A conversation is identified by its room path: a direct message's path names the two participating devices, so one peer is one conversation. `src/conversations.ts` holds the state of every conversation and the pure reducer that evolves it, and `src/hooks/use-room-messaging.ts` feeds it from live sessions. A conversation reaches its peer by up to two routes. Clicking Message opens it at once over the hub the peer was found through (a relay pairing and a secure channel, which needs only the connection both sides already have), and a WebRTC connection is negotiated in the background; when it opens, the conversation moves onto it, and when it closes the conversation falls back to the hub. `src/request-demux.ts` splits a hub connection's incoming requests between the negotiator and the conversations reached through that hub. A conversation restored from `src/message-store.ts` at start-up has no route, is shown as offline, and takes one when a peer reconnects. Durable notices sync with data frames, which only a direct connection carries, so they are unavailable while a conversation is on the hub. Received messages in a conversation that is not on screen count as unread; the conversation on screen is read as messages arrive. Both sent and received messages are persisted per room path. A message being sent shows in the conversation as sending; if delivery fails it stays there as not sent, with the reason, and can be retried or dismissed.
+Messages travel over `core/room`. A conversation is identified by its room path: a direct message's path names the two participating devices, so one peer is one conversation. `src/conversations.ts` holds the state of every conversation and the pure reducer that evolves it, and `src/hooks/use-room-messaging.ts` feeds it from live sessions. A conversation reaches its peer by up to two routes. Clicking Message opens it at once over the hub the peer was found through (a relay pairing and a secure channel, which needs only the connection both sides already have), and a WebRTC connection is negotiated in the background; when it opens, the conversation moves onto it, and when it closes the conversation falls back to the hub. `src/request-demux.ts` splits a hub connection's incoming requests between the negotiator and the conversations reached through that hub. A conversation restored from `src/message-store.ts` at start-up has no route, is shown as offline, and takes one when a peer reconnects. Durable notices are kept in the console's storage together with the conversation's room keys and tokens, so a reload keeps them, and they sync with data frames over whichever route is connected: a peer announces its log directly, and a hub that holds logs for absent devices (`--mailbox-dir`) is sent the own log when the connection opens and after each post and is asked for the conversation peers' logs, so a notice posted while the peer was offline is read after the peer reconnects. A hub does not announce a new notice to a peer that is already connected, so over a hub alone a notice appears at the peer's next sync, which a message from the author, a reconnect or opening the conversation triggers. Received messages in a conversation that is not on screen count as unread; the conversation on screen is read as messages arrive. Both sent and received messages are persisted per room path. A message being sent shows in the conversation as sending; if delivery fails it stays there as not sent, with the reason, and can be retried or dismissed.
 
 ## What is deliberately deferred
 

ts/packages/web-console/playwright.config.ts

@@ -3,6 +3,10 @@ import { defineConfig } from "@playwright/test";
 // A fixed, dedicated port pair for the e2e run (never the dev-time 8787/5173 pair) so a developer's own `pnpm dev` session never collides with CI or a concurrent local test run.
 export const RELAY_PORT = 8790;
 export const VITE_PORT = 8798;
+// A second hub that also holds other devices' logs while they are offline, for the spec that has one device away while the other writes.
+export const MAILBOX_RELAY_PORT = 8791;
+export const MAILBOX_RELAY_ADDRESS = `ws://127.0.0.1:${String(MAILBOX_RELAY_PORT)}`;
+const MAILBOX_DIR = "test-results/mailbox";
 export const RELAY_ADDRESS = `ws://127.0.0.1:${String(RELAY_PORT)}`;
 const RELAY_HEALTH_URL = `http://127.0.0.1:${String(RELAY_PORT)}/`;
 const HARNESS_URL = `http://localhost:${String(VITE_PORT)}/live-check/harness.html`;
@@ -26,6 +30,12 @@ export default defineConfig({
       reuseExistingServer: process.env.CI === undefined,
       timeout: 10_000,
     },
+    {
+      command: `node ${nodePackageDist} --bind 127.0.0.1:${String(MAILBOX_RELAY_PORT)} --mailbox-dir ${MAILBOX_DIR}`,
+      port: MAILBOX_RELAY_PORT,
+      reuseExistingServer: process.env.CI === undefined,
+      timeout: 10_000,
+    },
     {
       command: `npx vite --port ${String(VITE_PORT)} --strictPort`,
       url: HARNESS_URL,

ts/packages/web-console/src/App.tsx

@@ -57,7 +57,10 @@ import type { PreferencesStore } from "./preferences-store.js";
 import type { NameStore } from "./name-store.js";
 import { ConversationList } from "./components/ConversationList.js";
 import { RoomPanel } from "./components/RoomPanel.js";
-import { useRoomMessaging } from "./hooks/use-room-messaging.js";
+import {
+  useRoomMessaging,
+  type HubEndpoint,
+} from "./hooks/use-room-messaging.js";
 import { createRequestDemux } from "./request-demux.js";
 import { ROOM_MEMBER_CAPABILITY } from "wire-mesh-core/domain/room-token-verification";
 import { WEBRTC_SIGNAL_VERB } from "wire-mesh-core/domain/webrtc-signaling";
@@ -104,6 +107,8 @@ interface ConnectionFailure {
 interface ConnectionEntry {
   address: string;
   session: ReturnType<typeof createMeshSession>;
+  /** The session as the conversations reach it: one stable object per connection, so a conversation's route can be matched when the connection closes. */
+  hub: HubEndpoint;
   negotiator: WebrtcNegotiator;
 }
 
@@ -168,6 +173,11 @@ export function App({
   }, [selectedRoomPath, selectedUnread, markRead]);
 
   // A negotiator's own onIncomingConnection callback is registered once, at construction, and must still call whatever the *latest* attach is -- attach itself is a fresh function every time useRoomMessaging's own session map changes, so a ref (updated every render, read from the callback) is what keeps that call from closing over a stale, since-superseded attach.
+  const hubsBySession = useRef(new Map<MeshSession, HubEndpoint>());
+  const onHubFrameRef = useRef(roomMessaging.onHubFrame);
+  useEffect(() => {
+    onHubFrameRef.current = roomMessaging.onHubFrame;
+  }, [roomMessaging.onHubFrame]);
   const watchHubRef = useRef(roomMessaging.watchHub);
   useEffect(() => {
     watchHubRef.current = roomMessaging.watchHub;
@@ -238,13 +248,16 @@ export function App({
         }
       }
     })();
-    watchHubRef.current({
+    const hub: HubEndpoint = {
       sendManageRequest: async (...args) => session.sendManageRequest(...args),
+      sendDataFrame: async (frame) => session.sendDataFrame(frame),
       incomingManageRequests: demux.stream(ROOM_MEMBER_CAPABILITY),
-    });
+    };
+    hubsBySession.current.set(session, hub);
+    watchHubRef.current(hub);
     setConnections((current) => [
       ...current,
-      { address: entryAddress, session, negotiator },
+      { address: entryAddress, session, hub, negotiator },
     ]);
   }
 
@@ -287,6 +300,7 @@ export function App({
     const expansionRef: { current: GossipExpansion | null } = {
       current: null,
     };
+    const sessionRef: { current: MeshSession | null } = { current: null };
     const session = createMeshSession(
       createDialTransport(trust.memory),
       identity,
@@ -296,7 +310,18 @@ export function App({
       (advert) => {
         expansionRef.current?.considerAdvert(advert);
       },
+      (connection, frame) => {
+        // Frames that arrive before the connection is attached have no hub to go to yet; the notice logs are exchanged on the first frame after it is.
+        const hub =
+          sessionRef.current === null
+            ? undefined
+            : hubsBySession.current.get(sessionRef.current);
+        if (hub !== undefined) {
+          onHubFrameRef.current(hub, connection, frame);
+        }
+      },
     );
+    sessionRef.current = session;
     expansionRef.current = createGossipExpansion({
       selfDeviceId: identity.deviceId,
       shouldExpand: async (candidate) => confirmExpansion(candidate, via),
@@ -425,7 +450,7 @@ export function App({
   }
 
   function handleClose(target: Readonly<ConnectionEntry>): void {
-    roomMessaging.dropHub(target.session);
+    roomMessaging.dropHub(target.hub);
     void target.session.close();
     setConnections((current) =>
       current.filter((entry) => entry.session !== target.session),
@@ -437,7 +462,7 @@ export function App({
     device: DeviceId,
   ): void {
     // The conversation opens at once over the hub the peer was found through, and a direct connection is negotiated in the background: it takes over when it opens, and its failure costs nothing but the direct route.
-    roomMessaging.openRelay(entry.session, device);
+    roomMessaging.openRelay(entry.hub, device);
     setSelectedPath(
       dmRoomPath(deviceIdToHex(identity.deviceId), deviceIdToHex(device)),
     );

ts/packages/web-console/src/components/RoomPanel.tsx

@@ -107,7 +107,7 @@ export function RoomPanel({
         labelPosition="center"
       />
       <NoticesView notices={view.notices} />
-      {view.via === "direct" ? (
+      {view.via !== undefined ? (
         <Group>
           <SubmitRow
             ariaLabel="Post a durable notice"
@@ -118,7 +118,7 @@ export function RoomPanel({
         </Group>
       ) : (
         <Text size="sm" c="dimmed">
-          Durable notices need a direct connection.
+          Durable notices can be posted while the conversation is connected.
         </Text>
       )}
     </Stack>

ts/packages/web-console/src/conversations.ts

@@ -31,8 +31,11 @@ export type PendingOutgoing = {
     | { status: "failed"; error: string }
   );
 
-/** What a conversation needs from a hub connection to send through it: the relay peer is addressed by device-id on each request. */
-export type RelaySender = Pick<MeshSession, "sendManageRequest">;
+/** What a conversation needs from a hub connection to send through it: the relay peer is addressed by device-id on each request, and notice logs are exchanged with the hub as data frames. */
+export type RelaySender = Pick<
+  MeshSession,
+  "sendManageRequest" | "sendDataFrame"
+>;
 
 /** A conversation's route through a hub, where the peer is reached by a relay pairing and a secure channel instead of a connection of its own. */
 export interface RelayRoute {