authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-08 16:39:48-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-08 16:39:48-07:00
logfdf34c4018352df41aa22fbaa774c25c6c1ff7da
treeed787dce2fc05115ac7e37ad8133bde5a0b730e0
parent20e2e899ecded6b30e12fa23c8f18847c2bf12e1
signature Signed by PGP key B5690EEEBB952194

fix(relay): share notification policy and prioritize waiting agents (#10848)

Co-authored-by: Ryan Hughes <ryan@heyoodle.com>

6 files changed, 448 insertions(+), 303 deletions(-)

infra/relay/src/agentActivity/AgentActivityPublisher.ts+6-126
......@@ -1,5 +1,9 @@
1import { makeAggregateState } from "./agentActivityAggregate.ts";
2export {
3 makeAggregateState,
4 TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS,
5} from "./agentActivityAggregate.ts";
16import type {
2 RelayAgentActivityAggregateState,
37 RelayAgentActivityState,
48 RelayDeliveryResult,
59 RelayPublishResponse,
......@@ -8,14 +12,8 @@ import * as Context from "effect/Context";
812import * as DateTime from "effect/DateTime";
913import * as Effect from "effect/Effect";
1014import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
1215
13import {
14 isExpiredAgentActivityState,
15 isTerminalPhase,
16 MAX_ACTIVITY_ROWS,
17 sanitizeAgentActivityAggregateState,
18} from "./agentActivityPayloads.ts";
16import { isTerminalPhase } from "./agentActivityPayloads.ts";
1917
2018export { isExpiredAgentActivityState } from "./agentActivityPayloads.ts";
2119import * as AgentActivityRows from "./AgentActivityRows.ts";
......@@ -183,122 +181,4 @@ export const make = Effect.gen(function* () {
183181 });
184182});
185183
186function statusForPhase(phase: RelayAgentActivityState["phase"]): string {
187 switch (phase) {
188 case "waiting_for_approval":
189 return "Approval";
190 case "waiting_for_input":
191 return "Input";
192 case "completed":
193 return "Done";
194 case "failed":
195 return "Failed";
196 case "starting":
197 // Matches the web sidebar's pill wording (Sidebar.logic.ts) so the same
198 // thread reads the same across surfaces.
199 return "Connecting";
200 case "running":
201 return "Working";
202 case "stale":
203 return "Waiting";
204 }
205}
206
207function aggregateRowForState(state: RelayAgentActivityState) {
208 return {
209 environmentId: state.environmentId,
210 threadId: state.threadId,
211 projectTitle: state.projectTitle,
212 threadTitle: state.threadTitle,
213 modelTitle: state.modelTitle,
214 phase: state.phase,
215 status: statusForPhase(state.phase),
216 updatedAt: state.updatedAt,
217 deepLink: state.deepLink,
218 };
219}
220
221function terminalAggregateState(state: RelayAgentActivityState): RelayAgentActivityAggregateState {
222 return sanitizeAgentActivityAggregateState({
223 title: "T3 Code",
224 subtitle: state.phase === "failed" ? "Agent work failed" : "Agent work completed",
225 activeCount: 0,
226 updatedAt: state.updatedAt,
227 activities: [aggregateRowForState(state)],
228 });
229}
230
231// How long a finished thread keeps its Done/Failed row in the aggregate while
232// other agents are still active. Long enough to be seen on the lock screen,
233// short enough that the activity list stays about live work.
234export const TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS = 15 * 60 * 1_000;
235
236function isRecentTerminalState(state: RelayAgentActivityState, nowMs: number): boolean {
237 if (!isTerminalPhase(state)) {
238 return false;
239 }
240 const updatedAtMs = Option.match(DateTime.make(state.updatedAt), {
241 onNone: () => Number.NaN,
242 onSome: (dt) => dt.epochMilliseconds,
243 });
244 if (Number.isNaN(updatedAtMs)) {
245 return false;
246 }
247 return nowMs - updatedAtMs <= TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS;
248}
249
250export function makeAggregateState(input: {
251 readonly activeStates: ReadonlyArray<RelayAgentActivityState>;
252 readonly terminalState: RelayAgentActivityState | null;
253 readonly nowMs: number;
254}): RelayAgentActivityAggregateState | null {
255 const activeStates = input.activeStates.filter(
256 (state) => !isTerminalPhase(state) && !isExpiredAgentActivityState(state, input.nowMs),
257 );
258 if (activeStates.length === 0) {
259 if (input.terminalState !== null) {
260 return terminalAggregateState(input.terminalState);
261 }
262 // With no live work, recently finished threads keep the card showing
263 // Done/Failed content (an armed card never renders an empty state). The
264 // newly-terminal alert rules key off the previously delivered aggregate,
265 // so replays repaint this without buzzing. Once the terminal rows age
266 // out, the aggregate is null and the delivery layer ends the card.
267 const recentTerminal = input.activeStates
268 .filter((state) => isRecentTerminalState(state, input.nowMs))
269 .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
270 const newest = recentTerminal[0];
271 if (!newest) {
272 return null;
273 }
274 return sanitizeAgentActivityAggregateState({
275 title: "T3 Code",
276 subtitle: newest.phase === "failed" ? "Agent work failed" : "Agent work completed",
277 activeCount: 0,
278 updatedAt: newest.updatedAt,
279 activities: recentTerminal.slice(0, MAX_ACTIVITY_ROWS).map(aggregateRowForState),
280 });
281 }
282 // Recently finished threads ride along after the active ones (display slots
283 // permitting) so a completion is visible as Done/Failed instead of the row
284 // silently vanishing while other agents keep the activity alive.
285 const recentTerminalStates = input.activeStates
286 .filter((state) => isRecentTerminalState(state, input.nowMs))
287 .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
288 const displayedStates = [
289 ...activeStates.slice(0, MAX_ACTIVITY_ROWS),
290 ...recentTerminalStates,
291 ].slice(0, MAX_ACTIVITY_ROWS);
292 const updatedAt = [...activeStates, ...recentTerminalStates].reduce((latest, state) =>
293 state.updatedAt.localeCompare(latest.updatedAt) > 0 ? state : latest,
294 ).updatedAt;
295 return sanitizeAgentActivityAggregateState({
296 title: "T3 Code",
297 subtitle: "Agent work in progress",
298 activeCount: activeStates.length,
299 updatedAt,
300 activities: displayedStates.map(aggregateRowForState),
301 });
302}
303
304184export const layer = Layer.effect(AgentActivityPublisher, make);
infra/relay/src/agentActivity/ApnsDeliveries.ts+18-171
......@@ -20,6 +20,7 @@ import * as Schema from "effect/Schema";
2020import {
2121 isExpiredAgentActivityState,
2222 isTerminalPhase,
23 notificationForActivity,
2324 sanitizeAgentActivityAggregateState,
2425 sanitizeApnsNotificationPayload,
2526} from "./agentActivityPayloads.ts";
......@@ -42,6 +43,19 @@ import * as RelayConfiguration from "../Config.ts";
4243import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts";
4344import { withSpanAttributes } from "../observability.ts";
4445
46import {
47 alertForAttentionTransition,
48 alertForNewlyTerminal,
49 alertForTerminalAggregate,
50 newlyTerminalRows,
51 shouldAlertForActivity,
52} from "./agentActivityAlerts.ts";
53export {
54 alertForAttentionTransition,
55 alertForNewlyTerminal,
56 alertForTerminalAggregate,
57} from "./agentActivityAlerts.ts";
58
4559const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000;
4660// How long a just-armed card may sit with an empty aggregate before an end is
4761// warranted; covers the gap between arming on send and the environment's
......@@ -146,148 +160,6 @@ function aggregateNeedsAttention(aggregate: RelayAgentActivityAggregateState): b
146160 );
147161}
148162
149function isAttentionPhase(phase: string): boolean {
150 return phase === "waiting_for_approval" || phase === "waiting_for_input";
151}
152
153// Honors the same per-event notification switches the push channel uses; a
154// missing/corrupt preferences blob only disables nothing (matching how the
155// liveActivitiesEnabled check treats it), since every registration writes one.
156function alertAllowedForPhase(
157 preferences: RelayAgentAwarenessPreferences | null,
158 phase: string,
159): boolean {
160 if (preferences === null) {
161 return true;
162 }
163 switch (phase) {
164 case "waiting_for_approval":
165 return preferences.notifyOnApproval;
166 case "waiting_for_input":
167 return preferences.notifyOnInput;
168 case "completed":
169 return preferences.notifyOnCompletion;
170 case "failed":
171 return preferences.notifyOnFailure;
172 default:
173 return false;
174 }
175}
176
177// Alert copy for an update whose aggregate contains threads that were NOT in an
178// attention phase in the previously delivered aggregate. A null previous
179// aggregate means there is no known baseline (fresh registration, replay after
180// data loss) — alerting there would buzz on reconnect, not on a transition.
181export function alertForAttentionTransition(input: {
182 readonly previousAggregate: RelayAgentActivityAggregateState | null;
183 readonly nextAggregate: RelayAgentActivityAggregateState;
184 readonly preferences: RelayAgentAwarenessPreferences | null;
185}): ApnsLiveActivityAlert | null {
186 if (input.previousAggregate === null) {
187 return null;
188 }
189 const previouslyAttention = new Set(
190 input.previousAggregate.activities
191 .filter((row) => isAttentionPhase(row.phase))
192 .map((row) => row.threadId),
193 );
194 const newlyAttention = input.nextAggregate.activities.filter(
195 (row) =>
196 isAttentionPhase(row.phase) &&
197 !previouslyAttention.has(row.threadId) &&
198 alertAllowedForPhase(input.preferences, row.phase),
199 );
200 const first = newlyAttention[0];
201 if (!first) {
202 return null;
203 }
204 if (newlyAttention.length === 1) {
205 return { title: first.threadTitle, body: `${first.status}: ${first.projectTitle}` };
206 }
207 return {
208 title: `${newlyAttention.length} agents need attention`,
209 body: newlyAttention.map((row) => row.threadTitle).join(", "),
210 };
211}
212
213// Alert copy for an update whose aggregate contains threads that finished
214// (Done/Failed) since the previously delivered aggregate — the mid-flight
215// completion buzz while other agents keep the activity alive. Requires the
216// thread to have been present and non-terminal before, so a baseline-less
217// replay or a row that merely fell off the display cap never rings.
218function newlyTerminalRows(
219 previousAggregate: RelayAgentActivityAggregateState | null,
220 nextAggregate: RelayAgentActivityAggregateState,
221): ReadonlyArray<RelayAgentActivityAggregateState["activities"][number]> {
222 if (previousAggregate === null) {
223 return [];
224 }
225 const previousPhases = new Map(
226 previousAggregate.activities.map((row) => [row.threadId, row.phase]),
227 );
228 return nextAggregate.activities.filter((row) => {
229 if (row.phase !== "completed" && row.phase !== "failed") {
230 return false;
231 }
232 const previousPhase = previousPhases.get(row.threadId);
233 return (
234 previousPhase !== undefined && previousPhase !== "completed" && previousPhase !== "failed"
235 );
236 });
237}
238
239function isFreshTerminalRow(
240 row: RelayAgentActivityAggregateState["activities"][number],
241 nowMs: number,
242): boolean {
243 const updatedAtMs = Option.match(DateTime.make(row.updatedAt), {
244 onNone: () => null,
245 onSome: (dt) => dt.epochMilliseconds,
246 });
247 return updatedAtMs !== null && nowMs - updatedAtMs <= TERMINAL_NOTIFICATION_FRESHNESS_MS;
248}
249
250export function alertForNewlyTerminal(input: {
251 readonly previousAggregate: RelayAgentActivityAggregateState | null;
252 readonly nextAggregate: RelayAgentActivityAggregateState;
253 readonly preferences: RelayAgentAwarenessPreferences | null;
254 readonly nowMs: number;
255}): ApnsLiveActivityAlert | null {
256 const newlyTerminal = newlyTerminalRows(input.previousAggregate, input.nextAggregate).filter(
257 (row) =>
258 alertAllowedForPhase(input.preferences, row.phase) &&
259 // Replays of old aggregates (server restarts, redeliveries) repaint
260 // state without ringing; only fresh completions buzz.
261 isFreshTerminalRow(row, input.nowMs),
262 );
263 const first = newlyTerminal[0];
264 if (!first) {
265 return null;
266 }
267 if (newlyTerminal.length === 1) {
268 return { title: first.threadTitle, body: `${first.status}: ${first.projectTitle}` };
269 }
270 return {
271 title: `${newlyTerminal.length} agents finished`,
272 body: newlyTerminal.map((row) => row.threadTitle).join(", "),
273 };
274}
275
276// Alert copy for an end event carrying a terminal (Done/Failed) aggregate.
277export function alertForTerminalAggregate(input: {
278 readonly aggregate: RelayAgentActivityAggregateState | null;
279 readonly preferences: RelayAgentAwarenessPreferences | null;
280}): ApnsLiveActivityAlert | null {
281 const row = input.aggregate?.activities[0];
282 if (!row || (row.phase !== "completed" && row.phase !== "failed")) {
283 return null;
284 }
285 if (!alertAllowedForPhase(input.preferences, row.phase)) {
286 return null;
287 }
288 return { title: row.threadTitle, body: `${row.status}: ${row.projectTitle}` };
289}
290
291163function shouldUpdateLiveActivity(input: {
292164 readonly previousAggregate: RelayAgentActivityAggregateState | null;
293165 readonly nextAggregate: RelayAgentActivityAggregateState;
......@@ -328,7 +200,6 @@ function shouldUpdateLiveActivity(input: {
328200
329201// Completions replayed long after the fact (server restarts republish every
330202// recently-finished thread) must not ring the device again.
331const TERMINAL_NOTIFICATION_FRESHNESS_MS = 2 * 60 * 1_000;
332203
333204function notificationForAggregate(input: {
334205 readonly target: LiveActivities.TargetRow;
......@@ -346,32 +217,8 @@ function notificationForAggregate(input: {
346217 if (!activity) {
347218 return null;
348219 }
349 if (activity.phase === "completed" || activity.phase === "failed") {
350 const updatedAtMs = Option.match(DateTime.make(activity.updatedAt), {
351 onNone: () => null,
352 onSome: (dt) => dt.epochMilliseconds,
353 });
354 if (updatedAtMs === null || input.nowMs - updatedAtMs > TERMINAL_NOTIFICATION_FRESHNESS_MS) {
355 return null;
356 }
357 }
358 const enabled =
359 (activity.phase === "waiting_for_approval" && preferences.notifyOnApproval) ||
360 (activity.phase === "waiting_for_input" && preferences.notifyOnInput) ||
361 (activity.phase === "completed" && preferences.notifyOnCompletion) ||
362 (activity.phase === "failed" && preferences.notifyOnFailure);
363 if (!enabled) {
364 return null;
365 }
366 return {
367 title: activity.threadTitle,
368 body: `${activity.status}: ${activity.projectTitle}`,
369 environmentId: activity.environmentId,
370 threadId: activity.threadId,
371 deepLink: activity.deepLink,
372 phase: activity.phase,
373 updatedAt: activity.updatedAt,
374 };
220 if (!shouldAlertForActivity({ ...activity, preferences, nowMs: input.nowMs })) return null;
221 return notificationForActivity(activity);
375222}
376223
377224// "suppressed" means a Live Activity owns this state but no update is due
......@@ -584,9 +431,9 @@ interface LiveActivityDeliveryTarget {
584431// DeviceTokenNotForTopic/BadDeviceToken, so per-device values override the
585432// relay-wide defaults when present.
586433function credentialsForTarget(
587 credentials: RelayConfiguration.RelayConfiguration["Service"]["apns"],
434 credentials: RelayConfiguration.ApnsCredentials,
588435 target: LiveActivityDeliveryTarget,
589): RelayConfiguration.RelayConfiguration["Service"]["apns"] {
436): RelayConfiguration.ApnsCredentials {
590437 return {
591438 ...credentials,
592439 ...(target.bundle_id ? { bundleId: target.bundle_id } : {}),
infra/relay/src/agentActivity/agentActivityAggregate.ts created+139
......@@ -0,0 +1,139 @@
1import type {
2 RelayAgentActivityAggregateState,
3 RelayAgentActivityState,
4} from "@t3tools/contracts/relay";
5import * as DateTime from "effect/DateTime";
6import * as Option from "effect/Option";
7import {
8 isExpiredAgentActivityState,
9 isTerminalPhase,
10 MAX_ACTIVITY_ROWS,
11 sanitizeAgentActivityAggregateState,
12} from "./agentActivityPayloads.ts";
13
14export function statusForPhase(phase: RelayAgentActivityState["phase"]): string {
15 switch (phase) {
16 case "waiting_for_approval":
17 return "Approval";
18 case "waiting_for_input":
19 return "Input";
20 case "completed":
21 return "Done";
22 case "failed":
23 return "Failed";
24 case "starting":
25 // Matches the web sidebar's pill wording (Sidebar.logic.ts) so the same
26 // thread reads the same across surfaces.
27 return "Connecting";
28 case "running":
29 return "Working";
30 case "stale":
31 return "Waiting";
32 }
33}
34
35function aggregateRowForState(state: RelayAgentActivityState) {
36 return {
37 environmentId: state.environmentId,
38 threadId: state.threadId,
39 projectTitle: state.projectTitle,
40 threadTitle: state.threadTitle,
41 modelTitle: state.modelTitle,
42 phase: state.phase,
43 status: statusForPhase(state.phase),
44 updatedAt: state.updatedAt,
45 deepLink: state.deepLink,
46 };
47}
48
49function terminalAggregateState(state: RelayAgentActivityState): RelayAgentActivityAggregateState {
50 return sanitizeAgentActivityAggregateState({
51 title: "T3 Code",
52 subtitle: state.phase === "failed" ? "Agent work failed" : "Agent work completed",
53 activeCount: 0,
54 updatedAt: state.updatedAt,
55 activities: [aggregateRowForState(state)],
56 });
57}
58
59// How long a finished thread keeps its Done/Failed row in the aggregate while
60// other agents are still active. Long enough to be seen on the lock screen,
61// short enough that the activity list stays about live work.
62export const TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS = 15 * 60 * 1_000;
63
64function isRecentTerminalState(state: RelayAgentActivityState, nowMs: number): boolean {
65 if (!isTerminalPhase(state)) {
66 return false;
67 }
68 const updatedAtMs = Option.match(DateTime.make(state.updatedAt), {
69 onNone: () => Number.NaN,
70 onSome: (dt) => dt.epochMilliseconds,
71 });
72 if (Number.isNaN(updatedAtMs)) {
73 return false;
74 }
75 return nowMs - updatedAtMs <= TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS;
76}
77
78export function makeAggregateState(input: {
79 readonly activeStates: ReadonlyArray<RelayAgentActivityState>;
80 readonly terminalState: RelayAgentActivityState | null;
81 readonly nowMs: number;
82}): RelayAgentActivityAggregateState | null {
83 const activeStates = input.activeStates.filter(
84 (state) => !isTerminalPhase(state) && !isExpiredAgentActivityState(state, input.nowMs),
85 );
86 if (activeStates.length === 0) {
87 if (input.terminalState !== null) {
88 return terminalAggregateState(input.terminalState);
89 }
90 // With no live work, recently finished threads keep the card showing
91 // Done/Failed content (an armed card never renders an empty state). The
92 // newly-terminal alert rules key off the previously delivered aggregate,
93 // so replays repaint this without buzzing. Once the terminal rows age
94 // out, the aggregate is null and the delivery layer ends the card.
95 const recentTerminal = input.activeStates
96 .filter((state) => isRecentTerminalState(state, input.nowMs))
97 .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
98 const newest = recentTerminal[0];
99 if (!newest) {
100 return null;
101 }
102 return sanitizeAgentActivityAggregateState({
103 title: "T3 Code",
104 subtitle: newest.phase === "failed" ? "Agent work failed" : "Agent work completed",
105 activeCount: 0,
106 updatedAt: newest.updatedAt,
107 activities: recentTerminal.slice(0, MAX_ACTIVITY_ROWS).map(aggregateRowForState),
108 });
109 }
110 // Recently finished threads ride along after the active ones (display slots
111 // permitting) so a completion is visible as Done/Failed instead of the row
112 // silently vanishing while other agents keep the activity alive.
113 const recentTerminalStates = input.activeStates
114 .filter((state) => isRecentTerminalState(state, input.nowMs))
115 .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
116 const displayedStates = [
117 ...activeStates
118 .toSorted((a, b) => activityPhasePriority(a.phase) - activityPhasePriority(b.phase))
119 .slice(0, MAX_ACTIVITY_ROWS),
120 ...recentTerminalStates,
121 ].slice(0, MAX_ACTIVITY_ROWS);
122 const updatedAt = [...activeStates, ...recentTerminalStates].reduce((latest, state) =>
123 state.updatedAt.localeCompare(latest.updatedAt) > 0 ? state : latest,
124 ).updatedAt;
125 return sanitizeAgentActivityAggregateState({
126 title: "T3 Code",
127 subtitle: "Agent work in progress",
128 activeCount: activeStates.length,
129 updatedAt,
130 activities: displayedStates.map(aggregateRowForState),
131 });
132}
133
134export function activityPhasePriority(phase: RelayAgentActivityState["phase"]): number {
135 if (phase === "waiting_for_approval" || phase === "waiting_for_input") return 0;
136 if (phase === "failed") return 1;
137 if (phase === "starting" || phase === "running") return 2;
138 return 3;
139}
infra/relay/src/agentActivity/agentActivityAlerts.ts created+152
......@@ -0,0 +1,152 @@
1import type {
2 RelayAgentActivityAggregateRow,
3 RelayAgentActivityAggregateState,
4 RelayAgentAwarenessPreferences,
5} from "@t3tools/contracts/relay";
6import * as DateTime from "effect/DateTime";
7import * as Option from "effect/Option";
8
9export interface AgentActivityAlert {
10 readonly title: string;
11 readonly body: string;
12}
13
14export const TERMINAL_NOTIFICATION_FRESHNESS_MS = 2 * 60 * 1_000;
15
16export function isFreshTerminalNotification(updatedAt: string, nowMs: number): boolean {
17 const timestamp = Option.getOrNull(DateTime.make(updatedAt));
18 return (
19 timestamp !== null && nowMs - timestamp.epochMilliseconds <= TERMINAL_NOTIFICATION_FRESHNESS_MS
20 );
21}
22
23type TransitionInput = {
24 readonly previousAggregate: RelayAgentActivityAggregateState | null;
25 readonly nextAggregate: RelayAgentActivityAggregateState;
26 readonly preferences: RelayAgentAwarenessPreferences | null;
27};
28
29function rowKey(row: RelayAgentActivityAggregateRow): string {
30 return JSON.stringify([row.environmentId, row.threadId]);
31}
32
33function isAttentionPhase(phase: string): boolean {
34 return phase === "waiting_for_approval" || phase === "waiting_for_input";
35}
36
37export function alertAllowedForPhase(
38 preferences: RelayAgentAwarenessPreferences | null,
39 phase: string,
40): boolean {
41 if (preferences === null) return true;
42 switch (phase) {
43 case "waiting_for_approval":
44 return preferences.notifyOnApproval;
45 case "waiting_for_input":
46 return preferences.notifyOnInput;
47 case "completed":
48 return preferences.notifyOnCompletion;
49 case "failed":
50 return preferences.notifyOnFailure;
51 default:
52 return false;
53 }
54}
55
56// A missing baseline is a replay, not a transition that should buzz the phone.
57export function attentionTransitionRows(input: TransitionInput) {
58 if (input.previousAggregate === null) return [];
59 const previouslyAttention = new Set(
60 input.previousAggregate.activities.filter((row) => isAttentionPhase(row.phase)).map(rowKey),
61 );
62 return input.nextAggregate.activities.filter(
63 (row) =>
64 isAttentionPhase(row.phase) &&
65 !previouslyAttention.has(rowKey(row)) &&
66 alertAllowedForPhase(input.preferences, row.phase),
67 );
68}
69
70// Reconciliation uses only observed transitions. Event-driven delivery can
71// include fresh completions whose running update never reached the device.
72export function newlyTerminalRows(
73 previousAggregate: RelayAgentActivityAggregateState | null,
74 nextAggregate: RelayAgentActivityAggregateState,
75 includeUnobserved = false,
76): ReadonlyArray<RelayAgentActivityAggregateRow> {
77 if (previousAggregate === null) return [];
78 const previousPhases = new Map(
79 previousAggregate.activities.map((row) => [rowKey(row), row.phase]),
80 );
81 return nextAggregate.activities.filter((row) => {
82 if (row.phase !== "completed" && row.phase !== "failed") return false;
83 const previousPhase = previousPhases.get(rowKey(row));
84 return (
85 (includeUnobserved || previousPhase !== undefined) &&
86 previousPhase !== "completed" &&
87 previousPhase !== "failed"
88 );
89 });
90}
91
92export function terminalTransitionRows(
93 input: TransitionInput & { readonly nowMs: number; readonly includeUnobserved?: boolean },
94) {
95 return newlyTerminalRows(
96 input.previousAggregate,
97 input.nextAggregate,
98 input.includeUnobserved,
99 ).filter((row) => {
100 return (
101 alertAllowedForPhase(input.preferences, row.phase) &&
102 isFreshTerminalNotification(row.updatedAt, input.nowMs)
103 );
104 });
105}
106
107export function alertForActivityRows(
108 rows: ReadonlyArray<RelayAgentActivityAggregateRow>,
109): AgentActivityAlert | null {
110 const first = rows[0];
111 if (!first) return null;
112 if (rows.length === 1) {
113 return { title: first.threadTitle, body: `${first.status}: ${first.projectTitle}` };
114 }
115 return {
116 title: `${rows.length} agents ${isAttentionPhase(first.phase) ? "need attention" : "finished"}`,
117 body: rows.map((row) => row.threadTitle).join(", "),
118 };
119}
120
121export function alertForAttentionTransition(input: TransitionInput): AgentActivityAlert | null {
122 return alertForActivityRows(attentionTransitionRows(input));
123}
124
125export function alertForNewlyTerminal(
126 input: TransitionInput & { readonly nowMs: number; readonly includeUnobserved?: boolean },
127): AgentActivityAlert | null {
128 return alertForActivityRows(terminalTransitionRows(input));
129}
130
131export function alertForTerminalAggregate(input: {
132 readonly aggregate: RelayAgentActivityAggregateState | null;
133 readonly preferences: RelayAgentAwarenessPreferences | null;
134}): AgentActivityAlert | null {
135 const row = input.aggregate?.activities[0];
136 if (!row || (row.phase !== "completed" && row.phase !== "failed")) return null;
137 return alertAllowedForPhase(input.preferences, row.phase) ? alertForActivityRows([row]) : null;
138}
139
140export function shouldAlertForActivity(input: {
141 readonly phase: RelayAgentActivityAggregateRow["phase"];
142 readonly updatedAt: string;
143 readonly preferences: RelayAgentAwarenessPreferences | null;
144 readonly nowMs: number;
145}): boolean {
146 return (
147 input.preferences?.notificationsEnabled === true &&
148 alertAllowedForPhase(input.preferences, input.phase) &&
149 ((input.phase !== "completed" && input.phase !== "failed") ||
150 isFreshTerminalNotification(input.updatedAt, input.nowMs))
151 );
152}
infra/relay/src/agentActivity/agentActivityPayloads.ts+28-6
......@@ -22,22 +22,29 @@ export function isTerminalPhase(state: RelayAgentActivityState): boolean {
2222const RUNNING_AGENT_ACTIVITY_ROW_TTL_MS = 2 * 60 * 60 * 1_000;
2323const WAITING_AGENT_ACTIVITY_ROW_TTL_MS = 24 * 60 * 60 * 1_000;
2424
25export function isExpiredAgentActivityState(
26 state: RelayAgentActivityState,
27 nowMs: number,
28): boolean {
25export function agentActivityExpiresAt(
26 state: Pick<RelayAgentActivityState, "phase" | "updatedAt">,
27): number {
2928 const updatedAtMs = Option.match(DateTime.make(state.updatedAt), {
3029 onNone: () => Number.NaN,
3130 onSome: (dt) => dt.epochMilliseconds,
3231 });
3332 if (Number.isNaN(updatedAtMs)) {
34 return true;
33 return Number.NaN;
3534 }
3635 const ttlMs =
3736 state.phase === "running" || state.phase === "starting"
3837 ? RUNNING_AGENT_ACTIVITY_ROW_TTL_MS
3938 : WAITING_AGENT_ACTIVITY_ROW_TTL_MS;
40 return nowMs - updatedAtMs > ttlMs;
39 return updatedAtMs + ttlMs;
40}
41
42export function isExpiredAgentActivityState(
43 state: RelayAgentActivityState,
44 nowMs: number,
45): boolean {
46 const expiresAt = agentActivityExpiresAt(state);
47 return !Number.isFinite(expiresAt) || nowMs > expiresAt;
4148}
4249
4350const MAX_SUMMARY_TEXT_LENGTH = 120;
......@@ -99,3 +106,18 @@ export function sanitizeApnsNotificationPayload(
99106 deepLink: sanitizeDeepLink(notification.deepLink),
100107 };
101108}
109
110export function notificationForActivity(
111 row: RelayAgentActivityAggregateRow,
112): ApnsNotificationPayload {
113 const activity = sanitizeAgentActivityAggregateRow(row);
114 return sanitizeApnsNotificationPayload({
115 title: activity.threadTitle,
116 body: `${activity.status}: ${activity.projectTitle}`,
117 environmentId: activity.environmentId,
118 threadId: activity.threadId,
119 deepLink: activity.deepLink,
120 phase: activity.phase,
121 updatedAt: activity.updatedAt,
122 });
123}
infra/relay/src/agentActivity/agentActivityPolicy.test.ts created+105
......@@ -0,0 +1,105 @@
1import { describe, expect, it } from "@effect/vitest";
2import { EnvironmentId, ThreadId } from "@t3tools/contracts";
3import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
4import { makeAggregateState } from "./agentActivityAggregate.ts";
5import {
6 attentionTransitionRows,
7 terminalTransitionRows,
8 shouldAlertForActivity,
9} from "./agentActivityAlerts.ts";
10
11const state: RelayAgentActivityState = {
12 environmentId: EnvironmentId.make("env"),
13 threadId: ThreadId.make("thread"),
14 projectTitle: "Project",
15 threadTitle: "Thread",
16 modelTitle: "Model",
17 headline: "Working",
18 phase: "running",
19 updatedAt: "1970-01-01T00:00:00.000Z",
20 deepLink: "/threads/env/thread",
21};
22const preferences = {
23 notificationsEnabled: true,
24 liveActivitiesEnabled: true,
25 notifyOnApproval: true,
26 notifyOnInput: true,
27 notifyOnCompletion: true,
28 notifyOnFailure: true,
29};
30const aggregate = (states: RelayAgentActivityState[]) =>
31 makeAggregateState({ activeStates: states, terminalState: null, nowMs: 0 })!;
32
33describe("shared agent activity policy", () => {
34 it.each(["waiting_for_approval", "waiting_for_input"] as const)(
35 "keeps an older %s ahead of five running rows",
36 (phase) => {
37 const running = Array.from({ length: 5 }, (_, i) => ({
38 ...state,
39 threadId: ThreadId.make(`running-${i}`),
40 }));
41 const waiting = { ...state, phase, updatedAt: "1969-12-31T23:59:00.000Z" };
42 const next = aggregate([...running, waiting]);
43 expect(next.activities[0]?.threadId).toBe(state.threadId);
44 expect(next.activeCount).toBe(6);
45 expect(next.activities).toHaveLength(5);
46 expect(
47 attentionTransitionRows({
48 previousAggregate: aggregate(running),
49 nextAggregate: next,
50 preferences,
51 }),
52 ).toMatchObject([{ threadId: state.threadId }]);
53 },
54 );
55
56 it("allows fresh unobserved completions for events, but not reconciliation or repeats", () => {
57 const previousAggregate = aggregate([
58 { ...state, threadId: ThreadId.make("old"), phase: "completed" },
59 ]);
60 const nextAggregate = aggregate([{ ...state, phase: "completed" }]);
61 const input = { previousAggregate, nextAggregate, preferences, nowMs: 0 };
62 expect(terminalTransitionRows(input)).toEqual([]);
63 expect(terminalTransitionRows({ ...input, includeUnobserved: true })).toHaveLength(1);
64 expect(
65 terminalTransitionRows({
66 ...input,
67 includeUnobserved: true,
68 previousAggregate: nextAggregate,
69 }),
70 ).toEqual([]);
71 expect(terminalTransitionRows({ ...input, includeUnobserved: true, nowMs: 180_000 })).toEqual(
72 [],
73 );
74 });
75
76 it("distinguishes equal thread IDs across environments", () => {
77 const waiting = { ...state, phase: "waiting_for_input" as const };
78 const other = { ...waiting, environmentId: EnvironmentId.make("other") };
79 expect(
80 attentionTransitionRows({
81 previousAggregate: aggregate([waiting]),
82 nextAggregate: aggregate([waiting, other]),
83 preferences,
84 }),
85 ).toMatchObject([{ environmentId: other.environmentId }]);
86 });
87
88 it("checks current permission, event preferences, and freshness together", () => {
89 const input = { ...state, phase: "completed" as const, preferences, nowMs: 0 };
90 expect(shouldAlertForActivity(input)).toBe(true);
91 expect(
92 shouldAlertForActivity({
93 ...input,
94 preferences: { ...preferences, notificationsEnabled: false },
95 }),
96 ).toBe(false);
97 expect(
98 shouldAlertForActivity({
99 ...input,
100 preferences: { ...preferences, notifyOnCompletion: false },
101 }),
102 ).toBe(false);
103 expect(shouldAlertForActivity({ ...input, nowMs: 180_000 })).toBe(false);
104 });
105});