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

@@ -122,6 +122,8 @@ Agents send through the shared `message` tool to `reef:<handle>`; 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                                                         |

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<ReefAccount> = {
         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)}`),
         });

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();
+  });
+});

extensions/reef/src/flow.test-helpers.ts

@@ -111,6 +111,7 @@ export function trust(initial: Record<string, ReefPeerTrust>) {
       textHash?: string;
       recipient: ReefPeerIdentity;
       resendDisabled?: true;
+      overdueNotifiedAt?: number;
       rejection?: {
         category?: string;
         notice?: ReefRejectionNoticeState;

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

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<string>();
+    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);
+  });
+});

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<void>;
+  thresholdMs?: number;
+  now?: number;
+}): Promise<void> {
+  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"];

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);
+  });
+});

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,