fix(notices): keep wire-successful receipts owed and renew the hold during pending sends

Codex P2 x2 on #376:
- A send that succeeded but whose signalAvailability() commit failed was
  deduplicated only in memory, so a restarted signaller could resend it once
  the abandoned hold lapsed. Owed receipts are now retried with the same
  idempotency key before any later observation spends, and on close().
- A protocol write pending longer than the 30s TTL let another process take
  the hold and send too. The holder now renews under its key while send() is
  pending; the reducer refuses a foreign key while a hold is live and refuses a
  lapsed holder's late renewal once another key has taken over.

Author: ScriptedAlchemy

.changeset/notice-inbox-resource-updated.md

@@ -3,4 +3,4 @@
 "agent-bundle": patch
 ---
 
-Wire the #99 stage-4 `mcp-resource-updated` delivery route into generated stateful MCP servers. `@agent-bundle/runtime/notices` gains `createNoticeInboxSignaller` — one connection's subscription to the reserved inbox resource (`AGENT_NOTICE_INBOX_URI`) that, after each completed render, sends at most one `notifications/resources/updated` for the subscriber's newly eligible pending notices and records it through `signalAvailability()` as an availability receipt (never delivery), honouring `nextAttemptAt` and bounding signals per notice by `retryBudget` across restarts. The ledger gains `reserveAvailability()` / `releaseAvailability()` (and `AgentNotice.availabilityReservation`, `AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS`): the signaller holds a notice's budget slot by compare-and-swap before the wire write so concurrent processes cannot both send, finalizes the hold into the receipt only when the protocol write succeeds, and releases it when the write fails, so a failed send costs no budget and the receipt always means the write succeeded; `unsubscribe()` resolves only after in-flight observations settle; `@agent-bundle/runtime/mount` gains `createGeneratedNoticeRuntime` and `GeneratedRuntimeState.noticeLedger()` so a server process can hold its own handle on the durable store its worker mounts. Generated workspace-durable MCP entries now register `resources/subscribe`/`resources/unsubscribe` for the inbox URI only, advertise `resources.subscribe` exactly when that wiring is active, fail subscriptions closed when the store is unreadable, and the inbox projection exposes the `availability` receipt alongside `exposure`. Volatile lifetimes keep the store in the worker's heap and advertise no subscription capability.
+Wire the #99 stage-4 `mcp-resource-updated` delivery route into generated stateful MCP servers. `@agent-bundle/runtime/notices` gains `createNoticeInboxSignaller` — one connection's subscription to the reserved inbox resource (`AGENT_NOTICE_INBOX_URI`) that, after each completed render, sends at most one `notifications/resources/updated` for the subscriber's newly eligible pending notices and records it through `signalAvailability()` as an availability receipt (never delivery), honouring `nextAttemptAt` and bounding signals per notice by `retryBudget` across restarts. The ledger gains `reserveAvailability()` / `releaseAvailability()` (and `AgentNotice.availabilityReservation`, `AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS`): the signaller holds a notice's budget slot by compare-and-swap before the wire write so concurrent processes cannot both send, finalizes the hold into the receipt only when the protocol write succeeds, and releases it when the write fails, so a failed send costs no budget and the receipt always means the write succeeded. The hold is renewed while the write is pending (a different key may take it over only after the TTL, and a lapsed holder cannot steal it back), and a receipt whose commit failed after a successful send is retried idempotently before later observations and on `close()`; `unsubscribe()` resolves only after in-flight observations settle; `@agent-bundle/runtime/mount` gains `createGeneratedNoticeRuntime` and `GeneratedRuntimeState.noticeLedger()` so a server process can hold its own handle on the durable store its worker mounts. Generated workspace-durable MCP entries now register `resources/subscribe`/`resources/unsubscribe` for the inbox URI only, advertise `resources.subscribe` exactly when that wiring is active, fail subscriptions closed when the store is unreadable, and the inbox projection exposes the `availability` receipt alongside `exposure`. Volatile lifetimes keep the store in the worker's heap and advertise no subscription capability.

packages/rsc-runtime/README.md

@@ -266,10 +266,15 @@ protocol write failed, so the receipt only ever means the write succeeded and
 a failed send costs no budget. Eligibility is recipient-matched against the
 subscriber's observed identity, respects `nextAttemptAt`, skips notices whose
 slot another signaller currently holds (a hold older than
-`AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS` counts as abandoned), and is
+`AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS` counts as abandoned; a live
+holder renews it under its key while its write is pending, and the reducer
+refuses a renewal once another key has legitimately taken over), and is
 bounded by `retryBudget` (availability receipts per notice, durable across
 restarts); because the slot is held before the wire write, two server processes
-over one store can never both signal the same notice. Exposure and availability
+over one store can never both signal the same notice. A send that reached the
+wire but whose receipt commit failed stays owed: the same idempotent receipt is
+retried before any later observation spends (and on `close()`), so a restarted
+process cannot resend a notice the wire already carried. Exposure and availability
 receipts never re-trigger a signal, so a subscribed client cannot be driven into
 a refetch loop. Subscribing fails closed when the store is unreadable,
 `unsubscribe()` resolves only after in-flight observations settle, and only the

packages/rsc-runtime/src/notices/resource-updated.ts

@@ -28,6 +28,12 @@ export interface AgentNoticeInboxStore {
 export interface CreateNoticeInboxSignallerOptions {
   /** Clock injection for deterministic tests. */
   readonly now?: () => Date;
+  /**
+   * How often a hold is renewed while its `send()` is still pending. Defaults
+   * to a third of `AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS` so a live
+   * holder never lapses; only a holder whose process is gone does.
+   */
+  readonly reservationRenewalIntervalMs?: number;
   readonly store: AgentNoticeInboxStore;
 }
 
@@ -59,11 +65,14 @@ export type AgentNoticeInboxSignalOutcome =
 export interface AgentNoticeInboxSignaller {
   readonly inboxUri: typeof AGENT_NOTICE_INBOX_URI;
   readonly subscribed: boolean;
+  /** Commits any receipt still owed for a send that reached the wire, then closes the store. */
   close(): Promise<void>;
   /**
-   * Runs after one completed render: reads the ledger and, when the
-   * subscriber has newly eligible pending notices, sends exactly one
-   * `resources/updated` through `send` and records the availability receipt.
+   * Runs after one completed render: first commits any receipt still owed from
+   * an earlier send, then reads the ledger and, when the subscriber has newly
+   * eligible pending notices, holds their budget slot, sends exactly one
+   * `resources/updated` through `send` (renewing the hold while the write is
+   * pending), and records the availability receipt once the write succeeded.
    * Never throws; failures are returned so the render path stays unaffected.
    */
   observe(send: () => Promise<void>): Promise<AgentNoticeInboxSignalOutcome>;
@@ -89,6 +98,13 @@ interface InboxSubscription {
   readonly signalled: Set<string>;
 }
 
+/** A send that succeeded on the wire whose availability receipt has not been committed yet. */
+interface PendingReceipt {
+  readonly at: string;
+  readonly noticeIds: readonly string[];
+  readonly reservationKey: string;
+}
+
 const eligibleForSignal = (
   notice: AgentNotice,
   principal: AgentNoticePrincipal,
@@ -127,8 +143,16 @@ export const createNoticeInboxSignaller = (
   options: CreateNoticeInboxSignallerOptions,
 ): AgentNoticeInboxSignaller => {
   const now = options.now ?? ((): Date => new Date());
+  const renewalIntervalMs = options.reservationRenewalIntervalMs
+    ?? Math.floor(AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS / 3);
   let subscription: InboxSubscription | undefined;
   let signalSequence = 0;
+  // Sends that reached the wire but whose receipt commit failed. The send is
+  // a fact, so the receipt is retried with the same idempotency key on every
+  // later observation (and on close) until it lands; the memory dedupe alone
+  // would otherwise let a restarted signaller spend the same slot again once
+  // the abandoned hold lapsed.
+  const pendingReceipts = new Map<string, PendingReceipt>();
   // Observations and subscription changes serialize on one queue: two renders
   // completing together cannot both select the same notice and send two
   // signals for one revision, and an unsubscribe (or re-subscribe) that
@@ -210,9 +234,74 @@ export const createNoticeInboxSignaller = (
     };
   };
 
+  /** Commits the receipt for one wire-successful send; idempotent across retries. */
+  const commitReceipt = (
+    ledger: AgentNoticeLedger,
+    receipt: PendingReceipt,
+  ): Promise<Awaited<ReturnType<AgentNoticeLedger['read']>>> => ledger.signalAvailability({
+    at: receipt.at,
+    idempotencyKey: `agent-notices:availability:signal:${receipt.reservationKey}`,
+    noticeIds: receipt.noticeIds,
+    reservationKey: receipt.reservationKey,
+  });
+
+  /** Retries every outstanding receipt; the first failure is returned so the caller reports it. */
+  const drainPendingReceipts = async (ledger: AgentNoticeLedger): Promise<unknown> => {
+    for (const receipt of [...pendingReceipts.values()]) {
+      try {
+        await commitReceipt(ledger, receipt);
+        pendingReceipts.delete(receipt.reservationKey);
+      } catch (error) {
+        return error;
+      }
+    }
+    return undefined;
+  };
+
+  /**
+   * Keeps a hold alive while its protocol write is pending. A write that
+   * outlives the TTL would otherwise let another process treat the hold as
+   * abandoned and send too; renewing under the same key refreshes `at`, and
+   * the reducer refuses a renewal once a different key has legitimately taken
+   * over, so a holder that could not renew for a whole TTL never steals back.
+   * The timer exists only for the duration of one in-flight send.
+   */
+  const renewWhile = async <T>(
+    ledger: AgentNoticeLedger,
+    hold: { readonly noticeIds: readonly string[]; readonly reservationKey: string },
+    pending: Promise<T>,
+  ): Promise<T> => {
+    let renewals = 0;
+    let stopped = false;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const tick = (): void => {
+      if (stopped) return;
+      renewals += 1;
+      const renewal = renewals;
+      void ledger.reserveAvailability({
+        at: now().toISOString(),
+        idempotencyKey: `agent-notices:availability:renew:${hold.reservationKey}:${String(renewal)}`,
+        noticeIds: hold.noticeIds,
+        reservationKey: hold.reservationKey,
+      }).catch(() => undefined).then(() => {
+        if (stopped) return;
+        timer = setTimeout(tick, renewalIntervalMs);
+        timer.unref?.();
+      });
+    };
+    timer = setTimeout(tick, renewalIntervalMs);
+    timer.unref?.();
+    try {
+      return await pending;
+    } finally {
+      stopped = true;
+      if (timer !== undefined) clearTimeout(timer);
+    }
+  };
+
   const observeOnce = async (send: () => Promise<void>): Promise<AgentNoticeInboxSignalOutcome> => {
     const current = subscription;
-    if (current === undefined) {
+    if (current === undefined && pendingReceipts.size === 0) {
       return Object.freeze({ kind: 'idle', reason: 'no-subscription', revision: undefined });
     }
     let ledger: AgentNoticeLedger;
@@ -221,7 +310,13 @@ export const createNoticeInboxSignaller = (
     } catch (error) {
       return Object.freeze({ error, kind: 'failed' as const, stage: 'read' as const });
     }
-    if (pendingUnsubscribes > 0) {
+    // Receipts owed from earlier sends come first: they are facts about the
+    // wire, and a ledger that cannot take them is not one to spend against.
+    const owed = await drainPendingReceipts(ledger);
+    if (owed !== undefined) {
+      return Object.freeze({ error: owed, kind: 'failed' as const, stage: 'record' as const });
+    }
+    if (current === undefined || pendingUnsubscribes > 0) {
       return Object.freeze({ kind: 'idle', reason: 'no-subscription', revision: undefined });
     }
     const claimed = await claim(ledger, current);
@@ -245,7 +340,7 @@ export const createNoticeInboxSignaller = (
     // send finalizes the hold into the availability receipt. Only the receipt
     // means the protocol write succeeded.
     try {
-      await send();
+      await renewWhile(ledger, claimed, send());
     } catch (error) {
       try {
         await ledger.releaseAvailability({
@@ -259,14 +354,18 @@ export const createNoticeInboxSignaller = (
       }
       return Object.freeze({ error, kind: 'failed' as const, stage: 'send' as const });
     }
+    // The wire write succeeded: this subscription never sends for these notices
+    // again, and the receipt is owed until it commits.
     for (const id of claimed.noticeIds) current.signalled.add(id);
+    const receipt: PendingReceipt = Object.freeze({
+      at: claimed.at,
+      noticeIds: claimed.noticeIds,
+      reservationKey: claimed.reservationKey,
+    });
+    pendingReceipts.set(receipt.reservationKey, receipt);
     try {
-      const committed = await ledger.signalAvailability({
-        at: claimed.at,
-        idempotencyKey: `agent-notices:availability:signal:${claimed.reservationKey}`,
-        noticeIds: claimed.noticeIds,
-        reservationKey: claimed.reservationKey,
-      });
+      const committed = await commitReceipt(ledger, receipt);
+      pendingReceipts.delete(receipt.reservationKey);
       return Object.freeze({ kind: 'signalled', noticeIds: claimed.noticeIds, revision: committed.revision });
     } catch (error) {
       return Object.freeze({ error, kind: 'failed' as const, stage: 'record' as const });
@@ -279,8 +378,18 @@ export const createNoticeInboxSignaller = (
       return subscription !== undefined;
     },
     close(): Promise<void> {
-      subscription = undefined;
-      return options.store.close();
+      return serialized(async () => {
+        subscription = undefined;
+        if (pendingReceipts.size > 0) {
+          try {
+            await drainPendingReceipts(await options.store.noticeLedger());
+          } catch {
+            // A receipt still owed at close is lost with the process; the hold
+            // it left behind lapses after the TTL.
+          }
+        }
+        await options.store.close();
+      });
     },
     observe(send: () => Promise<void>): Promise<AgentNoticeInboxSignalOutcome> {
       return serialized(() => observeOnce(send));

packages/rsc-runtime/src/notices/state.ts

@@ -11,10 +11,11 @@ import type {
   AgentStateLifetime,
 } from '../state/contract.js';
 import { canonicalJson, defineState } from '../state/index.js';
-import type {
-  AgentNotice,
-  AgentNoticePrincipal,
-  AgentRecipient,
+import {
+  AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS,
+  type AgentNotice,
+  type AgentNoticePrincipal,
+  type AgentRecipient,
 } from './contract.js';
 
 const observed = <T extends z.ZodType>(value: T) => z.discriminatedUnion('state', [
@@ -299,10 +300,12 @@ const transitionAvailability = (
 };
 
 /**
- * Holds a budget slot without spending it. A live notice may carry one
- * reservation; a newer reserver replaces an older one outright because the
- * ledger cannot tell abandoned from slow, so the signaller decides staleness
- * by `at` against the reservation TTL before it ever reserves.
+ * Holds a budget slot without spending it. A live notice carries at most one
+ * reservation: the holder renews it by reserving again under the same key,
+ * and a different key takes it over only once the current hold is older than
+ * the reservation TTL at the event's own `at` — a rule the reducer can apply
+ * deterministically on replay, so a slow holder's renewal can never steal a
+ * hold back from the process that legitimately took over after it lapsed.
  */
 const transitionAvailabilityReservation = (
   notice: AgentNotice,
@@ -311,11 +314,20 @@ const transitionAvailabilityReservation = (
   if (!input.noticeIds.has(notice.id)) return notice;
   switch (notice.state) {
     case 'pending':
-    case 'attempted':
+    case 'attempted': {
+      const held = notice.availabilityReservation;
+      if (
+        held !== undefined
+        && held.key !== input.reservationKey
+        && Date.parse(held.at) + AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS > Date.parse(input.at)
+      ) {
+        return notice;
+      }
       return Object.freeze({
         ...notice,
         availabilityReservation: Object.freeze({ at: input.at, key: input.reservationKey }),
       });
+    }
     case 'expired':
     case 'unavailable':
     case 'withdrawn':

packages/rsc-runtime/tests/notices-ledger.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from '@rstest/core';
 
 import {
+  AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS,
   AGENT_NOTICE_STATES,
   AgentNoticeError,
   selectNoticeDeliveryRoutes,
@@ -882,6 +883,46 @@ describe('notice delivery routing receipts (#99 stage 4)', () => {
       reservationKey: 'holder-b:1',
     })).rejects.toMatchObject({ code: 'revision-conflict' });
 
+    // A live hold is not overwritten by a different key even without a
+    // compare-and-swap; the holder's own key renews it; a lapsed hold is taken.
+    const contested = await ledger.reserveAvailability({
+      at: '2026-09-01T19:04:02.000Z',
+      idempotencyKey: 'reserve:b-uncontended',
+      noticeIds: [id],
+      reservationKey: 'holder-b:1',
+    });
+    expect(contested.notices[0]?.availabilityReservation).toEqual({ at: '2026-09-01T19:04:00.000Z', key: 'holder-a:1' });
+    const renewed = await ledger.reserveAvailability({
+      at: '2026-09-01T19:04:10.000Z',
+      idempotencyKey: 'renew:a:1',
+      noticeIds: [id],
+      reservationKey: 'holder-a:1',
+    });
+    expect(renewed.notices[0]?.availabilityReservation).toEqual({ at: '2026-09-01T19:04:10.000Z', key: 'holder-a:1' });
+    const lapsedAt = new Date(Date.parse('2026-09-01T19:04:10.000Z') + AGENT_NOTICE_AVAILABILITY_RESERVATION_TTL_MS).toISOString();
+    const takenOver = await ledger.reserveAvailability({
+      at: lapsedAt,
+      idempotencyKey: 'reserve:b-after-lapse',
+      noticeIds: [id],
+      reservationKey: 'holder-b:1',
+    });
+    expect(takenOver.notices[0]?.availabilityReservation).toEqual({ at: lapsedAt, key: 'holder-b:1' });
+    // The lapsed holder's late renewal cannot steal the hold back.
+    const lateRenewal = await ledger.reserveAvailability({
+      at: new Date(Date.parse(lapsedAt) + 1_000).toISOString(),
+      idempotencyKey: 'renew:a:2',
+      noticeIds: [id],
+      reservationKey: 'holder-a:1',
+    });
+    expect(lateRenewal.notices[0]?.availabilityReservation).toEqual({ at: lapsedAt, key: 'holder-b:1' });
+    await ledger.releaseAvailability({ idempotencyKey: 'release:b-takeover', noticeIds: [id], reservationKey: 'holder-b:1' });
+    await ledger.reserveAvailability({
+      at: '2026-09-01T19:04:00.000Z',
+      idempotencyKey: 'reserve:a-again',
+      noticeIds: [id],
+      reservationKey: 'holder-a:1',
+    });
+
     // Releasing with another holder's key leaves the hold intact; the owner's key clears it.
     const foreignRelease = await ledger.releaseAvailability({
       idempotencyKey: 'release:b',