diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index e98c2b639598..9c277c170055 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -146,6 +146,12 @@ const target: LiveActivities.TargetRow = { last_live_activity_delivery_at: null, }; +/** + * Test layer: capture delivery attempts and queued APNs jobs for assertions. + * + * @param input - Sinks and optional overrides for rows, config, and HTTP + * @returns Layer providing ApnsDeliveries and its test doubles + */ function makeLayer(input: { readonly attempts: Array; readonly sourceJobClaims?: ReadonlyMap; @@ -256,6 +262,100 @@ function makeLayer(input: { ); } +/** + * Send starting then running four seconds later and assert both + * `live_activity_update` jobs, with phase running / Working on the second. + * + * @param input - Aggregates and the queued-job sink for assertions + */ +function* sendStartingThenRunningLiveActivityUpdates(input: { + readonly startingAggregate: RelayAgentActivityAggregateState; + readonly runningAggregate: RelayAgentActivityAggregateState; + readonly queuedJobs: Array; +}) { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const first = yield* deliveries.sendForTarget({ + target, + aggregate: input.startingAggregate, + nowMs: 0, + }); + expect(first?.kind).toBe("live_activity_update"); + + const second = yield* deliveries.sendForTarget({ + target: { + ...target, + last_aggregate_json: JSON.stringify(input.startingAggregate), + last_live_activity_delivery_at: "1970-01-01T00:00:00.000Z", + }, + aggregate: input.runningAggregate, + nowMs: 4_000, + }); + + expect(second?.kind).toBe("live_activity_update"); + expect(input.queuedJobs).toMatchObject([ + { + payload: { + kind: "live_activity_update", + target: { + token: "activity-token", + }, + }, + }, + { + payload: { + kind: "live_activity_update", + target: { + token: "activity-token", + }, + aggregate: { + activities: [{ phase: "running", status: "Working" }], + }, + }, + }, + ]); +} + +/** + * Regression for starting→running inside the 15s Live Activity throttle: + * both updates must queue, and the second payload is the running aggregate. + * + * @returns Effect that drives both deliveries and asserts the queued jobs + */ +function queuesLiveActivityUpdateOnStartingToRunningPhaseChange() { + const attempts: Array = []; + const queuedJobs: Array = []; + const startingAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "starting", + status: "Connecting", + }, + ], + }; + const runningAggregate: RelayAgentActivityAggregateState = { + ...startingAggregate, + updatedAt: "1970-01-01T00:00:04.000Z", + activities: [ + { + ...startingAggregate.activities[0]!, + phase: "running", + status: "Working", + updatedAt: "1970-01-01T00:00:04.000Z", + }, + ], + }; + + return Effect.gen( + sendStartingThenRunningLiveActivityUpdates.bind(undefined, { + startingAggregate, + runningAggregate, + queuedJobs, + }), + ).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); +} + describe("ApnsDeliveries", () => { it.effect("skips Apple delivery when an Android-only relay disables APNs", () => { const attempts: Array = []; @@ -621,6 +721,222 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("throttles timestamp-only changes while an activity already awaits input", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const waitingAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "waiting_for_input", + status: "Input", + }, + ], + }; + const laterWaitingAggregate: RelayAgentActivityAggregateState = { + ...waitingAggregate, + updatedAt: "1970-01-01T00:00:04.000Z", + activities: [ + { + ...waitingAggregate.activities[0]!, + updatedAt: "1970-01-01T00:00:04.000Z", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + last_aggregate_json: JSON.stringify(waitingAggregate), + last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z", + }, + aggregate: laterWaitingAggregate, + nowMs: 5_000, + }); + + expect(result).toBeNull(); + expect(queuedJobs).toEqual([]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); + }); + + it.effect("throttles ordering-only changes while an activity already awaits input", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const waitingRow = { + ...aggregate.activities[0]!, + phase: "waiting_for_input" as const, + status: "Input", + }; + const runningRow = { + ...aggregate.activities[0]!, + threadId: "thread-2" as RelayAgentActivityState["threadId"], + threadTitle: "Other thread", + phase: "running" as const, + status: "Working", + }; + const waitingAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activeCount: 2, + activities: [waitingRow, runningRow], + }; + const reorderedAggregate: RelayAgentActivityAggregateState = { + ...waitingAggregate, + activities: [runningRow, waitingRow], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + last_aggregate_json: JSON.stringify(waitingAggregate), + last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z", + }, + aggregate: reorderedAggregate, + nowMs: 5_000, + }); + + expect(result).toBeNull(); + expect(queuedJobs).toEqual([]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); + }); + + it.effect("queues an update when a new thread awaits input while another already does", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const waitingRow = { + ...aggregate.activities[0]!, + phase: "waiting_for_input" as const, + status: "Input", + }; + const runningRow = { + ...aggregate.activities[0]!, + threadId: "thread-2" as RelayAgentActivityState["threadId"], + threadTitle: "Other thread", + phase: "running" as const, + status: "Working", + }; + const newWaitingRow = { + ...aggregate.activities[0]!, + threadId: "thread-3" as RelayAgentActivityState["threadId"], + threadTitle: "New waiting thread", + phase: "waiting_for_input" as const, + status: "Input", + updatedAt: "1970-01-01T00:00:04.000Z", + }; + const previousAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activeCount: 2, + activities: [waitingRow, runningRow], + }; + const nextAggregate: RelayAgentActivityAggregateState = { + ...previousAggregate, + updatedAt: "1970-01-01T00:00:04.000Z", + activities: [waitingRow, newWaitingRow], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + last_aggregate_json: JSON.stringify(previousAggregate), + last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z", + }, + aggregate: nextAggregate, + nowMs: 5_000, + }); + + expect(result?.kind).toBe("live_activity_update"); + expect(queuedJobs).toMatchObject([ + { + payload: { + kind: "live_activity_update", + target: { + token: "activity-token", + }, + aggregate: { + activities: [ + { phase: "waiting_for_input", threadId: "thread" }, + { phase: "waiting_for_input", threadId: "thread-3" }, + ], + }, + }, + }, + ]); + }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); + }); + + it.effect( + "queues an update when a waiting thread leaves while another still awaits input", + () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const waitingRow = { + ...aggregate.activities[0]!, + phase: "waiting_for_input" as const, + status: "Input", + }; + const otherWaitingRow = { + ...aggregate.activities[0]!, + threadId: "thread-2" as RelayAgentActivityState["threadId"], + threadTitle: "Other thread", + phase: "waiting_for_input" as const, + status: "Input", + }; + const replacementRunningRow = { + ...aggregate.activities[0]!, + threadId: "thread-3" as RelayAgentActivityState["threadId"], + threadTitle: "Replacement thread", + phase: "running" as const, + status: "Working", + }; + const previousAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activeCount: 2, + activities: [waitingRow, otherWaitingRow], + }; + const nextAggregate: RelayAgentActivityAggregateState = { + ...previousAggregate, + updatedAt: "1970-01-01T00:00:04.000Z", + activities: [waitingRow, replacementRunningRow], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + last_aggregate_json: JSON.stringify(previousAggregate), + last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z", + }, + aggregate: nextAggregate, + nowMs: 5_000, + }); + + expect(result?.kind).toBe("live_activity_update"); + expect(queuedJobs).toMatchObject([ + { + payload: { + kind: "live_activity_update", + aggregate: { + activities: [ + { phase: "waiting_for_input", threadId: "thread" }, + { phase: "running", threadId: "thread-3" }, + ], + }, + }, + }, + ]); + }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); + }, + ); + it.effect( "throttles updates for changed aggregates with stable counts and no pending attention", () => { @@ -650,6 +966,11 @@ describe("ApnsDeliveries", () => { }, ); + it.effect( + "queues an update when phase changes from starting to running inside the throttle window", + queuesLiveActivityUpdateOnStartingToRunningPhaseChange, + ); + it.effect("queues an end for an active Live Activity when Live Activities are disabled", () => { const attempts: Array = []; const queuedJobs: Array = []; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 651f031f0efa..7493f66936ca 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -47,16 +47,15 @@ import { alertForAttentionTransition, alertForNewlyTerminal, alertForTerminalAggregate, - newlyTerminalRows, shouldAlertForActivity, } from "./agentActivityAlerts.ts"; +import { shouldUpdateLiveActivity } from "./liveActivityUpdateThrottle.ts"; export { alertForAttentionTransition, alertForNewlyTerminal, alertForTerminalAggregate, } from "./agentActivityAlerts.ts"; -const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000; // How long a just-armed card may sit with an empty aggregate before an end is // warranted; covers the gap between arming on send and the environment's // first publish reaching the relay. @@ -143,6 +142,12 @@ const decodeRelayAgentAwarenessPreferencesJson = Schema.decodeUnknownOption( ); const decodeSignedApnsDeliveryJob = Schema.decodeUnknownEffect(SignedApnsDeliveryJob); +/** + * Decode a stored Live Activity aggregate JSON payload. + * + * @param value - Serialized aggregate, if any + * @returns The aggregate, or null when missing or invalid + */ function parseAggregate(value: string | null): RelayAgentActivityAggregateState | null { if (!value) { return null; @@ -150,54 +155,16 @@ function parseAggregate(value: string | null): RelayAgentActivityAggregateState return Option.getOrNull(decodeRelayAgentActivityAggregateStateJson(value)); } +/** + * Decode a device's awareness preference JSON payload. + * + * @param value - Serialized preferences + * @returns The preferences, or null when invalid + */ function parsePreferences(value: string): RelayAgentAwarenessPreferences | null { return Option.getOrNull(decodeRelayAgentAwarenessPreferencesJson(value)); } -function aggregateNeedsAttention(aggregate: RelayAgentActivityAggregateState): boolean { - return aggregate.activities.some( - (row) => row.phase === "waiting_for_approval" || row.phase === "waiting_for_input", - ); -} - -function shouldUpdateLiveActivity(input: { - readonly previousAggregate: RelayAgentActivityAggregateState | null; - readonly nextAggregate: RelayAgentActivityAggregateState; - readonly lastDeliveryAt: string | null; - readonly nowMs: number; -}): boolean { - if (!input.previousAggregate) { - return true; - } - if (JSON.stringify(input.previousAggregate) === JSON.stringify(input.nextAggregate)) { - return false; - } - if (input.previousAggregate.activeCount !== input.nextAggregate.activeCount) { - return true; - } - if (aggregateNeedsAttention(input.nextAggregate)) { - return true; - } - // A thread finishing must never be throttled away: when a completion and a - // new start land in the same window, activeCount is unchanged and the Done - // transition (and its alert) would otherwise be suppressed. - if (newlyTerminalRows(input.previousAggregate, input.nextAggregate, true).length > 0) { - return true; - } - const lastDeliveryAtMs = - input.lastDeliveryAt === null - ? null - : Option.match(DateTime.make(input.lastDeliveryAt), { - onNone: () => Number.NaN, - onSome: (dt) => dt.epochMilliseconds, - }); - return ( - lastDeliveryAtMs === null || - Number.isNaN(lastDeliveryAtMs) || - input.nowMs - lastDeliveryAtMs >= MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS - ); -} - // Completions replayed long after the fact (server restarts republish every // recently-finished thread) must not ring the device again. diff --git a/infra/relay/src/agentActivity/liveActivityUpdateThrottle.ts b/infra/relay/src/agentActivity/liveActivityUpdateThrottle.ts new file mode 100644 index 000000000000..b08445792e6b --- /dev/null +++ b/infra/relay/src/agentActivity/liveActivityUpdateThrottle.ts @@ -0,0 +1,138 @@ +import type { RelayAgentActivityAggregateState } from "@t3tools/contracts/relay"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; + +import { newlyTerminalRows } from "./agentActivityAlerts.ts"; + +const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000; + +/** + * True for approval and input phases that show lock-screen attention. + * + * @param phase - Activity phase + * @returns Whether the row needs user attention + */ +function isAttentionPhase( + phase: RelayAgentActivityAggregateState["activities"][number]["phase"], +): boolean { + return phase === "waiting_for_approval" || phase === "waiting_for_input"; +} + +/** + * True when the set of waiting threads changed. + * Timestamp and ordering churn on the same waiting rows is not a transition. + * + * @param previous - Aggregate already delivered to this Live Activity + * @param next - Newly observed aggregate + * @returns Whether attention appeared, disappeared, or moved to another thread + */ +function aggregateHasAttentionTransition( + previous: RelayAgentActivityAggregateState, + next: RelayAgentActivityAggregateState, +): boolean { + const previouslyAttention = new Set(); + for (const row of previous.activities) { + if (isAttentionPhase(row.phase)) { + previouslyAttention.add(`${row.environmentId}\0${row.threadId}`); + } + } + let nextCount = 0; + for (const row of next.activities) { + if (!isAttentionPhase(row.phase)) { + continue; + } + nextCount += 1; + if (!previouslyAttention.has(`${row.environmentId}\0${row.threadId}`)) { + return true; + } + } + return nextCount !== previouslyAttention.size; +} + +/** + * True when a previously observed thread changed phase. + * Rows are matched by `environmentId` and `threadId`. + * + * @param previous - Aggregate already delivered to this Live Activity + * @param next - Newly observed aggregate + * @returns Whether any matched thread changed phase + */ +function aggregateHasPhaseChange( + previous: RelayAgentActivityAggregateState, + next: RelayAgentActivityAggregateState, +): boolean { + const previousPhases = new Map(); + for (const row of previous.activities) { + previousPhases.set(`${row.environmentId}\0${row.threadId}`, row.phase); + } + for (const row of next.activities) { + const previousPhase = previousPhases.get(`${row.environmentId}\0${row.threadId}`); + if (previousPhase !== undefined && previousPhase !== row.phase) { + return true; + } + } + return false; +} + +/** + * Epoch ms for the last Live Activity delivery, or NaN if the timestamp is invalid. + * + * @param lastDeliveryAt - ISO timestamp from the target row, if any + * @returns null when unset, NaN when unparseable, otherwise epoch milliseconds + */ +function lastLiveActivityDeliveryAtMs(lastDeliveryAt: string | null): number | null { + if (lastDeliveryAt === null) { + return null; + } + const parsed = DateTime.make(lastDeliveryAt); + if (Option.isNone(parsed)) { + return Number.NaN; + } + return parsed.value.epochMilliseconds; +} + +/** + * Queue a Live Activity update on first delivery, exempt changes + * (activeCount, attention row transition, newly-terminal, or phase), or after the 15s throttle. + * + * @param input - Previous/next aggregates, last delivery time, and now + * @returns Whether an update should be queued + */ +export function shouldUpdateLiveActivity(input: { + readonly previousAggregate: RelayAgentActivityAggregateState | null; + readonly nextAggregate: RelayAgentActivityAggregateState; + readonly lastDeliveryAt: string | null; + readonly nowMs: number; +}): boolean { + if (!input.previousAggregate) { + return true; + } + if (JSON.stringify(input.previousAggregate) === JSON.stringify(input.nextAggregate)) { + return false; + } + if (input.previousAggregate.activeCount !== input.nextAggregate.activeCount) { + return true; + } + // Waiting already on the lock screen must stay throttled for timestamp and + // ordering churn. Exempt when the waiting-thread set changes. + if (aggregateHasAttentionTransition(input.previousAggregate, input.nextAggregate)) { + return true; + } + // A thread finishing must never be throttled away: when a completion and a + // new start land in the same window, activeCount is unchanged and the Done + // transition (and its alert) would otherwise be suppressed. + if (newlyTerminalRows(input.previousAggregate, input.nextAggregate, true).length > 0) { + return true; + } + // starting→running keeps activeCount at 1 and is not attention/terminal, but + // the lock-screen copy changes (Connecting→Working) and is never republished. + if (aggregateHasPhaseChange(input.previousAggregate, input.nextAggregate)) { + return true; + } + const lastDeliveryAtMs = lastLiveActivityDeliveryAtMs(input.lastDeliveryAt); + return ( + lastDeliveryAtMs === null || + Number.isNaN(lastDeliveryAtMs) || + input.nowMs - lastDeliveryAtMs >= MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS + ); +}