From ba371bd9a151c53954aef996a75bb05c84f8d2f2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 14 Jul 2026 10:27:19 +0200 Subject: [PATCH] Skip stale working-task notifications - Stop publishing start-intent awareness snapshots - Recheck queued APNs jobs against current activity state --- .../src/relay/AgentAwarenessRelay.test.ts | 13 +- apps/server/src/relay/AgentAwarenessRelay.ts | 8 +- .../AgentActivityPublisher.test.ts | 1 + .../agentActivity/AgentActivityRows.test.ts | 9 + .../src/agentActivity/AgentActivityRows.ts | 53 +++++ .../src/agentActivity/ApnsDeliveries.test.ts | 211 +++++++++++++++++- .../relay/src/agentActivity/ApnsDeliveries.ts | 112 ++++++++++ .../agentActivity/MobileRegistrations.test.ts | 1 + .../src/agentActivity/apnsDeliveryJobs.ts | 11 +- 9 files changed, 406 insertions(+), 13 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 40ed694723d..9ca17dbc2dd 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -137,7 +137,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { expect(AgentAwarenessRelay.eventThreadId(event)).toBe(threadId); }); - it("does not publish streaming content or non-awareness activity events", () => { + it("does not publish start intents, streaming content, or non-awareness activity events", () => { const now = "2026-05-25T00:00:00.000Z"; const base = { sequence: 1, @@ -191,7 +191,16 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { streaming: false, }, } as unknown as OrchestrationEvent), - ).toBe(true); + ).toBe(false); + expect( + AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ + ...base, + type: "thread.turn-start-requested", + payload: { + threadId: "thread-1" as ThreadId, + }, + } as unknown as OrchestrationEvent), + ).toBe(false); }); it("deduplicates awareness state updates whose only change is their event timestamp", () => { diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index d4bcd2096f1..58de98f1ca1 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -67,7 +67,13 @@ export function eventThreadId(event: OrchestrationEvent): ThreadId | null { export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boolean { switch (event.type) { case "thread.message-sent": - return !event.payload.streaming; + case "thread.turn-start-requested": + // These events express intent to start work, but the shell still contains + // the previous turn's terminal state until the provider acknowledges the + // new turn. Publishing that snapshot can queue a fresh "Done" alert just + // before the real running state arrives. Provider lifecycle events publish + // the authoritative starting/running state instead. + return false; case "thread.proposed-plan-upserted": case "thread.runtime-mode-set": case "thread.interaction-mode-set": diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 1fcc9d0cb99..4215f834654 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -64,6 +64,7 @@ function makeAgentActivityRows( remove: () => Effect.void, pruneTerminal: () => Effect.void, listForUser: () => Effect.succeed([state]), + getForUserThread: () => Effect.succeed(state), ...overrides, }; } diff --git a/infra/relay/src/agentActivity/AgentActivityRows.test.ts b/infra/relay/src/agentActivity/AgentActivityRows.test.ts index be976d16bbb..65b0c2a0a90 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.test.ts @@ -75,6 +75,15 @@ describe("AgentActivityRows", () => { const listError = yield* rows.listForUser({ userId: "user-2" }).pipe(Effect.flip); expect(listError).toMatchObject({ userId: "user-2", cause }); expect(listError.message).toBe("Failed to list agent activity state for user user-2."); + + const getError = yield* rows + .getForUserThread({ + userId: "user-2", + environmentId: "env-1", + threadId: "thread-1", + }) + .pipe(Effect.flip); + expect(getError).toMatchObject({ userId: "user-2", cause }); }).pipe( Effect.provide( AgentActivityRows.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, failingDb))), diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index 63868ad2d72..14245075e10 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -83,6 +83,11 @@ export class AgentActivityRows extends Context.Service< ReadonlyArray, AgentActivityRowListPersistenceError >; + readonly getForUserThread: (input: { + readonly userId: string; + readonly environmentId: string; + readonly threadId: string; + }) => Effect.Effect; } >()("t3code-relay/agentActivity/AgentActivityRows") {} @@ -241,6 +246,54 @@ export const make = Effect.gen(function* () { ), ); }), + + getForUserThread: Effect.fn("relay.agent_activity_rows.get_for_user_thread")(function* (input) { + return yield* db + .select({ stateJson: relayAgentActivityRows.stateJson }) + .from(relayAgentActivityRows) + .innerJoin( + relayEnvironmentLinks, + and( + eq(relayEnvironmentLinks.environmentId, relayAgentActivityRows.environmentId), + eq( + relayEnvironmentLinks.environmentPublicKey, + relayAgentActivityRows.environmentPublicKey, + ), + ), + ) + .where( + and( + eq(relayEnvironmentLinks.userId, input.userId), + isNull(relayEnvironmentLinks.revokedAt), + eq(relayAgentActivityRows.environmentId, input.environmentId), + eq(relayAgentActivityRows.threadId, input.threadId), + ), + ) + .orderBy(desc(relayAgentActivityRows.updatedAt)) + .pipe( + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => encodeJsonValue(row.stateJson), { + concurrency: "unbounded", + }), + ), + Effect.map((rows) => { + for (const row of rows) { + const decoded = decodeRelayAgentActivityStateJson(row); + if (Option.isSome(decoded)) { + return decoded.value; + } + } + return null; + }), + Effect.mapError( + (cause) => + new AgentActivityRowListPersistenceError({ + userId: input.userId, + cause, + }), + ), + ); + }), }); }); diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index cdafc668269..1c89b56da5c 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -5,7 +5,6 @@ import type { import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto"; import { describe, expect, it } from "@effect/vitest"; import * as NodeCrypto from "node:crypto"; -import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; @@ -161,10 +160,13 @@ function makeLayer(input: { >; readonly currentTargets?: ReadonlyArray; readonly config?: RelayConfiguration.RelayConfiguration["Service"]; - // Live agent-activity rows returned by the delivery-time start recheck. - // Defaults to one freshly-updated running thread so queued starts stay - // deliverable in tests that don't care about the recheck. + // Live agent-activity rows returned by delivery-time state rechecks. + // Defaults to the fixture row so queued updates match unless a test is + // explicitly exercising stale-state behavior. readonly activityStates?: ReadonlyArray; + // Current per-thread rows used to reject queued updates/notifications that + // have been superseded before APNs delivery. + readonly currentActivityStates?: ReadonlyArray; readonly execute?: ( request: HttpClientRequest.HttpClientRequest, ) => Effect.Effect; @@ -182,9 +184,14 @@ function makeLayer(input: { listForUser: () => input.activityStates !== undefined ? Effect.succeed([...input.activityStates]) - : DateTime.now.pipe( - Effect.map((now) => [{ ...state, updatedAt: DateTime.formatIso(now) }]), - ), + : Effect.succeed([state]), + getForUserThread: ({ environmentId, threadId }) => + Effect.succeed( + (input.currentActivityStates ?? input.activityStates ?? [state]).find( + (current) => + current.environmentId === environmentId && current.threadId === threadId, + ) ?? null, + ), } satisfies AgentActivityRows.AgentActivityRows["Service"]), Layer.succeed(ApnsDeliveryQueue.ApnsDeliveryQueueSender, { send: (body) => @@ -674,6 +681,8 @@ describe("ApnsDeliveries", () => { environmentId: "env", threadId: "thread", deepLink: "/", + phase: "waiting_for_input", + updatedAt: state.updatedAt, }, }, }, @@ -778,6 +787,30 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("does not queue a push notification when a thread starts working", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + push_token: "apns-device-token", + push_to_start_token: null, + activity_push_token: null, + remote_started_at: null, + }, + aggregate, + nowMs: 5_000, + }); + + expect(result).toBeNull(); + expect(queuedJobs).toEqual([]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs }))); + }); + it.effect("queues bounded alert notification payloads", () => { const attempts: Array = []; const queuedJobs: Array = []; @@ -1176,6 +1209,150 @@ describe("ApnsDeliveries", () => { ); }); + it.effect("skips a queued Done Live Activity update after the thread resumes working", () => { + const attempts: Array = []; + let executeCount = 0; + const completedAt = "1970-01-01T00:00:01.000Z"; + const completedAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activeCount: 0, + updatedAt: completedAt, + activities: [ + { + ...aggregate.activities[0]!, + phase: "completed", + status: "Done", + updatedAt: completedAt, + }, + ], + }; + const payload = makeApnsDeliveryJobPayload({ + kind: "live_activity_update", + userId: target.user_id, + deviceId: target.device_id, + token: target.activity_push_token ?? "activity-token", + aggregate: completedAggregate, + alert: { title: "Thread", body: "Done: Project" }, + createdAt: completedAt, + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-update-superseded-by-running", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + executeCount += 1; + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.processSignedJob(signed); + + expect(result).toMatchObject({ + kind: "live_activity_update", + ok: true, + apnsStatus: null, + apnsReason: "Stale APNs delivery job skipped.", + }); + expect(executeCount).toBe(0); + expect(attempts).toMatchObject([ + { + sourceJobId: "job-update-superseded-by-running", + apnsReason: "Stale agent activity state skipped.", + }, + ]); + }).pipe( + Effect.provide( + makeLayer({ + attempts, + activityStates: [ + { + ...state, + phase: "running", + headline: "Running", + updatedAt: "1970-01-01T00:00:02.000Z", + }, + ], + config: signingConfig, + execute, + }), + ), + ); + }); + + it.effect("skips a queued Done notification after the thread resumes working", () => { + const attempts: Array = []; + let executeCount = 0; + const completedAt = "1970-01-01T00:00:01.000Z"; + const payload = makeApnsDeliveryJobPayload({ + kind: "push_notification", + userId: target.user_id, + deviceId: target.device_id, + token: "apns-device-token", + aggregate: null, + notification: { + title: "Thread", + body: "Done: Project", + environmentId: "env", + threadId: "thread", + deepLink: "/", + phase: "completed", + updatedAt: completedAt, + }, + createdAt: completedAt, + expiresAt: "1970-01-01T00:10:00.000Z", + jobId: "job-push-superseded-by-running", + }); + const signed = signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload, + }); + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + executeCount += 1; + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.processSignedJob(signed); + + expect(result).toMatchObject({ + kind: "push_notification", + ok: true, + apnsStatus: null, + apnsReason: "Stale APNs delivery job skipped.", + }); + expect(executeCount).toBe(0); + expect(attempts).toMatchObject([ + { + sourceJobId: "job-push-superseded-by-running", + apnsReason: "Stale agent activity state skipped.", + }, + ]); + }).pipe( + Effect.provide( + makeLayer({ + attempts, + currentTargets: [{ ...target, push_token: "apns-device-token" }], + currentActivityStates: [ + { + ...state, + phase: "running", + headline: "Running", + updatedAt: "1970-01-01T00:00:02.000Z", + }, + ], + config: signingConfig, + execute, + }), + ), + ); + }); + it.effect("retries signed queue jobs that are already claimed but not completed", () => { const attempts: Array = []; let executeCount = 0; @@ -1327,7 +1504,15 @@ describe("ApnsDeliveries", () => { deviceId: target.device_id, }, ]); - }).pipe(Effect.provide(makeLayer({ attempts, clearedStarts }))); + }).pipe( + Effect.provide( + makeLayer({ + attempts, + clearedStarts, + activityStates: [{ ...state, updatedAt: "9999-01-01T00:00:00.000Z" }], + }), + ), + ); }); it.effect("invalidates dead push-to-start tokens after permanent APNs start failures", () => { @@ -1373,7 +1558,15 @@ describe("ApnsDeliveries", () => { }, ]); }).pipe( - Effect.provide(makeLayer({ attempts, invalidatedTokens, config: signingConfig, execute })), + Effect.provide( + makeLayer({ + attempts, + invalidatedTokens, + activityStates: [{ ...state, updatedAt: "9999-01-01T00:00:00.000Z" }], + config: signingConfig, + execute, + }), + ), ); }); diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 9b86ed46417..9db9be3f168 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -369,6 +369,8 @@ function notificationForAggregate(input: { environmentId: activity.environmentId, threadId: activity.threadId, deepLink: activity.deepLink, + phase: activity.phase, + updatedAt: activity.updatedAt, }; } @@ -722,6 +724,87 @@ export const make = Effect.gen(function* () { ); }); + const stateIdentityIsCurrent = Effect.fnUntraced(function* (input: { + readonly userId: string; + readonly environmentId: string; + readonly threadId: string; + readonly phase: RelayAgentActivityAggregateState["activities"][number]["phase"]; + readonly updatedAt: string; + }) { + return yield* activityRows + .getForUserThread({ + userId: input.userId, + environmentId: input.environmentId, + threadId: input.threadId, + }) + .pipe( + Effect.map( + (current) => + current !== null && + current.phase === input.phase && + current.updatedAt === input.updatedAt, + ), + // A transient persistence failure must not permanently discard a + // legitimate alert. Fail open and let the signed job's retry/dedupe + // protections handle transport failures as usual. + Effect.catchCause((cause) => + Effect.logWarning("agent-activity state recheck failed; allowing queued delivery", { + cause, + environmentId: input.environmentId, + threadId: input.threadId, + }).pipe(Effect.as(true)), + ), + ); + }); + + const aggregateRowsAreCurrent = Effect.fnUntraced(function* (input: { + readonly userId: string; + readonly aggregate: RelayAgentActivityAggregateState; + }) { + return yield* activityRows.listForUser({ userId: input.userId }).pipe( + Effect.map((currentStates) => { + const currentByThread = new Map( + currentStates.map((current) => [ + `${current.environmentId}\u0000${current.threadId}`, + current, + ]), + ); + return input.aggregate.activities.every((row) => { + const current = currentByThread.get(`${row.environmentId}\u0000${row.threadId}`); + return ( + current !== undefined && + current.phase === row.phase && + current.updatedAt === row.updatedAt + ); + }); + }), + Effect.catchCause((cause) => + Effect.logWarning("agent-activity aggregate recheck failed; allowing queued delivery", { + cause, + userId: input.userId, + }).pipe(Effect.as(true)), + ), + ); + }); + + const notificationStateIsCurrent = Effect.fnUntraced(function* (input: { + readonly userId: string; + readonly notification: ApnsNotificationPayload; + }) { + // Jobs from older relay versions do not carry a state identity. Preserve + // backwards compatibility and only revalidate newly queued jobs. + if (input.notification.phase === undefined || input.notification.updatedAt === undefined) { + return true; + } + return yield* stateIdentityIsCurrent({ + userId: input.userId, + environmentId: input.notification.environmentId, + threadId: input.notification.threadId, + phase: input.notification.phase, + updatedAt: input.notification.updatedAt, + }); + }); + const isCurrentSignedJobToken = Effect.fnUntraced(function* (input: { readonly target: LiveActivityDeliveryTarget; readonly kind: RelayDeliveryKind; @@ -791,6 +874,20 @@ export const make = Effect.gen(function* () { }); return staleJobResult({ deviceId: input.target.device_id, kind: input.kind }); } + if ( + input.kind !== "live_activity_start" && + aggregate !== null && + !(yield* aggregateRowsAreCurrent({ + userId: input.target.user_id, + aggregate, + })) + ) { + yield* attempts.completeSourceJob({ + sourceJobId: input.sourceJobId, + apnsReason: "Stale agent activity state skipped.", + }); + return staleJobResult({ deviceId: input.target.device_id, kind: input.kind }); + } } if ( input.kind === "live_activity_start" && @@ -930,6 +1027,21 @@ export const make = Effect.gen(function* () { kind: "push_notification", }); } + if ( + !(yield* notificationStateIsCurrent({ + userId: input.target.user_id, + notification, + })) + ) { + yield* attempts.completeSourceJob({ + sourceJobId: input.sourceJobId, + apnsReason: "Stale agent activity state skipped.", + }); + return staleJobResult({ + deviceId: input.target.device_id, + kind: "push_notification", + }); + } } const result = yield* apns .sendPushNotificationRequest({ diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 1f1256e3567..ca39484373e 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -86,6 +86,7 @@ function makeAgentActivityRows( }; return Effect.succeed([activeState]); }, + getForUserThread: () => Effect.succeed(null), ...overrides, }; } diff --git a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts index 027b17bd5e0..256ca3edb5c 100644 --- a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts +++ b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts @@ -1,6 +1,10 @@ import * as NodeCrypto from "node:crypto"; -import { RelayAgentActivityAggregateState, type RelayDeliveryKind } from "@t3tools/contracts/relay"; +import { + RelayAgentActivityAggregateState, + RelayAgentAwarenessPhase, + type RelayDeliveryKind, +} from "@t3tools/contracts/relay"; import { stableStringify } from "@t3tools/shared/relaySigning"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; @@ -38,6 +42,11 @@ export const ApnsNotificationPayload = Schema.Struct({ environmentId: Schema.String, threadId: Schema.String, deepLink: Schema.String, + // Optional so delivery jobs queued by older relay builds still decode. + // New jobs use these fields to avoid delivering a stale Done/attention + // notification after the thread has moved to another phase. + phase: Schema.optional(RelayAgentAwarenessPhase), + updatedAt: Schema.optional(Schema.String), }); export type ApnsNotificationPayload = typeof ApnsNotificationPayload.Type;