fix(relay): end idle Live Activities after the display window

infra/relay/src/agentActivity/AgentActivityPublisher.test.ts

@@ -58,6 +58,7 @@ function makeLiveActivities(
   return {
     register: () => Effect.void,
     listTargets: () => Effect.succeed([]),
+    listIdleArmedTargets: () => Effect.succeed([]),
     markDelivery: () => Effect.void,
     markStartQueued: () => Effect.void,
     clearStartQueued: () => Effect.void,
@@ -268,6 +269,81 @@ describe("AgentActivityPublisher", () => {
     });
   });
 
+  it.effect("ends armed cards left showing finished work once the display window passes", () => {
+    const idleTarget: LiveActivities.TargetRow = {
+      ...target("device-1"),
+      activity_push_token: "activity-token",
+      remote_started_at: "1970-01-01T00:00:01.000Z",
+      last_live_activity_delivery_at: "1970-01-01T00:30:00.000Z",
+    };
+    const listedBefore: Array<string> = [];
+    const sent: Array<Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]> =
+      [];
+    const nowMs = 60 * 60 * 1_000;
+    let activeStates: ReadonlyArray<RelayAgentActivityState> = [state];
+
+    const endIdle = Effect.gen(function* () {
+      const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
+      yield* publisher.endIdleLiveActivities({ nowMs });
+    }).pipe(
+      Effect.provide(
+        publisherLayer.pipe(
+          Layer.provide(
+            Layer.mergeAll(
+              Layer.succeed(
+                AgentActivityRows.AgentActivityRows,
+                makeAgentActivityRows({ listForUser: () => Effect.sync(() => activeStates) }),
+              ),
+              Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()),
+              Layer.succeed(
+                LiveActivities.LiveActivities,
+                makeLiveActivities({
+                  listIdleArmedTargets: (input) =>
+                    Effect.sync(() => {
+                      listedBefore.push(input.deliveredBefore);
+                      return [{ user_id: "dev:julius", device_id: "device-1" }];
+                    }),
+                  listTargets: () => Effect.succeed([idleTarget, target("device-2")]),
+                }),
+              ),
+              Layer.succeed(
+                ApnsDeliveries.ApnsDeliveries,
+                makeApnsDeliveries({
+                  sendForTarget: (input) =>
+                    Effect.sync(() => {
+                      sent.push(input);
+                      return null;
+                    }),
+                }),
+              ),
+            ),
+          ),
+        ),
+      ),
+    );
+
+    return Effect.gen(function* () {
+      // Work that started between the scan and the replay is left to its own
+      // publish: a silent repaint here would make that publish look unchanged
+      // and swallow its alert.
+      yield* endIdle;
+      expect(listedBefore).toEqual(["1970-01-01T00:45:00.000Z"]);
+      expect(sent).toEqual([]);
+
+      // The finished rows have aged out (or been pruned), so the aggregate is
+      // empty and the delivery layer ends the card. The replay is silent so
+      // ending cannot buzz the phone.
+      activeStates = [];
+      yield* endIdle;
+      expect(sent).toHaveLength(1);
+      expect(sent[0]).toMatchObject({
+        target: { device_id: "device-1" },
+        aggregate: null,
+        replay: true,
+      });
+    });
+  });
+
   it.effect("publishes listed targets through the APNs delivery service", () => {
     const firstTarget = target("device-1");
     const secondTarget = target("device-2");

infra/relay/src/agentActivity/AgentActivityPublisher.ts

@@ -1,4 +1,7 @@
-import { makeAggregateState } from "./agentActivityAggregate.ts";
+import {
+  makeAggregateState,
+  TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS,
+} from "./agentActivityAggregate.ts";
 export {
   makeAggregateState,
   TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS,
@@ -29,6 +32,7 @@ export type AgentActivityPublishError =
   | AgentActivityRows.AgentActivityRowListPersistenceError
   | EnvironmentLinks.EnvironmentLinkUserListPersistenceError
   | LiveActivities.LiveActivityTargetListPersistenceError
+  | LiveActivities.LiveActivityIdleTargetListPersistenceError
   | ApnsDeliveries.ApnsDeliveryError;
 
 export class AgentActivityPublisher extends Context.Service<
@@ -44,6 +48,9 @@ export class AgentActivityPublisher extends Context.Service<
       readonly userId: string;
       readonly deviceId: string;
     }) => Effect.Effect<RelayDeliveryResult | null, AgentActivityPublishError>;
+    readonly endIdleLiveActivities: (input: {
+      readonly nowMs: number;
+    }) => Effect.Effect<void, AgentActivityPublishError>;
   }
 >()("t3code-relay/agentActivity/AgentActivityPublisher") {}
 
@@ -108,6 +115,48 @@ export const make = Effect.gen(function* () {
     return deliveriesByTarget.flat();
   });
 
+  // Silently re-delivers the current aggregate to one device: repaints drifted
+  // content, or ends the card when nothing is left to show. With `endOnly`
+  // the content is never repainted: a silent repaint would make the real
+  // publish for that same state look unchanged and swallow its alert.
+  const replayForTarget = Effect.fnUntraced(function* (input: {
+    readonly userId: string;
+    readonly deviceId: string;
+    readonly endOnly?: boolean;
+  }) {
+    const { activeStates, targets } = yield* Effect.all(
+      {
+        activeStates: rows.listForUser({ userId: input.userId }),
+        targets: liveActivities.listTargets({ userId: input.userId }),
+      },
+      { concurrency: 2 },
+    );
+    const target = targets.find((row) => row.device_id === input.deviceId) ?? null;
+    if (target === null) {
+      return null;
+    }
+    if (target.platform === "android") {
+      return input.endOnly
+        ? null
+        : yield* fcmDeliveries.enqueue({ target, state: null, replay: true });
+    }
+    const now = yield* DateTime.now;
+    const aggregate = makeAggregateState({
+      activeStates,
+      terminalState: null,
+      nowMs: now.epochMilliseconds,
+    });
+    if (input.endOnly && aggregate !== null) {
+      return null;
+    }
+    return yield* apnsDeliveries.sendForTarget({
+      target,
+      aggregate,
+      nowMs: now.epochMilliseconds,
+      replay: true,
+    });
+  });
+
   return AgentActivityPublisher.of({
     replayForLiveActivityRegistration: Effect.fn(
       "relay.agent_activity_publisher.replay_for_live_activity_registration",
@@ -116,33 +165,39 @@ export const make = Effect.gen(function* () {
         "relay.mobile.device_id": input.deviceId,
         "relay.operation": "replayForLiveActivityRegistration",
       });
-      const { activeStates, targets } = yield* Effect.all(
-        {
-          activeStates: rows.listForUser({ userId: input.userId }),
-          targets: liveActivities.listTargets({ userId: input.userId }),
-        },
-        { concurrency: 2 },
-      );
-      const target = targets.find((row) => row.device_id === input.deviceId) ?? null;
-      if (target === null) {
-        return null;
-      }
-      if (target.platform === "android") {
-        return yield* fcmDeliveries.enqueue({ target, state: null, replay: true });
-      }
-      const now = yield* DateTime.now;
-      const aggregate = makeAggregateState({
-        activeStates,
-        terminalState: null,
-        nowMs: now.epochMilliseconds,
-      });
-      return yield* apnsDeliveries.sendForTarget({
-        target,
-        aggregate,
-        nowMs: now.epochMilliseconds,
-        replay: true,
-      });
+      return yield* replayForTarget({ userId: input.userId, deviceId: input.deviceId });
     }),
+    // Deliveries only happen when an environment publishes or the app
+    // re-registers, so a card left showing finished work would otherwise keep
+    // its Done rows past the display window until one of those happens.
+    // Replaying the aggregate once the window has passed ends it.
+    endIdleLiveActivities: Effect.fn("relay.agent_activity_publisher.end_idle_live_activities")(
+      function* (input) {
+        const deliveredBefore = DateTime.formatIso(
+          DateTime.makeUnsafe(input.nowMs - TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS),
+        );
+        const targets = yield* liveActivities.listIdleArmedTargets({ deliveredBefore });
+        yield* Effect.annotateCurrentSpan({ "relay.live_activities.idle_count": targets.length });
+        yield* Effect.forEach(
+          targets,
+          (target) =>
+            replayForTarget({
+              userId: target.user_id,
+              deviceId: target.device_id,
+              endOnly: true,
+            }).pipe(
+              Effect.tapError((error) =>
+                Effect.logWarning("idle live activity replay failed", {
+                  deviceId: target.device_id,
+                  errorTag: error._tag,
+                }),
+              ),
+              Effect.ignore,
+            ),
+          { concurrency: 4, discard: true },
+        );
+      },
+    ),
     publish: Effect.fn("relay.agent_activity_publisher.publish")(function* (input) {
       yield* Effect.annotateCurrentSpan({
         "relay.environment_id": input.environmentId,

infra/relay/src/agentActivity/ApnsDeliveries.test.ts

@@ -230,6 +230,7 @@ function makeLayer(input: {
         Layer.succeed(LiveActivities.LiveActivities, {
           register: () => Effect.void,
           listTargets: () => Effect.succeed(input.currentTargets ?? [target]),
+          listIdleArmedTargets: () => Effect.succeed([]),
           markStartQueued: (queued) =>
             Effect.sync(() => {
               input.queuedStarts?.push(queued);
@@ -916,6 +917,51 @@ describe("ApnsDeliveries", () => {
     }).pipe(Effect.provide(makeLayer({ attempts })));
   });
 
+  it.effect("skips a queued contentless end when the user has live work again", () => {
+    const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
+    const requests: Array<HttpClientRequest.HttpClientRequest> = [];
+    const payload = makeApnsDeliveryJobPayload({
+      kind: "live_activity_end",
+      userId: target.user_id,
+      deviceId: target.device_id,
+      token: "activity-token",
+      aggregate: null,
+      createdAt: "1970-01-01T00:00:00.000Z",
+      expiresAt: "1970-01-01T00:10:00.000Z",
+      jobId: "job-end-1",
+    });
+    const signed = signApnsDeliveryJob({
+      secret: config.apnsDeliveryJobSigningSecret,
+      payload,
+    });
+    const execute = (request: HttpClientRequest.HttpClientRequest) =>
+      Effect.sync(() => {
+        requests.push(request);
+        return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
+      });
+
+    return Effect.gen(function* () {
+      const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
+      const result = yield* deliveries.processSignedJob(signed);
+
+      // The end was decided while nothing was running; work that started
+      // since owns the card now, and its own update has repainted it.
+      expect(result).toMatchObject({
+        kind: "live_activity_end",
+        ok: true,
+        apnsStatus: null,
+      });
+      expect(requests).toEqual([]);
+      expect(attempts).toMatchObject([
+        {
+          kind: "live_activity_end",
+          sourceJobId: "job-end-1",
+          apnsReason: "Stale APNs end job skipped.",
+        },
+      ]);
+    }).pipe(Effect.provide(makeLayer({ attempts, activityStates: [state], execute })));
+  });
+
   it.effect("skips a queued start when the user no longer has live work", () => {
     const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
     const clearedStarts: Array<

infra/relay/src/agentActivity/ApnsDeliveries.ts

@@ -768,6 +768,22 @@ export const make = Effect.gen(function* () {
         return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
       }
     }
+    // A contentless end retires the token. Work that started after the end
+    // was queued may already have painted the card through its own update;
+    // ending now would drop that card and strand the new work.
+    if (
+      input.kind === "live_activity_end" &&
+      aggregate === null &&
+      (yield* userStillHasLiveWork(input.target.user_id))
+    ) {
+      if (input.sourceJobId) {
+        yield* attempts.completeSourceJob({
+          sourceJobId: input.sourceJobId,
+          apnsReason: "Stale APNs end job skipped.",
+        });
+      }
+      return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
+    }
     if (
       input.kind === "live_activity_start" &&
       !(yield* userStillHasLiveWork(input.target.user_id))

infra/relay/src/agentActivity/FcmDeliveries.test.ts

@@ -127,6 +127,7 @@ function harness() {
     Layer.succeed(LiveActivities, {
       register: () => Effect.void,
       listTargets: () => Effect.sync(() => [current.target]),
+      listIdleArmedTargets: () => Effect.succeed([]),
       markDelivery: (input) =>
         Effect.sync(() => {
           marked.push(input);

infra/relay/src/agentActivity/LiveActivities.test.ts

@@ -249,6 +249,46 @@ describe("LiveActivities", () => {
     );
   });
 
+  it.effect("lists armed cards that showed no live work for the display window", () => {
+    const conditions: Array<SQL> = [];
+    const fakeDb = {
+      select: () => ({
+        from: () => ({
+          where: (condition: SQL) => {
+            conditions.push(condition);
+            return Effect.succeed([{ user_id: "user-2", device_id: "device-1" }]);
+          },
+        }),
+      }),
+    } as unknown as RelayDb.RelayDb["Service"];
+
+    return Effect.gen(function* () {
+      const liveActivities = yield* LiveActivities.LiveActivities;
+      const targets = yield* liveActivities.listIdleArmedTargets({
+        deliveredBefore: "2026-05-25T00:15:00.000Z",
+      });
+
+      expect(targets).toEqual([{ user_id: "user-2", device_id: "device-1" }]);
+      // Only cards with a live token count, and only when the last delivered
+      // content had no active agents (or never arrived) and nothing has been
+      // delivered since the cutoff.
+      expect(conditions.map((condition) => new PgDialect().sqlToQuery(condition))).toEqual([
+        {
+          sql:
+            '((("relay_live_activities"."activity_push_token" is not null)) and ' +
+            "(coalesce(\"relay_live_activities\".\"last_aggregate_json\" ->> 'activeCount', '0') = '0') and " +
+            '(coalesce("relay_live_activities"."last_live_activity_delivery_at", ' +
+            '"relay_live_activities"."remote_started_at", "relay_live_activities"."updated_at") < $1))',
+          params: ["2026-05-25T00:15:00.000Z"],
+        },
+      ]);
+    }).pipe(
+      Effect.provide(
+        LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
+      ),
+    );
+  });
+
   it.effect("preserves correlation context and causes for persistence failures", () => {
     const cause = new Error("database unavailable");
     const registration: RelayLiveActivityRegistrationRequest = {