From 2bcbe3a612525ddbd0ee64a76bbda10722305fb9 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 21:08:10 +0100 Subject: [PATCH] feat(reef): follow up on unconfirmed deliveries instead of staying silent (#110938) * feat(reef): follow up on unconfirmed deliveries instead of staying silent * test(reef): satisfy lint and test-type gates for delivery notices * test(reef): keep rejection notifier mock signature untyped --- docs/channels/reef.md | 2 + extensions/reef/src/channel.ts | 17 +++++- extensions/reef/src/flow-receipts.test.ts | 68 +++++++++++++++++++++++ extensions/reef/src/flow.test-helpers.ts | 1 + extensions/reef/src/flow.ts | 14 +++++ extensions/reef/src/owner-notice.test.ts | 68 +++++++++++++++++++++++ extensions/reef/src/owner-notice.ts | 51 +++++++++++++++++ extensions/reef/src/trust-store.test.ts | 57 ++++++++++++++++++- extensions/reef/src/trust-store.ts | 55 +++++++++++++++++- 9 files changed, 329 insertions(+), 4 deletions(-) diff --git a/docs/channels/reef.md b/docs/channels/reef.md index 494c1041fc32..e89a33eacc9a 100644 --- a/docs/channels/reef.md +++ b/docs/channels/reef.md @@ -122,6 +122,8 @@ Agents send through the shared `message` tool to `reef:`; humans can tes openclaw message send --channel reef --target @friend --message "hello from my claw" ``` +A send never fails silently. Local guard or relay errors fail the send immediately, replies and peer guard rejections come back through the flows below, and if the peer's claw confirms nothing for about 10 minutes the sending agent receives a delivery-delay notice, plus a follow-up once the message is finally delivered or rejected. A peer that accepts a message and simply does not reply (for example a `notify-only` friend) is a successful delivery, not an error. + Inbound messages arrive as untrusted third-party data: provenance-framed, command-unauthorized, with URLs inert. Depending on the friend's autonomy tier, OpenClaw notifies you or sends a bounded guarded reply: | Tier | Behavior | diff --git a/extensions/reef/src/channel.ts b/extensions/reef/src/channel.ts index c2faf4e1f556..9ea923988f29 100644 --- a/extensions/reef/src/channel.ts +++ b/extensions/reef/src/channel.ts @@ -25,6 +25,7 @@ import { resolveReefInboundDispatchContent } from "./inbound.js"; import { reefMessageAdapter, reefOutboundAdapter } from "./outbound.js"; import { createReefOwnerNoticeHandler, + notifyOverdueReefDeliveries, processReefInboxEntriesInOrder, ReefReceiptNotifier, } from "./owner-notice.js"; @@ -454,7 +455,21 @@ export const reefPlugin: ChannelPlugin = { await runReefChannelLifecycle({ parentSignal: ctx.abortSignal, startInbox: (signal) => inbox.start(signal), - reconcile, + reconcile: async () => { + // The overdue sweep must run even while the relay is unreachable: + // that outage is exactly when queued sends go unconfirmed, and the + // notices themselves are local. + let reconcileError: Error | undefined; + try { + await reconcile(); + } catch (error) { + reconcileError = error instanceof Error ? error : new Error(String(error)); + } + await notifyOverdueReefDeliveries({ trust, ownerNotice }); + if (reconcileError) { + throw reconcileError; + } + }, onReconcileError: (error) => ctx.log?.error?.(`reef friend reconcile failed: ${String(error)}`), }); diff --git a/extensions/reef/src/flow-receipts.test.ts b/extensions/reef/src/flow-receipts.test.ts index 8b0c979bb770..565db2cb92c4 100644 --- a/extensions/reef/src/flow-receipts.test.ts +++ b/extensions/reef/src/flow-receipts.test.ts @@ -771,3 +771,71 @@ describe("ReefMessageFlow delivery receipts", () => { }); }); }); + +describe("ReefMessageFlow overdue delivery follow-up", () => { + it("notifies the owner when an accepted receipt closes an overdue notice", async () => { + const alice = generateIdentity(); + const bob = reefKeys(); + const trusted = trust({ alice: peerTrust(alice) }); + const relay = transport(); + const onOwnerNotice = vi.fn(async (_text: string) => {}); + const flow = new ReefMessageFlow({ + config: config(), + trust: trusted.store, + keys: bob, + transport: relay as unknown as ReefTransportClient, + guard: guard(allow), + audit: new MemoryAuditStore(new Uint8Array(32).fill(23)), + replay: new MemoryReplayStore(), + ...flowStores(), + onIngress: async () => {}, + onOwnerNotice, + }); + const id = await flow.send("alice", "are you there?"); + const record = trusted.deliveries.get(`alice:${id}`)!; + record.overdueNotifiedAt = Date.now(); + const receipt = signReceipt( + { id, bodyHash: record.bodyHash, auditHead: "a".repeat(64), status: "accepted" }, + alice.signing.secretKey, + ); + + await expect( + flow.processEntries([{ seq: 1, peer: "alice", id, kind: "receipt", receipt, ts: 1 }]), + ).resolves.toEqual([]); + + expect(trusted.deliveries.has(`alice:${id}`)).toBe(false); + expect(onOwnerNotice).toHaveBeenCalledOnce(); + expect(onOwnerNotice.mock.calls[0]?.[0]).toContain("delivered after"); + }); + + it("stays silent for accepted receipts that were never reported overdue", async () => { + const alice = generateIdentity(); + const bob = reefKeys(); + const trusted = trust({ alice: peerTrust(alice) }); + const relay = transport(); + const onOwnerNotice = vi.fn(async (_text: string) => {}); + const flow = new ReefMessageFlow({ + config: config(), + trust: trusted.store, + keys: bob, + transport: relay as unknown as ReefTransportClient, + guard: guard(allow), + audit: new MemoryAuditStore(new Uint8Array(32).fill(24)), + replay: new MemoryReplayStore(), + ...flowStores(), + onIngress: async () => {}, + onOwnerNotice, + }); + const id = await flow.send("alice", "quick ping"); + const record = trusted.deliveries.get(`alice:${id}`)!; + const receipt = signReceipt( + { id, bodyHash: record.bodyHash, auditHead: "a".repeat(64), status: "accepted" }, + alice.signing.secretKey, + ); + + await expect( + flow.processEntries([{ seq: 1, peer: "alice", id, kind: "receipt", receipt, ts: 1 }]), + ).resolves.toEqual([]); + expect(onOwnerNotice).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/reef/src/flow.test-helpers.ts b/extensions/reef/src/flow.test-helpers.ts index 7a768914a82f..0b728e7396d8 100644 --- a/extensions/reef/src/flow.test-helpers.ts +++ b/extensions/reef/src/flow.test-helpers.ts @@ -111,6 +111,7 @@ export function trust(initial: Record) { textHash?: string; recipient: ReefPeerIdentity; resendDisabled?: true; + overdueNotifiedAt?: number; rejection?: { category?: string; notice?: ReefRejectionNoticeState; diff --git a/extensions/reef/src/flow.ts b/extensions/reef/src/flow.ts index 07f26722b902..879eb1872e51 100644 --- a/extensions/reef/src/flow.ts +++ b/extensions/reef/src/flow.ts @@ -260,6 +260,20 @@ export class ReefMessageFlow { return undefined; } if (receipt.status === "accepted") { + // The owner was told this send looked undelivered; close that loop so + // silence after an overdue notice always means "still undelivered". + // Notify before consuming the binding: a failed dispatch leaves the + // record for the retried receipt, while a duplicate enqueue stays + // deduped by its context key. Skip conflicted records — the rejection + // notice path owns their follow-up. A rejection cannot appear during + // this await: receipts are the only rejection writer and the inbox + // dispatches entries strictly serially (ReefInboxConnection.serialize), + // so this snapshot stays authoritative until the consume below. + if (delivery.overdueNotifiedAt !== undefined && !delivery.rejection) { + await this.options.onOwnerNotice( + `Reef message ${entry.id} to @${entry.peer} was delivered after the earlier delay notice; the peer's claw is reachable again.`, + ); + } if ( !this.options.trust.consumeOutboundDelivery(entry.peer, entry.id, delivery) && this.options.trust.outboundDelivery(entry.peer, entry.id)?.rejection diff --git a/extensions/reef/src/owner-notice.test.ts b/extensions/reef/src/owner-notice.test.ts index 699a019bd100..c9697d914bc2 100644 --- a/extensions/reef/src/owner-notice.test.ts +++ b/extensions/reef/src/owner-notice.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import type { ReefPeerIdentity } from "./friend-types.js"; import { createReefOwnerNoticeHandler, + notifyOverdueReefDeliveries, processReefInboxEntriesInOrder, ReefReceiptNotifier, } from "./owner-notice.js"; @@ -457,3 +458,70 @@ describe("ReefReceiptNotifier", () => { expect(notices.records.get(`${pending.peer}:${pending.id}`)?.phase).toBe("consumed"); }); }); + +describe("notifyOverdueReefDeliveries", () => { + function overdueStore(overdue: Array<{ peer: string; id: string; sentAt: number }>) { + const marked = new Set(); + return { + marked, + overdueOutboundDeliveries: vi.fn(() => overdue.filter((entry) => !marked.has(entry.id))), + markOutboundDeliveryOverdueNotified: vi.fn((_peer: string, id: string) => { + if (marked.has(id)) { + return false; + } + marked.add(id); + return true; + }), + }; + } + + it("wakes the sender agent once per overdue delivery", async () => { + const sentAt = Date.now() - 12 * 60_000; + const store = overdueStore([{ peer: "clawd", id: "01JZ0000000000000000000150", sentAt }]); + const ownerNotice = vi.fn( + async (_notice: { + text: string; + peer?: string; + contextKey: string; + wakeAgent?: boolean; + }) => {}, + ); + + await notifyOverdueReefDeliveries({ trust: store, ownerNotice }); + await notifyOverdueReefDeliveries({ trust: store, ownerNotice }); + + expect(ownerNotice).toHaveBeenCalledOnce(); + expect(ownerNotice.mock.calls[0]?.[0]).toMatchObject({ + peer: "clawd", + wakeAgent: true, + contextKey: "reef:delivery-overdue:clawd:01JZ0000000000000000000150", + }); + expect(ownerNotice.mock.calls[0]?.[0]?.text).toContain("not been confirmed delivered"); + }); + + it("does not mark a delivery notified when dispatch fails, so the next sweep retries", async () => { + const store = overdueStore([ + { peer: "clawd", id: "01JZ0000000000000000000151", sentAt: Date.now() - 20 * 60_000 }, + ]); + const failing = vi.fn(async () => { + throw new Error("enqueue failed"); + }); + + await expect( + notifyOverdueReefDeliveries({ trust: store, ownerNotice: failing }), + ).rejects.toThrow("enqueue failed"); + expect(store.markOutboundDeliveryOverdueNotified).not.toHaveBeenCalled(); + + const ownerNotice = vi.fn( + async (_notice: { + text: string; + peer?: string; + contextKey: string; + wakeAgent?: boolean; + }) => {}, + ); + await notifyOverdueReefDeliveries({ trust: store, ownerNotice }); + expect(ownerNotice).toHaveBeenCalledOnce(); + expect(store.marked.has("01JZ0000000000000000000151")).toBe(true); + }); +}); diff --git a/extensions/reef/src/owner-notice.ts b/extensions/reef/src/owner-notice.ts index a8f73a208e3a..a5da3a84f1de 100644 --- a/extensions/reef/src/owner-notice.ts +++ b/extensions/reef/src/owner-notice.ts @@ -398,6 +398,57 @@ export async function processReefInboxEntriesInOrder(params: { } } +// Long enough to ride out transport reconnects and peer gateway restarts; +// short enough that a human waiting on a cross-claw errand hears about a dead +// peer in minutes instead of never. +const REEF_DELIVERY_OVERDUE_NOTICE_MS = 10 * 60 * 1_000; + +interface ReefOverdueDeliveryStore { + overdueOutboundDeliveries( + olderThanMs: number, + now?: number, + ): Array<{ peer: string; id: string; sentAt: number }>; + markOutboundDeliveryOverdueNotified(peer: string, id: string): boolean; +} + +/** + * Follow-up for sends that produced no receipt at all (peer offline, peer + * inbox dead). Every other outcome already reports back: replies and + * rejection receipts dispatch turns, and local send failures reject the + * message tool call. Without this sweep an unacknowledged send is silent + * until its record ages out. + */ +export async function notifyOverdueReefDeliveries(params: { + trust: ReefOverdueDeliveryStore; + ownerNotice: (notice: ReefOwnerNotice) => Promise; + thresholdMs?: number; + now?: number; +}): Promise { + const thresholdMs = params.thresholdMs ?? REEF_DELIVERY_OVERDUE_NOTICE_MS; + for (const overdue of params.trust.overdueOutboundDeliveries(thresholdMs, params.now)) { + const elapsedMs = (params.now ?? Date.now()) - overdue.sentAt; + const minutes = Math.max(1, Math.round(elapsedMs / 60_000)); + // Dispatch before marking: a crash in between re-sends one deduped notice + // on the next tick, whereas marking first could silence it permanently — + // the exact failure this sweep exists to report. The context key keeps + // redispatch idempotent while the event is still queued, and a receipt + // only sees overdueNotifiedAt after the notice really went out. + await params.ownerNotice({ + text: `Reef message ${overdue.id} to @${overdue.peer} has not been confirmed delivered after ${minutes} minute${minutes === 1 ? "" : "s"}; the peer's claw looks offline or unreachable. The relay keeps it queued and you will get a follow-up if it is delivered or rejected. If your owner was waiting on this, let them know now.`, + peer: overdue.peer, + contextKey: `reef:delivery-overdue:${overdue.peer}:${overdue.id}`, + wakeAgent: true, + }); + // A failed mark means the record vanished mid-dispatch (receipt consumed + // it, keys changed, or it aged out). None of that is positive delivery + // evidence, so claim nothing here: only the accepted-receipt path may say + // "delivered after all", and a reply that races this notice corrects it + // naturally. Accepted tradeoff: in that sliver the delay notice stands + // uncorrected rather than risking a false delivery claim. + params.trust.markOutboundDeliveryOverdueNotified(overdue.peer, overdue.id); + } +} + export function createReefOwnerNoticeHandler(params: { runtime: PluginRuntime; cfg: ResolveAgentRouteParams["cfg"]; diff --git a/extensions/reef/src/trust-store.test.ts b/extensions/reef/src/trust-store.test.ts index 4878c2cf4d26..fe38159aacd7 100644 --- a/extensions/reef/src/trust-store.test.ts +++ b/extensions/reef/src/trust-store.test.ts @@ -134,14 +134,14 @@ describe("ReefTrustStore", () => { openReefTrustStore(runtime(), config()).recordOutboundDelivery("clawd", id, binding); const reopened = openReefTrustStore(runtime(), config()); - expect(reopened.outboundDelivery("clawd", id)).toEqual(binding); + expect(reopened.outboundDelivery("clawd", id)).toMatchObject(binding); expect( reopened.consumeOutboundDelivery("clawd", id, { ...binding, bodyHash: "b".repeat(64) }), ).toBe(false); expect( reopened.consumeOutboundDelivery("clawd", id, { ...binding, textHash: "c".repeat(64) }), ).toBe(false); - expect(reopened.outboundDelivery("clawd", id)).toEqual(binding); + expect(reopened.outboundDelivery("clawd", id)).toMatchObject(binding); expect(reopened.consumeOutboundDelivery("clawd", id, binding)).toBe(true); expect(reopened.outboundDelivery("clawd", id)).toBeUndefined(); expect(reopened.consumeOutboundDelivery("clawd", id, binding)).toBe(false); @@ -361,3 +361,56 @@ describe("ReefTrustStore", () => { expect(molty.matchesPairingApproval(token, friend)).toBe(false); }); }); + +describe("ReefTrustStore overdue outbound deliveries", () => { + const OVERDUE_MS = 10 * 60 * 1_000; + + it("reports an unacknowledged delivery overdue exactly once", () => { + const id = "01JZ0000000000000000000140"; + const store = openReefTrustStore(runtime(), config()); + const trustedPeer = peerTrust(); + store.set("clawd", trustedPeer); + const binding = { + bodyHash: "a".repeat(64), + textHash: "b".repeat(64), + recipient: reefPeerIdentity(trustedPeer), + }; + store.recordOutboundDelivery("clawd", id, binding); + + expect(store.overdueOutboundDeliveries(OVERDUE_MS)).toEqual([]); + const later = Date.now() + OVERDUE_MS + 1_000; + expect(store.overdueOutboundDeliveries(OVERDUE_MS, later)).toMatchObject([ + { peer: "clawd", id }, + ]); + + expect(store.markOutboundDeliveryOverdueNotified("clawd", id)).toBe(true); + expect(store.markOutboundDeliveryOverdueNotified("clawd", id)).toBe(false); + expect(store.overdueOutboundDeliveries(OVERDUE_MS, later)).toEqual([]); + }); + + it("excludes rejected and unpinned deliveries from the overdue sweep", () => { + const store = openReefTrustStore(runtime(), config()); + const trustedPeer = peerTrust(); + store.set("clawd", trustedPeer); + const later = Date.now() + OVERDUE_MS + 1_000; + + const rejectedId = "01JZ0000000000000000000141"; + const rejectedBinding = { + bodyHash: "c".repeat(64), + recipient: reefPeerIdentity(trustedPeer), + }; + store.recordOutboundDelivery("clawd", rejectedId, rejectedBinding); + expect(store.recordOutboundRejection("clawd", rejectedId, rejectedBinding, "guard_deny")).toBe( + true, + ); + + const unpinnedId = "01JZ0000000000000000000142"; + store.recordOutboundDelivery("stranger", unpinnedId, { + bodyHash: "d".repeat(64), + recipient: reefPeerIdentity(peerTrust()), + }); + + expect(store.overdueOutboundDeliveries(OVERDUE_MS, later)).toEqual([]); + expect(store.markOutboundDeliveryOverdueNotified("clawd", rejectedId)).toBe(false); + }); +}); diff --git a/extensions/reef/src/trust-store.ts b/extensions/reef/src/trust-store.ts index 6942b76168ae..f78bd5c7f8a2 100644 --- a/extensions/reef/src/trust-store.ts +++ b/extensions/reef/src/trust-store.ts @@ -48,6 +48,10 @@ const ReefOutboundDeliveryBindingSchema = z const ReefOutboundDeliverySchema = ReefOutboundDeliveryBindingSchema.extend({ resendDisabled: z.literal(true).optional(), rejection: ReefOutboundRejectionSchema.optional(), + // sentAt is absent on records written before overdue notices shipped; those + // legacy sends age out via TTL without an overdue follow-up. + sentAt: z.number().int().positive().optional(), + overdueNotifiedAt: z.number().int().positive().optional(), }).strict(); const ReefPeerStateSchema = z .object({ @@ -335,12 +339,61 @@ export class ReefTrustStore { options: { resendDisabled?: true } = {}, ): void { const key = this.#deliveryKey(peer, id); - const value = ReefOutboundDeliverySchema.parse({ ...binding, ...options }); + const value = ReefOutboundDeliverySchema.parse({ ...binding, ...options, sentAt: Date.now() }); if (!this.stores.deliveries.registerIfAbsent(key, value)) { throw new Error(`Duplicate outbound Reef delivery id ${id}`); } } + /** + * Sends that never produced any receipt. Rejections have their own notice + * path, and each delivery is reported overdue at most once. + */ + overdueOutboundDeliveries( + olderThanMs: number, + now: number = Date.now(), + ): Array<{ peer: string; id: string; sentAt: number }> { + return this.stores.deliveries + .entries() + .filter((entry) => entry.key.startsWith(this.#prefix)) + .flatMap((entry) => { + const parsed = ReefOutboundDeliverySchema.safeParse(entry.value); + if ( + !parsed.success || + parsed.data.rejection || + parsed.data.overdueNotifiedAt !== undefined || + parsed.data.sentAt === undefined || + parsed.data.sentAt + olderThanMs > now + ) { + return []; + } + const separator = entry.key.lastIndexOf(":"); + const peer = requirePeer(entry.key.slice(this.#prefix.length, separator)); + const id = entry.key.slice(separator + 1); + if ( + !MESSAGE_ID_PATTERN.test(id) || + !matchesReefPeerIdentity(this.get(peer), parsed.data.recipient) + ) { + return []; + } + return [{ peer, id, sentAt: parsed.data.sentAt }]; + }); + } + + markOutboundDeliveryOverdueNotified(peer: string, id: string): boolean { + const update = this.stores.deliveries.update; + if (!update) { + throw new Error("Reef outbound delivery state requires atomic plugin-state updates"); + } + return update(this.#deliveryKey(peer, id), (value) => { + const parsed = ReefOutboundDeliverySchema.safeParse(value); + if (!parsed.success || parsed.data.rejection || parsed.data.overdueNotifiedAt !== undefined) { + return undefined; + } + return { ...parsed.data, overdueNotifiedAt: Date.now() }; + }); + } + outboundDelivery( peer: string, id: string,