1import { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
3import { describe, expect, it } from "@effect/vitest";
4import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
5import * as Effect from "effect/Effect";
6import * as DateTime from "effect/DateTime";
7import * as Layer from "effect/Layer";
8import * as Redacted from "effect/Redacted";
9import * as Schema from "effect/Schema";
10import * as Stream from "effect/Stream";
11import type * as Cloudflare from "alchemy/Cloudflare";
12import * as FcmDeliveryQueueConsumer from "./FcmDeliveryQueueConsumer.ts";
13
14import * as RelayConfiguration from "../Config.ts";
15import * as RelayDb from "../db.ts";
16import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
17import * as AgentActivityRows from "./AgentActivityRows.ts";
18import * as LiveActivities from "./LiveActivities.ts";
19import * as FcmDeliveryQueueSender from "./FcmDeliveryQueueSender.ts";
20import * as FcmClient from "./FcmClient.ts";
21import * as FcmDeliveries from "./FcmDeliveries.ts";
22import { TestClock } from "effect/testing";
23import { androidActivityData, fitFcmData } from "./fcmPayloads.ts";
24import { makeAggregateState } from "./agentActivityAggregate.ts";
25
26const aggregateFor = (states: ReadonlyArray<RelayAgentActivityState>) =>
27 makeAggregateState({ activeStates: states, terminalState: null, nowMs: 0 })!;
28
29const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
30const state: RelayAgentActivityState = {
31 environmentId: EnvironmentId.make("env"),
32 threadId: ThreadId.make("thread"),
33 projectTitle: "Project",
34 threadTitle: "Fix notifications",
35 phase: "running",
36 headline: "Working",
37 modelTitle: "Codex",
38 updatedAt: "1970-01-01T00:00:00.000Z",
39 deepLink: "/threads/env/thread",
40};
41const preferences = {
42 notificationsEnabled: true,
43 liveActivitiesEnabled: true,
44 notifyOnApproval: true,
45 notifyOnInput: true,
46 notifyOnCompletion: true,
47 notifyOnFailure: true,
48};
49const target: LiveActivities.TargetRow = {
50 user_id: "user",
51 device_id: "phone",
52 platform: "android",
53 ios_major_version: null,
54 app_version: null,
55 bundle_id: "com.t3tools.t3code.dev",
56 aps_environment: null,
57 push_token: "fcm-token",
58 push_to_start_token: null,
59 preferences_json: encodeJson(preferences),
60 activity_push_token: null,
61 remote_start_queued_at: null,
62 remote_started_at: null,
63 ended_at: null,
64 last_aggregate_json: null,
65 last_live_activity_delivery_at: null,
66};
67const config = {
68 relayIssuer: "https://relay.test",
69 fcmServiceAccount: Redacted.make("configured"),
70 apns: {
71 environment: "sandbox",
72 teamId: "unused",
73 keyId: "unused",
74 bundleId: "unused",
75 privateKey: Redacted.make("unused"),
76 },
77 clerkSecretKey: Redacted.make("unused"),
78 clerkPublishableKey: "unused",
79 clerkJwtAudience: "unused",
80 apnsDeliveryJobSigningSecret: Redacted.make("unused"),
81 cloudMintPrivateKey: Redacted.make("unused"),
82 cloudMintPublicKey: "unused",
83 managedEndpointBaseDomain: undefined,
84 managedEndpointNamespace: undefined,
85} satisfies RelayConfiguration.RelayConfiguration["Service"];
86
87function harness() {
88 const sent: Array<Parameters<FcmClient.FcmClient["Service"]["send"]>[0]> = [];
89 const queued: FcmDeliveries.FcmDeliveryJob[] = [];
90 const marked: Array<Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]> = [];
91 const current = {
92 target: { ...target },
93 state: { ...state } as RelayAgentActivityState | null,
94 otherStates: [] as RelayAgentActivityState[],
95 mutedEnvironments: [] as string[],
96 notificationOnlyEnvironments: [] as string[],
97 revokedEnvironments: [] as string[],
98 linked: true,
99 deliveryFailure: null as FcmClient.FcmClientError | null,
100 };
101 const layerServices = Layer.mergeAll(
102 NodeCryptoLayer.layer,
103 Layer.succeed(RelayConfiguration.RelayConfiguration, config),
104 Layer.succeed(FcmDeliveryQueueSender.FcmDeliveryQueueSender, {
105 send: (job) =>
106 Effect.sync(() => {
107 queued.push(job);
108 }),
109 }),
110 Layer.succeed(FcmClient.FcmClient, {
111 send: (input) =>
112 Effect.suspend(() =>
113 current.deliveryFailure
114 ? Effect.fail(current.deliveryFailure)
115 : Effect.sync(() => {
116 sent.push(input);
117 return { unregistered: false };
118 }),
119 ),
120 }),
121 Layer.succeed(LiveActivities.LiveActivities, {
122 register: () => Effect.void,
123 listTargets: () => Effect.sync(() => [current.target]),
124 markDelivery: (input) =>
125 Effect.sync(() => {
126 marked.push(input);
127 current.target.last_aggregate_json = input.aggregate ? encodeJson(input.aggregate) : null;
128 }),
129 markStartQueued: () => Effect.void,
130 clearStartQueued: () => Effect.void,
131 invalidateDeliveryToken: () => Effect.void,
132 }),
133 Layer.succeed(AgentActivityRows.AgentActivityRows, {
134 upsert: () => Effect.void,
135 remove: () => Effect.void,
136 pruneTerminal: () => Effect.void,
137 listForUser: () =>
138 Effect.sync(() =>
139 current.linked
140 ? [...(current.state ? [current.state] : []), ...current.otherStates].filter(
141 (row) =>
142 !current.revokedEnvironments.includes(row.environmentId) &&
143 !current.notificationOnlyEnvironments.includes(row.environmentId),
144 )
145 : [],
146 ),
147 getForUserThread: (input) =>
148 Effect.sync(() =>
149 current.linked
150 ? ([current.state, ...current.otherStates].find(
151 (row) =>
152 row?.environmentId === input.environmentId && row.threadId === input.threadId,
153 ) ?? null)
154 : null,
155 ),
156 }),
157 Layer.succeed(EnvironmentLinks.EnvironmentLinks, {
158 upsert: () => Effect.void,
159 listDeliveryUsersForEnvironment: (input) =>
160 Effect.sync(() =>
161 current.linked && !current.revokedEnvironments.includes(input.environmentId)
162 ? [
163 {
164 userId: "user",
165 notificationsEnabled: !current.mutedEnvironments.includes(input.environmentId),
166 liveActivitiesEnabled: !current.notificationOnlyEnvironments.includes(
167 input.environmentId,
168 ),
169 },
170 ]
171 : [],
172 ),
173 listForUser: () => Effect.succeed([]),
174 findActiveManagedForEnvironment: () => Effect.succeed([]),
175 setHoldWebhooksWhileOffline: () => Effect.void,
176 revokeForUser: () => Effect.succeed(false),
177 getForUser: (input) =>
178 Effect.sync(() =>
179 current.linked && !current.revokedEnvironments.includes(input.environmentId)
180 ? {
181 environmentId: EnvironmentId.make(input.environmentId),
182 label: "Desktop",
183 environmentPublicKey: "key",
184 linkedAt: state.updatedAt,
185 endpoint: {
186 httpBaseUrl: "https://env.test",
187 wsBaseUrl: "wss://env.test",
188 providerKind: "manual",
189 },
190 }
191 : null,
192 ),
193 }),
194 Layer.succeed(RelayDb.RelayDb, {} as RelayDb.RelayDb["Service"]),
195 );
196 return {
197 sent,
198 queued,
199 marked,
200 current,
201 layer: FcmDeliveries.layer.pipe(Layer.provide(layerServices)),
202 job: {
203 userId: "user",
204 deviceId: "phone",
205 token: "fcm-token",
206 state,
207 queuedAt: 0,
208 } satisfies FcmDeliveries.FcmDeliveryJob,
209 };
210}
211
212describe("Android delivery routing", () => {
213 const secondState: RelayAgentActivityState = {
214 ...state,
215 threadId: ThreadId.make("second-thread"),
216 threadTitle: "Second thread",
217 deepLink: "/threads/env/second-thread",
218 };
219
220 it.effect.each([
221 {
222 firstPhase: "waiting_for_approval",
223 secondPhase: "waiting_for_input",
224 title: "2 agents need attention",
225 active: "true",
226 },
227 { firstPhase: "completed", secondPhase: "failed", title: "2 agents finished", active: "false" },
228 ] as const)(
229 "routes grouped $firstPhase and $secondPhase to the overview once across their queued jobs",
230 ({ firstPhase, secondPhase, title, active }) => {
231 const h = harness();
232 h.current.otherStates = [secondState];
233 return Effect.gen(function* () {
234 const delivery = yield* FcmDeliveries.FcmDeliveries;
235 yield* delivery.process(h.job);
236 h.current.state = { ...state, phase: firstPhase };
237 h.current.otherStates = [{ ...secondState, phase: secondPhase }];
238 yield* delivery.process({ ...h.job, state: h.current.state });
239 yield* delivery.process({ ...h.job, state: h.current.otherStates[0] });
240 yield* delivery.process({ ...h.job, state: h.current.state });
241 const alerts = h.sent.filter((message) => message.alert);
242 expect(alerts).toHaveLength(1);
243 expect(alerts[0]?.data).toMatchObject({
244 alert_title: title,
245 alert_body: "Fix notifications, Second thread",
246 alert_path: "/",
247 active,
248 });
249 expect(h.marked.at(-1)?.aggregate?.activities).toHaveLength(2);
250 }).pipe(Effect.provide(h.layer));
251 },
252 );
253
254 it.effect("filters disabled event types before counting a group", () => {
255 const h = harness();
256 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, secondState]));
257 h.current.target.preferences_json = encodeJson({ ...preferences, notifyOnApproval: false });
258 h.current.state = { ...state, phase: "waiting_for_approval" };
259 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
260 return Effect.gen(function* () {
261 const delivery = yield* FcmDeliveries.FcmDeliveries;
262 yield* delivery.process({ ...h.job, state: h.current.state });
263 expect(h.sent[0]?.data).toMatchObject({
264 alert_title: "Second thread",
265 alert_body: "Input: Project",
266 alert_path: "/threads/env/second-thread",
267 });
268 }).pipe(Effect.provide(h.layer));
269 });
270
271 it.effect("excludes already delivered attention and stale completions from groups", () => {
272 const h = harness();
273 h.current.state = { ...state, phase: "waiting_for_approval" };
274 h.current.target.last_aggregate_json = encodeJson(aggregateFor([h.current.state, secondState]));
275 h.current.otherStates = [
276 { ...secondState, phase: "completed", updatedAt: "1969-12-31T23:57:00.000Z" },
277 ];
278 return Effect.gen(function* () {
279 const delivery = yield* FcmDeliveries.FcmDeliveries;
280 yield* delivery.process({ ...h.job, state: h.current.state });
281 expect(h.sent.every((message) => !message.alert)).toBe(true);
282 }).pipe(Effect.provide(h.layer));
283 });
284
285 it.effect("prioritizes attention over simultaneous completions like iOS", () => {
286 const h = harness();
287 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, secondState]));
288 h.current.state = { ...state, phase: "completed" };
289 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
290 return Effect.gen(function* () {
291 const delivery = yield* FcmDeliveries.FcmDeliveries;
292 yield* delivery.process({ ...h.job, state: h.current.state });
293 expect(h.sent[0]?.data).toMatchObject({
294 alert_title: "Second thread",
295 alert_body: "Input: Project",
296 });
297 yield* delivery.process({ ...h.job, state: h.current.otherStates[0] });
298 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
299 }).pipe(Effect.provide(h.layer));
300 });
301
302 it.effect.each(["completed", "waiting_for_approval", "waiting_for_input", "failed"] as const)(
303 "deleting one thread preserves another thread's %s alert",
304 (phase) => {
305 const h = harness();
306 h.current.otherStates = [secondState];
307 return Effect.gen(function* () {
308 const delivery = yield* FcmDeliveries.FcmDeliveries;
309 yield* delivery.process(h.job);
310 h.current.otherStates = [];
311 h.current.state = { ...state, phase };
312 yield* delivery.enqueue({ target, state: null });
313 yield* delivery.process(h.queued[0]);
314 expect(h.sent.at(-1)?.alert).toBe(false);
315 yield* delivery.process({ ...h.job, state: h.current.state });
316 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
317 yield* delivery.process({ ...h.job, state: h.current.state });
318 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
319 }).pipe(Effect.provide(h.layer));
320 },
321 );
322
323 it.effect("registration replay establishes a baseline without alerting", () => {
324 const h = harness();
325 h.current.state = { ...state, phase: "waiting_for_approval" };
326 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
327 return Effect.gen(function* () {
328 const delivery = yield* FcmDeliveries.FcmDeliveries;
329 yield* delivery.enqueue({ target, state: null, replay: true });
330 yield* delivery.process(h.queued[0]);
331 yield* delivery.process({ ...h.job, state: h.current.state });
332 expect(h.sent.every((message) => !message.alert)).toBe(true);
333 expect(h.marked[0]?.aggregate?.activities).toHaveLength(2);
334 }).pipe(Effect.provide(h.layer));
335 });
336
337 it.effect.each(["mutedEnvironments", "revokedEnvironments"] as const)(
338 "excludes %s when forming cross-environment groups",
339 (restriction) => {
340 const h = harness();
341 const other = { ...secondState, environmentId: EnvironmentId.make("other-env") };
342 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, other]));
343 h.current.state = { ...state, phase: "waiting_for_approval" };
344 h.current.otherStates = [{ ...other, phase: "waiting_for_input" }];
345 h.current[restriction] = [other.environmentId];
346 return Effect.gen(function* () {
347 const delivery = yield* FcmDeliveries.FcmDeliveries;
348 yield* delivery.process({ ...h.job, state: h.current.state });
349 expect(h.sent[0]?.data).toMatchObject({
350 alert_title: "Fix notifications",
351 alert_body: "Approval: Project",
352 });
353 }).pipe(Effect.provide(h.layer));
354 },
355 );
356
357 it("gives a group a stable retry identity independent of row order", () => {
358 const other = {
359 ...secondState,
360 environmentId: EnvironmentId.make("other-env"),
361 threadId: state.threadId,
362 };
363 const input = {
364 previousAggregate: aggregateFor([state, other]),
365 nextAggregate: aggregateFor([
366 { ...state, phase: "completed" },
367 { ...other, phase: "failed" },
368 ]),
369 preferences,
370 nowMs: 0,
371 };
372 const alert = FcmDeliveries.androidAlertForAggregate(input);
373 expect(alert?.alert_title).toBe("2 agents finished");
374 expect(
375 FcmDeliveries.androidAlertForAggregate({
376 ...input,
377 nextAggregate: {
378 ...input.nextAggregate,
379 activities: input.nextAggregate.activities.toReversed(),
380 },
381 })?.alert_id,
382 ).toBe(alert?.alert_id);
383 expect(
384 FcmDeliveries.androidAlertForAggregate({
385 ...input,
386 nextAggregate: {
387 ...input.nextAggregate,
388 activities: input.nextAggregate.activities.map((row) => ({
389 ...row,
390 updatedAt: "1970-01-01T00:00:01.000Z",
391 })),
392 },
393 })?.alert_id,
394 ).not.toBe(alert?.alert_id);
395 });
396
397 it("distinguishes matching thread IDs in different environments", () => {
398 const other = {
399 ...secondState,
400 environmentId: EnvironmentId.make("other-env"),
401 threadId: state.threadId,
402 };
403 const alreadyWaiting = { ...state, phase: "waiting_for_approval" as const };
404 expect(
405 FcmDeliveries.androidAlertForAggregate({
406 previousAggregate: aggregateFor([alreadyWaiting, other]),
407 nextAggregate: aggregateFor([alreadyWaiting, { ...other, phase: "waiting_for_input" }]),
408 preferences,
409 nowMs: 0,
410 }),
411 ).toMatchObject({ alert_title: "Second thread", alert_body: "Input: Project" });
412 });
413
414 it.effect.each([
415 { phase: "waiting_for_approval", body: "Approval: Project", preference: "notifyOnApproval" },
416 { phase: "waiting_for_input", body: "Input: Project", preference: "notifyOnInput" },
417 { phase: "completed", body: "Done: Project", preference: "notifyOnCompletion" },
418 { phase: "failed", body: "Failed: Project", preference: "notifyOnFailure" },
419 ] as const)(
420 "uses iOS alert wording for $phase and honors its preference",
421 ({ phase, body, preference }) => {
422 const h = harness();
423 h.current.state = { ...state, phase };
424 return Effect.gen(function* () {
425 const delivery = yield* FcmDeliveries.FcmDeliveries;
426 yield* delivery.process({ ...h.job, state: h.current.state });
427 expect(h.sent[0]?.data).toMatchObject({
428 alert_title: "Fix notifications",
429 alert_body: body,
430 alert_path: "/threads/env/thread",
431 });
432 h.current.target.preferences_json = encodeJson({
433 ...preferences,
434 [preference]: false,
435 });
436 yield* delivery.process({ ...h.job, state: h.current.state });
437 expect(h.sent.slice(1).every((sent) => !sent.alert && !sent.data.alert_id)).toBe(true);
438 }).pipe(Effect.provide(h.layer));
439 },
440 );
441
442 it("trims and truncates alert text like iOS", () => {
443 expect(
444 FcmDeliveries.androidAlertForState(
445 {
446 ...state,
447 phase: "completed",
448 threadTitle: ` ${"T".repeat(150)} `,
449 projectTitle: ` ${"P".repeat(150)} `,
450 },
451 preferences,
452 0,
453 ),
454 ).toMatchObject({
455 alert_title: `${"T".repeat(117)}...`,
456 alert_body: `Done: ${"P".repeat(111)}...`,
457 alert_group: JSON.stringify([state.environmentId, state.threadId]),
458 });
459 });
460
461 it("keeps notification groups distinct when identifiers contain slashes", () => {
462 const left = FcmDeliveries.androidAlertForState(
463 {
464 ...state,
465 environmentId: EnvironmentId.make("a/b"),
466 threadId: ThreadId.make("c"),
467 phase: "completed",
468 },
469 preferences,
470 0,
471 );
472 const right = FcmDeliveries.androidAlertForState(
473 {
474 ...state,
475 environmentId: EnvironmentId.make("a"),
476 threadId: ThreadId.make("b/c"),
477 phase: "completed",
478 },
479 preferences,
480 0,
481 );
482 expect(left?.alert_group).not.toBe(right?.alert_group);
483 });
484
485 it.effect(
486 "queues Android devices and sends the latest aggregate instead of a stale running state",
487 () => {
488 const h = harness();
489 return Effect.gen(function* () {
490 const delivery = yield* FcmDeliveries.FcmDeliveries;
491 yield* delivery.enqueue({ target, state });
492 h.current.state = { ...state, phase: "completed" };
493 yield* delivery.process(h.queued[0]);
494 expect(h.sent).toHaveLength(0);
495 yield* delivery.process({ ...h.job, state: h.current.state });
496 expect(h.sent[0]?.data).toMatchObject({
497 active: "false",
498 alert_title: "Fix notifications",
499 alert_body: "Done: Project",
500 alert_path: "/threads/env/thread",
501 });
502 }).pipe(Effect.provide(h.layer));
503 },
504 );
505
506 it.effect("keeps completion alerts working with ongoing activity disabled", () => {
507 const h = harness();
508 h.current.state = { ...state, phase: "completed" };
509 h.current.target.preferences_json = encodeJson({
510 ...preferences,
511 liveActivitiesEnabled: false,
512 });
513 return Effect.gen(function* () {
514 const delivery = yield* FcmDeliveries.FcmDeliveries;
515 yield* delivery.process({ ...h.job, state: h.current.state });
516 expect(h.sent[0]?.data).toMatchObject({
517 active: "false",
518 alert_title: "Fix notifications",
519 alert_body: "Done: Project",
520 });
521 expect(h.sent[0]?.data.user_id).toBe("user");
522 }).pipe(Effect.provide(h.layer));
523 });
524
525 it.effect("drops jobs for rotated tokens, expired jobs, and revoked links", () => {
526 const h = harness();
527 return Effect.gen(function* () {
528 const delivery = yield* FcmDeliveries.FcmDeliveries;
529 yield* delivery.process({ ...h.job, token: "old-token" });
530 yield* delivery.process({ ...h.job, queuedAt: -400_000 });
531 h.current.linked = false;
532 yield* delivery.process(h.job);
533 expect(h.sent).toHaveLength(0);
534 }).pipe(Effect.provide(h.layer));
535 });
536
537 it.effect("ends an existing ongoing notification when the environment stops publishing", () => {
538 const h = harness();
539 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state]));
540 h.current.state = null;
541 return Effect.gen(function* () {
542 const delivery = yield* FcmDeliveries.FcmDeliveries;
543 yield* delivery.process({ ...h.job, state: null });
544 expect(h.sent[0]?.data).toMatchObject({ active: "false" });
545 expect(h.sent[0]?.alert).toBe(false);
546 expect(h.marked[0]?.kind).toBe("live_activity_end");
547 }).pipe(Effect.provide(h.layer));
548 });
549
550 it.effect("leaves iOS devices on their existing delivery path", () => {
551 const h = harness();
552 return Effect.gen(function* () {
553 const delivery = yield* FcmDeliveries.FcmDeliveries;
554 expect(yield* delivery.enqueue({ target: { ...target, platform: "ios" }, state })).toBeNull();
555 expect(h.queued).toHaveLength(0);
556 }).pipe(Effect.provide(h.layer));
557 });
558
559 it.effect("honors disabled alert preferences while continuing ongoing activity", () => {
560 const h = harness();
561 h.current.state = { ...state, phase: "waiting_for_approval" };
562 h.current.target.preferences_json = encodeJson({ ...preferences, notifyOnApproval: false });
563 return Effect.gen(function* () {
564 const delivery = yield* FcmDeliveries.FcmDeliveries;
565 yield* delivery.process({ ...h.job, state: h.current.state });
566 expect(h.sent[0]?.data.active).toBe("true");
567 expect(h.sent[0]?.data.alert_id).toBeUndefined();
568 h.current.target.preferences_json = encodeJson({
569 ...preferences,
570 notificationsEnabled: false,
571 });
572 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state]));
573 yield* delivery.process({ ...h.job, state: h.current.state });
574 expect(h.sent[1]?.data.active).toBe("false");
575 expect(h.sent[1]?.data.alert_id).toBeUndefined();
576 }).pipe(Effect.provide(h.layer));
577 });
578 it.effect.each(
579 [true, false].flatMap((ongoing) =>
580 (["completed", "failed"] as const).map((phase) => ({ ongoing, phase })),
581 ),
582 )("does not alert a stale $phase without a baseline (ongoing=$ongoing)", ({ ongoing, phase }) => {
583 const h = harness();
584 h.current.state = { ...state, phase, updatedAt: "1969-12-31T23:57:00.000Z" };
585 h.current.target.preferences_json = encodeJson({
586 ...preferences,
587 liveActivitiesEnabled: ongoing,
588 });
589 return Effect.gen(function* () {
590 const delivery = yield* FcmDeliveries.FcmDeliveries;
591 yield* delivery.process({ ...h.job, state: h.current.state });
592 expect(h.sent.every((message) => !message.alert)).toBe(true);
593 }).pipe(Effect.provide(h.layer));
594 });
595
596 it.effect(
597 "retains finished results without extending expiry on replay and clears expired cards",
598 () => {
599 const h = harness();
600 h.current.state = { ...state, phase: "failed" };
601 return Effect.gen(function* () {
602 const delivery = yield* FcmDeliveries.FcmDeliveries;
603 yield* delivery.process({ ...h.job, state: h.current.state });
604 expect(h.sent[0]?.data).toMatchObject({
605 active: "false",
606 activity_title: "Agent work failed",
607 activity_expires_at: "900000",
608 });
609 yield* TestClock.adjust("5 minutes");
610 yield* delivery.process({ ...h.job, queuedAt: 300000, state: null, replay: true });
611 expect(h.sent[1]?.data.activity_expires_at).toBe("900000");
612 expect(h.sent[1]?.alert).toBe(false);
613 yield* TestClock.adjust("11 minutes");
614 yield* delivery.process({ ...h.job, queuedAt: 960000, state: null, replay: true });
615 expect(h.sent[2]?.data).toMatchObject({ active: "false", activity_expires_at: "0" });
616 expect(h.marked.at(-1)?.aggregate).toBeNull();
617 }).pipe(Effect.provide(h.layer));
618 },
619 );
620
621 it.effect("replays an empty card to repair an orphan without a delivery baseline", () => {
622 const h = harness();
623 h.current.state = null;
624 return Effect.gen(function* () {
625 const delivery = yield* FcmDeliveries.FcmDeliveries;
626 yield* delivery.process({ ...h.job, state: null });
627 expect(h.sent[0]?.data.activity_expires_at).toBe("0");
628 expect(h.sent[0]?.alert).toBe(false);
629 }).pipe(Effect.provide(h.layer));
630 });
631
632 it("shows five rows with attention then failure first, including project and status", () => {
633 const aggregate = aggregateFor([
634 state,
635 { ...secondState, phase: "failed" },
636 { ...state, threadId: ThreadId.make("approval"), phase: "waiting_for_approval" },
637 { ...state, threadId: ThreadId.make("input"), phase: "waiting_for_input" },
638 { ...state, threadId: ThreadId.make("done"), phase: "completed" },
639 ]);
640 const data = androidActivityData(aggregate);
641 expect(data.activity_title).toBe("3 active agents · 2 need attention");
642 expect(data.activity_chip).toBe("Review");
643 expect(data.activity_phase).toBe("waiting_for_approval");
644 expect(data.activity_active_count).toBe("3");
645 expect(data.activity_attention_count).toBe("2");
646 const input = androidActivityData(aggregateFor([{ ...state, phase: "waiting_for_input" }]));
647 expect(input.activity_phase).toBe("waiting_for_input");
648 expect(input.activity_active_count).toBe("1");
649 expect(androidActivityData(aggregateFor([state])).activity_chip).toBe("Active");
650 expect(
651 androidActivityData(aggregateFor([{ ...state, phase: "completed" }])).activity_chip,
652 ).toBe("");
653 expect(androidActivityData(null).activity_chip).toBe("");
654 expect(
655 Object.entries(data)
656 .filter(([key]) => key.startsWith("activity_line_"))
657 .map(([, value]) => value),
658 ).toEqual([
659 "Approval\tFix notifications\tProject",
660 "Input\tFix notifications\tProject",
661 "Failed\tSecond thread\tProject",
662 "Working\tFix notifications\tProject",
663 "Done\tFix notifications\tProject",
664 ]);
665 expect(data.activity_expires_at).toBe(String(24 * 60 * 60 * 1000));
666 expect(androidActivityData(aggregateFor([state])).activity_expires_at).toBe(
667 String(2 * 60 * 60 * 1000),
668 );
669 });
670
671 it("fits five Unicode rows and a grouped alert in the FCM budget without corrupting text or routes", () => {
672 const longTitle = 'šŸ¤–ę¼¢å­—"\\'.repeat(30);
673 const aggregate = aggregateFor(
674 Array.from({ length: 5 }, (_, i) => ({
675 ...state,
676 threadId: ThreadId.make(`thread-${i}`),
677 threadTitle: longTitle,
678 projectTitle: longTitle,
679 })),
680 );
681 const data = fitFcmData({
682 ...androidActivityData(aggregate),
683 t3_kind: "agent_activity",
684 device_id: "d".repeat(128),
685 user_id: "u".repeat(128),
686 updated_at: "1788780000000",
687 alert_id: "a".repeat(64),
688 alert_title: "5 agents finished",
689 alert_body: Array(5).fill(longTitle).join(", "),
690 alert_path: "/threads/env/thread",
691 });
692 expect(new TextEncoder().encode(JSON.stringify(data)).length).toBeLessThanOrEqual(3800);
693 expect(data.alert_path).toBe("/threads/env/thread");
694 expect(data.activity_path).toBe("/threads/env/thread");
695 expect(data.alert_id).toBe("a".repeat(64));
696 expect(data.activity_line_4).toContain("Working\t");
697 for (const value of Object.values(data))
698 expect(new TextDecoder().decode(new TextEncoder().encode(value))).toBe(value);
699 });
700});
701
702describe("delivery policy regressions", () => {
703 it.effect("alerts a quick completion even with a previous unrelated card", () => {
704 const h = harness();
705 const old = { ...state, threadId: ThreadId.make("old"), phase: "completed" as const };
706 h.current.target.last_aggregate_json = encodeJson(aggregateFor([old]));
707 h.current.otherStates = [old];
708 h.current.state = { ...state, phase: "completed" };
709 return Effect.gen(function* () {
710 const d = yield* FcmDeliveries.FcmDeliveries;
711 yield* d.process(h.job);
712 yield* d.process({ ...h.job, state: h.current.state });
713 expect(h.sent.filter((x) => x.alert)).toHaveLength(1);
714 }).pipe(Effect.provide(h.layer));
715 });
716 it.effect("a muted environment job cannot consume another environment's attention alert", () => {
717 const h = harness();
718 const other = {
719 ...state,
720 environmentId: EnvironmentId.make("other"),
721 threadId: ThreadId.make("other"),
722 };
723 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, other]));
724 h.current.otherStates = [{ ...other, phase: "waiting_for_input" }];
725 h.current.mutedEnvironments = [state.environmentId];
726 return Effect.gen(function* () {
727 const d = yield* FcmDeliveries.FcmDeliveries;
728 yield* d.process(h.job);
729 yield* d.process({ ...h.job, state: h.current.otherStates[0]! });
730 expect(h.sent.filter((x) => x.alert)).toHaveLength(1);
731 }).pipe(Effect.provide(h.layer));
732 });
733 it("keeps an older waiting thread visible and alertable beyond five running rows", () => {
734 const running = Array.from({ length: 5 }, (_, i) => ({
735 ...state,
736 threadId: ThreadId.make(`running-${i}`),
737 }));
738 const waiting = {
739 ...state,
740 phase: "waiting_for_approval" as const,
741 updatedAt: "1969-12-31T23:59:00.000Z",
742 };
743 const next = aggregateFor([...running, waiting]);
744 expect(
745 FcmDeliveries.androidAlertForAggregate({
746 previousAggregate: aggregateFor(running),
747 nextAggregate: next,
748 preferences,
749 nowMs: 0,
750 }),
751 ).not.toBeNull();
752 });
753});
754
755describe("notification-only environments", () => {
756 it.effect("alerts independently while another environment has a live card", () => {
757 const h = harness();
758 const other = {
759 ...state,
760 environmentId: EnvironmentId.make("other"),
761 threadId: ThreadId.make("other"),
762 };
763 h.current.target.last_aggregate_json = encodeJson(aggregateFor([other]));
764 h.current.otherStates = [other];
765 h.current.state = { ...state, phase: "waiting_for_input" };
766 h.current.notificationOnlyEnvironments = [state.environmentId];
767 return Effect.gen(function* () {
768 const deliveries = yield* FcmDeliveries.FcmDeliveries;
769 yield* deliveries.process({ ...h.job, state: h.current.state });
770 expect(h.sent).toHaveLength(1);
771 expect(h.sent[0]?.alert).toBe(true);
772 expect(h.sent[0]?.data.alert_path).toBe(state.deepLink);
773 }).pipe(Effect.provide(h.layer));
774 });
775});
776
777it("continues shrinking text when a longer activity line is already minimal", () => {
778 const data = fitFcmData({
779 activity_line_0: "Approval\tšŸ˜€šŸ˜€šŸ˜€šŸ˜€\tšŸ˜€šŸ˜€šŸ˜€šŸ˜€",
780 alert_body: "x".repeat(30),
781 device_id: "x".repeat(3680),
782 });
783 expect(new TextEncoder().encode(encodeJson(data)).length).toBeLessThanOrEqual(3800);
784 expect(data.activity_line_0).toBe("Approval\tšŸ˜€šŸ˜€šŸ˜€šŸ˜€\tšŸ˜€šŸ˜€šŸ˜€šŸ˜€");
785});
786
787it.effect("notification-only jobs do not consume another environment's card alert", () => {
788 const h = harness();
789 const other = {
790 ...state,
791 environmentId: EnvironmentId.make("other"),
792 threadId: ThreadId.make("other"),
793 };
794 h.current.target.last_aggregate_json = encodeJson(aggregateFor([other]));
795 h.current.otherStates = [{ ...other, phase: "waiting_for_approval" }];
796 h.current.state = { ...state, phase: "waiting_for_input" };
797 h.current.notificationOnlyEnvironments = [state.environmentId];
798 return Effect.gen(function* () {
799 const deliveries = yield* FcmDeliveries.FcmDeliveries;
800 yield* deliveries.process({ ...h.job, state: h.current.state });
801 yield* deliveries.process({ ...h.job, state: h.current.otherStates[0]! });
802 expect(h.sent.filter((delivery) => delivery.alert)).toHaveLength(2);
803 }).pipe(Effect.provide(h.layer));
804});
805
806it.effect("preserves a structured Firebase failure through the queue consumer", () => {
807 const h = harness();
808 const failure = new FcmClient.FcmClientError({ operation: "send", status: 503 });
809 h.current.deliveryFailure = failure;
810 return Effect.gen(function* () {
811 const deliveries = yield* FcmDeliveries.FcmDeliveries;
812 expect(yield* deliveries.process(h.job).pipe(Effect.flip)).toBe(failure);
813 }).pipe(Effect.provide(h.layer));
814});
815
816it("stops reducing five-character row fields and fits the remaining alert", () => {
817 const data = fitFcmData({
818 device_id: "x".repeat(3710),
819 activity_line_0: "Approval\taaaaa\tbbbbb",
820 alert_body: "y".repeat(200),
821 });
822 expect(data.activity_line_0).toBe("Approval\taaaaa\tbbbbb");
823 expect(new TextEncoder().encode(encodeJson(data)).length).toBeLessThanOrEqual(3800);
824});
825
826describe("FCM queue message isolation", () => {
827 it.effect.each(["invalid-job", "fcm-rejection"] as const)(
828 "retries only the %s message and delivers the rest of its batch",
829 (failure) => {
830 const h = harness();
831 const outcomes = new Map<string, "ack" | "retry">();
832 const message = (id: string, body: unknown): Cloudflare.Queues.Message<unknown> => ({
833 id,
834 body,
835 timestamp: DateTime.toDateUtc(DateTime.makeUnsafe(0)),
836 attempts: 1,
837 ack: () => {
838 if (!outcomes.has(id)) outcomes.set(id, "ack");
839 },
840 retry: () => {
841 if (!outcomes.has(id)) outcomes.set(id, "retry");
842 },
843 });
844 const batch = [
845 message("failed", failure === "invalid-job" ? {} : h.job),
846 message("healthy", h.job),
847 ];
848 return Effect.gen(function* () {
849 yield* Stream.fromIterable(batch).pipe(
850 Stream.tap((item) =>
851 Effect.sync(() => {
852 h.current.deliveryFailure =
853 item.id === "failed" && failure === "fcm-rejection"
854 ? new FcmClient.FcmClientError({ operation: "send", status: 400 })
855 : null;
856 }),
857 ),
858 Stream.runForEach(FcmDeliveryQueueConsumer.processMessage),
859 );
860 // Alchemy acknowledges the batch after a successful stream. Cloudflare
861 // ignores those acknowledgements for messages explicitly retried earlier.
862 for (const item of batch) item.ack();
863 expect([...outcomes]).toEqual([
864 ["failed", "retry"],
865 ["healthy", "ack"],
866 ]);
867 expect(h.sent).toHaveLength(1);
868 expect(h.marked).toHaveLength(1);
869 }).pipe(Effect.provide(h.layer));
870 },
871 );
872});