diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 68e2289524c09..59119e80ca183 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -369,12 +369,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"); }), ); @@ -658,6 +673,52 @@ 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"], + ], + ); + + // 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"); + }), + ); + 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 76f854548deb6..7bac5b3395fb2 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -179,15 +179,40 @@ function agentAwarenessPublishIdentity( ); } -const CLEARED_PUBLISH_IDENTITY = agentAwarenessPublishIdentity(null); +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. +// A child's wait keeps its own per-request time instead, so each new child request still alerts. +function withPublishedPhaseTime( + state: RelayAgentActivityState, + published: PublishedAwareness | undefined, + occurrence: string | null, +): RelayAgentActivityState { + return published?.state?.phase === state.phase && + !published.childDriven && + published.occurrence === occurrence + ? { ...state, updatedAt: published.state.updatedAt } + : state; +} // A published tombstone stays recorded so repeats are skipped, but it is not live state. function hasLivePublishedState( - publishedStateByThread: ReadonlyMap, + publishedStateByThread: ReadonlyMap, threadId: ThreadId, ): boolean { - const identity = publishedStateByThread.get(threadId); - return identity !== undefined && identity !== CLEARED_PUBLISH_IDENTITY; + return (publishedStateByThread.get(threadId)?.state ?? null) !== null; } function resolveAgentActivityPublishingStartupState(input: { @@ -414,7 +439,7 @@ export const make = Effect.gen(function* () { const scope = yield* Effect.scope; const startedAt = (yield* DateTime.now).epochMilliseconds; const catchUpRequests = yield* Queue.dropping(1); - const publishedStateByThreadRef = yield* Ref.make(new Map()); + const publishedStateByThreadRef = yield* Ref.make(new Map()); // Android keys alerts by phase and time, so a child's unchanged request keeps // its first alert time through parent renames, relinks and the parent's own requests. const childAlertByThread = new Map(); @@ -585,6 +610,8 @@ 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) @@ -592,7 +619,7 @@ export const make = Effect.gen(function* () { if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return; } if ( - publishedStateByThread.get(threadId) === publishIdentity && + published?.identity === publishIdentity && // A held completion settled after an earlier clear still has to alert. !(snapshot.state === null && heldCompletions.has(threadId)) ) { @@ -652,7 +679,7 @@ export const make = Effect.gen(function* () { }) : null; if (unsettled?.state?.phase === "completed") { - const completion = unsettled.state; + const completion = withPublishedPhaseTime(unsettled.state, published, occurrence); yield* publishState({ projectId: unsettled.projectId, state: completion, @@ -661,7 +688,12 @@ 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, { + identity: agentAwarenessPublishIdentity(completion), + state: completion, + childDriven: false, + occurrence, + }), ); yield* schedulePublishConfirm(threadId); return; @@ -687,11 +719,13 @@ export const make = Effect.gen(function* () { const childRequestId = snapshot.subagentWait?.requestId; const childAlert = childAlertByThread.get(threadId); const state = - snapshot.state !== null && - childRequestId !== undefined && - childAlert?.requestId === childRequestId - ? { ...snapshot.state, updatedAt: childAlert.updatedAt } - : snapshot.state; + snapshot.state === null + ? null + : childRequestId === undefined + ? withPublishedPhaseTime(snapshot.state, published, occurrence) + : childAlert?.requestId === childRequestId + ? { ...snapshot.state, updatedAt: childAlert.updatedAt } + : snapshot.state; yield* publishState({ projectId: snapshot.projectId, state, @@ -703,7 +737,12 @@ export const make = Effect.gen(function* () { publishConfirmDeadlines.delete(threadId); heldCompletions.delete(threadId); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => - new Map(publishedStates).set(threadId, publishIdentity), + new Map(publishedStates).set(threadId, { + identity: publishIdentity, + state, + childDriven: childRequestId !== undefined, + occurrence, + }), ); }); diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index e9d2ded1b2862..fcf5e769f9d48 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -684,6 +684,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(yield* 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("processes signed push notification jobs through APNs and records attempts", () => { const attempts: Array = []; const payload = makeApnsDeliveryJobPayload({ diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 4f19fa3ecee05..5f61db5a35381 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -23,6 +23,7 @@ import { isExpiredAgentActivityState, notificationForActivity, sanitizeAgentActivityAggregateState, + sanitizeAgentActivityAggregateRow, sanitizeApnsNotificationPayload, } from "./agentActivityPayloads.ts"; import { makeAggregateState, statusForPhase } from "./agentActivityAggregate.ts"; @@ -572,10 +573,15 @@ 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 ); }); }), diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts index 358698dd197b3..b97d9c401ce70 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[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)); + }); + 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 09cff5184b058..8af93ec97b26c 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts @@ -1,5 +1,6 @@ import * as Alchemy from "alchemy"; import * as Cloudflare from "@alchemy.run/cloudflare"; +import * as Digest from "effect/crypto/Digest"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -19,6 +20,7 @@ import { } from "./agentActivityPayloads.ts"; import { expiresAtForJob, + notificationJobIdentity, makeApnsDeliveryJobPayload, signApnsDeliveryJob, type ApnsDeliveryJobPayload, @@ -80,6 +82,7 @@ export class ApnsDeliveryQueue extends Context.Service< const make = Effect.gen(function* () { const sender = yield* ApnsDeliveryQueueSender; const random = yield* SecureRandom.SecureRandom; + const digest = yield* Digest.Digest; const config = yield* RelayConfiguration.RelayConfiguration; const mac = yield* Mac.Mac; @@ -150,19 +153,23 @@ const make = Effect.gen(function* () { "relay.thread_id": input.notification.threadId, }); const now = yield* DateTime.now; - const jobId = yield* random.uuidV4().pipe( - Effect.mapError( - (cause) => - new ApnsDeliveryQueueSendError({ - operation: "generate-job-id", - jobId: null, - kind: "push_notification", - userId: input.userId, - deviceId: input.deviceId, - cause, - }), - ), - ); + const identity = notificationJobIdentity(input); + const jobId = + identity !== null + ? yield* digest.digestHex(Digest.sha256, new TextEncoder().encode(identity)) + : yield* random.uuidV4().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/agentActivityPayloads.ts b/infra/relay/src/agentActivity/agentActivityPayloads.ts index 4195c81171803..dc03bf17dcbbf 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 { diff --git a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts index b732507a731a5..b030486d9dff8 100644 --- a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts +++ b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts @@ -40,6 +40,30 @@ export const ApnsNotificationPayload = Schema.Struct({ }); export type ApnsNotificationPayload = typeof ApnsNotificationPayload.Type; +// One alert per target and phase time, so retries and same-phase renames reuse the delivery claim. +export function notificationJobIdentity(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 JSON.stringify([ + input.userId, + input.deviceId, + input.token, + input.bundleId ?? null, + input.apsEnvironment ?? null, + notification.environmentId, + notification.threadId, + notification.phase, + notification.updatedAt, + ]); +} + export const ApnsLiveActivityAlert = Schema.Struct({ title: Schema.String, body: Schema.String,