fix(notices): serialize unsubscribe with in-flight inbox observations

An unsubscribe that overlaps an observation parked on the durable store now
wins: the observation yields before claiming any budget or sending, and the
unsubscribe resolves (and the MCP request is acknowledged) only after in-flight
observations settle, so a client is never signalled after its unsubscribe
succeeded. Subscribe and unsubscribe share the observation queue.

Author: ScriptedAlchemy

packages/agent-bundle/src/mcp-server-runtime.ts

@@ -524,9 +524,11 @@ const installNoticeInboxSubscriptions = (
     }
     return {};
   }) as never);
-  protocol.setRequestHandler('resources/unsubscribe', ((request: ResourceSubscriptionRequest) => {
+  protocol.setRequestHandler('resources/unsubscribe', (async (request: ResourceSubscriptionRequest) => {
     assertInboxUri(request.params.uri);
-    notices.unsubscribe();
+    // Acknowledged only once in-flight observations have settled, so the
+    // client never receives a signal after its unsubscribe succeeded.
+    await notices.unsubscribe();
     return {};
   }) as never);
   const send = async (): Promise<void> => {

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

@@ -73,7 +73,13 @@ export interface AgentNoticeInboxSignaller {
    * subscription that could never be honoured.
    */
   subscribe(principal: AgentNoticePrincipal): Promise<void>;
-  unsubscribe(): void;
+  /**
+   * Ends the subscription. Resolves only after every observation already in
+   * flight has settled, so once it resolves no further signal is sent or
+   * budget spent for the connection — a `resources/unsubscribe` acknowledged
+   * to the client is honoured even while the store is slow.
+   */
+  unsubscribe(): Promise<void>;
 }
 
 interface InboxSubscription {
@@ -115,9 +121,21 @@ export const createNoticeInboxSignaller = (
   const now = options.now ?? ((): Date => new Date());
   let subscription: InboxSubscription | undefined;
   let signalSequence = 0;
-  // Observations serialize so two renders completing together cannot both
-  // select the same notice and send two signals for one revision.
+  // 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
+  // overlaps an observation awaiting the store takes effect only after that
+  // observation settles, never between its eligibility read and its send.
   let queue: Promise<unknown> = Promise.resolve();
+  // Unsubscribes requested while an observation awaits the store are counted
+  // synchronously so that observation yields before spending any budget: the
+  // client asked to stop, so nothing is claimed or sent on its behalf.
+  let pendingUnsubscribes = 0;
+  const serialized = <T>(step: () => Promise<T>): Promise<T> => {
+    const run = queue.then(step);
+    queue = run.catch(() => undefined);
+    return run;
+  };
 
   /**
    * Claims the budget for every newly eligible notice as one compare-and-swap
@@ -133,6 +151,7 @@ export const createNoticeInboxSignaller = (
   ): Promise<
     | { readonly at: string; readonly kind: 'claimed'; readonly noticeIds: readonly string[]; readonly revision: number }
     | { readonly kind: 'nothing-eligible'; readonly revision: number }
+    | { readonly kind: 'unsubscribed' }
     | { readonly error: unknown; readonly kind: 'failed'; readonly stage: 'read' | 'record' }
   > => {
     for (let attempt = 0; attempt < MAX_CLAIM_ATTEMPTS; attempt += 1) {
@@ -142,6 +161,7 @@ export const createNoticeInboxSignaller = (
       } catch (error) {
         return { error, kind: 'failed', stage: 'read' };
       }
+      if (pendingUnsubscribes > 0 || subscription !== current) return { kind: 'unsubscribed' };
       const at = now().toISOString();
       const nowMs = Date.parse(at);
       const noticeIds = Object.freeze(snapshot.notices
@@ -183,12 +203,17 @@ export const createNoticeInboxSignaller = (
     } catch (error) {
       return Object.freeze({ error, kind: 'failed' as const, stage: 'read' as const });
     }
+    if (pendingUnsubscribes > 0) {
+      return Object.freeze({ kind: 'idle', reason: 'no-subscription', revision: undefined });
+    }
     const claimed = await claim(ledger, current);
     switch (claimed.kind) {
       case 'failed':
         return Object.freeze({ error: claimed.error, kind: 'failed' as const, stage: claimed.stage });
       case 'nothing-eligible':
         return Object.freeze({ kind: 'idle', reason: 'nothing-eligible', revision: claimed.revision });
+      case 'unsubscribed':
+        return Object.freeze({ kind: 'idle', reason: 'no-subscription', revision: undefined });
       case 'claimed':
         break;
       default: {
@@ -219,21 +244,25 @@ export const createNoticeInboxSignaller = (
       return options.store.close();
     },
     observe(send: () => Promise<void>): Promise<AgentNoticeInboxSignalOutcome> {
-      const run = queue.then(() => observeOnce(send));
-      queue = run.catch(() => undefined);
-      return run;
+      return serialized(() => observeOnce(send));
     },
-    async subscribe(principal: AgentNoticePrincipal): Promise<void> {
-      const ledger = await options.store.noticeLedger();
-      await ledger.read();
-      subscription = Object.freeze({
-        id: randomUUID(),
-        principal,
-        signalled: new Set<string>(),
+    subscribe(principal: AgentNoticePrincipal): Promise<void> {
+      return serialized(async () => {
+        const ledger = await options.store.noticeLedger();
+        await ledger.read();
+        subscription = Object.freeze({
+          id: randomUUID(),
+          principal,
+          signalled: new Set<string>(),
+        });
       });
     },
-    unsubscribe(): void {
-      subscription = undefined;
+    unsubscribe(): Promise<void> {
+      pendingUnsubscribes += 1;
+      return serialized(async () => {
+        pendingUnsubscribes -= 1;
+        subscription = undefined;
+      });
     },
   });
 };

packages/rsc-runtime/tests/notices-resource-updated.test.ts

@@ -155,12 +155,96 @@ describe('notice inbox resources/updated signaller', () => {
     await driver.close();
   });
 
+  it('honours an unsubscribe that overlaps an observation awaiting the store', async () => {
+    const { driver, ledger } = await openLedger();
+    await publish(ledger, { sessionId: 's1' });
+    // A slow ledger: the first read parks until the test releases it, standing
+    // in for a contended durable store.
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    let parkNextRead = false;
+    const slowLedger: AgentNoticeLedger = Object.freeze({
+      ...ledger,
+      read: async () => {
+        if (parkNextRead) {
+          parkNextRead = false;
+          await gate;
+        }
+        return ledger.read();
+      },
+    });
+    const signaller = signallerOver(slowLedger);
+    const { send, sends } = sender();
+    await signaller.subscribe(principal('s1'));
+
+    const events: string[] = [];
+    parkNextRead = true;
+    const observing = signaller.observe(async () => {
+      events.push('send');
+      await send();
+    });
+    // `resources/unsubscribe` arrives while the observation is parked on the read.
+    const unsubscribing = signaller.unsubscribe().then(() => {
+      events.push('unsubscribed');
+    });
+    release!();
+    await expect(observing).resolves.toEqual({ kind: 'idle', reason: 'no-subscription', revision: undefined });
+    await unsubscribing;
+    expect(events).toEqual(['unsubscribed']);
+    expect(sends).toEqual([]);
+    expect(signaller.subscribed).toBe(false);
+    // Nothing was claimed on the client's behalf: the durable budget is intact,
+    // so a later subscriber still receives the signal.
+    expect((await ledger.read()).notices[0]?.availability).toBeUndefined();
+    await signaller.subscribe(principal('s1'));
+    await expect(signaller.observe(send)).resolves.toMatchObject({ kind: 'signalled' });
+    expect(sends).toHaveLength(1);
+    await driver.close();
+  });
+
+  it('acknowledges unsubscribe only after an in-flight send has settled', async () => {
+    const { driver, ledger } = await openLedger();
+    await publish(ledger, { sessionId: 's1' });
+    const signaller = signallerOver(ledger);
+    await signaller.subscribe(principal('s1'));
+    let releaseSend: (() => void) | undefined;
+    const sendGate = new Promise<void>((resolve) => {
+      releaseSend = resolve;
+    });
+    const events: string[] = [];
+    const observing = signaller.observe(async () => {
+      await sendGate;
+      events.push('send');
+    });
+    // Let the observation pass its eligibility read and claim before the
+    // unsubscribe arrives mid-send.
+    await new Promise<void>((resolve) => {
+      setTimeout(resolve, 0);
+    });
+    const unsubscribing = signaller.unsubscribe().then(() => {
+      events.push('unsubscribed');
+    });
+    releaseSend!();
+    await expect(observing).resolves.toMatchObject({ kind: 'signalled' });
+    await unsubscribing;
+    // The wire write that was already committed completes first; the client is
+    // told it is unsubscribed only afterwards, never the other way round.
+    expect(events).toEqual(['send', 'unsubscribed']);
+    await expect(signaller.observe(async () => {
+      events.push('late');
+    })).resolves.toMatchObject({ kind: 'idle', reason: 'no-subscription' });
+    expect(events).toEqual(['send', 'unsubscribed']);
+    await driver.close();
+  });
+
   it('stops signalling once unsubscribed and resets tracking on re-subscribe', async () => {
     const { driver, ledger } = await openLedger();
     const signaller = signallerOver(ledger);
     const { send, sends } = sender();
     await signaller.subscribe(principal('s1'));
-    signaller.unsubscribe();
+    await signaller.unsubscribe();
     expect(signaller.subscribed).toBe(false);
 
     await publish(ledger, { retryBudget: 2, sessionId: 's1' });