From ddf5114394466406af5c23503a1f47e953a3dfed Mon Sep 17 00:00:00 2001 From: Antonio Pantano Date: Mon, 19 Oct 2026 17:33:47 +0200 Subject: [PATCH] feat(core): expose durable room-send options --- README.md | 16 ++++ src/core/comms-store.ts | 2 + src/core/index.ts | 2 +- src/core/room-messaging.ts | 7 +- ...esh-store-durable-send.integration.test.ts | 76 +++++++++++++++++++ 5 files changed, 96 insertions(+), 7 deletions(-) create mode 100644 src/test/mesh-store-durable-send.integration.test.ts diff --git a/README.md b/README.md index 911efa68..b5eed700 100644 --- a/README.md +++ b/README.md @@ -426,6 +426,22 @@ When an agent's status changes (active / idle / busy / offline), all rooms it be - Graceful shutdown - Stale agent cleanup (the elected coordinator's PID probe) +## Durable room sends from the library + +Library callers can opt a room message into the sender's local room-notice log: + +```ts +import type { SendRoomMessageOptions } from "agent-comms"; + +const options: SendRoomMessageOptions = { durable: true }; +await store.sendRoomMessage(roomId, senderId, "Keep this message", options); +``` + +`MeshStore` records a signed room-notice in the sender's own log only when `durable` is `true`. +This only writes the notice locally: nothing currently announces or fetches these room-notice +entries. Automatic offline catch-up still needs a catch-up policy and bridge wiring. +`FileStore` ignores the option because it already persists every message. + ## Delivery status and read receipts Messages carry a `readBy` field tracking which agents have consumed them. Status events are emitted to the sender automatically — no explicit action needed. diff --git a/src/core/comms-store.ts b/src/core/comms-store.ts index 2a2c0d98..756cdff4 100644 --- a/src/core/comms-store.ts +++ b/src/core/comms-store.ts @@ -23,6 +23,8 @@ import type { export interface SendRoomMessageOptions { replyTo?: string | undefined; streamingBehavior?: StreamingBehavior | undefined; + /** When true, `MeshStore` also records the message as a signed room-notice in the sender's own log for offline catch-up. `FileStore` ignores it, since it already persists every message. */ + durable?: boolean | undefined; } /** What became of a room message for one specific member of that room. Delivery is per member because a room send fans out one directed request each, and a member that is unreachable right now is independent of every other. */ diff --git a/src/core/index.ts b/src/core/index.ts index 99e1f744..d66271e5 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -5,7 +5,7 @@ * Bridges (pi extension, Claude Code channel) are in ../bridges/. */ -export type { CommsStore } from "./comms-store.js"; +export type { CommsStore, SendRoomMessageOptions } from "./comms-store.js"; export { FileStore, CommsError } from "./store.js"; export { MeshStore } from "./mesh-store.js"; export { CommsTool } from "./tool.js"; diff --git a/src/core/room-messaging.ts b/src/core/room-messaging.ts index 833b919c..9d63fb35 100644 --- a/src/core/room-messaging.ts +++ b/src/core/room-messaging.ts @@ -51,11 +51,6 @@ export interface RoomMessagingDeps { resolveAgent: (id: string) => Promise; } -/** RoomMessaging's own richer sendRoomMessage options, extending the public CommsStore-facing SendRoomMessageOptions with durable -- see sendRoomMessage's own doc comment for what durable means. */ -export interface RoomMessagingSendOptions extends SendRoomMessageOptions { - durable?: boolean; -} - export class RoomMessaging { constructor(private readonly deps: RoomMessagingDeps) {} @@ -70,7 +65,7 @@ export class RoomMessaging { roomIdOrName: string, from: string, content: string, - options?: RoomMessagingSendOptions, + options?: SendRoomMessageOptions, ): Promise { const { replyTo, streamingBehavior, durable } = options ?? {}; const roomId = this.deps.resolveRoomId(roomIdOrName); diff --git a/src/test/mesh-store-durable-send.integration.test.ts b/src/test/mesh-store-durable-send.integration.test.ts new file mode 100644 index 00000000..bc4416fa --- /dev/null +++ b/src/test/mesh-store-durable-send.integration.test.ts @@ -0,0 +1,76 @@ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; +import { decode, cdeDecodeOptions } from "cbor2"; +import { createNodeFsStorage } from "wire-mesh-core/adapters/node-fs-storage"; +import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; +import { readEntries } from "wire-mesh-core/domain/data-sync"; +import { bytesFromHex } from "wire-mesh-core/domain/device-id"; +import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { verifyRoomNotice } from "wire-mesh-core/domain/room"; +import type { RoomNotice } from "wire-mesh-core/generated/protocol"; +import { + createBridgeMesh, + loadIdentityForFront, + releaseIdentityLock, + type CommsStore, + type SendRoomMessageOptions, +} from "../core/index.js"; +import { oplogDirFor } from "../core/identity-store.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; + +describe("public room-send durability", () => { + it.each([undefined, false, true])( + "records a notice only when durable is true (received %j)", + async (durable) => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "mesh-durable-")); + const slot = { harness: "durable-test", cwd: root, dataDir: root }; + const mesh = await createBridgeMesh(slot, { + fetchLatestVersion: async () => undefined, + }); + try { + const store: CommsStore = mesh.store; + const room = await store.createRoom({ + name: "durable-room", + type: "public", + owner: mesh.store.peerId, + description: "", + }); + const options: SendRoomMessageOptions = { durable }; + const sent = await store.sendRoomMessage( + room.id, + mesh.store.peerId, + "public durable message", + options, + ); + expect(sent.message.content).toBe("public durable message"); + + const entries = await readEntries( + createNodeFsStorage({ dir: oplogDirFor(slot) }), + Uint8Array.from(bytesFromHex(mesh.store.peerId)), + 0, + ); + expect(entries).toHaveLength(durable === true ? 1 : 0); + if (entries[0] !== undefined) { + const notice = decode(entries[0], cdeDecodeOptions); + const verdict = await verifyRoomNotice(notice, { + identity: await toIdentityPort(loadIdentityForFront(slot)), + clock: createSystemClock(), + revocation: createRevocationView(), + }); + expect(verdict.ok).toBe(true); + if (!verdict.ok) return; + expect(verdict.claims.room).toBe(room.id); + expect(new TextDecoder().decode(verdict.claims.content)).toBe( + sent.message.content, + ); + } + } finally { + await mesh.store.shutdown(); + releaseIdentityLock(slot); + fs.rmSync(root, { recursive: true, force: true }); + } + }, + ); +});