Open History to see the last commit
1import { makeAggregateState } from "./agentActivityAggregate.ts";
2import type { RelayAgentActivityState, RelayDeliveryResult } from "@t3tools/contracts/relay";
3import { describe, expect, it } from "@effect/vitest";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Redacted from "effect/Redacted";
7import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
8import { FetchHttpClient } from "effect/http";
9
10import * as RelayConfiguration from "../Config.ts";
11import * as ApnsClient from "./ApnsClient.ts";
12import * as ApnsProviderTokens from "./ApnsProviderTokens.ts";
13import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts";
14import * as DeliveryAttempts from "./DeliveryAttempts.ts";
15import type { SignedApnsDeliveryJob } from "./apnsDeliveryJobs.ts";
16
17import * as AgentActivityRows from "./AgentActivityRows.ts";
18import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
19import * as LiveActivities from "./LiveActivities.ts";
20import * as AgentActivityPublisher from "./AgentActivityPublisher.ts";
21import * as FcmDeliveries from "./FcmDeliveries.ts";
22import * as ApnsDeliveries from "./ApnsDeliveries.ts";
23
24const layerPublisher = AgentActivityPublisher.layer.pipe(
25 Layer.provide(
26 Layer.succeed(FcmDeliveries.FcmDeliveries, {
27 enqueue: () => Effect.succeed(null),
28 process: () => Effect.void,
29 }),
30 ),
31);
32
33const state: RelayAgentActivityState = {
34 environmentId: "env" as RelayAgentActivityState["environmentId"],
35 threadId: "thread" as RelayAgentActivityState["threadId"],
36 projectTitle: "Project",
37 threadTitle: "Thread",
38 modelTitle: "gpt-5.4",
39 phase: "running",
40 headline: "Running",
41 updatedAt: "1970-01-01T00:00:00.000Z",
42 deepLink: "/threads/env/thread",
43};
44
45function target(deviceId: string): LiveActivities.TargetRow {
46 return {
47 user_id: "dev:julius",
48 device_id: deviceId,
49 platform: "ios",
50 ios_major_version: 18,
51 app_version: "1.0.0",
52 bundle_id: null,
53 aps_environment: null,
54 push_token: null,
55 push_to_start_token: "start-token",
56 preferences_json: "{}",
57 activity_push_token: null,
58 remote_start_queued_at: null,
59 remote_started_at: null,
60 ended_at: null,
61 last_aggregate_json: null,
62 last_live_activity_delivery_at: null,
63 };
64}
65
66function makeLiveActivities(
67 overrides: Partial<LiveActivities.LiveActivities["Service"]> = {},
68): LiveActivities.LiveActivities["Service"] {
69 return {
70 register: () => Effect.void,
71 listTargets: () => Effect.succeed([]),
72 listIdleArmedTargets: () => Effect.succeed([]),
73 markDelivery: () => Effect.void,
74
75 invalidateDeliveryToken: () => Effect.void,
76 ...overrides,
77 };
78}
79
80function makeAgentActivityRows(
81 overrides: Partial<AgentActivityRows.AgentActivityRows["Service"]> = {},
82): AgentActivityRows.AgentActivityRows["Service"] {
83 return {
84 upsert: () => Effect.void,
85 remove: () => Effect.void,
86 pruneTerminal: () => Effect.void,
87 listForUser: () => Effect.succeed([state]),
88 getForUserThread: () => Effect.succeed(state),
89 ...overrides,
90 };
91}
92
93function makeEnvironmentLinks(
94 overrides: Partial<EnvironmentLinks.EnvironmentLinks["Service"]> = {},
95): EnvironmentLinks.EnvironmentLinks["Service"] {
96 return {
97 upsert: () => Effect.void,
98 listDeliveryUsersForEnvironment: () =>
99 Effect.succeed([
100 {
101 userId: "dev:julius",
102 notificationsEnabled: true,
103 liveActivitiesEnabled: true,
104 },
105 ]),
106 listForUser: () => Effect.succeed([]),
107 getForUser: () => Effect.succeed(null),
108 findActiveManagedForEnvironment: () => Effect.succeed([]),
109 setHoldWebhooksWhileOffline: () => Effect.void,
110 revokeForUser: () => Effect.succeed(false),
111 ...overrides,
112 };
113}
114
115function makeApnsDeliveries(
116 overrides: Partial<ApnsDeliveries.ApnsDeliveries["Service"]> = {},
117): ApnsDeliveries.ApnsDeliveries["Service"] {
118 return {
119 processBatch: () => Effect.void,
120 sendForTarget: () => Effect.succeed(null),
121 sendLiveActivity: () =>
122 Effect.succeed({
123 deviceId: "device",
124 kind: "live_activity_start",
125 ok: true,
126 apnsStatus: 200,
127 apnsReason: null,
128 apnsId: "apns-id",
129 }),
130 sendPushNotification: () =>
131 Effect.succeed({
132 deviceId: "device",
133 kind: "push_notification",
134 ok: true,
135 apnsStatus: 200,
136 apnsReason: null,
137 apnsId: "apns-id",
138 }),
139 processSignedJob: () =>
140 Effect.succeed({
141 deviceId: "device",
142 kind: "live_activity_start",
143 ok: true,
144 apnsStatus: 200,
145 apnsReason: null,
146 apnsId: "apns-id",
147 }),
148 ...overrides,
149 };
150}
151
152describe("AgentActivityPublisher", () => {
153 it.effect("routes Android publication and registration replay to FCM alongside iOS", () => {
154 const android = { ...target("android"), platform: "android" as const, ios_major_version: null };
155 const ios = target("ios");
156 const fcmCalls: Array<Parameters<FcmDeliveries.FcmDeliveries["Service"]["enqueue"]>[0]> = [];
157 const appleDevices: string[] = [];
158 return Effect.gen(function* () {
159 const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
160 yield* publisher.publish({
161 environmentId: state.environmentId,
162 environmentPublicKey: "key",
163 threadId: state.threadId,
164 state,
165 });
166 yield* publisher.replayForLiveActivityRegistration({
167 userId: android.user_id,
168 deviceId: android.device_id,
169 });
170 expect(fcmCalls).toEqual([
171 { target: android, state },
172 { target: android, state: null, replay: true },
173 ]);
174 expect(appleDevices).toEqual(["ios"]);
175 }).pipe(
176 Effect.provide(
177 AgentActivityPublisher.layer.pipe(
178 Layer.provide(
179 Layer.mergeAll(
180 Layer.succeed(AgentActivityRows.AgentActivityRows, makeAgentActivityRows()),
181 Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()),
182 Layer.succeed(
183 LiveActivities.LiveActivities,
184 makeLiveActivities({ listTargets: () => Effect.succeed([android, ios]) }),
185 ),
186 Layer.succeed(
187 ApnsDeliveries.ApnsDeliveries,
188 makeApnsDeliveries({
189 sendForTarget: (input) =>
190 Effect.sync(() => {
191 appleDevices.push(input.target.device_id);
192 return null;
193 }),
194 }),
195 ),
196 Layer.succeed(FcmDeliveries.FcmDeliveries, {
197 enqueue: (input) =>
198 Effect.sync(() => {
199 fcmCalls.push(input);
200 return null;
201 }),
202 process: () => Effect.void,
203 }),
204 ),
205 ),
206 ),
207 ),
208 );
209 });
210
211 it.effect.each([
212 {
213 name: "skips the activity query without a running iOS Live Activity",
214 activityToken: null,
215 listCalls: 0,
216 },
217 {
218 name: "aggregates activity for a running iOS Live Activity",
219 activityToken: "activity-token",
220 listCalls: 1,
221 },
222 ])("$name", ({ activityToken, listCalls }) => {
223 const android = { ...target("android"), platform: "android" as const, ios_major_version: null };
224 const ios = { ...target("ios"), activity_push_token: activityToken };
225 let listed = 0;
226 const hasAggregate: boolean[] = [];
227 return Effect.gen(function* () {
228 const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
229 yield* publisher.publish({
230 environmentId: state.environmentId,
231 environmentPublicKey: "key",
232 threadId: state.threadId,
233 state,
234 });
235 expect(listed).toBe(listCalls);
236 expect(hasAggregate).toEqual([listCalls > 0]);
237 }).pipe(
238 Effect.provide(
239 AgentActivityPublisher.layer.pipe(
240 Layer.provide(
241 Layer.mergeAll(
242 Layer.succeed(
243 AgentActivityRows.AgentActivityRows,
244 makeAgentActivityRows({
245 listForUser: () =>
246 Effect.sync(() => {
247 listed += 1;
248 return [state];
249 }),
250 }),
251 ),
252 Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()),
253 Layer.succeed(
254 LiveActivities.LiveActivities,
255 makeLiveActivities({ listTargets: () => Effect.succeed([android, ios]) }),
256 ),
257 Layer.succeed(
258 ApnsDeliveries.ApnsDeliveries,
259 makeApnsDeliveries({
260 sendForTarget: (input) =>
261 Effect.sync(() => {
262 hasAggregate.push(input.aggregate !== null);
263 return null;
264 }),
265 }),
266 ),
267 Layer.succeed(FcmDeliveries.FcmDeliveries, {
268 enqueue: () => Effect.succeed(null),
269 process: () => Effect.void,
270 }),
271 ),
272 ),
273 ),
274 ),
275 );
276 });
277
278 it.effect("replays the latest aggregate when a Live Activity token registers", () => {
279 const registeredTarget: LiveActivities.TargetRow = {
280 ...target("device-1"),
281 push_to_start_token: null,
282 activity_push_token: "activity-token",
283 remote_start_queued_at: null,
284 remote_started_at: "1970-01-01T00:00:01.000Z",
285 };
286 const sent: Array<Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]> =
287 [];
288 const deliveryResult: RelayDeliveryResult = {
289 deviceId: "device-1",
290 kind: "live_activity_update",
291 ok: true,
292 apnsStatus: null,
293 apnsReason: null,
294 apnsId: "queued",
295 };
296
297 return Effect.gen(function* () {
298 const result = yield* Effect.gen(function* () {
299 const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
300 return yield* publisher.replayForLiveActivityRegistration({
301 userId: "dev:julius",
302 deviceId: "device-1",
303 });
304 }).pipe(
305 Effect.provide(
306 layerPublisher.pipe(
307 Layer.provide(
308 Layer.mergeAll(
309 Layer.succeed(AgentActivityRows.AgentActivityRows, makeAgentActivityRows()),
310 Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()),
311 Layer.succeed(
312 LiveActivities.LiveActivities,
313 makeLiveActivities({
314 listTargets: () => Effect.succeed([registeredTarget, target("device-2")]),
315 }),
316 ),
317 Layer.succeed(
318 ApnsDeliveries.ApnsDeliveries,
319 makeApnsDeliveries({
320 sendForTarget: (input) =>
321 Effect.sync(() => {
322 sent.push(input);
323 return deliveryResult;
324 }),
325 }),
326 ),
327 ),
328 ),
329 ),
330 ),
331 );
332
333 expect(result).toEqual(deliveryResult);
334 expect(sent).toHaveLength(1);
335 expect(sent[0]?.target.device_id).toBe("device-1");
336 expect(sent[0]?.aggregate).toMatchObject({
337 activeCount: 1,
338 activities: [
339 {
340 environmentId: state.environmentId,
341 threadId: state.threadId,
342 status: "Working",
343 },
344 ],
345 });
346 });
347 });
348
349 it.effect.each([
350 { phase: "waiting_for_input", pushToStartToken: null, body: "Input: Project" },
351 { phase: "waiting_for_approval", pushToStartToken: "start-token", body: "Approval: Project" },
352 ] as const)(
353 "queues $phase notifications for notification-only links and respects link opt-out",
354 ({ phase, pushToStartToken, body }) => {
355 const publishedState: RelayAgentActivityState = { ...state, phase };
356 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
357 let notificationsEnabled = true;
358 let liveActivitiesEnabled = false;
359 const dependencies = Layer.mergeAll(
360 Layer.succeed(
361 AgentActivityRows.AgentActivityRows,
362 makeAgentActivityRows({
363 listForUser: () =>
364 liveActivitiesEnabled
365 ? Effect.succeed([publishedState])
366 : Effect.die("Notification-only delivery must not read rows"),
367 }),
368 ),
369 Layer.succeed(
370 EnvironmentLinks.EnvironmentLinks,
371 makeEnvironmentLinks({
372 listDeliveryUsersForEnvironment: () =>
373 Effect.succeed([
374 { userId: "dev:julius", notificationsEnabled, liveActivitiesEnabled },
375 ]),
376 }),
377 ),
378 Layer.succeed(
379 LiveActivities.LiveActivities,
380 makeLiveActivities({
381 listTargets: () =>
382 Effect.succeed([
383 {
384 ...target("device-1"),
385 push_token: "apns-device-token",
386 push_to_start_token: pushToStartToken,
387 preferences_json: JSON.stringify({
388 liveActivitiesEnabled: true,
389 notificationsEnabled: true,
390 notifyOnApproval: true,
391 notifyOnInput: true,
392 notifyOnCompletion: true,
393 notifyOnFailure: true,
394 }),
395 },
396 ]),
397 }),
398 ),
399 Layer.succeed(ApnsDeliveryQueue.ApnsDeliveryQueueSender, {
400 send: (job) =>
401 Effect.sync(() => {
402 queuedJobs.push(job);
403 }),
404 }),
405 Layer.succeed(DeliveryAttempts.DeliveryAttempts, {
406 pruneExpired: Effect.succeed(0),
407 claimSourceJob: () => Effect.succeed("claimed" as const),
408 completeSourceJob: () => Effect.void,
409 }),
410 RelayConfiguration.layer({
411 relayIssuer: "https://relay.example.test",
412 apns: {
413 environment: "sandbox",
414 teamId: "team-id",
415 keyId: "key-id",
416 privateKey: Redacted.make("unused-key"),
417 bundleId: "com.t3tools.t3code.dev",
418 },
419 apnsDeliveryJobSigningSecret: Redacted.make("job-signing-secret"),
420 clerkSecretKey: Redacted.make("clerk-secret"),
421 clerkPublishableKey: "pk_test_test",
422 clerkJwtAudience: "t3-code-relay",
423 cloudMintPrivateKey: Redacted.make("cloud-private-key"),
424 cloudMintPublicKey: "cloud-public-key",
425 managedEndpointBaseDomain: undefined,
426 managedEndpointNamespace: undefined,
427 }),
428 );
429 const deliveriesLayer = ApnsDeliveries.layer.pipe(
430 Layer.provide(NodeCryptoLayer.layerMac),
431 Layer.provide(
432 ApnsClient.layer.pipe(
433 Layer.provide(ApnsProviderTokens.layer),
434 Layer.provide(FetchHttpClient.layer),
435 ),
436 ),
437 Layer.provide(ApnsDeliveryQueue.layer.pipe(Layer.provide(NodeCryptoLayer.layer))),
438 );
439
440 return Effect.gen(function* () {
441 const publisher = yield* AgentActivityPublisher.AgentActivityPublisher;
442 const publish = () =>
443 publisher.publish({
444 environmentId: state.environmentId,
445 environmentPublicKey: "environment-public-key",
446 threadId: state.threadId,
447 state: publishedState,
448 });
449 expect((yield* publish()).deliveries).toMatchObject([
450 { deviceId: "device-1", kind: "push_notification", queued: true },
451 ]);
452 expect(queuedJobs.map((job) => job.payload)).toMatchObject([
453 { kind: "push_notification", notification: { body, threadId: state.threadId } },
454 ]);
455
456 notificationsEnabled = false;
457 liveActivitiesEnabled = true;
458 expect((yield* publish()).deliveries).toEqual([]);
459 expect(queuedJobs).toHaveLength(1);
460 }).pipe(
461 Effect.provide(
462 layerPublisher.pipe(Layer.provide(deliveriesLayer), Layer.provide(dependencies)),
463 ),
464 );
465 },
466 );
467});
468
469describe("isExpiredAgentActivityState", () => {
470 it("treats rows with unparseable timestamps as expired", () => {
471 expect(
472 AgentActivityPublisher.isExpiredAgentActivityState({ ...state, updatedAt: "not-a-date" }, 0),
473 ).toBe(true);
474 });
475});
476
477describe("makeAggregateState", () => {
478 const hourMs = 60 * 60 * 1_000;
479
480 it("drops expired rows from the aggregate", () => {
481 const fresh: RelayAgentActivityState = {
482 ...state,
483 threadId: "thread-fresh" as RelayAgentActivityState["threadId"],
484 updatedAt: "1970-01-01T03:00:00.000Z",
485 };
486 const aggregate = makeAggregateState({
487 activeStates: [state, fresh],
488 terminalState: null,
489 nowMs: 3 * hourMs,
490 });
491
492 expect(aggregate?.activeCount).toBe(1);
493 expect(aggregate?.activities).toMatchObject([{ threadId: "thread-fresh" }]);
494 });
495
496 it("returns null when every row has expired and nothing terminal remains", () => {
497 expect(
498 makeAggregateState({
499 activeStates: [state],
500 terminalState: null,
501 nowMs: 3 * hourMs,
502 }),
503 ).toBeNull();
504 });
505
506 it("still reports the terminal state when active rows have expired", () => {
507 const terminalState: RelayAgentActivityState = {
508 ...state,
509 phase: "completed",
510 updatedAt: "1970-01-01T03:00:00.000Z",
511 };
512 const aggregate = makeAggregateState({
513 activeStates: [state],
514 terminalState,
515 nowMs: 3 * hourMs,
516 });
517
518 expect(aggregate?.activeCount).toBe(0);
519 expect(aggregate?.activities).toMatchObject([{ phase: "completed" }]);
520 });
521
522 it("keeps a recently finished thread visible as Done beside active agents", () => {
523 const active: RelayAgentActivityState = {
524 ...state,
525 threadId: "thread-active" as RelayAgentActivityState["threadId"],
526 updatedAt: "1970-01-01T00:58:00.000Z",
527 };
528 const justCompleted: RelayAgentActivityState = {
529 ...state,
530 threadId: "thread-done" as RelayAgentActivityState["threadId"],
531 phase: "completed",
532 updatedAt: "1970-01-01T00:59:00.000Z",
533 };
534 const aggregate = makeAggregateState({
535 activeStates: [active, justCompleted],
536 terminalState: null,
537 nowMs: hourMs,
538 });
539
540 expect(aggregate?.activeCount).toBe(1);
541 expect(aggregate?.subtitle).toBe("Agent work in progress");
542 expect(aggregate?.activities).toMatchObject([
543 { threadId: "thread-active", phase: "running" },
544 { threadId: "thread-done", phase: "completed", status: "Done" },
545 ]);
546 expect(aggregate?.updatedAt).toBe("1970-01-01T00:59:00.000Z");
547 });
548
549 it("keeps showing recently finished work when nothing is active", () => {
550 const lingeringCompleted: RelayAgentActivityState = {
551 ...state,
552 phase: "completed",
553 updatedAt: "1970-01-01T00:59:00.000Z",
554 };
555 const aggregate = makeAggregateState({
556 activeStates: [lingeringCompleted],
557 terminalState: null,
558 nowMs: hourMs,
559 });
560
561 expect(aggregate).toMatchObject({
562 activeCount: 0,
563 subtitle: "Agent work completed",
564 activities: [{ phase: "completed", status: "Done" }],
565 });
566 expect(
567 makeAggregateState({
568 activeStates: [{ ...lingeringCompleted, updatedAt: "1970-01-01T00:44:00.000Z" }],
569 terminalState: null,
570 nowMs: hourMs,
571 }),
572 ).toBeNull();
573 });
574
575 it("gives active agents the display slots before finished ones", () => {
576 const mkActive = (id: string): RelayAgentActivityState => ({
577 ...state,
578 threadId: id as RelayAgentActivityState["threadId"],
579 updatedAt: "1970-01-01T00:58:00.000Z",
580 });
581 const justCompleted: RelayAgentActivityState = {
582 ...state,
583 threadId: "thread-done" as RelayAgentActivityState["threadId"],
584 phase: "completed",
585 updatedAt: "1970-01-01T00:59:00.000Z",
586 };
587 const aggregate = makeAggregateState({
588 activeStates: [
589 mkActive("a-1"),
590 mkActive("a-2"),
591 mkActive("a-3"),
592 mkActive("a-4"),
593 mkActive("a-5"),
594 justCompleted,
595 ],
596 terminalState: null,
597 nowMs: hourMs,
598 });
599
600 expect(aggregate?.activeCount).toBe(5);
601 expect(aggregate?.activities).toMatchObject([
602 { threadId: "a-1" },
603 { threadId: "a-2" },
604 { threadId: "a-3" },
605 { threadId: "a-4" },
606 { threadId: "a-5" },
607 ]);
608 });
609});