diff --git a/.changeset/fix-notice-admission-replay.md b/.changeset/fix-notice-admission-replay.md new file mode 100644 index 000000000..34317a92b --- /dev/null +++ b/.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. diff --git a/packages/rsc-runtime/src/notices/ledger.ts b/packages/rsc-runtime/src/notices/ledger.ts index 6bb4d4aed..b86657e1f 100644 --- a/packages/rsc-runtime/src/notices/ledger.ts +++ b/packages/rsc-runtime/src/notices/ledger.ts @@ -271,6 +271,7 @@ 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' @@ -278,6 +279,7 @@ export const createAgentNoticeLedger = ( && 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 { diff --git a/packages/rsc-runtime/tests/notices-ledger.test.ts b/packages/rsc-runtime/tests/notices-ledger.test.ts index 6d4fd0ac3..5afff97af 100644 --- a/packages/rsc-runtime/tests/notices-ledger.test.ts +++ b/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((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((_, 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',