diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index ae3d4e4ad24d..7e5ab8035e51 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/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 = []; + const sent: Array[0]> = + []; + const nowMs = 60 * 60 * 1_000; + let activeStates: ReadonlyArray = [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"); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 61a420dd7858..b6a793449bf8 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/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; + readonly endIdleLiveActivities: (input: { + readonly nowMs: number; + }) => Effect.Effect; } >()("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, diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index e98c2b639598..fe938fe9fac4 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/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 = []; + const requests: Array = []; + 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 = []; const clearedStarts: Array< diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 651f031f0efa..625d5b24937f 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/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)) diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index 73b90f99b0a1..084d057ff043 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/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); diff --git a/infra/relay/src/agentActivity/LiveActivities.test.ts b/infra/relay/src/agentActivity/LiveActivities.test.ts index 7f2bce87431e..d81ab0984c38 100644 --- a/infra/relay/src/agentActivity/LiveActivities.test.ts +++ b/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 = []; + 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 = { diff --git a/infra/relay/src/agentActivity/LiveActivities.ts b/infra/relay/src/agentActivity/LiveActivities.ts index 0c88fbb73c39..a9c74a1d5a6c 100644 --- a/infra/relay/src/agentActivity/LiveActivities.ts +++ b/infra/relay/src/agentActivity/LiveActivities.ts @@ -13,7 +13,7 @@ import * as Effect from "effect/Effect"; import * as Function from "effect/Function"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, isNotNull, sql } from "drizzle-orm"; import * as RelayDb from "../db.ts"; import { relayLiveActivities, relayMobileDevices } from "../persistence/schema.ts"; @@ -43,6 +43,18 @@ export class LiveActivityTargetListPersistenceError extends Schema.TaggedError()( + "LiveActivityIdleTargetListPersistenceError", + { + deliveredBefore: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to list idle armed Live Activities delivered before ${this.deliveredBefore}.`; + } +} + export class LiveActivityDeliveryMarkPersistenceError extends Schema.TaggedError()( "LiveActivityDeliveryMarkPersistenceError", { @@ -97,6 +109,12 @@ export class LiveActivities extends Context.Service< readonly listTargets: (input: { readonly userId: string; }) => Effect.Effect, LiveActivityTargetListPersistenceError>; + readonly listIdleArmedTargets: (input: { + readonly deliveredBefore: string; + }) => Effect.Effect< + ReadonlyArray<{ readonly user_id: string; readonly device_id: string }>, + LiveActivityIdleTargetListPersistenceError + >; readonly markDelivery: (input: { readonly userId: string; readonly deviceId: string; @@ -250,6 +268,38 @@ export const make = Effect.gen(function* () { ); }), + // Armed cards whose last delivered content had no live work (or no + // content yet) and that have heard nothing since the finished-row display + // window. Nothing else ends such a card: deliveries only run when an + // environment publishes or the app re-registers, so a Done card with no + // follow-up activity would otherwise sit on the lock screen for hours. + listIdleArmedTargets: Effect.fn("relay.live_activities.list_idle_armed_targets")( + function* (input) { + return yield* db + .select({ + user_id: relayLiveActivities.userId, + device_id: relayLiveActivities.deviceId, + }) + .from(relayLiveActivities) + .where( + and( + isNotNull(relayLiveActivities.activityPushToken), + sql`coalesce(${relayLiveActivities.lastAggregateJson} ->> 'activeCount', '0') = '0'`, + sql`coalesce(${relayLiveActivities.lastLiveActivityDeliveryAt}, ${relayLiveActivities.remoteStartedAt}, ${relayLiveActivities.updatedAt}) < ${input.deliveredBefore}`, + ), + ) + .pipe( + Effect.mapError( + (cause) => + new LiveActivityIdleTargetListPersistenceError({ + deliveredBefore: input.deliveredBefore, + cause, + }), + ), + ); + }, + ), + markDelivery: Effect.fn("relay.live_activities.mark_delivery")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.mobile.device_id": input.deviceId, diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 74d5c3740a51..0c5a9b1d71d3 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -67,6 +67,7 @@ function makeLiveActivities( return { register: () => Effect.void, listTargets: () => Effect.succeed([]), + listIdleArmedTargets: () => Effect.succeed([]), markDelivery: () => Effect.void, markStartQueued: () => Effect.void, clearStartQueued: () => Effect.void, @@ -192,6 +193,7 @@ function makeAgentActivityPublisher( return { publish: () => Effect.succeed({ ok: true, deliveries: [] }), replayForLiveActivityRegistration: () => Effect.succeed(null), + endIdleLiveActivities: () => Effect.void, ...overrides, }; } diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index fcaeca640420..25fde57f72be 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1044,6 +1044,7 @@ const RelayCommonPersistenceError = Schema.Union([ EnvironmentCredentials.EnvironmentCredentialRevokePersistenceError, DpopProofs.DpopProofReplayPersistenceError, LiveActivities.LiveActivityTargetListPersistenceError, + LiveActivities.LiveActivityIdleTargetListPersistenceError, AgentActivityRows.AgentActivityRowUpsertPersistenceError, AgentActivityRows.AgentActivityRowDeletePersistenceError, AgentActivityRows.AgentActivityRowListPersistenceError, diff --git a/infra/relay/src/worker.ts b/infra/relay/src/worker.ts index caa5425b083f..2a33ff38c2e6 100644 --- a/infra/relay/src/worker.ts +++ b/infra/relay/src/worker.ts @@ -329,6 +329,15 @@ export const ApiLive = Api.make( ), ), ), + // Armed cards showing only finished work end once the display window + // passes, even when no environment publishes again. + Effect.andThen( + Effect.all([AgentActivityPublisher.AgentActivityPublisher, DateTime.now]).pipe( + Effect.flatMap(([publisher, now]) => + publisher.endIdleLiveActivities({ nowMs: now.epochMilliseconds }), + ), + ), + ), Effect.withSpan("relay.cron.prune_expired_state"), Effect.provide(runtimeLayer), ),