From 25f5e96fe444efee910362bde4c1fc5c6f55ac1b Mon Sep 17 00:00:00 2001 From: Tomasz Torz Date: Mon, 26 Oct 2026 20:54:33 +0100 Subject: [PATCH 1/9] fix(server): renaming a thread no longer re-sends phone alerts A rename republished the thread's awareness state with a fresh updatedAt, and the relay's notification paths treat a fresh time as a new alert. The server now keeps the last published time while the phase is unchanged, and the relay skips notification pushes when a publish repeats the stored row's phase. Live Activity alerts already compare phases and are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../src/relay/AgentAwarenessRelay.test.ts | 31 +++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 33 ++++++---- .../AgentActivityPublisher.test.ts | 65 ++++++++++++++++++- .../agentActivity/AgentActivityPublisher.ts | 9 ++- .../src/agentActivity/AgentActivityRows.ts | 24 ++++++- .../src/agentActivity/ApnsDeliveries.test.ts | 20 +++++- .../relay/src/agentActivity/ApnsDeliveries.ts | 21 +++--- .../src/agentActivity/FcmDeliveries.test.ts | 2 +- .../agentActivity/MobileRegistrations.test.ts | 2 +- 9 files changed, 181 insertions(+), 26 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index e101b7475cdf8..a8a0f241eed30 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -352,6 +352,37 @@ describe("AgentAwarenessRelay", () => { }), ); + it.effect("keeps the phase's published time when a finished thread is renamed", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + yield* relay.publishThread(THREAD_ID); + const completedAt = DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"); + yield* Ref.set(currentShell, shell({ status: "completed", updatedAt: completedAt })); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + yield* Ref.set( + currentShell, + shell({ + status: "completed", + title: "Renamed thread", + updatedAt: DateTime.makeUnsafe("2026-09-04T13:00:00.000Z"), + }), + ); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.deepEqual( + publications.map(({ state }) => [state?.phase, state?.threadTitle, state?.updatedAt]), + [ + ["running", "Thread", NOW], + ["completed", "Thread", "2026-09-04T12:01:00.000Z"], + ["completed", "Renamed thread", "2026-09-04T12:01:00.000Z"], + ], + ); + }), + ); + it.effect("republishes unchanged state with fresh credentials after relinking", () => Effect.gen(function* () { const { relay, secrets, publications } = yield* makeTestRelay(); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index da72320af27ae..95fed00c76fc7 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -163,6 +163,14 @@ function agentAwarenessPublishIdentity(state: RelayAgentActivityState | null): s return JSON.stringify(meaningfulState); } +// The relay treats a new time as a new alert, so same-phase changes like renames keep the old one. +function withPublishedPhaseTime( + state: RelayAgentActivityState, + published: RelayAgentActivityState | undefined, +): RelayAgentActivityState { + return published?.phase === state.phase ? { ...state, updatedAt: published.updatedAt } : state; +} + function resolveAgentActivityPublishingStartupState(input: { readonly relayConfigured: boolean; readonly publishEnabled: boolean; @@ -371,7 +379,7 @@ export const make = Effect.gen(function* () { const startedAt = (yield* DateTime.now).epochMilliseconds; const activeSnapshotPublishedRef = yield* Ref.make(false); const catchUpRequests = yield* Queue.dropping(1); - const publishedStateByThreadRef = yield* Ref.make(new Map()); + const publishedStateByThreadRef = yield* Ref.make(new Map()); const readSecretString = (name: string) => secrets @@ -534,15 +542,19 @@ export const make = Effect.gen(function* () { thread, project, }); - const publishIdentity = agentAwarenessPublishIdentity(snapshot.state); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); + const publishedState = publishedStateByThread.get(threadId); + const state = snapshot.state && withPublishedPhaseTime(snapshot.state, publishedState); if ( (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") && !publishedStateByThread.has(threadId) ) { if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return; } - if (publishedStateByThread.get(threadId) === publishIdentity) { + if ( + publishedState !== undefined && + agentAwarenessPublishIdentity(publishedState) === agentAwarenessPublishIdentity(state) + ) { // Clear moot deadlines so later transient nulls cannot skip their deferral window. publishConfirmDeadlines.delete(threadId); heldCompletions.delete(threadId); @@ -557,10 +569,7 @@ export const make = Effect.gen(function* () { // Confirm tombstones over live state to avoid removing active cards. // Confirm completion: a run can end moments before its continuation starts, and the relay // alerts on every move into completed. - const requiresConfirmation = - (snapshot.state === null && - publishedStateByThread.get(threadId) !== agentAwarenessPublishIdentity(null)) || - snapshot.state?.phase === "completed"; + const requiresConfirmation = snapshot.state === null || snapshot.state.phase === "completed"; if (requiresConfirmation) { if (snapshot.state?.phase === "completed") heldCompletions.add(threadId); const nowMs = (yield* DateTime.now).epochMilliseconds; @@ -602,7 +611,7 @@ export const make = Effect.gen(function* () { }) : null; if (unsettled?.state?.phase === "completed") { - const completion = unsettled.state; + const completion = withPublishedPhaseTime(unsettled.state, publishedState); yield* publishState({ projectId: unsettled.projectId, state: completion, @@ -611,7 +620,7 @@ export const make = Effect.gen(function* () { heldCompletions.delete(threadId); publishConfirmDeadlines.set(threadId, nowMs + 5_000); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => - new Map(publishedStates).set(threadId, agentAwarenessPublishIdentity(completion)), + new Map(publishedStates).set(threadId, completion), ); yield* schedulePublishConfirm(threadId); return; @@ -636,17 +645,17 @@ export const make = Effect.gen(function* () { yield* publishState({ projectId: snapshot.projectId, - state: snapshot.state, + state, reason: snapshot.reason, }).pipe(withRelayClientTracing); publishConfirmDeadlines.delete(threadId); heldCompletions.delete(threadId); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => { const nextPublishedStates = new Map(publishedStates); - if (snapshot.state === null) { + if (state === null) { nextPublishedStates.delete(threadId); } else { - nextPublishedStates.set(threadId, publishIdentity); + nextPublishedStates.set(threadId, state); } return nextPublishedStates; }); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 1b7ffe4a0aed2..8d177fc3b1e6c 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -71,7 +71,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: () => Effect.void, + upsert: () => Effect.succeed(null), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => Effect.succeed([state]), @@ -140,6 +140,69 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { + it.effect("does not notify again when a publish repeats the stored phase", () => { + const renamed = { ...state, phase: "completed" as const, threadTitle: "Renamed" }; + const sends: string[] = []; + const publishOver = (previous: RelayAgentActivityState) => + Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + yield* publisher.publish({ + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: renamed, + }); + }).pipe( + Effect.provide( + layerPublisher.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ upsert: () => Effect.succeed(previous) }), + ), + Layer.succeed( + EnvironmentLinks.EnvironmentLinks, + makeEnvironmentLinks({ + listDeliveryUsersForEnvironment: () => + Effect.succeed([ + { userId: "live", notificationsEnabled: true, liveActivitiesEnabled: true }, + { + userId: "push", + notificationsEnabled: true, + liveActivitiesEnabled: false, + }, + ]), + }), + ), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ listTargets: () => Effect.succeed([target("ios")]) }), + ), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendForTarget: (input) => + Effect.sync(() => void sends.push(`target:${input.repeatsAlert}`)).pipe( + Effect.as(null), + ), + sendPushNotificationForTarget: () => + Effect.sync(() => void sends.push("notification")).pipe(Effect.as(null)), + }), + ), + ), + ), + ), + ), + ); + return Effect.gen(function* () { + yield* publishOver({ ...renamed, threadTitle: "Thread", updatedAt: "1969-12-31T00:00:00Z" }); + expect(sends.splice(0).sort()).toEqual(["target:true", "target:true"]); + yield* publishOver({ ...renamed, phase: "running" }); + expect(sends.splice(0).sort()).toEqual(["notification", "target:false", "target:false"]); + }); + }); + it.effect("routes Android publication and registration replay to FCM alongside iOS", () => { const android = { ...target("android"), platform: "android" as const, ios_major_version: null }; const ios = target("ios"); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index d793cb2e08d34..a367967145086 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -65,6 +65,7 @@ export const make = Effect.gen(function* () { const publishForDeliveryUser = Effect.fnUntraced(function* (input: { readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord; readonly state: RelayAgentActivityState | null; + readonly repeatsAlert: boolean; readonly nowMs: number; }) { const activeStates = input.deliveryUser.liveActivitiesEnabled @@ -80,6 +81,7 @@ export const make = Effect.gen(function* () { const notificationOnlyAggregate = input.deliveryUser.notificationsEnabled && !input.deliveryUser.liveActivitiesEnabled && + !input.repeatsAlert && input.state !== null ? makeAggregateState({ activeStates: isTerminalPhase(input.state) ? [] : [input.state], @@ -100,6 +102,7 @@ export const make = Effect.gen(function* () { target, aggregate: liveActivityAggregate, nowMs: input.nowMs, + repeatsAlert: input.repeatsAlert, }), notificationOnlyAggregate === null ? Effect.succeed(null) @@ -210,12 +213,15 @@ export const make = Effect.gen(function* () { "relay.thread_id": input.threadId, "relay.agent_activity.phase": input.state?.phase ?? "deleted", }); + let repeatsAlert = false; if (input.state) { // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones. - yield* rows.upsert({ + const previous = yield* rows.upsert({ environmentPublicKey: input.environmentPublicKey, state: input.state, }); + // Same-phase publishes such as renames or restart snapshots are not new alerts. + repeatsAlert = previous?.phase === input.state.phase; } else { yield* rows.remove({ environmentId: input.environmentId, @@ -235,6 +241,7 @@ export const make = Effect.gen(function* () { publishForDeliveryUser({ deliveryUser, state: input.state, + repeatsAlert, nowMs: now.epochMilliseconds, }), { concurrency: 4 }, diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index 5ed288b8e1ada..d24c3924eac63 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -65,10 +65,11 @@ export class AgentActivityRowListPersistenceError extends Schema.TaggedError Effect.Effect; + }) => Effect.Effect; readonly pruneTerminal: (input: { readonly updatedBefore: string; }) => Effect.Effect; @@ -121,6 +122,26 @@ export const make = Effect.gen(function* () { }), ), ); + const [previous] = yield* db + .select({ stateJson: relayAgentActivityRows.stateJson }) + .from(relayAgentActivityRows) + .where( + and( + eq(relayAgentActivityRows.environmentId, input.state.environmentId), + eq(relayAgentActivityRows.environmentPublicKey, input.environmentPublicKey), + eq(relayAgentActivityRows.threadId, input.state.threadId), + ), + ) + .pipe( + Effect.mapError( + (cause) => + new AgentActivityRowUpsertPersistenceError({ + environmentId: input.state.environmentId, + threadId: input.state.threadId, + cause, + }), + ), + ); yield* db .insert(relayAgentActivityRows) .values({ @@ -152,6 +173,7 @@ export const make = Effect.gen(function* () { }), ), ); + return previous ? Option.getOrNull(decodeRelayAgentActivityState(previous.stateJson)) : null; }), remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) { diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index ddabfe557fa51..f13d0145cca62 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -168,7 +168,7 @@ function layerFor(input: { Layer.provide( Layer.mergeAll( Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: () => Effect.void, + upsert: () => Effect.succeed(null), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => @@ -547,6 +547,24 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("skips the fallback push notification when a publish repeats the phase", () => { + const queuedJobs: Array = []; + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { ...target, push_token: "apns-device-token", activity_push_token: null }, + aggregate: { + ...aggregate, + activities: [{ ...aggregate.activities[0]!, phase: "waiting_for_input" }], + }, + nowMs: 5_000, + repeatsAlert: true, + }); + expect(result).toBeNull(); + expect(queuedJobs).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs }))); + }); + it.effect("queues bounded alert notification payloads", () => { const attempts: Array = []; const queuedJobs: Array = []; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 5fb7b235206db..59f3b570d51d2 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -291,6 +291,7 @@ function chooseDelivery(input: { readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + readonly repeatsAlert?: boolean; }): ChosenDelivery | null { const liveActivityDelivery = chooseLiveActivityDelivery(input); if (liveActivityDelivery === "suppressed") { @@ -299,7 +300,7 @@ function chooseDelivery(input: { if (liveActivityDelivery) { return liveActivityDelivery; } - const notification = input.replay ? null : notificationForAggregate(input); + const notification = input.replay || input.repeatsAlert ? null : notificationForAggregate(input); return notification && input.target.push_token ? { kind: "push_notification", @@ -501,6 +502,8 @@ export class ApnsDeliveries extends Context.Service< readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + /** Skips notification pushes but keeps Live Activity alerts, which compare phases. */ + readonly repeatsAlert?: boolean; }) => Effect.Effect; readonly sendPushNotificationForTarget: (input: { readonly target: LiveActivities.TargetRow; @@ -1127,6 +1130,7 @@ export const make = Effect.gen(function* () { aggregate: input.aggregate, nowMs: input.nowMs, replay: input.replay ?? false, + repeatsAlert: input.repeatsAlert ?? false, }); if (!delivery) { return null; @@ -1142,13 +1146,14 @@ export const make = Effect.gen(function* () { }); return result; } - const notification = input.replay - ? null - : notificationForAggregate({ - target: input.target, - aggregate: input.aggregate, - nowMs: input.nowMs, - }); + const notification = + input.replay || input.repeatsAlert + ? null + : notificationForAggregate({ + target: input.target, + aggregate: input.aggregate, + nowMs: input.nowMs, + }); // Use the end alert only when no companion notification will ring, preserving feedback for Live-Activity-only users. const alert = input.replay ? null diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index 5448bc2eb84b7..7535c235d3366 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -131,7 +131,7 @@ function harness() { invalidateDeliveryToken: () => Effect.void, }), Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: () => Effect.void, + upsert: () => Effect.succeed(null), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 38840cf02b5b6..d2435f891f92f 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -80,7 +80,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: () => Effect.void, + upsert: () => Effect.succeed(null), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => { From 82682cdc63dfa1f5367a30fa5f52827390fecc29 Mon Sep 17 00:00:00 2001 From: Tomasz Torz Date: Mon, 26 Oct 2026 21:37:17 +0100 Subject: [PATCH 2/9] fix(relay): a failed delivery no longer turns its retry into a repeat Restore the replaced activity row when delivery fails after the upsert, so the environment's retry of the same state still alerts. Co-Authored-By: Claude Opus 5.5 --- .../AgentActivityPublisher.test.ts | 41 +++++++++++++++++++ .../agentActivity/AgentActivityPublisher.ts | 27 ++++++++---- 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 8d177fc3b1e6c..6dbcdcd622d37 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -140,6 +140,47 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { + it.effect("restores the replaced row when delivery fails so the retry still alerts", () => { + const upserts: string[] = []; + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + const exit = yield* Effect.exit( + publisher.publish({ + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: { ...state, phase: "completed" }, + }), + ); + expect(exit._tag).toBe("Failure"); + expect(upserts).toEqual(["completed", "running"]); + }).pipe( + Effect.provide( + layerPublisher.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ + upsert: (input) => + Effect.sync(() => void upserts.push(input.state.phase)).pipe(Effect.as(state)), + }), + ), + Layer.succeed( + EnvironmentLinks.EnvironmentLinks, + makeEnvironmentLinks({ + listDeliveryUsersForEnvironment: () => Effect.die("offline"), + }), + ), + Layer.succeed(LiveActivities.LiveActivities, makeLiveActivities()), + Layer.succeed(ApnsDeliveries.ApnsDeliveries, makeApnsDeliveries()), + ), + ), + ), + ), + ); + }); + it.effect("does not notify again when a publish repeats the stored phase", () => { const renamed = { ...state, phase: "completed" as const, threadTitle: "Renamed" }; const sends: string[] = []; diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index a367967145086..2edd0e8793e65 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -213,15 +213,13 @@ export const make = Effect.gen(function* () { "relay.thread_id": input.threadId, "relay.agent_activity.phase": input.state?.phase ?? "deleted", }); - let repeatsAlert = false; + let previous: RelayAgentActivityState | null = null; if (input.state) { // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones. - const previous = yield* rows.upsert({ + previous = yield* rows.upsert({ environmentPublicKey: input.environmentPublicKey, state: input.state, }); - // Same-phase publishes such as renames or restart snapshots are not new alerts. - repeatsAlert = previous?.phase === input.state.phase; } else { yield* rows.remove({ environmentId: input.environmentId, @@ -229,11 +227,22 @@ export const make = Effect.gen(function* () { threadId: input.threadId, }); } + // Same-phase publishes such as renames or restart snapshots are not new alerts. + const repeatsAlert = previous !== null && previous.phase === input.state?.phase; + // The environment retries failed publishes; restore the row so the retry still alerts. + const restorePrevious = + input.state === null || repeatsAlert + ? Effect.void + : previous === null + ? rows.remove(input) + : rows.upsert({ environmentPublicKey: input.environmentPublicKey, state: previous }); - const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({ - environmentId: input.environmentId, - environmentPublicKey: input.environmentPublicKey, - }); + const deliveryUsers = yield* links + .listDeliveryUsersForEnvironment({ + environmentId: input.environmentId, + environmentPublicKey: input.environmentPublicKey, + }) + .pipe(Effect.onError(() => Effect.ignore(restorePrevious))); const now = yield* DateTime.now; const deliveriesByUser = yield* Effect.forEach( deliveryUsers, @@ -245,7 +254,7 @@ export const make = Effect.gen(function* () { nowMs: now.epochMilliseconds, }), { concurrency: 4 }, - ); + ).pipe(Effect.onError(() => Effect.ignore(restorePrevious))); const deliveries = deliveriesByUser.flat(); return { ok: true, From 3be565a20179d08082acbb8f1b2727f93cc4e8e7 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 27 Oct 2026 02:34:09 +0000 Subject: [PATCH 3/9] fix(relay): deduplicate phase alerts without rolling back activity --- .../AgentActivityPublisher.test.ts | 164 ++++++++---------- .../agentActivity/AgentActivityPublisher.ts | 35 ++-- .../src/agentActivity/AgentActivityRows.ts | 34 +--- .../src/agentActivity/ApnsDeliveries.test.ts | 20 +-- .../relay/src/agentActivity/ApnsDeliveries.ts | 21 +-- .../agentActivity/ApnsDeliveryQueue.test.ts | 50 ++++++ .../src/agentActivity/ApnsDeliveryQueue.ts | 29 ++-- .../src/agentActivity/FcmDeliveries.test.ts | 2 +- .../agentActivity/MobileRegistrations.test.ts | 2 +- .../src/agentActivity/apnsDeliveryJobs.ts | 27 +++ 10 files changed, 190 insertions(+), 194 deletions(-) diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 6dbcdcd622d37..f3d7dce2cc03c 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -71,7 +71,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: () => Effect.succeed(null), + upsert: ({ state }) => Effect.succeed(state), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => Effect.succeed([state]), @@ -140,109 +140,81 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { - it.effect("restores the replaced row when delivery fails so the retry still alerts", () => { - const upserts: string[] = []; - return Effect.gen(function* () { - const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; - const exit = yield* Effect.exit( - publisher.publish({ - environmentId: state.environmentId, - environmentPublicKey: "key", - threadId: state.threadId, - state: { ...state, phase: "completed" }, + it.effect( + "fans out the persisted phase time to iOS and Android without rolling it back on failure", + () => { + const persisted = { ...state, phase: "waiting_for_input" as const }; + const renamed = { + ...persisted, + threadTitle: "Renamed", + updatedAt: "1970-01-01T00:01:00.000Z", + }; + const writes: RelayAgentActivityState[] = []; + const androidStates: Array = []; + const appleTimes: string[] = []; + let fail = false; + const dependencies = Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ + upsert: (input) => + Effect.sync(() => { + writes.push(input.state); + return { ...input.state, updatedAt: persisted.updatedAt }; + }), + }), + ), + Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ + listTargets: () => + Effect.succeed([ + target("ios"), + { ...target("android"), platform: "android" as const }, + ]), + }), + ), + Layer.succeed(FcmDeliveries.FcmDeliveries, { + enqueue: (input) => + Effect.sync(() => { + androidStates.push(input.state); + return null; + }), + process: () => Effect.void, }), - ); - expect(exit._tag).toBe("Failure"); - expect(upserts).toEqual(["completed", "running"]); - }).pipe( - Effect.provide( - layerPublisher.pipe( - Layer.provide( - Layer.mergeAll( - Layer.succeed( - AgentActivityRows.AgentActivityRows, - makeAgentActivityRows({ - upsert: (input) => - Effect.sync(() => void upserts.push(input.state.phase)).pipe(Effect.as(state)), - }), - ), - Layer.succeed( - EnvironmentLinks.EnvironmentLinks, - makeEnvironmentLinks({ - listDeliveryUsersForEnvironment: () => Effect.die("offline"), - }), - ), - Layer.succeed(LiveActivities.LiveActivities, makeLiveActivities()), - Layer.succeed(ApnsDeliveries.ApnsDeliveries, makeApnsDeliveries()), - ), - ), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendForTarget: (input) => + fail + ? Effect.die("offline") + : Effect.sync(() => { + appleTimes.push(input.aggregate!.activities[0]!.updatedAt); + return null; + }), + }), ), - ), - ); - }); - - it.effect("does not notify again when a publish repeats the stored phase", () => { - const renamed = { ...state, phase: "completed" as const, threadTitle: "Renamed" }; - const sends: string[] = []; - const publishOver = (previous: RelayAgentActivityState) => - Effect.gen(function* () { + ); + return Effect.gen(function* () { const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; - yield* publisher.publish({ + const publish = publisher.publish({ environmentId: state.environmentId, environmentPublicKey: "key", threadId: state.threadId, state: renamed, }); - }).pipe( - Effect.provide( - layerPublisher.pipe( - Layer.provide( - Layer.mergeAll( - Layer.succeed( - AgentActivityRows.AgentActivityRows, - makeAgentActivityRows({ upsert: () => Effect.succeed(previous) }), - ), - Layer.succeed( - EnvironmentLinks.EnvironmentLinks, - makeEnvironmentLinks({ - listDeliveryUsersForEnvironment: () => - Effect.succeed([ - { userId: "live", notificationsEnabled: true, liveActivitiesEnabled: true }, - { - userId: "push", - notificationsEnabled: true, - liveActivitiesEnabled: false, - }, - ]), - }), - ), - Layer.succeed( - LiveActivities.LiveActivities, - makeLiveActivities({ listTargets: () => Effect.succeed([target("ios")]) }), - ), - Layer.succeed( - ApnsDeliveries.ApnsDeliveries, - makeApnsDeliveries({ - sendForTarget: (input) => - Effect.sync(() => void sends.push(`target:${input.repeatsAlert}`)).pipe( - Effect.as(null), - ), - sendPushNotificationForTarget: () => - Effect.sync(() => void sends.push("notification")).pipe(Effect.as(null)), - }), - ), - ), - ), - ), - ), - ); - return Effect.gen(function* () { - yield* publishOver({ ...renamed, threadTitle: "Thread", updatedAt: "1969-12-31T00:00:00Z" }); - expect(sends.splice(0).sort()).toEqual(["target:true", "target:true"]); - yield* publishOver({ ...renamed, phase: "running" }); - expect(sends.splice(0).sort()).toEqual(["notification", "target:false", "target:false"]); - }); - }); + yield* publish; + expect(appleTimes).toEqual([persisted.updatedAt]); + expect(androidStates).toMatchObject([ + { threadTitle: "Renamed", updatedAt: persisted.updatedAt }, + ]); + fail = true; + expect((yield* Effect.exit(publish))._tag).toBe("Failure"); + expect(writes).toEqual([renamed, renamed]); + }).pipe(Effect.provide(AgentActivityPublisher.layer.pipe(Layer.provide(dependencies)))); + }, + ); it.effect("routes Android publication and registration replay to FCM alongside iOS", () => { const android = { ...target("android"), platform: "android" as const, ios_major_version: null }; diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 2edd0e8793e65..72c7b32410e3c 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -65,7 +65,6 @@ export const make = Effect.gen(function* () { const publishForDeliveryUser = Effect.fnUntraced(function* (input: { readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord; readonly state: RelayAgentActivityState | null; - readonly repeatsAlert: boolean; readonly nowMs: number; }) { const activeStates = input.deliveryUser.liveActivitiesEnabled @@ -81,7 +80,6 @@ export const make = Effect.gen(function* () { const notificationOnlyAggregate = input.deliveryUser.notificationsEnabled && !input.deliveryUser.liveActivitiesEnabled && - !input.repeatsAlert && input.state !== null ? makeAggregateState({ activeStates: isTerminalPhase(input.state) ? [] : [input.state], @@ -102,7 +100,6 @@ export const make = Effect.gen(function* () { target, aggregate: liveActivityAggregate, nowMs: input.nowMs, - repeatsAlert: input.repeatsAlert, }), notificationOnlyAggregate === null ? Effect.succeed(null) @@ -213,12 +210,12 @@ export const make = Effect.gen(function* () { "relay.thread_id": input.threadId, "relay.agent_activity.phase": input.state?.phase ?? "deleted", }); - let previous: RelayAgentActivityState | null = null; - if (input.state) { + let state = input.state; + if (state) { // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones. - previous = yield* rows.upsert({ + state = yield* rows.upsert({ environmentPublicKey: input.environmentPublicKey, - state: input.state, + state, }); } else { yield* rows.remove({ @@ -227,34 +224,22 @@ export const make = Effect.gen(function* () { threadId: input.threadId, }); } - // Same-phase publishes such as renames or restart snapshots are not new alerts. - const repeatsAlert = previous !== null && previous.phase === input.state?.phase; - // The environment retries failed publishes; restore the row so the retry still alerts. - const restorePrevious = - input.state === null || repeatsAlert - ? Effect.void - : previous === null - ? rows.remove(input) - : rows.upsert({ environmentPublicKey: input.environmentPublicKey, state: previous }); - const deliveryUsers = yield* links - .listDeliveryUsersForEnvironment({ - environmentId: input.environmentId, - environmentPublicKey: input.environmentPublicKey, - }) - .pipe(Effect.onError(() => Effect.ignore(restorePrevious))); + const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({ + environmentId: input.environmentId, + environmentPublicKey: input.environmentPublicKey, + }); const now = yield* DateTime.now; const deliveriesByUser = yield* Effect.forEach( deliveryUsers, (deliveryUser) => publishForDeliveryUser({ deliveryUser, - state: input.state, - repeatsAlert, + state, nowMs: now.epochMilliseconds, }), { concurrency: 4 }, - ).pipe(Effect.onError(() => Effect.ignore(restorePrevious))); + ); const deliveries = deliveriesByUser.flat(); return { ok: true, diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index d24c3924eac63..38caf853409fa 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -65,11 +65,11 @@ export class AgentActivityRowListPersistenceError extends Schema.TaggedError Effect.Effect; + }) => Effect.Effect; readonly pruneTerminal: (input: { readonly updatedBefore: string; }) => Effect.Effect; @@ -122,27 +122,8 @@ export const make = Effect.gen(function* () { }), ), ); - const [previous] = yield* db - .select({ stateJson: relayAgentActivityRows.stateJson }) - .from(relayAgentActivityRows) - .where( - and( - eq(relayAgentActivityRows.environmentId, input.state.environmentId), - eq(relayAgentActivityRows.environmentPublicKey, input.environmentPublicKey), - eq(relayAgentActivityRows.threadId, input.state.threadId), - ), - ) - .pipe( - Effect.mapError( - (cause) => - new AgentActivityRowUpsertPersistenceError({ - environmentId: input.state.environmentId, - threadId: input.state.threadId, - cause, - }), - ), - ); - yield* db + const samePhase = sql`${relayAgentActivityRows.stateJson} ->> 'phase' = excluded.state_json ->> 'phase'`; + const [stored] = yield* db .insert(relayAgentActivityRows) .values({ environmentId: input.state.environmentId, @@ -159,10 +140,11 @@ export const make = Effect.gen(function* () { relayAgentActivityRows.threadId, ], set: { - stateJson, - updatedAt: input.state.updatedAt, + stateJson: sql`case when ${samePhase} then excluded.state_json || jsonb_build_object('updatedAt', ${relayAgentActivityRows.stateJson} ->> 'updatedAt') else excluded.state_json end`, + updatedAt: sql`case when ${samePhase} then ${relayAgentActivityRows.updatedAt} else excluded.updated_at end`, }, }) + .returning({ stateJson: relayAgentActivityRows.stateJson }) .pipe( Effect.mapError( (cause) => @@ -173,7 +155,7 @@ export const make = Effect.gen(function* () { }), ), ); - return previous ? Option.getOrNull(decodeRelayAgentActivityState(previous.stateJson)) : null; + return stored!.stateJson; }), remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) { diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index f13d0145cca62..61600a5edeeeb 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -168,7 +168,7 @@ function layerFor(input: { Layer.provide( Layer.mergeAll( Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: () => Effect.succeed(null), + upsert: ({ state }) => Effect.succeed(state), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => @@ -547,24 +547,6 @@ describe("ApnsDeliveries", () => { }, ); - it.effect("skips the fallback push notification when a publish repeats the phase", () => { - const queuedJobs: Array = []; - return Effect.gen(function* () { - const deliveries = yield* ApnsDeliveries.ApnsDeliveries; - const result = yield* deliveries.sendForTarget({ - target: { ...target, push_token: "apns-device-token", activity_push_token: null }, - aggregate: { - ...aggregate, - activities: [{ ...aggregate.activities[0]!, phase: "waiting_for_input" }], - }, - nowMs: 5_000, - repeatsAlert: true, - }); - expect(result).toBeNull(); - expect(queuedJobs).toEqual([]); - }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs }))); - }); - it.effect("queues bounded alert notification payloads", () => { const attempts: Array = []; const queuedJobs: Array = []; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 59f3b570d51d2..5fb7b235206db 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -291,7 +291,6 @@ function chooseDelivery(input: { readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; - readonly repeatsAlert?: boolean; }): ChosenDelivery | null { const liveActivityDelivery = chooseLiveActivityDelivery(input); if (liveActivityDelivery === "suppressed") { @@ -300,7 +299,7 @@ function chooseDelivery(input: { if (liveActivityDelivery) { return liveActivityDelivery; } - const notification = input.replay || input.repeatsAlert ? null : notificationForAggregate(input); + const notification = input.replay ? null : notificationForAggregate(input); return notification && input.target.push_token ? { kind: "push_notification", @@ -502,8 +501,6 @@ export class ApnsDeliveries extends Context.Service< readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; - /** Skips notification pushes but keeps Live Activity alerts, which compare phases. */ - readonly repeatsAlert?: boolean; }) => Effect.Effect; readonly sendPushNotificationForTarget: (input: { readonly target: LiveActivities.TargetRow; @@ -1130,7 +1127,6 @@ export const make = Effect.gen(function* () { aggregate: input.aggregate, nowMs: input.nowMs, replay: input.replay ?? false, - repeatsAlert: input.repeatsAlert ?? false, }); if (!delivery) { return null; @@ -1146,14 +1142,13 @@ export const make = Effect.gen(function* () { }); return result; } - const notification = - input.replay || input.repeatsAlert - ? null - : notificationForAggregate({ - target: input.target, - aggregate: input.aggregate, - nowMs: input.nowMs, - }); + const notification = input.replay + ? null + : notificationForAggregate({ + target: input.target, + aggregate: input.aggregate, + nowMs: input.nowMs, + }); // Use the end alert only when no companion notification will ring, preserving feedback for Live-Activity-only users. const alert = input.replay ? null diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts index b7210b07d58bf..9acf4c5ba90b5 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts @@ -8,6 +8,7 @@ import * as Redacted from "effect/Redacted"; import * as RelayConfiguration from "../Config.ts"; import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts"; +import type { SignedApnsDeliveryJob } from "./apnsDeliveryJobs.ts"; const config: RelayConfiguration.RelayConfiguration["Service"] = { relayIssuer: "https://relay.example.com", @@ -29,6 +30,55 @@ const config: RelayConfiguration.RelayConfiguration["Service"] = { }; describe("ApnsDeliveryQueue", () => { + it.effect("keeps alert jobs stable across renames and retries, independently per target", () => { + const sent: SignedApnsDeliveryJob[] = []; + const layer = ApnsDeliveryQueue.layer.pipe( + Layer.provide(NodeCryptoLayer.layer), + Layer.provide(RelayConfiguration.layer(config)), + Layer.provide( + Layer.succeed(ApnsDeliveryQueue.ApnsDeliveryQueueSender, { + send: (job) => + Effect.sync(() => { + sent.push(job); + }), + }), + ), + ); + return Effect.gen(function* () { + const queue = yield* ApnsDeliveryQueue.ApnsDeliveryQueue; + const input = { + userId: "user-1", + deviceId: "device-1", + token: "push-token", + notification: { + title: "Thread", + body: "Input: Project", + environmentId: "env-1", + threadId: "thread-1", + deepLink: "/threads/env-1/thread-1", + phase: "waiting_for_input" as const, + updatedAt: "2026-09-04T12:00:00.000Z", + }, + }; + yield* queue.enqueuePushNotification(input); + yield* queue.enqueuePushNotification({ + ...input, + notification: { ...input.notification, title: "Renamed" }, + }); + yield* queue.enqueuePushNotification({ ...input, deviceId: "device-2" }); + yield* queue.enqueuePushNotification({ + ...input, + notification: { ...input.notification, updatedAt: "2026-09-04T13:00:00.000Z" }, + }); + yield* queue.enqueuePushNotification({ ...input, token: "rotated-token" }); + expect(sent[4]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); + expect(sent[1]!.payload.jobId).toBe(sent[0]!.payload.jobId); + expect(sent[2]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); + expect(sent[3]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); + expect(sent[1]!.payload.notification?.title).toBe("Renamed"); + }).pipe(Effect.provide(layer)); + }); + it.effect("does not require the deployment RuntimeContext when building the Worker layer", () => { const sent: unknown[] = []; const sender: Cloudflare.Queues.WriteQueueClient = { diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts index 6b608802a3421..d7fcd894371c4 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts @@ -18,6 +18,7 @@ import { } from "./agentActivityPayloads.ts"; import { expiresAtForJob, + notificationJobId, makeApnsDeliveryJobPayload, signApnsDeliveryJob, type ApnsDeliveryJobPayload, @@ -148,19 +149,21 @@ export const make = Effect.gen(function* () { "relay.thread_id": input.notification.threadId, }); const now = yield* DateTime.now; - const jobId = yield* crypto.randomUUIDv4().pipe( - Effect.mapError( - (cause) => - new ApnsDeliveryQueueSendError({ - operation: "generate-job-id", - jobId: null, - kind: "push_notification", - userId: input.userId, - deviceId: input.deviceId, - cause, - }), - ), - ); + const jobId = + notificationJobId(input) ?? + (yield* crypto.randomUUIDv4().pipe( + Effect.mapError( + (cause) => + new ApnsDeliveryQueueSendError({ + operation: "generate-job-id", + jobId: null, + kind: "push_notification", + userId: input.userId, + deviceId: input.deviceId, + cause, + }), + ), + )); yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": jobId }); const payload = makeApnsDeliveryJobPayload({ kind: "push_notification", diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index 7535c235d3366..759a9bd7e97a1 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -131,7 +131,7 @@ function harness() { invalidateDeliveryToken: () => Effect.void, }), Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: () => Effect.succeed(null), + upsert: ({ state }) => Effect.succeed(state), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index d2435f891f92f..33bf610e8e8db 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -80,7 +80,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: () => Effect.succeed(null), + upsert: ({ state }) => Effect.succeed(state), remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => { diff --git a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts index ed6651509ee19..4848e08f1730d 100644 --- a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts +++ b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts @@ -48,6 +48,33 @@ export const ApnsNotificationPayload = Schema.Struct({ }); export type ApnsNotificationPayload = typeof ApnsNotificationPayload.Type; +export function notificationJobId(input: { + readonly userId: string; + readonly deviceId: string; + readonly token: string; + readonly bundleId?: string | null; + readonly apsEnvironment?: "sandbox" | "production" | null; + readonly notification: ApnsNotificationPayload; +}): string | null { + const { notification } = input; + if (notification.phase === undefined || notification.updatedAt === undefined) return null; + return NodeCrypto.createHash("sha256") + .update( + JSON.stringify([ + input.userId, + input.deviceId, + input.token, + input.bundleId ?? null, + input.apsEnvironment ?? null, + notification.environmentId, + notification.threadId, + notification.phase, + notification.updatedAt, + ]), + ) + .digest("hex"); +} + export const ApnsLiveActivityAlert = Schema.Struct({ title: Schema.String, body: Schema.String, From 742748eb92f71cc7a9c50ee17403ef9a5b8b710a Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 27 Oct 2026 02:41:47 +0000 Subject: [PATCH 4/9] fix(relay): reject queued cards with stale thread metadata --- .../src/agentActivity/ApnsDeliveries.test.ts | 42 +++++++++++++++++++ .../relay/src/agentActivity/ApnsDeliveries.ts | 10 ++++- 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 61600a5edeeeb..015e84d1f7d46 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -645,6 +645,48 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("skips queued card metadata from before a same-phase rename", () => { + const attempts: DeliveryAttempts.DeliveryAttemptInput[] = []; + const requests: HttpClientRequest.HttpClientRequest[] = []; + const waiting = { ...state, phase: "waiting_for_input" as const, threadTitle: "Renamed" }; + const stale = { + ...aggregate, + activities: [{ ...aggregate.activities[0]!, phase: waiting.phase }], + }; + const job = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload: makeApnsDeliveryJobPayload({ + kind: "live_activity_update", + userId: target.user_id, + deviceId: target.device_id, + token: target.activity_push_token!, + aggregate: stale, + createdAt: "1970-01-01T00:00:00.000Z", + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "stale-title", + }), + }); + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.processSignedJob(job); + expect(result.apnsReason).toBe("Stale APNs delivery job skipped."); + expect(requests).toHaveLength(0); + }).pipe( + Effect.provide( + layerFor({ + attempts, + activityStates: [waiting], + config: signingConfig, + execute: (request) => + Effect.sync(() => { + requests.push(request); + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }), + }), + ), + ); + }); + 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 5fb7b235206db..d6c350fe2c7fa 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -22,6 +22,7 @@ import { isTerminalPhase, notificationForActivity, sanitizeAgentActivityAggregateState, + sanitizeAgentActivityAggregateRow, sanitizeApnsNotificationPayload, } from "./agentActivityPayloads.ts"; import { makeAggregateState } from "./agentActivityAggregate.ts"; @@ -611,10 +612,15 @@ export const make = Effect.gen(function* () { ); return input.aggregate.activities.every((row) => { const current = currentByThread.get(`${row.environmentId}\u0000${row.threadId}`); + if (current === undefined) return false; + const metadata = sanitizeAgentActivityAggregateRow({ ...row, ...current }); return ( - current !== undefined && current.phase === row.phase && - current.updatedAt === row.updatedAt + current.updatedAt === row.updatedAt && + metadata.threadTitle === row.threadTitle && + metadata.projectTitle === row.projectTitle && + metadata.modelTitle === row.modelTitle && + metadata.deepLink === row.deepLink ); }); }), From 3650479e6f8d55ba3e82e4c07727d69ecf1d840b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 08:17:58 +0000 Subject: [PATCH 5/9] fix(relay): preserve aggregate row sanitizer export --- infra/relay/src/agentActivity/agentActivityPayloads.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/infra/relay/src/agentActivity/agentActivityPayloads.ts b/infra/relay/src/agentActivity/agentActivityPayloads.ts index b2086426a9ec0..8176609ccc7c1 100644 --- a/infra/relay/src/agentActivity/agentActivityPayloads.ts +++ b/infra/relay/src/agentActivity/agentActivityPayloads.ts @@ -63,7 +63,7 @@ function sanitizeDeepLink(value: string): string { return truncateText(trimmed, MAX_DEEP_LINK_LENGTH); } -function sanitizeAgentActivityAggregateRow( +export function sanitizeAgentActivityAggregateRow( row: RelayAgentActivityAggregateRow, ): RelayAgentActivityAggregateRow { return { From 5206a4138cecc11c4f7148cc1ecc7ab6baac1b8c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 21:28:30 +0000 Subject: [PATCH 6/9] fix(server): a subagent's next request still alerts after the phase-time change Carrying the published time across same-phase publishes would give a child's next request the time of its previous one, and the relay keys alerts by phase and time, so the new wait would never reach the phone. Child-driven waits keep their per-request time; the carryover applies only between publishes no child drove. Also drops the publisher test's rollback half, left from a reverted approach. AI-assisted; verified with the tests and tools listed above. --- .../src/relay/AgentAwarenessRelay.test.ts | 19 ++- .../AgentActivityPublisher.test.ts | 122 ++++++++---------- 2 files changed, 69 insertions(+), 72 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 8120a8aa4b195..b5ca7394206be 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -368,12 +368,27 @@ describe("AgentAwarenessRelay", () => { assert.equal(publications[2]?.state?.threadTitle, "Renamed thread"); assert.equal(publications[2]?.state?.updatedAt, publications[1]?.state?.updatedAt); + // The child's next request is a new wait in the same phase, so it gets a new alert time. + yield* setShell( + waitingChild({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make("child-request-2"), + kind: "command", + createdAt: DateTime.makeUnsafe("2026-09-04T12:02:00.000Z"), + }, + }), + ); + yield* relay.publishThread(CHILD_THREAD_ID); + assert.equal(publications.length, 4); + assert.equal(publications[3]?.state?.phase, "waiting_for_approval"); + assert.notEqual(publications[3]?.state?.updatedAt, publications[2]?.state?.updatedAt); + yield* setShell(waitingChild({ pendingRuntimeRequest: null })); yield* relay.publishThread(CHILD_THREAD_ID); yield* TestClock.adjust("5 seconds"); yield* relay.drain; - assert.equal(publications.length, 4); - assert.equal(publications[3]?.state?.phase, "completed"); + assert.equal(publications.length, 5); + assert.equal(publications[4]?.state?.phase, "completed"); }), ); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 4c7676d7445a7..793a2d2399da4 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -140,81 +140,63 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { - it.effect( - "fans out the persisted phase time to iOS and Android without rolling it back on failure", - () => { - const persisted = { ...state, phase: "waiting_for_input" as const }; - const renamed = { - ...persisted, - threadTitle: "Renamed", - updatedAt: "1970-01-01T00:01:00.000Z", - }; - const writes: RelayAgentActivityState[] = []; - const androidStates: Array = []; - const appleTimes: string[] = []; - let fail = false; - const dependencies = Layer.mergeAll( - Layer.succeed( - AgentActivityRows.AgentActivityRows, - makeAgentActivityRows({ - upsert: (input) => - Effect.sync(() => { - writes.push(input.state); - return { ...input.state, updatedAt: persisted.updatedAt }; - }), - }), - ), - Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), - Layer.succeed( - LiveActivities.LiveActivities, - makeLiveActivities({ - listTargets: () => - Effect.succeed([ - target("ios"), - { ...target("android"), platform: "android" as const }, - ]), + it.effect("fans out the persisted phase time to iOS and Android", () => { + const persisted = { ...state, phase: "waiting_for_input" as const }; + const renamed = { + ...persisted, + threadTitle: "Renamed", + updatedAt: "1970-01-01T00:01:00.000Z", + }; + const androidStates: Array = []; + const appleTimes: string[] = []; + const dependencies = Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ + upsert: (input) => Effect.succeed({ ...input.state, updatedAt: persisted.updatedAt }), + }), + ), + Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ + listTargets: () => + Effect.succeed([target("ios"), { ...target("android"), platform: "android" as const }]), + }), + ), + Layer.succeed(FcmDeliveries.FcmDeliveries, { + enqueue: (input) => + Effect.sync(() => { + androidStates.push(input.state); + return null; }), - ), - Layer.succeed(FcmDeliveries.FcmDeliveries, { - enqueue: (input) => + process: () => Effect.void, + }), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendForTarget: (input) => Effect.sync(() => { - androidStates.push(input.state); + appleTimes.push(input.aggregate!.activities[0]!.updatedAt); return null; }), - process: () => Effect.void, }), - Layer.succeed( - ApnsDeliveries.ApnsDeliveries, - makeApnsDeliveries({ - sendForTarget: (input) => - fail - ? Effect.die("offline") - : Effect.sync(() => { - appleTimes.push(input.aggregate!.activities[0]!.updatedAt); - return null; - }), - }), - ), - ); - return Effect.gen(function* () { - const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; - const publish = publisher.publish({ - environmentId: state.environmentId, - environmentPublicKey: "key", - threadId: state.threadId, - state: renamed, - }); - yield* publish; - expect(appleTimes).toEqual([persisted.updatedAt]); - expect(androidStates).toMatchObject([ - { threadTitle: "Renamed", updatedAt: persisted.updatedAt }, - ]); - fail = true; - expect((yield* Effect.exit(publish))._tag).toBe("Failure"); - expect(writes).toEqual([renamed, renamed]); - }).pipe(Effect.provide(AgentActivityPublisher.layer.pipe(Layer.provide(dependencies)))); - }, - ); + ), + ); + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + yield* publisher.publish({ + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: renamed, + }); + expect(appleTimes).toEqual([persisted.updatedAt]); + expect(androidStates).toMatchObject([ + { threadTitle: "Renamed", updatedAt: persisted.updatedAt }, + ]); + }).pipe(Effect.provide(AgentActivityPublisher.layer.pipe(Layer.provide(dependencies)))); + }); it.effect("routes Android publication and registration replay to FCM alongside iOS", () => { const android = { ...target("android"), platform: "android" as const, ios_major_version: null }; From 8d9f6ec83620f06f43ab17c1f5dd2724ffe63c3e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 21:40:04 +0000 Subject: [PATCH 7/9] fix(relay): store the server's phase time instead of carrying it over The relay upsert kept the stored updatedAt whenever the phase matched, which overwrote the fresh time the server sends for a subagent's next request. Android then rebuilt an already-seen alert_id and iOS reused a completed job claim, so the new request never alerted. The server already carries the time across same-phase publishes like renames and knows about subagent requests, so the relay goes back to storing what it receives. AI-assisted; verified with the tests and tools listed above. --- .../AgentActivityPublisher.test.ts | 60 +------------------ .../agentActivity/AgentActivityPublisher.ts | 9 ++- .../src/agentActivity/AgentActivityRows.ts | 12 ++-- .../src/agentActivity/ApnsDeliveries.test.ts | 2 +- .../src/agentActivity/FcmDeliveries.test.ts | 2 +- .../agentActivity/MobileRegistrations.test.ts | 2 +- 6 files changed, 12 insertions(+), 75 deletions(-) diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 793a2d2399da4..33f1518ec30ae 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -71,7 +71,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: ({ state }) => Effect.succeed(state), + upsert: () => Effect.void, remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => Effect.succeed([state]), @@ -140,64 +140,6 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { - it.effect("fans out the persisted phase time to iOS and Android", () => { - const persisted = { ...state, phase: "waiting_for_input" as const }; - const renamed = { - ...persisted, - threadTitle: "Renamed", - updatedAt: "1970-01-01T00:01:00.000Z", - }; - const androidStates: Array = []; - const appleTimes: string[] = []; - const dependencies = Layer.mergeAll( - Layer.succeed( - AgentActivityRows.AgentActivityRows, - makeAgentActivityRows({ - upsert: (input) => Effect.succeed({ ...input.state, updatedAt: persisted.updatedAt }), - }), - ), - Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), - Layer.succeed( - LiveActivities.LiveActivities, - makeLiveActivities({ - listTargets: () => - Effect.succeed([target("ios"), { ...target("android"), platform: "android" as const }]), - }), - ), - Layer.succeed(FcmDeliveries.FcmDeliveries, { - enqueue: (input) => - Effect.sync(() => { - androidStates.push(input.state); - return null; - }), - process: () => Effect.void, - }), - Layer.succeed( - ApnsDeliveries.ApnsDeliveries, - makeApnsDeliveries({ - sendForTarget: (input) => - Effect.sync(() => { - appleTimes.push(input.aggregate!.activities[0]!.updatedAt); - return null; - }), - }), - ), - ); - return Effect.gen(function* () { - const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; - yield* publisher.publish({ - environmentId: state.environmentId, - environmentPublicKey: "key", - threadId: state.threadId, - state: renamed, - }); - expect(appleTimes).toEqual([persisted.updatedAt]); - expect(androidStates).toMatchObject([ - { threadTitle: "Renamed", updatedAt: persisted.updatedAt }, - ]); - }).pipe(Effect.provide(AgentActivityPublisher.layer.pipe(Layer.provide(dependencies)))); - }); - it.effect("routes Android publication and registration replay to FCM alongside iOS", () => { const android = { ...target("android"), platform: "android" as const, ios_major_version: null }; const ios = target("ios"); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index f5c977bbd8134..6ef258acdb6d3 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -206,12 +206,11 @@ const make = Effect.gen(function* () { "relay.thread_id": input.threadId, "relay.agent_activity.phase": input.state?.phase ?? "deleted", }); - let state = input.state; - if (state) { + if (input.state) { // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones. - state = yield* rows.upsert({ + yield* rows.upsert({ environmentPublicKey: input.environmentPublicKey, - state, + state: input.state, }); } else { yield* rows.remove({ @@ -231,7 +230,7 @@ const make = Effect.gen(function* () { (deliveryUser) => publishForDeliveryUser({ deliveryUser, - state, + state: input.state, nowMs: now.epochMilliseconds, }), { concurrency: 4 }, diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index 9f0fd8d41d0c1..0faf5b9da358e 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -65,11 +65,10 @@ export class AgentActivityRowListPersistenceError extends Schema.TaggedError Effect.Effect; + }) => Effect.Effect; readonly pruneTerminal: (input: { readonly updatedBefore: string; }) => Effect.Effect; @@ -122,8 +121,7 @@ const make = Effect.gen(function* () { }), ), ); - const samePhase = sql`${relayAgentActivityRows.stateJson} ->> 'phase' = excluded.state_json ->> 'phase'`; - const [stored] = yield* db + yield* db .insert(relayAgentActivityRows) .values({ environmentId: input.state.environmentId, @@ -140,11 +138,10 @@ const make = Effect.gen(function* () { relayAgentActivityRows.threadId, ], set: { - stateJson: sql`case when ${samePhase} then excluded.state_json || jsonb_build_object('updatedAt', ${relayAgentActivityRows.stateJson} ->> 'updatedAt') else excluded.state_json end`, - updatedAt: sql`case when ${samePhase} then ${relayAgentActivityRows.updatedAt} else excluded.updated_at end`, + stateJson, + updatedAt: input.state.updatedAt, }, }) - .returning({ stateJson: relayAgentActivityRows.stateJson }) .pipe( Effect.mapError( (cause) => @@ -155,7 +152,6 @@ const make = Effect.gen(function* () { }), ), ); - return stored!.stateJson; }), remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) { diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index fc0f3638e4542..9a6015a1c2fd3 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -164,7 +164,7 @@ function layerFor(input: { Layer.provide( Layer.mergeAll( Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: ({ state }) => Effect.succeed(state), + upsert: () => Effect.void, remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index d0b86c835faab..f1ce50d5832b6 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -130,7 +130,7 @@ function harness() { invalidateDeliveryToken: () => Effect.void, }), Layer.succeed(AgentActivityRows.AgentActivityRows, { - upsert: ({ state }) => Effect.succeed(state), + upsert: () => Effect.void, remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index ccadc8805ffb3..1865a633d5715 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -79,7 +79,7 @@ function makeAgentActivityRows( overrides: Partial = {}, ): AgentActivityRows.AgentActivityRows["Service"] { return { - upsert: ({ state }) => Effect.succeed(state), + upsert: () => Effect.void, remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => { From 8cd1de1121387fc60fc0907af1e5ffdcc7c6e1f9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 00:01:18 +0000 Subject: [PATCH 8/9] fix(server): a coalesced new run's completion keeps its own alert time Carry a published phase time over only when the run and the thread's own request are unchanged, so queued updates that coalesce a new run into the same phase still alert. Co-Authored-By: Claude Opus 5.5 --- .../src/relay/AgentAwarenessRelay.test.ts | 15 ++++++++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 20 ++++++++++++++++--- 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index b5ca7394206be..9c78900d2325d 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -546,6 +546,21 @@ describe("AgentAwarenessRelay", () => { ["completed", "Renamed thread", "2026-09-04T12:01:00.000Z"], ], ); + + // A new run that completed before the worker caught up is a new completion, not a rename. + yield* Ref.set( + currentShell, + shell({ + status: "completed", + title: "Renamed again", + latestRunId: RunId.make("run:next"), + updatedAt: DateTime.makeUnsafe("2026-09-04T13:05:00.000Z"), + }), + ); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications[3]?.state?.updatedAt, "2026-09-04T13:05:00.000Z"); }), ); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 4b130528093dd..f3b5524412954 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -177,6 +177,14 @@ interface PublishedAwareness { readonly identity: string; readonly state: RelayAgentActivityState | null; readonly childDriven: boolean; + /** The run and own request behind the state; queued updates can coalesce a whole new run. */ + readonly occurrence: string | null; +} + +function awarenessOccurrence(thread: Option.Option): string | null { + return Option.isSome(thread) + ? JSON.stringify([thread.value.latestRunId, thread.value.pendingRuntimeRequest?.id ?? null]) + : null; } // The relay treats a new time as a new alert, so same-phase changes like renames keep the old one. @@ -184,8 +192,11 @@ interface PublishedAwareness { function withPublishedPhaseTime( state: RelayAgentActivityState, published: PublishedAwareness | undefined, + occurrence: string | null, ): RelayAgentActivityState { - return published?.state?.phase === state.phase && !published.childDriven + return published?.state?.phase === state.phase && + !published.childDriven && + published.occurrence === occurrence ? { ...state, updatedAt: published.state.updatedAt } : state; } @@ -594,6 +605,7 @@ export const make = Effect.gen(function* () { const publishIdentity = agentAwarenessPublishIdentity(snapshot.state, snapshot.subagentWait); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); const published = publishedStateByThread.get(threadId); + const occurrence = awarenessOccurrence(thread); if ( (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") && !hasLivePublishedState(publishedStateByThread, threadId) @@ -661,7 +673,7 @@ export const make = Effect.gen(function* () { }) : null; if (unsettled?.state?.phase === "completed") { - const completion = withPublishedPhaseTime(unsettled.state, published); + const completion = withPublishedPhaseTime(unsettled.state, published, occurrence); yield* publishState({ projectId: unsettled.projectId, state: completion, @@ -674,6 +686,7 @@ export const make = Effect.gen(function* () { identity: agentAwarenessPublishIdentity(completion), state: completion, childDriven: false, + occurrence, }), ); yield* schedulePublishConfirm(threadId); @@ -703,7 +716,7 @@ export const make = Effect.gen(function* () { snapshot.state === null ? null : childRequestId === undefined - ? withPublishedPhaseTime(snapshot.state, published) + ? withPublishedPhaseTime(snapshot.state, published, occurrence) : childAlert?.requestId === childRequestId ? { ...snapshot.state, updatedAt: childAlert.updatedAt } : snapshot.state; @@ -722,6 +735,7 @@ export const make = Effect.gen(function* () { identity: publishIdentity, state, childDriven: childRequestId !== undefined, + occurrence, }), ); }); From 2b452b8fbcd974477d7495f3b4a2bea02adf8f5c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 23:41:52 +0000 Subject: [PATCH 9/9] test(relay): assert alert job ids in enqueue order Co-Authored-By: Claude Opus 5.5 --- infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts index ed238acd43c1f..b97d9c401ce70 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts @@ -71,10 +71,10 @@ describe("ApnsDeliveryQueue", () => { notification: { ...input.notification, updatedAt: "2026-09-04T13:00:00.000Z" }, }); yield* queue.enqueuePushNotification({ ...input, token: "rotated-token" }); - expect(sent[4]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); expect(sent[1]!.payload.jobId).toBe(sent[0]!.payload.jobId); expect(sent[2]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); expect(sent[3]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); + expect(sent[4]!.payload.jobId).not.toBe(sent[0]!.payload.jobId); expect(sent[1]!.payload.notification?.title).toBe("Renamed"); }).pipe(Effect.provide(layer)); });