fix(notices): stable replay deliveries, admission ordering, abort-aware authorization

.changeset/fix-notice-admission-replay.md

@@ -0,0 +1,8 @@
+---
+"@agent-bundle/runtime": patch
+---
+
+Keep next-event notice admission deterministic across invocation replays,
+scope replayed deliveries to the matching principal, exclude notices created
+after an event started, and interrupt delivery authorization when the request
+is aborted.

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

@@ -271,13 +271,15 @@ export const createAgentNoticeLedger = (
       let deliveries: readonly AgentNoticeDelivery[] = Object.freeze([]);
       if (request.invocation.kind === 'event') {
         const before = yield* storeEffect(() => store.read({ signal: request.signal }));
+        let admitted = before.state;
         const admissionTime = Date.parse(request.invocation.startedAt);
         const expiring = before.state.notices.filter((notice) =>
           notice.state === 'pending'
           && notice.expiresAt !== undefined
           && Date.parse(notice.expiresAt) <= admissionTime);
         const candidates = before.state.notices.filter((notice) =>
           notice.state === 'pending'
+          && Date.parse(notice.createdAt) <= admissionTime
           && (notice.expiresAt === undefined
             || Date.parse(notice.expiresAt) > admissionTime)
           && recipientMatchesPrincipal(notice.recipient, request.principal));
@@ -307,10 +309,12 @@ export const createAgentNoticeLedger = (
               signal: request.signal,
             },
           ));
-          deliveries = Object.freeze(committed.state.notices
-            .map((notice) => deliveryFor(notice, request.invocation.id))
-            .filter((delivery): delivery is AgentNoticeDelivery => delivery !== undefined));
+          admitted = committed.state;
         }
+        deliveries = Object.freeze(admitted.notices
+          .filter((notice) => recipientMatchesPrincipal(notice.recipient, request.principal))
+          .map((notice) => deliveryFor(notice, request.invocation.id))
+          .filter((delivery): delivery is AgentNoticeDelivery => delivery !== undefined));
       }
 
       let closed = false;
@@ -334,7 +338,7 @@ export const createAgentNoticeLedger = (
         },
         handle,
       });
-    }));
+    }), { signal: request.signal });
   },
 
   async read(): Promise<AgentNoticeLedgerSnapshot> {

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

@@ -258,6 +258,115 @@ describe('next-event delivery', () => {
     await driver.close();
   });
 
+  it('replays the same admitted notice set only for the matching principal', async () => {
+    const { driver, ledger } = await openLedger();
+    const published = await run(ledger, {
+      actorId: 'publisher',
+      id: 'publish-replay',
+      kind: 'tool',
+      startedAt: '2026-09-01T19:00:00.000Z',
+    }, async () => (await agent()).notices!.publish({
+      content: document('replay'),
+      priority: 'normal',
+      recipient: { actor: { id: 'recipient' } },
+    }, { idempotencyKey: 'publish:replay' }));
+    const invocation = {
+      actorId: 'recipient',
+      id: 'event-replay',
+      kind: 'event' as const,
+      startedAt: '2026-09-01T19:02:00.000Z',
+    };
+
+    const first = await run(ledger, invocation, async () => (await agent()).notices!.read());
+    const otherPrincipal = await run(ledger, {
+      ...invocation,
+      actorId: 'other',
+    }, async () => (await agent()).notices!.read());
+    const replay = await run(ledger, invocation, async () => (await agent()).notices!.read());
+
+    expect(first.map(({ notice }) => notice.id)).toEqual([published.notice.id]);
+    expect(otherPrincipal).toEqual([]);
+    expect(replay.map(({ notice }) => notice.id)).toEqual(
+      first.map(({ notice }) => notice.id),
+    );
+    await driver.close();
+  });
+
+  it('does not admit notices created after the event started', async () => {
+    const { driver, ledger } = await openLedger();
+    await run(ledger, {
+      actorId: 'publisher',
+      id: 'publish-future',
+      kind: 'tool',
+      startedAt: '2026-09-01T19:10:00.000Z',
+    }, async () => (await agent()).notices!.publish({
+      content: document('future'),
+      priority: 'normal',
+      recipient: { actor: { id: 'recipient' } },
+    }, { idempotencyKey: 'publish:future' }));
+
+    const observed = await run(ledger, {
+      actorId: 'recipient',
+      id: 'event-before-publish',
+      kind: 'event',
+      startedAt: '2026-09-01T19:05:00.000Z',
+    }, async () => (await agent()).notices!.read());
+
+    expect(observed).toEqual([]);
+    expect((await ledger.read()).notices[0]?.state).toBe('pending');
+    await driver.close();
+  });
+
+  it('interrupts delivery authorization when the request is aborted', { timeout: 5_000 }, async () => {
+    let authorizationStarted!: () => void;
+    const started = new Promise<void>((resolve) => {
+      authorizationStarted = resolve;
+    });
+    const { driver, ledger } = await openLedger((request) => {
+      if (request.phase === 'publish') return { state: 'authorized' };
+      authorizationStarted();
+      return new Promise(() => undefined);
+    });
+    await run(ledger, {
+      actorId: 'publisher',
+      id: 'publish-abort',
+      kind: 'tool',
+      startedAt: '2026-09-01T19:00:00.000Z',
+    }, async () => (await agent()).notices!.publish({
+      content: document('abort'),
+      priority: 'normal',
+      recipient: { actor: { id: 'recipient' } },
+    }, { idempotencyKey: 'publish:abort' }));
+    const controller = new AbortController();
+    const opening = runAgentRequest({
+      actor: actor('recipient'),
+      host,
+      invocation: {
+        id: 'event-abort',
+        kind: 'event',
+        startedAt: '2026-09-01T19:02:00.000Z',
+      },
+      noticeLedger: ledger,
+      session,
+      signal: controller.signal,
+      workspace,
+    }, async () => (await agent()).notices!.read());
+
+    await started;
+    controller.abort('test abort');
+    const guarded = Promise.race([
+      opening,
+      new Promise<never>((_, reject) => {
+        AbortSignal.timeout(1_000).addEventListener('abort', () => {
+          reject(new Error('Timed out waiting for authorization interruption'));
+        }, { once: true });
+      }),
+    ]);
+
+    await expect(guarded).rejects.toMatchObject({ name: 'AbortError' });
+    await driver.close();
+  });
+
   it('marks a matched notice unavailable when delivery-time authorization is unavailable', async () => {
     const { driver, ledger } = await openLedger((request) => ({
       state: request.phase === 'publish' ? 'authorized' : 'unavailable',