fix(mobile): Live Activities keep updating after iOS rotates their push token

apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts

@@ -8,6 +8,7 @@ import Constants from "expo-constants";
 import * as Cause from "effect/Cause";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import { Cookies, FetchHttpClient, HttpClientRequest, HttpClientResponse } from "effect/http";
 import { ManagedRelay } from "@t3tools/client-runtime/relay";
@@ -391,26 +392,144 @@ describe("makeRelayDeviceRegistrationRequest", () => {
     ).toEqual({ liveActivitiesEnabled: true, baseFontSize: 18 });
   });
 
-  it.effect("registers at most one listener while a Live Activity push token is pending", () => {
+  it.effect("registers one push-token listener per Live Activity across wrappers", () => {
     registerAgentAwarenessConnection(savedConnection());
     const addPushTokenListener = vi.fn();
-    const activity = {
+    // getInstances() returns a new wrapper for the same native activity each call.
+    const wrapper = () => ({
+      getId: () => "activity-1",
       getPushToken: vi.fn(() => Promise.resolve(null)),
       addPushTokenListener,
-    };
+    });
 
     return Effect.gen(function* () {
-      expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(false);
-      expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(false);
+      expect(yield* registerLiveActivityPushToken({ activity: wrapper() as never })).toBe(false);
+      expect(yield* registerLiveActivityPushToken({ activity: wrapper() as never })).toBe(false);
 
-      expect(activity.getPushToken).toHaveBeenCalledTimes(2);
       expect(addPushTokenListener).toHaveBeenCalledTimes(1);
     }).pipe(Effect.provide(layerRelayTest));
   });
 
+  it.effect("registers a Live Activity push token that rotates after registration", () => {
+    const registeredTokens: Array<string> = [];
+    vi.stubGlobal("fetch", async (request: RequestInfo | URL, init?: RequestInit) => {
+      const url = request instanceof Request ? request.url : String(request);
+      if (url.endsWith("/v1/client/dpop-token")) {
+        return Response.json({
+          access_token: "relay-dpop-token",
+          issued_token_type: "urn:ietf:params:oauth:token-type:access_token",
+          token_type: "DPoP",
+          expires_in: 300,
+          scope: "mobile:registration",
+        });
+      }
+      if (url.endsWith("/v1/mobile/live-activities")) {
+        const body = request instanceof Request ? await request.text() : String(init?.body);
+        registeredTokens.push(
+          (JSON.parse(body) as { readonly activityPushToken: string }).activityPushToken,
+        );
+      }
+      return Response.json({ ok: true });
+    });
+    Constants.expoConfig!.extra = { relay: { url: "https://relay.example.test/" } };
+    setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk-token-user-a"));
+    let tokenListener: ((event: { readonly pushToken: string }) => void) | undefined;
+    const activity = {
+      getId: () => "activity-1",
+      getPushToken: vi.fn(() => Promise.resolve("activity-token")),
+      addPushTokenListener: vi.fn((listener: typeof tokenListener) => {
+        tokenListener = listener;
+        return { remove: vi.fn() };
+      }),
+    };
+
+    return Effect.gen(function* () {
+      yield* runBackgroundOperations();
+      expect(yield* registerLiveActivityPushToken({ activity: activity as never })).toBe(true);
+
+      tokenListener?.({ pushToken: "rotated-activity-token" });
+      yield* runBackgroundOperations();
+
+      expect(registeredTokens).toEqual(["activity-token", "rotated-activity-token"]);
+    }).pipe(
+      Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch),
+      Effect.provide(layerRelayTest),
+    );
+  });
+
+  it.effect("registers a rotated Live Activity token only after an older registration", () => {
+    const arrivals: Array<string> = [];
+    let openOldRequest: (() => void) | undefined;
+    const oldRequestGate = new Promise<void>((resolve) => {
+      openOldRequest = resolve;
+    });
+    let markOldRequestArrived: (() => void) | undefined;
+    const oldRequestArrived = new Promise<void>((resolve) => {
+      markOldRequestArrived = resolve;
+    });
+    vi.stubGlobal("fetch", async (request: RequestInfo | URL, init?: RequestInit) => {
+      const url = request instanceof Request ? request.url : String(request);
+      if (url.endsWith("/v1/client/dpop-token")) {
+        return Response.json({
+          access_token: "relay-dpop-token",
+          issued_token_type: "urn:ietf:params:oauth:token-type:access_token",
+          token_type: "DPoP",
+          expires_in: 300,
+          scope: "mobile:registration",
+        });
+      }
+      if (url.endsWith("/v1/mobile/live-activities")) {
+        const body = request instanceof Request ? await request.text() : String(init?.body);
+        const token = (JSON.parse(body) as { readonly activityPushToken: string })
+          .activityPushToken;
+        arrivals.push(token);
+        if (token === "activity-token") {
+          markOldRequestArrived?.();
+          await oldRequestGate;
+        }
+      }
+      return Response.json({ ok: true });
+    });
+    Constants.expoConfig!.extra = { relay: { url: "https://relay.example.test/" } };
+    setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk-token-user-a"));
+    let tokenListener: ((event: { readonly pushToken: string }) => void) | undefined;
+    const activity = {
+      getId: () => "activity-1",
+      getPushToken: vi.fn(() => Promise.resolve("activity-token")),
+      addPushTokenListener: vi.fn((listener: typeof tokenListener) => {
+        tokenListener = listener;
+        return { remove: vi.fn() };
+      }),
+    };
+
+    return Effect.gen(function* () {
+      yield* runBackgroundOperations();
+      const lookup = yield* Effect.forkChild(
+        registerLiveActivityPushToken({ activity: activity as never }),
+      );
+      yield* Effect.promise(() => oldRequestArrived);
+
+      tokenListener?.({ pushToken: "rotated-activity-token" });
+      const listenerRegistration = yield* Effect.forkChild(runBackgroundOperations());
+      for (let pass = 0; pass < 5; pass++) {
+        yield* Effect.promise(() => new Promise<void>((resolve) => setImmediate(resolve)));
+      }
+      expect(arrivals).toEqual(["activity-token"]);
+
+      openOldRequest?.();
+      yield* Fiber.join(lookup);
+      yield* Fiber.join(listenerRegistration);
+      expect(arrivals).toEqual(["activity-token", "rotated-activity-token"]);
+    }).pipe(
+      Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch),
+      Effect.provide(layerRelayTest),
+    );
+  });
+
   it.effect("preserves Live Activity push-token lookup failures", () => {
     const cause = new Error("native token lookup failed");
     const activity = {
+      getId: () => "activity-1",
       getPushToken: vi.fn(() => Promise.reject(cause)),
       addPushTokenListener: vi.fn(),
     };
@@ -435,6 +554,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
     () => {
       registerAgentAwarenessConnection(savedConnection());
       const activity = {
+        getId: () => "activity-1",
         getPushToken: vi.fn(() => Promise.resolve("activity-token")),
         addPushTokenListener: vi.fn(),
       };
@@ -449,6 +569,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
     "registers APNS-started Live Activities for relay updates without mutating them locally",
     () => {
       const activity = {
+        getId: () => "activity-1",
         getPushToken: vi.fn(() => Promise.resolve("activity-token")),
         addPushTokenListener: vi.fn(),
         start: vi.fn(),
@@ -473,6 +594,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
     "re-registers active Live Activity tokens when the app returns to the foreground",
     () => {
       const activity = {
+        getId: () => "activity-1",
         getPushToken: vi.fn(() => Promise.resolve("activity-token")),
         addPushTokenListener: vi.fn(),
       };
@@ -502,6 +624,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
   it("ends local Live Activities and stops foreground reconciliation on cloud sign-out", () => {
     const end = vi.fn(() => Promise.resolve());
     const activity = {
+      getId: () => "activity-1",
       getPushToken: vi.fn(() => Promise.resolve("activity-token")),
       addPushTokenListener: vi.fn(),
       end,
@@ -613,6 +736,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
   it("releases the provider without ending activities or clearing the registration", () => {
     const end = vi.fn(() => Promise.resolve());
     const activity = {
+      getId: () => "activity-1",
       getPushToken: vi.fn(() => Promise.resolve("activity-token")),
       addPushTokenListener: vi.fn(),
       end,
@@ -708,6 +832,7 @@ describe("makeRelayDeviceRegistrationRequest", () => {
       },
     };
     const activity = {
+      getId: () => "activity-1",
       getPushToken: vi.fn(() => Promise.resolve("activity-token")),
       addPushTokenListener: vi.fn(),
     };

apps/mobile/src/features/agent-awareness/remoteRegistration.ts

@@ -7,6 +7,7 @@ import Constants from "expo-constants";
 import * as Notifications from "expo-notifications";
 import * as Effect from "effect/Effect";
 import * as Schema from "effect/Schema";
+import * as Semaphore from "effect/Semaphore";
 import { AppState, Platform } from "react-native";
 import type { EnvironmentId } from "@t3tools/contracts";
 import {
@@ -72,7 +73,12 @@ export class AgentAwarenessOperationError extends Schema.TaggedError<AgentAwaren
 }
 
 const environmentConnections = new Map<EnvironmentId, SavedRemoteConnection>();
-const activityPushTokenListeners = new WeakSet<LiveActivity<AgentActivityProps>>();
+// Keyed by ActivityKit id: getAgentLiveActivities() wraps the same native
+// activity in a new object on every call. The relay keeps only the latest token
+// a device registers, so each activity's token reads and registrations run one
+// at a time; otherwise a lookup that read the old token could land after the
+// rotated one.
+const activityPushTokenLocks = new Map<string, Semaphore.Semaphore>();
 // But the registration is not a pure no-op: the relay replays the current
 // aggregate to this device on every accepted registration, and that replay is
 // the foreground reconciliation that repairs drifted or orphaned activities.
@@ -893,6 +899,7 @@ export function __resetAgentAwarenessRemoteRegistrationForTest(): void {
   androidDeviceReplayedAt = null;
   registrationStatusListeners.clear();
   registeredActivityPushTokens.clear();
+  activityPushTokenLocks.clear();
 }
 
 export function unregisterAgentAwarenessDeviceForCurrentUser(
@@ -928,52 +935,53 @@ export function registerLiveActivityPushToken(input: {
       return false;
     }
 
-    const activityPushToken = yield* Effect.tryPromise({
-      try: () => input.activity.getPushToken(),
-      catch: (cause) =>
-        new AgentAwarenessOperationError({
-          operation: "read-live-activity-push-token",
-          cause,
-        }),
-    });
-    if (!activityPushToken) {
-      if (activityPushTokenListeners.has(input.activity)) {
-        logRegistrationDebug(
-          "live activity push token not available yet; token listener already registered",
-          {
-            connectionCount: environmentConnections.size,
-          },
-        );
-        return false;
-      }
-
-      logRegistrationDebug(
-        "live activity push token not available yet; listening for token event",
-        {
-          connectionCount: environmentConnections.size,
-        },
-      );
-      activityPushTokenListeners.add(input.activity);
-      input.activity.addPushTokenListener((event) => {
-        if (event.pushToken) {
-          logRegistrationDebug("live activity push token event received", {
-            tokenSuffix: event.pushToken.slice(-8),
-          });
-          runRegistrationInBackground(
-            registerLiveActivityPushTokenValue({
-              activityPushToken: event.pushToken,
+    const lock = observeLiveActivityPushTokens(input.activity);
+    return yield* lock.withPermit(
+      Effect.gen(function* () {
+        const activityPushToken = yield* Effect.tryPromise({
+          try: () => input.activity.getPushToken(),
+          catch: (cause) =>
+            new AgentAwarenessOperationError({
+              operation: "read-live-activity-push-token",
+              cause,
             }),
-            "live activity token listener registration failed",
+        });
+        if (!activityPushToken) {
+          logRegistrationDebug(
+            "live activity push token not available yet; waiting for token event",
+            { connectionCount: environmentConnections.size },
           );
+          return false;
         }
-      });
-      return false;
-    }
 
-    return yield* registerLiveActivityPushTokenValue({
-      activityPushToken,
+        return yield* registerLiveActivityPushTokenValue({ activityPushToken });
+      }),
+    );
+  });
+}
+
+// iOS can replace an activity's push token while it runs, not only deliver a
+// late first one. The relay keeps pushing to the old token until APNs rejects
+// it, so every activity needs a listener that registers each new token.
+function observeLiveActivityPushTokens(
+  activity: LiveActivity<AgentActivityProps>,
+): Semaphore.Semaphore {
+  const activityId = activity.getId();
+  const existing = activityPushTokenLocks.get(activityId);
+  if (existing) return existing;
+  const lock = Semaphore.makeUnsafe(1);
+  activityPushTokenLocks.set(activityId, lock);
+  activity.addPushTokenListener((event) => {
+    if (!event.pushToken) return;
+    logRegistrationDebug("live activity push token event received", {
+      tokenSuffix: event.pushToken.slice(-8),
     });
+    runRegistrationInBackground(
+      lock.withPermit(registerLiveActivityPushTokenValue({ activityPushToken: event.pushToken })),
+      "live activity token listener registration failed",
+    );
   });
+  return lock;
 }
 
 function registerLiveActivityPushTokenValue(input: {