authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-07-14 18:00:16+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-14 18:00:16+02:00
log735240f3162470032120603474605b8a6e42580b
tree9186fdac724db040037f7b1aab1fa284f78133e8
parent8de254a685f4356c719e744926397362c800f426
signature Signed by PGP key B5690EEEBB952194

Skip stale working-task notifications (#3961)

9 files changed, 406 insertions(+), 13 deletions(-)

apps/server/src/relay/AgentAwarenessRelay.test.ts+11-2
......@@ -137,7 +137,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
137137 expect(AgentAwarenessRelay.eventThreadId(event)).toBe(threadId);
138138 });
139139
140 it("does not publish streaming content or non-awareness activity events", () => {
140 it("does not publish start intents, streaming content, or non-awareness activity events", () => {
141141 const now = "2026-05-25T00:00:00.000Z";
142142 const base = {
143143 sequence: 1,
......@@ -191,7 +191,16 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
191191 streaming: false,
192192 },
193193 } as unknown as OrchestrationEvent),
194 ).toBe(true);
194 ).toBe(false);
195 expect(
196 AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({
197 ...base,
198 type: "thread.turn-start-requested",
199 payload: {
200 threadId: "thread-1" as ThreadId,
201 },
202 } as unknown as OrchestrationEvent),
203 ).toBe(false);
195204 });
196205
197206 it("deduplicates awareness state updates whose only change is their event timestamp", () => {
apps/server/src/relay/AgentAwarenessRelay.ts+7-1
......@@ -67,7 +67,13 @@ export function eventThreadId(event: OrchestrationEvent): ThreadId | null {
6767export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boolean {
6868 switch (event.type) {
6969 case "thread.message-sent":
70 return !event.payload.streaming;
70 case "thread.turn-start-requested":
71 // These events express intent to start work, but the shell still contains
72 // the previous turn's terminal state until the provider acknowledges the
73 // new turn. Publishing that snapshot can queue a fresh "Done" alert just
74 // before the real running state arrives. Provider lifecycle events publish
75 // the authoritative starting/running state instead.
76 return false;
7177 case "thread.proposed-plan-upserted":
7278 case "thread.runtime-mode-set":
7379 case "thread.interaction-mode-set":
infra/relay/src/agentActivity/AgentActivityPublisher.test.ts+1
......@@ -64,6 +64,7 @@ function makeAgentActivityRows(
6464 remove: () => Effect.void,
6565 pruneTerminal: () => Effect.void,
6666 listForUser: () => Effect.succeed([state]),
67 getForUserThread: () => Effect.succeed(state),
6768 ...overrides,
6869 };
6970}
infra/relay/src/agentActivity/AgentActivityRows.test.ts+9
......@@ -75,6 +75,15 @@ describe("AgentActivityRows", () => {
7575 const listError = yield* rows.listForUser({ userId: "user-2" }).pipe(Effect.flip);
7676 expect(listError).toMatchObject({ userId: "user-2", cause });
7777 expect(listError.message).toBe("Failed to list agent activity state for user user-2.");
78
79 const getError = yield* rows
80 .getForUserThread({
81 userId: "user-2",
82 environmentId: "env-1",
83 threadId: "thread-1",
84 })
85 .pipe(Effect.flip);
86 expect(getError).toMatchObject({ userId: "user-2", cause });
7887 }).pipe(
7988 Effect.provide(
8089 AgentActivityRows.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, failingDb))),
infra/relay/src/agentActivity/AgentActivityRows.ts+53
......@@ -83,6 +83,11 @@ export class AgentActivityRows extends Context.Service<
8383 ReadonlyArray<RelayAgentActivityState>,
8484 AgentActivityRowListPersistenceError
8585 >;
86 readonly getForUserThread: (input: {
87 readonly userId: string;
88 readonly environmentId: string;
89 readonly threadId: string;
90 }) => Effect.Effect<RelayAgentActivityState | null, AgentActivityRowListPersistenceError>;
8691 }
8792>()("t3code-relay/agentActivity/AgentActivityRows") {}
8893
......@@ -241,6 +246,54 @@ export const make = Effect.gen(function* () {
241246 ),
242247 );
243248 }),
249
250 getForUserThread: Effect.fn("relay.agent_activity_rows.get_for_user_thread")(function* (input) {
251 return yield* db
252 .select({ stateJson: relayAgentActivityRows.stateJson })
253 .from(relayAgentActivityRows)
254 .innerJoin(
255 relayEnvironmentLinks,
256 and(
257 eq(relayEnvironmentLinks.environmentId, relayAgentActivityRows.environmentId),
258 eq(
259 relayEnvironmentLinks.environmentPublicKey,
260 relayAgentActivityRows.environmentPublicKey,
261 ),
262 ),
263 )
264 .where(
265 and(
266 eq(relayEnvironmentLinks.userId, input.userId),
267 isNull(relayEnvironmentLinks.revokedAt),
268 eq(relayAgentActivityRows.environmentId, input.environmentId),
269 eq(relayAgentActivityRows.threadId, input.threadId),
270 ),
271 )
272 .orderBy(desc(relayAgentActivityRows.updatedAt))
273 .pipe(
274 Effect.flatMap((rows) =>
275 Effect.forEach(rows, (row) => encodeJsonValue(row.stateJson), {
276 concurrency: "unbounded",
277 }),
278 ),
279 Effect.map((rows) => {
280 for (const row of rows) {
281 const decoded = decodeRelayAgentActivityStateJson(row);
282 if (Option.isSome(decoded)) {
283 return decoded.value;
284 }
285 }
286 return null;
287 }),
288 Effect.mapError(
289 (cause) =>
290 new AgentActivityRowListPersistenceError({
291 userId: input.userId,
292 cause,
293 }),
294 ),
295 );
296 }),
244297 });
245298});
246299
infra/relay/src/agentActivity/ApnsDeliveries.test.ts+202-9
......@@ -5,7 +5,6 @@ import type {
55import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
66import { describe, expect, it } from "@effect/vitest";
77import * as NodeCrypto from "node:crypto";
8import * as DateTime from "effect/DateTime";
98import * as Effect from "effect/Effect";
109import * as Layer from "effect/Layer";
1110import * as Logger from "effect/Logger";
......@@ -161,10 +160,13 @@ function makeLayer(input: {
161160 >;
162161 readonly currentTargets?: ReadonlyArray<LiveActivities.TargetRow>;
163162 readonly config?: RelayConfiguration.RelayConfiguration["Service"];
164 // Live agent-activity rows returned by the delivery-time start recheck.
165 // Defaults to one freshly-updated running thread so queued starts stay
166 // deliverable in tests that don't care about the recheck.
163 // Live agent-activity rows returned by delivery-time state rechecks.
164 // Defaults to the fixture row so queued updates match unless a test is
165 // explicitly exercising stale-state behavior.
167166 readonly activityStates?: ReadonlyArray<RelayAgentActivityState>;
167 // Current per-thread rows used to reject queued updates/notifications that
168 // have been superseded before APNs delivery.
169 readonly currentActivityStates?: ReadonlyArray<RelayAgentActivityState>;
168170 readonly execute?: (
169171 request: HttpClientRequest.HttpClientRequest,
170172 ) => Effect.Effect<HttpClientResponse.HttpClientResponse>;
......@@ -182,9 +184,14 @@ function makeLayer(input: {
182184 listForUser: () =>
183185 input.activityStates !== undefined
184186 ? Effect.succeed([...input.activityStates])
185 : DateTime.now.pipe(
186 Effect.map((now) => [{ ...state, updatedAt: DateTime.formatIso(now) }]),
187 ),
187 : Effect.succeed([state]),
188 getForUserThread: ({ environmentId, threadId }) =>
189 Effect.succeed(
190 (input.currentActivityStates ?? input.activityStates ?? [state]).find(
191 (current) =>
192 current.environmentId === environmentId && current.threadId === threadId,
193 ) ?? null,
194 ),
188195 } satisfies AgentActivityRows.AgentActivityRows["Service"]),
189196 Layer.succeed(ApnsDeliveryQueue.ApnsDeliveryQueueSender, {
190197 send: (body) =>
......@@ -674,6 +681,8 @@ describe("ApnsDeliveries", () => {
674681 environmentId: "env",
675682 threadId: "thread",
676683 deepLink: "/",
684 phase: "waiting_for_input",
685 updatedAt: state.updatedAt,
677686 },
678687 },
679688 },
......@@ -778,6 +787,30 @@ describe("ApnsDeliveries", () => {
778787 },
779788 );
780789
790 it.effect("does not queue a push notification when a thread starts working", () => {
791 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
792 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
793
794 return Effect.gen(function* () {
795 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
796 const result = yield* deliveries.sendForTarget({
797 target: {
798 ...target,
799 push_token: "apns-device-token",
800 push_to_start_token: null,
801 activity_push_token: null,
802 remote_started_at: null,
803 },
804 aggregate,
805 nowMs: 5_000,
806 });
807
808 expect(result).toBeNull();
809 expect(queuedJobs).toEqual([]);
810 expect(attempts).toEqual([]);
811 }).pipe(Effect.provide(makeLayer({ attempts, queuedJobs })));
812 });
813
781814 it.effect("queues bounded alert notification payloads", () => {
782815 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
783816 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
......@@ -1176,6 +1209,150 @@ describe("ApnsDeliveries", () => {
11761209 );
11771210 });
11781211
1212 it.effect("skips a queued Done Live Activity update after the thread resumes working", () => {
1213 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
1214 let executeCount = 0;
1215 const completedAt = "1970-01-01T00:00:01.000Z";
1216 const completedAggregate: RelayAgentActivityAggregateState = {
1217 ...aggregate,
1218 activeCount: 0,
1219 updatedAt: completedAt,
1220 activities: [
1221 {
1222 ...aggregate.activities[0]!,
1223 phase: "completed",
1224 status: "Done",
1225 updatedAt: completedAt,
1226 },
1227 ],
1228 };
1229 const payload = makeApnsDeliveryJobPayload({
1230 kind: "live_activity_update",
1231 userId: target.user_id,
1232 deviceId: target.device_id,
1233 token: target.activity_push_token ?? "activity-token",
1234 aggregate: completedAggregate,
1235 alert: { title: "Thread", body: "Done: Project" },
1236 createdAt: completedAt,
1237 expiresAt: "1970-01-01T00:10:00.000Z",
1238 jobId: "job-update-superseded-by-running",
1239 });
1240 const signed = signApnsDeliveryJob({
1241 secret: config.apnsDeliveryJobSigningSecret,
1242 payload,
1243 });
1244 const execute = (request: HttpClientRequest.HttpClientRequest) =>
1245 Effect.sync(() => {
1246 executeCount += 1;
1247 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
1248 });
1249
1250 return Effect.gen(function* () {
1251 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1252 const result = yield* deliveries.processSignedJob(signed);
1253
1254 expect(result).toMatchObject({
1255 kind: "live_activity_update",
1256 ok: true,
1257 apnsStatus: null,
1258 apnsReason: "Stale APNs delivery job skipped.",
1259 });
1260 expect(executeCount).toBe(0);
1261 expect(attempts).toMatchObject([
1262 {
1263 sourceJobId: "job-update-superseded-by-running",
1264 apnsReason: "Stale agent activity state skipped.",
1265 },
1266 ]);
1267 }).pipe(
1268 Effect.provide(
1269 makeLayer({
1270 attempts,
1271 activityStates: [
1272 {
1273 ...state,
1274 phase: "running",
1275 headline: "Running",
1276 updatedAt: "1970-01-01T00:00:02.000Z",
1277 },
1278 ],
1279 config: signingConfig,
1280 execute,
1281 }),
1282 ),
1283 );
1284 });
1285
1286 it.effect("skips a queued Done notification after the thread resumes working", () => {
1287 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
1288 let executeCount = 0;
1289 const completedAt = "1970-01-01T00:00:01.000Z";
1290 const payload = makeApnsDeliveryJobPayload({
1291 kind: "push_notification",
1292 userId: target.user_id,
1293 deviceId: target.device_id,
1294 token: "apns-device-token",
1295 aggregate: null,
1296 notification: {
1297 title: "Thread",
1298 body: "Done: Project",
1299 environmentId: "env",
1300 threadId: "thread",
1301 deepLink: "/",
1302 phase: "completed",
1303 updatedAt: completedAt,
1304 },
1305 createdAt: completedAt,
1306 expiresAt: "1970-01-01T00:10:00.000Z",
1307 jobId: "job-push-superseded-by-running",
1308 });
1309 const signed = signApnsDeliveryJob({
1310 secret: config.apnsDeliveryJobSigningSecret,
1311 payload,
1312 });
1313 const execute = (request: HttpClientRequest.HttpClientRequest) =>
1314 Effect.sync(() => {
1315 executeCount += 1;
1316 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
1317 });
1318
1319 return Effect.gen(function* () {
1320 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1321 const result = yield* deliveries.processSignedJob(signed);
1322
1323 expect(result).toMatchObject({
1324 kind: "push_notification",
1325 ok: true,
1326 apnsStatus: null,
1327 apnsReason: "Stale APNs delivery job skipped.",
1328 });
1329 expect(executeCount).toBe(0);
1330 expect(attempts).toMatchObject([
1331 {
1332 sourceJobId: "job-push-superseded-by-running",
1333 apnsReason: "Stale agent activity state skipped.",
1334 },
1335 ]);
1336 }).pipe(
1337 Effect.provide(
1338 makeLayer({
1339 attempts,
1340 currentTargets: [{ ...target, push_token: "apns-device-token" }],
1341 currentActivityStates: [
1342 {
1343 ...state,
1344 phase: "running",
1345 headline: "Running",
1346 updatedAt: "1970-01-01T00:00:02.000Z",
1347 },
1348 ],
1349 config: signingConfig,
1350 execute,
1351 }),
1352 ),
1353 );
1354 });
1355
11791356 it.effect("retries signed queue jobs that are already claimed but not completed", () => {
11801357 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
11811358 let executeCount = 0;
......@@ -1327,7 +1504,15 @@ describe("ApnsDeliveries", () => {
13271504 deviceId: target.device_id,
13281505 },
13291506 ]);
1330 }).pipe(Effect.provide(makeLayer({ attempts, clearedStarts })));
1507 }).pipe(
1508 Effect.provide(
1509 makeLayer({
1510 attempts,
1511 clearedStarts,
1512 activityStates: [{ ...state, updatedAt: "9999-01-01T00:00:00.000Z" }],
1513 }),
1514 ),
1515 );
13311516 });
13321517
13331518 it.effect("invalidates dead push-to-start tokens after permanent APNs start failures", () => {
......@@ -1373,7 +1558,15 @@ describe("ApnsDeliveries", () => {
13731558 },
13741559 ]);
13751560 }).pipe(
1376 Effect.provide(makeLayer({ attempts, invalidatedTokens, config: signingConfig, execute })),
1561 Effect.provide(
1562 makeLayer({
1563 attempts,
1564 invalidatedTokens,
1565 activityStates: [{ ...state, updatedAt: "9999-01-01T00:00:00.000Z" }],
1566 config: signingConfig,
1567 execute,
1568 }),
1569 ),
13771570 );
13781571 });
13791572
infra/relay/src/agentActivity/ApnsDeliveries.ts+112
......@@ -369,6 +369,8 @@ function notificationForAggregate(input: {
369369 environmentId: activity.environmentId,
370370 threadId: activity.threadId,
371371 deepLink: activity.deepLink,
372 phase: activity.phase,
373 updatedAt: activity.updatedAt,
372374 };
373375}
374376
......@@ -722,6 +724,87 @@ export const make = Effect.gen(function* () {
722724 );
723725 });
724726
727 const stateIdentityIsCurrent = Effect.fnUntraced(function* (input: {
728 readonly userId: string;
729 readonly environmentId: string;
730 readonly threadId: string;
731 readonly phase: RelayAgentActivityAggregateState["activities"][number]["phase"];
732 readonly updatedAt: string;
733 }) {
734 return yield* activityRows
735 .getForUserThread({
736 userId: input.userId,
737 environmentId: input.environmentId,
738 threadId: input.threadId,
739 })
740 .pipe(
741 Effect.map(
742 (current) =>
743 current !== null &&
744 current.phase === input.phase &&
745 current.updatedAt === input.updatedAt,
746 ),
747 // A transient persistence failure must not permanently discard a
748 // legitimate alert. Fail open and let the signed job's retry/dedupe
749 // protections handle transport failures as usual.
750 Effect.catchCause((cause) =>
751 Effect.logWarning("agent-activity state recheck failed; allowing queued delivery", {
752 cause,
753 environmentId: input.environmentId,
754 threadId: input.threadId,
755 }).pipe(Effect.as(true)),
756 ),
757 );
758 });
759
760 const aggregateRowsAreCurrent = Effect.fnUntraced(function* (input: {
761 readonly userId: string;
762 readonly aggregate: RelayAgentActivityAggregateState;
763 }) {
764 return yield* activityRows.listForUser({ userId: input.userId }).pipe(
765 Effect.map((currentStates) => {
766 const currentByThread = new Map(
767 currentStates.map((current) => [
768 `${current.environmentId}\u0000${current.threadId}`,
769 current,
770 ]),
771 );
772 return input.aggregate.activities.every((row) => {
773 const current = currentByThread.get(`${row.environmentId}\u0000${row.threadId}`);
774 return (
775 current !== undefined &&
776 current.phase === row.phase &&
777 current.updatedAt === row.updatedAt
778 );
779 });
780 }),
781 Effect.catchCause((cause) =>
782 Effect.logWarning("agent-activity aggregate recheck failed; allowing queued delivery", {
783 cause,
784 userId: input.userId,
785 }).pipe(Effect.as(true)),
786 ),
787 );
788 });
789
790 const notificationStateIsCurrent = Effect.fnUntraced(function* (input: {
791 readonly userId: string;
792 readonly notification: ApnsNotificationPayload;
793 }) {
794 // Jobs from older relay versions do not carry a state identity. Preserve
795 // backwards compatibility and only revalidate newly queued jobs.
796 if (input.notification.phase === undefined || input.notification.updatedAt === undefined) {
797 return true;
798 }
799 return yield* stateIdentityIsCurrent({
800 userId: input.userId,
801 environmentId: input.notification.environmentId,
802 threadId: input.notification.threadId,
803 phase: input.notification.phase,
804 updatedAt: input.notification.updatedAt,
805 });
806 });
807
725808 const isCurrentSignedJobToken = Effect.fnUntraced(function* (input: {
726809 readonly target: LiveActivityDeliveryTarget;
727810 readonly kind: RelayDeliveryKind;
......@@ -791,6 +874,20 @@ export const make = Effect.gen(function* () {
791874 });
792875 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
793876 }
877 if (
878 input.kind !== "live_activity_start" &&
879 aggregate !== null &&
880 !(yield* aggregateRowsAreCurrent({
881 userId: input.target.user_id,
882 aggregate,
883 }))
884 ) {
885 yield* attempts.completeSourceJob({
886 sourceJobId: input.sourceJobId,
887 apnsReason: "Stale agent activity state skipped.",
888 });
889 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
890 }
794891 }
795892 if (
796893 input.kind === "live_activity_start" &&
......@@ -930,6 +1027,21 @@ export const make = Effect.gen(function* () {
9301027 kind: "push_notification",
9311028 });
9321029 }
1030 if (
1031 !(yield* notificationStateIsCurrent({
1032 userId: input.target.user_id,
1033 notification,
1034 }))
1035 ) {
1036 yield* attempts.completeSourceJob({
1037 sourceJobId: input.sourceJobId,
1038 apnsReason: "Stale agent activity state skipped.",
1039 });
1040 return staleJobResult({
1041 deviceId: input.target.device_id,
1042 kind: "push_notification",
1043 });
1044 }
9331045 }
9341046 const result = yield* apns
9351047 .sendPushNotificationRequest({
infra/relay/src/agentActivity/MobileRegistrations.test.ts+1
......@@ -86,6 +86,7 @@ function makeAgentActivityRows(
8686 };
8787 return Effect.succeed([activeState]);
8888 },
89 getForUserThread: () => Effect.succeed(null),
8990 ...overrides,
9091 };
9192}
infra/relay/src/agentActivity/apnsDeliveryJobs.ts+10-1
......@@ -1,6 +1,10 @@
11import * as NodeCrypto from "node:crypto";
22
3import { RelayAgentActivityAggregateState, type RelayDeliveryKind } from "@t3tools/contracts/relay";
3import {
4 RelayAgentActivityAggregateState,
5 RelayAgentAwarenessPhase,
6 type RelayDeliveryKind,
7} from "@t3tools/contracts/relay";
48import { stableStringify } from "@t3tools/shared/relaySigning";
59import * as DateTime from "effect/DateTime";
610import * as Option from "effect/Option";
......@@ -38,6 +42,11 @@ export const ApnsNotificationPayload = Schema.Struct({
3842 environmentId: Schema.String,
3943 threadId: Schema.String,
4044 deepLink: Schema.String,
45 // Optional so delivery jobs queued by older relay builds still decode.
46 // New jobs use these fields to avoid delivering a stale Done/attention
47 // notification after the thread has moved to another phase.
48 phase: Schema.optional(RelayAgentAwarenessPhase),
49 updatedAt: Schema.optional(Schema.String),
4150});
4251export type ApnsNotificationPayload = typeof ApnsNotificationPayload.Type;
4352