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}