fix(notifications): resolve agent-push dispatch review findings from #4627

services/notifications/src/__tests__/dispatch-push.test.ts

@@ -1039,27 +1039,34 @@ describe('NotificationChannelDO.dispatchPush — local push sink', () => {
     const payload = sinkCall?.[1] as {
       idempotencyKey: string;
       payload: {
-        title: string;
-        body: string;
-        data: unknown;
+        titleLength: number;
+        bodyLength: number;
+        dataKeys: string[];
+        dataType: unknown;
         sound: string | null;
         priority: string;
       };
       to: string;
     };
     expect(payload.idempotencyKey).toBe(idemKey);
+    // Content-free: sizes + key names + discriminator, never the values.
     expect(payload.payload).toEqual({
-      title: 'My title',
-      body: 'My body',
-      data: { type: 'cloud_agent_session', cliSessionId: 'ses_sink' },
+      titleLength: 'My title'.length,
+      bodyLength: 'My body'.length,
+      dataKeys: ['cliSessionId', 'type'],
+      dataType: 'cloud_agent_session',
       sound: 'default',
       priority: 'high',
     });
     expect(payload.to).toBe('<redacted>');
-    // Token must never appear in the sink payload.
+    // Neither the device token nor the user/agent-controlled content may appear.
     const serialized = JSON.stringify(payload);
     expect(serialized).not.toContain('tok-real');
     expect(serialized).not.toContain('ExponentPushToken');
+    // User/agent-controlled content values must never appear (identifiers such
+    // as the idempotency key are fine and intentionally retained for tracing).
+    expect(serialized).not.toContain('My title');
+    expect(serialized).not.toContain('My body');
 
     // No receipt work created.
     expect(env.RECEIPTS_QUEUE.send).not.toHaveBeenCalled();
@@ -1159,4 +1166,55 @@ describe('NotificationChannelDO.dispatchPush — local push sink', () => {
     expect(logSpy.mock.calls.some(call => call[0] === 'agent_push_sink_payload')).toBe(false);
     logSpy.mockRestore();
   });
+
+  it('does not double-send when the same key is dispatched concurrently (in-flight claim)', async () => {
+    installDbMock({ tokens: [{ user_id: 'user-concurrent', token: 'tok1' }] });
+    vi.spyOn(env.EVENT_SERVICE, 'isUserInContext').mockResolvedValue(false);
+
+    // Hold the first Expo send open so a concurrent replay of the same key
+    // interleaves while the first attempt is mid-flight — i.e. after it wrote
+    // its `pending` marker but before it recorded a terminal state. Without the
+    // in-flight claim the replay would read `pending`, treat it as a reusable
+    // retry slot, and submit a second OS push.
+    let releaseExpo: () => void = () => undefined;
+    const expoGate = new Promise<void>(resolve => {
+      releaseExpo = resolve;
+    });
+    vi.mocked(sendPushNotifications).mockImplementationOnce(async () => {
+      await expoGate;
+      return {
+        ticketTokenPairs: [{ ticketId: 't1', token: 'tok1' }],
+        staleTokens: [],
+        ticketErrors: [],
+      };
+    });
+
+    const stub = getDO('user-concurrent');
+    const input: DispatchPushInput = {
+      userId: 'user-concurrent',
+      presenceContext: null,
+      idempotencyKey: 'agent-notification:ses_cc:n-cc',
+      badge: null,
+      push: {
+        title: 'T',
+        body: 'B',
+        data: { type: 'cloud_agent_session', cliSessionId: 'ses_cc' },
+        sound: 'default',
+        priority: 'high',
+      },
+    };
+
+    const [first, second] = await runInDurableObject(stub, async instance => {
+      const p1 = instance.dispatchPush(input);
+      // Let p1 progress to the awaiting-Expo point before the replay enters.
+      await new Promise(resolve => setTimeout(resolve, 0));
+      const p2 = instance.dispatchPush(input);
+      releaseExpo();
+      return Promise.all([p1, p2]);
+    });
+
+    expect([first.kind, second.kind].sort()).toEqual(['delivered', 'duplicate']);
+    // The core guarantee: exactly one OS push submission for the key.
+    expect(sendPushNotifications).toHaveBeenCalledTimes(1);
+  });
 });

services/notifications/src/dos/NotificationChannelDO.ts

@@ -43,7 +43,29 @@ const IDEM_TTL_MS = 60 * 60 * 1000; // 1 hour
 const ACCEPTED_BOOKKEEPING_RETRY_DELAY_MS = 30_000;
 
 export class NotificationChannelDO extends DurableObject<Env> {
+  // In-flight claims by idempotency key. The DO serializes turns, but a turn
+  // yields the isolate on every external `await` (presence RPC, Expo send), so a
+  // concurrent replay of the same key can interleave: it would read the storage
+  // `pending` marker the first attempt left before its send, treat it as a
+  // reusable retry slot, and send a second time. This in-memory claim closes
+  // that window — a replay seeing a live claim returns `duplicate`, while a
+  // sequential post-failure retry (claim already released) still reuses the
+  // `pending` slot as before.
+  private readonly inFlight = new Set<string>();
+
   async dispatchPush(input: DispatchPushInput): Promise<DispatchPushOutcome> {
+    if (this.inFlight.has(input.idempotencyKey)) {
+      return { kind: 'duplicate' };
+    }
+    this.inFlight.add(input.idempotencyKey);
+    try {
+      return await this.dispatchPushInner(input);
+    } finally {
+      this.inFlight.delete(input.idempotencyKey);
+    }
+  }
+
+  private async dispatchPushInner(input: DispatchPushInput): Promise<DispatchPushOutcome> {
     // 1. Idempotency. DO is single-threaded — requests for a given
     //    user serialize on this instance. Retryable send failures leave the
     //    record at `pending` so upstream can retry the send without
@@ -160,12 +182,22 @@ export class NotificationChannelDO extends DurableObject<Env> {
     //    means the sink captured the payload, never device delivery.
     if (isPushSinkEnabled(this.env)) {
       const ts = Date.now();
+      // Content-free observability only. Title, body, and the `data` object are
+      // agent/user-controlled and can carry credentials or private customer
+      // content; logging their values would retain that in local, CI, and
+      // aggregated Worker logs. Record sizes, `data` key names, and the fixed
+      // `type` discriminator instead — enough to trace the send without content.
+      const data = input.push.data;
       console.log('agent_push_sink_payload', {
         idempotencyKey: input.idempotencyKey,
         payload: {
-          title: input.push.title,
-          body: input.push.body,
-          data: input.push.data,
+          titleLength: input.push.title.length,
+          bodyLength: input.push.body.length,
+          dataKeys: data && typeof data === 'object' ? Object.keys(data).sort() : [],
+          dataType:
+            data && typeof data === 'object' && 'type' in data
+              ? (data as { type?: unknown }).type
+              : undefined,
           sound: input.push.sound ?? null,
           priority: input.push.priority ?? 'default',
         },